Allow decoding of blocks in parallel

src.java/crosby/binary/file/ParallelBlockInputStream.java

@@ -0,0 +1,157 @@
+/** Copyright (c) 2026 Leonard Ehrenfried. <mail@leonard.io>
+
+   This program is free software: you can redistribute it and/or modify
+   it under the terms of the GNU Lesser General Public License as
+   published by the Free Software Foundation, either version 3 of the
+   License, or (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+package crosby.binary.file;
+
+import java.io.Closeable;
+import java.io.DataInputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayDeque;
+import java.util.Queue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+/**
+ * A drop-in, parallel replacement for {@link BlockInputStream}.
+ *
+ * Reading a fileblock's header and raw bytes must stay strictly sequential,
+ * since it consumes an {@link InputStream}, but decompressing and parsing a
+ * block's contents (the CPU-heavy step, dominated by zlib inflate for large
+ * files) is independent per block. This class overlaps that work across a
+ * thread pool while still delivering completed blocks to the adaptor one at
+ * a time, in file order -- exactly the contract {@link BlockReaderAdapter}
+ * documents -- so an existing {@link BlockReaderAdapter} (such as a
+ * {@code crosby.binary.BinaryParser} subclass) needs no changes to benefit:
+ * its {@code handleBlock}/{@code skipBlock}/{@code complete} calls all still
+ * happen on a single thread, one at a time, in order.
+ *
+ * Memory use is bounded by keeping at most {@code pipelineDepth} decompressed
+ * blocks in flight at once (queued or in progress); tune it down for memory
+ * constrained environments, or up to keep more threads fed on files with many
+ * small blocks.
+ */
+public class ParallelBlockInputStream implements Closeable {
+    public ParallelBlockInputStream(InputStream input, BlockReaderAdapter adaptor) {
+        this(input, adaptor, Runtime.getRuntime().availableProcessors());
+    }
+
+    public ParallelBlockInputStream(InputStream input, BlockReaderAdapter adaptor, int numThreads) {
+        this(input, adaptor, Executors.newFixedThreadPool(numThreads), true, numThreads * 2);
+    }
+
+    /**
+     * Use a caller-supplied executor, e.g. to share a thread pool across
+     * several files. The executor is not shut down by {@link #close()}.
+     */
+    public ParallelBlockInputStream(InputStream input, BlockReaderAdapter adaptor,
+            ExecutorService executor, int pipelineDepth) {
+        this(input, adaptor, executor, false, pipelineDepth);
+    }
+
+    private ParallelBlockInputStream(InputStream input, BlockReaderAdapter adaptor,
+            ExecutorService executor, boolean ownsExecutor, int pipelineDepth) {
+        this.input = input;
+        this.adaptor = adaptor;
+        this.executor = executor;
+        this.ownsExecutor = ownsExecutor;
+        this.pipelineDepth = Math.max(1, pipelineDepth);
+    }
+
+    public void process() throws IOException {
+        Queue<Future<FileBlock>> inflight = new ArrayDeque<>();
+        try {
+            while (true) {
+                FileBlockHead head;
+                try {
+                    head = FileBlockHead.readHead(input);
+                } catch (EOFException e) {
+                    break;
+                }
+                if (adaptor.skipBlock(head)) {
+                    head.skipContents(input);
+                    continue;
+                }
+
+                byte[] buf = new byte[head.getDatasize()];
+                new DataInputStream(input).readFully(buf);
+                inflight.add(submit(head, buf));
+
+                if (inflight.size() >= pipelineDepth) {
+                    adaptor.handleBlock(take(inflight));
+                }
+            }
+            while (!inflight.isEmpty()) {
+                adaptor.handleBlock(take(inflight));
+            }
+            adaptor.complete();
+        } finally {
+            if (ownsExecutor) {
+                executor.shutdown();
+            }
+        }
+    }
+
+    private Future<FileBlock> submit(final FileBlockHead head, final byte[] buf) {
+        return executor.submit(new Callable<FileBlock>() {
+            @Override
+            public FileBlock call() throws IOException {
+                return head.parseData(buf);
+            }
+        });
+    }
+
+    private static FileBlock take(Queue<Future<FileBlock>> inflight) throws IOException {
+        Future<FileBlock> future = inflight.poll();
+        try {
+            return future.get();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException(e);
+        } catch (ExecutionException e) {
+            Throwable cause = e.getCause();
+            if (cause instanceof IOException) {
+                throw (IOException) cause;
+            }
+            if (cause instanceof RuntimeException) {
+                throw (RuntimeException) cause;
+            }
+            throw new IOException(cause);
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        try {
+            input.close();
+        } finally {
+            if (ownsExecutor) {
+                executor.shutdownNow();
+            }
+        }
+    }
+
+    private final InputStream input;
+    private final BlockReaderAdapter adaptor;
+    private final ExecutorService executor;
+    private final boolean ownsExecutor;
+    private final int pipelineDepth;
+}

test.java/crosby/binary/ParallelBlockInputStreamTest.java

@@ -0,0 +1,29 @@
+package crosby.binary;
+
+import crosby.binary.file.ParallelBlockInputStream;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.InputStream;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+/**
+ * Verifies that ParallelBlockInputStream still delivers blocks to the adaptor
+ * one at a time, in file order, producing output identical to the sequential
+ * BlockInputStream reader (see ReadFileTest).
+ */
+public class ParallelBlockInputStreamTest {
+
+    @Test
+    public void testParallel() throws Exception {
+        try (InputStream input = ReadFileTest.class.getResourceAsStream("/sample.pbf");
+             StringWriter stringWriter = new StringWriter();
+             PrintWriter printWriter = new PrintWriter(stringWriter);
+             ParallelBlockInputStream blockInput = new ParallelBlockInputStream(
+                     input, new ReadFileTest.TestBinaryParser(printWriter), 4)) {
+            blockInput.process();
+            Assert.assertEquals(ReadFileTest.EXPECTED, stringWriter.toString());
+        }
+    }
+}

test.java/crosby/binary/ReadFileTest.java

@@ -19,9 +19,7 @@
  */
 public class ReadFileTest {
 
-    @Test
-    public void test() throws Exception {
-        String expected = ("" +
+    static final String EXPECTED = ("" +
                 "Got header block.\n" +
                 "Dense node, ID 653970877 @ 51.763603,-0.228757\n" +
                 "Dense node, ID 647105170 @ 51.763591,-0.234465\n" +
@@ -447,20 +445,23 @@ public void test() throws Exception {
                 "  Key=value pairs: cycleway=track highway=cycleway name=Alban Way ncn_ref=61 \n" +
                 "Got some relations to parse.\n" +
                 "Complete!\n").replace("\n", System.lineSeparator());
+
+    @Test
+    public void test() throws Exception {
         try (InputStream input = ReadFileTest.class.getResourceAsStream("/sample.pbf");
              StringWriter stringWriter = new StringWriter();
              PrintWriter printWriter = new PrintWriter(stringWriter)) {
             BlockReaderAdapter brad = new TestBinaryParser(printWriter);
             new BlockInputStream(input, brad).process();
-            Assert.assertEquals(expected, stringWriter.toString());
+            Assert.assertEquals(EXPECTED, stringWriter.toString());
         }
     }
 
-    private static class TestBinaryParser extends BinaryParser {
+    static class TestBinaryParser extends BinaryParser {
 
         private final PrintWriter writer;
 
-        private TestBinaryParser(PrintWriter writer) {
+        TestBinaryParser(PrintWriter writer) {
             this.writer = writer;
         }