rdhabalia opened a new pull request, #8716:
URL: https://github.com/apache/hadoop/pull/8716
### Description of PR
Large sequential scans, such as analytics, bulk copies, and
columnar readers, open a `DFSInputStream` and read a file end-to-end.
Today, each block is fetched synchronously:
1. The reader thread opens a `BlockReader`.
2. It waits for the DataNode/network round trip.
3. It drains the block.
4. It repeats the process for the next block.
The per-block open latency and single-block-at-a-time pipeline leave the
client CPU idle while waiting on I/O and cap throughput well below what the
DataNodes and network can deliver.
This change adds an **opt-in, client-side read-ahead prefetcher** that
fetches
blocks ahead of the reader's cursor using a shared background thread pool.
By the time the reader reaches a prefetched block, the block is already
resident in memory and can be served as an in-memory copy.
## Approach
### BlockPrefetcher
- Introduces a new `BlockPrefetcher` that maintains a bounded, sliding window
of block-sized buffers ahead of the current read position.
- Blocks are filled in chunks using a shared, JVM-wide daemon thread pool.
- The thread pool uses:
- `SynchronousQueue`
- `AbortPolicy`
- When all workers are busy, a prefetch submission is rejected and skipped
rather than blocking the foreground reader thread.
### DFSInputStream Integration
- `DFSInputStream.read(byte[])` and `read(ByteBuffer)` first attempt to serve
data from the prefetch cache.
- On a cache hit:
- The read position is advanced.
- The stateful synchronous reader is invalidated.
- Read statistics are updated exactly as they are on the direct path.
- Locality is preserved across short-circuit, local, and remote reads.
- On a cache miss:
- The normal synchronous read path is used.
- Block locations are cached for subsequent prefetch operations.
### Prefetch Safety and Correctness
- Prefetch operates entirely using cached block locations.
- It never issues a `getBlockLocations` RPC.
- It does not mutate foreground retry state while a foreground read may hold
`infoLock`.
- The following cases are excluded for correctness:
- Non-uniform block sizes
- Striped (EC) files
- Under-construction files
- Single-block files
### Metrics and Observability
- A shared, opt-in scheduled task periodically logs per-stream cache
hit/miss ratios at `INFO` level.
- Metrics logging occurs only when metrics logging is enabled.
### Global Memory Budget
- A global byte budget bounds the total memory held by all prefetch buffers
across the JVM.
- The budget grows to the largest configured value across clients.
- Memory is strictly reserved and released on a per-stream basis.
## Configuration
All configuration is client-side. The feature is **disabled by default**.
| Configuration | Description |
|---|---|
| `dfs.client.prefetch.enabled` | Enables client-side read-ahead prefetching
|
| `dfs.client.prefetch.size` | Per-stream read-ahead window |
| `dfs.client.prefetch.max.bytes` | JVM-wide prefetch memory cap |
| `dfs.client.prefetch.chunk.size` | Prefetch fill granularity |
| `dfs.client.prefetch.threads` | Shared prefetch worker thread count |
| `dfs.client.prefetch.threadpool.size` | Shared prefetch worker thread pool
size |
| `dfs.client.prefetch.ttl.ms` | Buffer time-to-live |
| `dfs.client.prefetch.metrics.log.enabled` | Enables periodic cache
hit-ratio logging |
## Results
A single-client sequential read benchmark was run with:
- **File size:** 10 GB
- **Block size:** 512 MB
- **Prefetch threads:** 5
- **Prefetch window:** ~3 GB
### Performance
| Metric | Result |
|---|---:|
| Baseline throughput | 406.8 MB/s |
| Prefetch throughput | 1,419.5 MB/s |
| Average improvement | **~3.49x** |
| Peak throughput | **~1.58 GB/s** |
| Cache hit ratio | **~81%** |
The feature is **off by default** and has no impact on the read path until
explicitly enabled.
<!--
Thanks for sending a pull request!
1. If this is your first time, please read our contributor guidelines:
https://cwiki.apache.org/confluence/display/HADOOP/How+To+Contribute
2. Make sure your PR title starts with JIRA issue id, e.g.,
'HADOOP-17799. Your PR title ...'.
-->
### How was this patch tested?
Tested by newly added unit test
### For code changes:
- [ ] Does the title of this PR start with the corresponding JIRA issue id
(e.g. 'HADOOP-17799. Your PR title ...')?
- [ ] Object storage: Have the integration tests been executed and the
endpoint
declared according to the connector-specific documentation? *Note:
Automated CI
testing doesn't cover all cases so manual testing with cloud storage
is still
required.*
- [ ] If adding new dependencies to the code, are these dependencies
licensed in a way that is compatible for inclusion under [ASF
2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the `LICENSE`, `LICENSE-binary`,
`NOTICE-binary` files?
### AI Tooling
If an AI tool was used:
- [ ] The PR includes the phrase "Contains content generated by <tool>"
where <tool> is the name of the AI tool used.
- [ ] My use of AI contributions follows the ASF legal policy
https://www.apache.org/legal/generative-tooling.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
