From bdb1c8d5ed67f01fedb28651d87f216020160d19 Mon Sep 17 00:00:00 2001 From: Leonard Ehrenfried Date: Fri, 11 Sep 2026 15:31:12 +0200 Subject: [PATCH 1/4] Fixes for parallel block reading --- .../binary/file/ParallelBlockInputStream.java | 48 +++++++---- .../binary/ParallelBlockInputStreamTest.java | 82 +++++++++++++++++++ 2 files changed, 113 insertions(+), 17 deletions(-) diff --git a/src.java/crosby/binary/file/ParallelBlockInputStream.java b/src.java/crosby/binary/file/ParallelBlockInputStream.java index ac7217f..3d77fe4 100644 --- a/src.java/crosby/binary/file/ParallelBlockInputStream.java +++ b/src.java/crosby/binary/file/ParallelBlockInputStream.java @@ -37,12 +37,21 @@ 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#handleBlock} 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} blocks + * ahead of the block whose {@code handleBlock} result was most recently + * delivered. 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. * * 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 +89,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 +115,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(); + } + } } From 86eb3b515b3332e170f2dfec9185201ce80eab1f Mon Sep 17 00:00:00 2001 From: Leonard Ehrenfried Date: Mon, 14 Sep 2026 12:56:59 +0200 Subject: [PATCH 2/4] Address nit comments --- .../binary/file/ParallelBlockInputStream.java | 22 ++++++++++++------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/src.java/crosby/binary/file/ParallelBlockInputStream.java b/src.java/crosby/binary/file/ParallelBlockInputStream.java index 3d77fe4..88c9101 100644 --- a/src.java/crosby/binary/file/ParallelBlockInputStream.java +++ b/src.java/crosby/binary/file/ParallelBlockInputStream.java @@ -39,19 +39,25 @@ License, or (at your option) any later version. * files) is independent per block. This class overlaps that work across a * thread pool while still delivering {@code handleBlock} calls to the * adaptor one at a time, in file order, matching what - * {@link BlockReaderAdapter#handleBlock} documents -- so an existing + * {@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} blocks - * ahead of the block whose {@code handleBlock} result was most recently - * delivered. 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. + * 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 From a6a206b459202b4c62cc3523c2075d26af180772 Mon Sep 17 00:00:00 2001 From: Leonard Ehrenfried Date: Mon, 14 Sep 2026 12:57:18 +0200 Subject: [PATCH 3/4] Add changelog for 1.8.0 --- CHANGELOG.md | 5 +++++ 1 file changed, 5 insertions(+) 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) From c6e2be0ab56fd5a24e7ab736c2a008a94a670bc3 Mon Sep 17 00:00:00 2001 From: Leonard Ehrenfried Date: Mon, 14 Sep 2026 13:00:50 +0200 Subject: [PATCH 4/4] Bump Java version to 1.8.0 --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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