Fixes for parallel block reading

CHANGELOG.md

@@ -1,5 +1,10 @@
 ## Unreleased
 
+## Release notes for 1.8.0 (2026-09-14)
+
+- Java: Upgrade protobuf version [#97](https://github.com/openstreetmap/OSM-binary/pull/97)
+- Java: Experimental parallel block decoding [#96](https://github.com/openstreetmap/OSM-binary/pull/96)
+
 ## Release notes for 1.7.0 (2026-07-26)
 
 - C++: Fix buffer overflow in osmpbf-outline tool [#95](https://github.com/openstreetmap/OSM-binary/pull/95)

pom.xml

@@ -6,7 +6,7 @@
   <groupId>org.openstreetmap.pbf</groupId>
   <artifactId>osmpbf</artifactId>
   <packaging>jar</packaging>
-  <version>1.7.0</version>
+  <version>1.8.0</version>
   <name>OSM-Binary</name>
   <description>Library for the OpenStreetMap PBF format</description>
   <url>https://github.com/openstreetmap/OSM-binary</url>

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

@@ -37,12 +37,27 @@ License, or (at your option) any later version.
  * 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.
+ * thread pool while still delivering {@code handleBlock} calls to the
+ * adaptor one at a time, in file order, matching what
+ * {@link BlockReaderAdapter#skipBlock} documents -- so an existing
+ * {@link BlockReaderAdapter} (such as a {@code crosby.binary.BinaryParser}
+ * subclass) needs no changes to benefit.
+ *
+ * Note: a block's header must be read -- and {@code skipBlock} called on it
+ * -- before its body can be read or skipped and the stream advanced past it,
+ * so {@code skipBlock} may be invoked for up to {@code pipelineDepth}
+ * non-skipped blocks ahead of the block whose {@code handleBlock} result was
+ * most recently delivered; a run of skipped blocks doesn't count against
+ * that limit (nothing is queued for them), so {@code skipBlock} can in fact
+ * run further ahead than {@code pipelineDepth} when the file contains long
+ * stretches of skipped blocks. An adaptor whose skip decision depends on
+ * mutable state updated inside {@code handleBlock} (rather than just the
+ * block's own type or metadata) will see stale state under this read-ahead;
+ * such an adaptor should use {@code pipelineDepth} 1, which restores fully
+ * sequential delivery at the cost of the parallelism benefit -- pass it via
+ * the 4-argument constructor, since the convenience constructor that takes a
+ * thread count derives {@code pipelineDepth} as {@code numThreads * 2} and
+ * cannot be used to request a depth of 1.
  *
  * 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
@@ -80,20 +95,22 @@ public void process() throws IOException {
         Queue<Future<FileBlock>> inflight = new ArrayDeque<>();
         try {
             while (true) {
-                FileBlockHead head;
                 try {
-                    head = FileBlockHead.readHead(input);
+                    FileBlockHead head = FileBlockHead.readHead(input);
+                    if (adaptor.skipBlock(head)) {
+                        head.skipContents(input);
+                        continue;
+                    }
+                    byte[] buf = new byte[head.getDatasize()];
+                    new DataInputStream(input).readFully(buf);
+                    inflight.add(submit(head, buf));
                 } catch (EOFException e) {
+                    // Matches BlockInputStream: a clean end of stream, or a truncated
+                    // header/body, both just end the read -- the trailing partial
+                    // block (if any) is dropped, and anything already in flight is
+                    // still delivered below.
                     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));
@@ -104,8 +121,11 @@ public void process() throws IOException {
             }
             adaptor.complete();
         } finally {
+            for (Future<FileBlock> future : inflight) {
+                future.cancel(true);
+            }
             if (ownsExecutor) {
-                executor.shutdown();
+                executor.shutdownNow();
             }
         }
     }

test.java/crosby/binary/ParallelBlockInputStreamTest.java

@@ -1,12 +1,19 @@
 package crosby.binary;
 
+import crosby.binary.file.BlockInputStream;
 import crosby.binary.file.ParallelBlockInputStream;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.PrintWriter;
 import java.io.StringWriter;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Verifies that ParallelBlockInputStream still delivers blocks to the adaptor
@@ -26,4 +33,79 @@ public void testParallel() throws Exception {
             Assert.assertEquals(ReadFileTest.EXPECTED, stringWriter.toString());
         }
     }
+
+    /**
+     * sample.pbf only has 4 blocks, so the default pipelineDepth (2 *
+     * numThreads) never fills up mid-stream: every block ends up delivered
+     * from the final drain loop after EOF, never from the backpressure
+     * branch inside the read loop. Force a small pipelineDepth so that
+     * branch -- and the resulting handleBlock/skipBlock interleaving -- is
+     * actually exercised.
+     */
+    @Test
+    public void testParallelWithBackpressure() throws Exception {
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        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), executor, 2)) {
+            blockInput.process();
+            Assert.assertEquals(ReadFileTest.EXPECTED, stringWriter.toString());
+        } finally {
+            executor.shutdownNow();
+            executor.awaitTermination(5, TimeUnit.SECONDS);
+        }
+    }
+
+    /**
+     * A block body truncated mid-stream must be treated as a clean end of
+     * input -- same as BlockInputStream -- rather than aborting process()
+     * before complete() is called and dropping whatever was already
+     * in flight. Compares against BlockInputStream on the exact same bytes
+     * to pin the expected behavior instead of hard-coding it.
+     */
+    @Test
+    public void testTruncatedInputMatchesSequentialReader() throws Exception {
+        byte[] full = readAll(ReadFileTest.class.getResourceAsStream("/sample.pbf"));
+        byte[] truncated = new byte[full.length / 2];
+        System.arraycopy(full, 0, truncated, 0, truncated.length);
+
+        String sequential = runSequential(truncated);
+        String parallel = runParallel(truncated);
+
+        Assert.assertTrue("sequential reader should still complete", sequential.endsWith("Complete!" + System.lineSeparator()));
+        Assert.assertEquals(sequential, parallel);
+    }
+
+    private static String runSequential(byte[] bytes) throws IOException {
+        try (InputStream input = new ByteArrayInputStream(bytes);
+             StringWriter stringWriter = new StringWriter();
+             PrintWriter printWriter = new PrintWriter(stringWriter)) {
+            new BlockInputStream(input, new ReadFileTest.TestBinaryParser(printWriter)).process();
+            return stringWriter.toString();
+        }
+    }
+
+    private static String runParallel(byte[] bytes) throws IOException {
+        try (InputStream input = new ByteArrayInputStream(bytes);
+             StringWriter stringWriter = new StringWriter();
+             PrintWriter printWriter = new PrintWriter(stringWriter);
+             ParallelBlockInputStream blockInput = new ParallelBlockInputStream(
+                     input, new ReadFileTest.TestBinaryParser(printWriter), 2)) {
+            blockInput.process();
+            return stringWriter.toString();
+        }
+    }
+
+    private static byte[] readAll(InputStream in) throws IOException {
+        try (InputStream input = in; ByteArrayOutputStream out = new ByteArrayOutputStream()) {
+            byte[] buf = new byte[4096];
+            int n;
+            while ((n = input.read(buf)) >= 0) {
+                out.write(buf, 0, n);
+            }
+            return out.toByteArray();
+        }
+    }
 }