Merge pull request #4 from joshmoore/blosc-multithread

Add property for number of threads to blosc compressor.
zarr-developers · May 23, 2023 · d4ed22c · d4ed22c
2 parents 50b950c + 09ec1c7
commit d4ed22c
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 3 deletions.
diff --git a/src/main/java/com/bc/zarr/CompressorFactory.java b/src/main/java/com/bc/zarr/CompressorFactory.java
@@ -223,6 +223,8 @@ static class BloscCompressor extends Compressor {
         public final static int defaultShuffle = BYTESHUFFLE;
         public final static String keyBlocksize = "blocksize";
         public final static int defaultBlocksize = 0;
+        public final static String keyNumThreads = "nthreads";
+        public final static int defaultNumThreads = 1;
         public final static int[] supportedShuffle = new int[]{/*AUTOSHUFFLE, */NOSHUFFLE, BYTESHUFFLE, BITSHUFFLE};
         public final static String[] supportedCnames = new String[]{"zstd", "blosclz", defaultCname, "lz4hc", "zlib"/*, "snappy"*/};
 
@@ -232,12 +234,14 @@ static class BloscCompressor extends Compressor {
                     put(keyClevel, defaultCLevel);
                     put(keyShuffle, defaultShuffle);
                     put(keyBlocksize, defaultBlocksize);
+                    put(keyNumThreads, defaultNumThreads);
                 }});
 
         private final int clevel;
         private final int blocksize;
         private final int shuffle;
         private final String cname;
+        private final int nthreads;
 
         private BloscCompressor(Map<String, Object> map) {
             final Object cnameObj = map.get(keyCname);
@@ -285,6 +289,15 @@ private BloscCompressor(Map<String, Object> map) {
             } else {
                 this.blocksize = ((Number) blocksizeObj).intValue();
             }
+
+            final Object nthreadsObj = map.get(keyNumThreads);
+            if (nthreadsObj == null) {
+                this.nthreads = defaultNumThreads;
+            } else if (nthreadsObj instanceof String) {
+                this.nthreads = Integer.parseInt((String) nthreadsObj);
+            } else {
+                this.nthreads = ((Number) nthreadsObj).intValue();
+            }
         }
 
         @Override
@@ -308,6 +321,10 @@ public String getCname() {
             return cname;
         }
 
+        public int getNumThreads() {
+            return nthreads;
+        }
+
         @Override
         public String toString() {
             return "compressor=" + getId()
@@ -324,7 +341,7 @@ public void compress(InputStream is, OutputStream os) throws IOException {
             final int outputSize = inputSize + JBlosc.OVERHEAD;
             final ByteBuffer inputBuffer = ByteBuffer.wrap(inputBytes);
             final ByteBuffer outBuffer = ByteBuffer.allocate(outputSize);
-            final int i = JBlosc.compressCtx(clevel, shuffle, 1, inputBuffer, inputSize, outBuffer, outputSize, cname, blocksize, 1);
+            final int i = JBlosc.compressCtx(clevel, shuffle, 1, inputBuffer, inputSize, outBuffer, outputSize, cname, blocksize, nthreads);
             final BufferSizes bs = cbufferSizes(outBuffer);
             byte[] compressedChunk = Arrays.copyOfRange(outBuffer.array(), 0, (int) bs.getCbytes());
             os.write(compressedChunk);
@@ -341,7 +358,7 @@ public void uncompress(InputStream is, OutputStream os) throws IOException {
             byte[] inBytes = Arrays.copyOf(header, compressedSize);
             di.readFully(inBytes, header.length, compressedSize - header.length);
             ByteBuffer outBuffer = ByteBuffer.allocate(uncompressedSize);
-            JBlosc.decompressCtx(ByteBuffer.wrap(inBytes), outBuffer, outBuffer.limit(), 1);
+            JBlosc.decompressCtx(ByteBuffer.wrap(inBytes), outBuffer, outBuffer.limit(), nthreads);
             os.write(outBuffer.array());
         }
 

diff --git a/src/test/java/com/bc/zarr/CompressorFactoryTest.java b/src/test/java/com/bc/zarr/CompressorFactoryTest.java
@@ -42,18 +42,20 @@ public class CompressorFactoryTest {
     public void getDefaultCompressorProperties() {
         final Map<String, Object> map = CompressorFactory.getDefaultCompressorProperties();
         assertNotNull(map);
-        assertEquals(5, map.size());
+        assertEquals(6, map.size());
         assertThat(map.containsKey("id"), is(true));
         assertThat(map.containsKey("cname"), is(true));
         assertThat(map.containsKey("clevel"), is(true));
         assertThat(map.containsKey("blocksize"), is(true));
         assertThat(map.containsKey("shuffle"), is(true));
+        assertThat(map.containsKey("nthreads"), is(true));
 
         assertThat(map.get("id"), is("blosc"));
         assertThat(map.get("cname"), is("lz4"));
         assertThat(map.get("clevel"), is(5));
         assertThat(map.get("blocksize"), is(0));
         assertThat(map.get("shuffle"), is(1));
+        assertThat(map.get("nthreads"), is(1));
     }
 
     @Test