diff --git a/CHANGELOG.md b/CHANGELOG.md
index d2c4a9b..d5d5e92 100644
--- a/CHANGELOG.md
+++ b/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)
diff --git a/pom.xml b/pom.xml
index 9396e34..5a633cd 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,7 +6,7 @@
org.openstreetmap.pbf
osmpbf
jar
- 1.7.0
+ 1.8.0
OSM-Binary
Library for the OpenStreetMap PBF format
https://github.com/openstreetmap/OSM-binary
diff --git a/src.java/crosby/binary/file/ParallelBlockInputStream.java b/src.java/crosby/binary/file/ParallelBlockInputStream.java
index ac7217f..88c9101 100644
--- a/src.java/crosby/binary/file/ParallelBlockInputStream.java
+++ b/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> 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 future : inflight) {
+ future.cancel(true);
+ }
if (ownsExecutor) {
- executor.shutdown();
+ executor.shutdownNow();
}
}
}
diff --git a/test.java/crosby/binary/ParallelBlockInputStreamTest.java b/test.java/crosby/binary/ParallelBlockInputStreamTest.java
index 3c9e281..ed4622d 100644
--- a/test.java/crosby/binary/ParallelBlockInputStreamTest.java
+++ b/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();
+ }
+ }
}