HDDS-15424. Fix Concurrent positional read - #11102
Conversation
There was a problem hiding this comment.
Pull request overview
This PR introduces an opt-in mechanism to make positioned reads (pread) thread-safe in OzoneFS by serializing the seek-read-restore sequence inside OzoneFSInputStream, addressing intermittent checksum failures when multiple threads reuse the same input stream.
Changes:
- Add
ozone.fs.synchronize.positioned.reads.enabledto optionally synchronize positioned reads inOzoneFSInputStream. - Plumb the new flag through the OzoneFS implementations so the created FS input streams honor the configuration.
- Add a unit test that exercises concurrent positioned reads with the flag enabled/disabled.
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java | Passes the new positioned-read synchronization flag when constructing the FS input stream wrapper. |
| hadoop-ozone/ozonefs/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java | Passes the new positioned-read synchronization flag when constructing the FS input stream wrapper. |
| hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/RootedOzoneFileSystem.java | Same flag plumbing for the Hadoop3 variant. |
| hadoop-ozone/ozonefs-hadoop3/src/main/java/org/apache/hadoop/fs/ozone/OzoneFileSystem.java | Same flag plumbing for the Hadoop3 variant. |
| hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java | Adds concurrent positioned-read coverage to validate the opt-in synchronization behavior. |
| hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java | Implements optional serialization of positioned reads via an internal lock. |
| hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/CapableOzoneFSInputStream.java | Extends constructors to propagate the new positioned-read synchronization option. |
| hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java | Reads the new config key and uses it when creating OzoneFSInputStream. |
| hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java | Reads the new config key and uses it when creating OzoneFSInputStream. |
| hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java | Defines the new configuration key and default value. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
taklwu
changed the title
0b724bf to
ffb81d6
Compare
e03ea7b to
8293565
Compare
-align with HDDS-15920
|
the om integration test |
| return doPositionedRead(chunkRelativePosition, dst); | ||
| } | ||
| // Local (short-circuit) reads share a FileChannel cursor; serialize them. | ||
| synchronized (this) { |
There was a problem hiding this comment.
Could we synchronize local positioned reads on the shared block FileChannel, or use positional FileChannel reads? Each LocalChunkInputStream locks its own instance, while all chunks in the block receive the same blockFileInputStream, so preads to different chunks can still interleave position(...).read(...) and use the wrong offset.
There was a problem hiding this comment.
yup, you're right about the positional FileChannel, I made the change and please review again.
| int idx = chunkIndexForPosition(pos, offsets); | ||
| ChunkInputStream chunk = streams.get(idx); | ||
| long chunkPos = pos - offsets[idx]; | ||
| int n = chunk.readPositioned(chunkPos, dst); |
There was a problem hiding this comment.
Could we preserve BlockInputStream’s outer retry and refresh handling here? readChunk still tries the DNs in the current pipeline, but this native path cannot refresh an expired block token or fetch an updated pipeline after those attempts fail.
There was a problem hiding this comment.
thanks, this is an important miss.
|
@taklwu thanks for the patch! |
- use positional FileChannel reads for local read - fix retry with the same token when using block input stream - remove the opt-in flag, default to stateless pread for ExtendedInputStream and fallback to sync if it's all other input stream
…putStream. - Use ephemeral ChunkInputStreams with per-call retry counters for block pread so concurrent positioned reads do not share sequential retry state or chunk stream buffers. - Serialize cursor-moving and positioned reads on OzoneCryptoInputStream because CryptoInputStream is not thread-safe.
What changes were proposed in this pull request?
Fix Non-Stream read failed positional read checksum intermittently during concurrently access by multi-threads.
Please describe your PR in detail:
Fixing the concurrent positional read by introducing stateless pread for major input streams (BlockInputStream, ChunkInputStream, MultiPartInputStream) as the default pread method, such that the read could be getting the range / chunkinfo via readChunk (additional RPC) and use
copyRangewithout touching shared cursor / position and avoid race condition.in addition, introduce a new/fallback opt-in feature
ozone.fs.synchronize.positioned.reads.enabledfor concurrently positional read for when stateless pread is not implemented. e.g. any child class of ExtendedInputStream without implementing thereadFully.The problem without thread-safe inputstream, checksum may fail because the offset has moved from the current position.
note that hadoop's FSInputStream does have this
synchronizedblock and make sure the input stream is thread-safe. Also we're learning from DFSInputStream's getBlockRange and have this logic as an additional RPC that used by readChunk and copyRange.What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-15424
How was this patch tested?