[
https://issues.apache.org/jira/browse/HDFS-17975?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18111764#comment-18111764
]
ASF GitHub Bot commented on HDFS-17975:
---------------------------------------
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
> HDFS Client-Side Block Prefetch to Improve Large Sequential Read Throughput
> ---------------------------------------------------------------------------
>
> Key: HDFS-17975
> URL: https://issues.apache.org/jira/browse/HDFS-17975
> Project: Hadoop HDFS
> Issue Type: Improvement
> Components: hdfs-client
> Reporter: Rajan Dhabalia
> Priority: Major
>
> h2. Summary
> Add client-side block prefetching (parallel read-ahead) to `DFSInputStream`
> to improve throughput for large and sequential read workloads.
> h2. Issue Type
> Improvement
> h2. Component/s
> hdfs-client
> h2. Description
> h3. Motivation
> Large scan and sequential read workloads on `DFSInputStream` can be limited
> by synchronous, one-block-at-a-time reads. The latency of each remote block
> read can stall the consumer and prevent full utilization of available network
> and DataNode capacity.
> Client-side prefetching hides this latency by fetching upcoming blocks in
> parallel while the consumer processes previously read data.
> h3. Approach
> Introduce a config-gated, default-off `BlockPrefetcher` in `DFSInputStream`.
> When enabled:
> * Asynchronously prefetch upcoming data using a bounded thread pool.
> * Store prefetched data in a bounded in-memory cache.
> * Serve reads directly from the cache when available.
> * Fall back to the existing synchronous read path on cache misses.
> * Support configurable prefetch window, chunk size, cache size, worker
> threads, and TTL.
> * Provide optional periodic metrics logging for cache hit ratio and
> prefetched bytes.
> * Add `PrefetchReadExample` and unit tests.
> The existing read path remains unchanged when prefetching is disabled.
> h3. Configuration
> ||Property||Default||Description||
> |`dfs.client.prefetch.enabled`|false|Master switch for client-side
> prefetching|
> |`dfs.client.prefetch.size`|See docs|Prefetch window size|
> |`dfs.client.prefetch.max.bytes`|See docs|Maximum bytes held in the prefetch
> cache|
> |`dfs.client.prefetch.chunk.size`|See docs|Size of each prefetched chunk|
> |`dfs.client.prefetch.threads`|See docs|Number of prefetch worker threads|
> |`dfs.client.prefetch.threadpool.size`|See docs|Prefetch thread-pool size|
> |`dfs.client.prefetch.ttl.ms`|See docs|TTL for cached prefetched data|
> |`dfs.client.prefetch.metrics.log.enabled`|false|Enable periodic prefetch
> metrics logging|
> |`dfs.client.prefetch.metrics.log.interval.ms`|See docs|Metrics logging
> interval|
> h3. Results
> Testing on large sequential read workloads showed:
> * {*}Up to 3.49x higher average read throughput{*}, approximately {*}249%
> improvement{*}.
> * Peak throughput of approximately {*}1.58 GB/s{*}.
> * Approximately {*}81% prefetch cache hit ratio{*}.
> * Approximately *4 out of 5 reads* served from the prefetch cache.
> * No change to the existing read path when the feature is disabled.
> h3. Estimated Improvement
> The benefit is workload-dependent and increases with read sequentiality and
> the amount of latency that can be hidden through parallel prefetching.
> ||Workload||Expected Benefit||Rationale||
> |Large sequential scan|~3x to 3.5x|Upcoming blocks can be fetched in parallel
> while earlier data is processed|
> |Mixed sequential + occasional seek|~1.8x to 2.5x|Sequential portions
> benefit, while seeks reduce cache effectiveness|
> |Random / small reads|~1x|Limited sequentiality provides little opportunity
> for prefetching|
> h3. Resource Protection
> Prefetching is bounded to prevent uncontrolled memory or CPU consumption:
> * Maximum prefetch cache size
> * Configurable prefetch window and chunk size
> * Bounded worker threads and thread-pool size
> * Entry TTL
> * Optional metrics logging
> For workloads with limited sequentiality, these limits reduce unnecessary
> memory usage and background work.
> h3. Expected Impact
> * Improve throughput for large sequential and scan workloads.
> * Hide remote block-read latency through parallelism.
> * Improve utilization of network and DataNode capacity.
> * Reduce the impact of per-block RPC latency on a single reader.
> * Preserve existing behavior when disabled.
> * Provide tunable resource limits for different workloads.
> h2. Backward Compatibility
> The feature is {*}disabled by default{*}. Existing `DFSInputStream` behavior
> and read paths remain unchanged unless client-side prefetching is explicitly
> enabled.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
