[I] LocalFileSystem: read_range zero-fills each buffer before read_at, costing 8-19% of read throughput for 64 KiB-1 MiB ranges [arrow-rs-object-store]

["andygrove (via GitHub)" <gi...@apache.org>]

andygrove opened a new issue, #880:
URL: https://github.com/apache/arrow-rs-object-store/issues/880

   **Is your feature request related to a problem or challenge? Please describe what you are trying to do.**
   
   Since #628, `read_range` in `LocalFileSystem` zero-fills each buffer before reading into it: it calls `buf.resize(requested, 0)` and then fills the buffer with `read_at`. Before #628 it did a `seek` and then `take(requested).read_to_end(&mut buf)`, which reads straight into the `Vec`'s uninitialized capacity. #628 saved the `lseek` per range, but added a pass over every byte read.
   
   I benchmarked the three ways to read a range of a 474 MB file that is already in the page cache, with a fresh `Vec` per range as `read_range` uses (std only, source below, Apple M3 Max, macOS, median of 15 interleaved rounds, GB/s):
   
   | Range | zero-fill + `read_at` (main) | `seek` + `read_to_end` (before #628) | `read_at` into uninitialized capacity |
   |---:|---:|---:|---:|
   | 16 KiB | 9.68 | 7.40 | 10.68 |
   | 64 KiB | 11.12 | 11.85 | 13.75 |
   | 256 KiB | 13.66 | 14.74 | 15.38 |
   | 1 MiB | 14.59 | 15.58 | 15.80 |
   | 8 MiB | 11.42 | 11.04 | 11.04 |
   
   Reading into uninitialized capacity is 8% to 19% faster from 64 KiB to 1 MiB, the range of most Parquet column chunk and page reads. At 16 KiB, saving the `lseek` matters more than the zero-fill, consistent with #628's benchmark, which read ranges of about 8 KiB. At 8 MiB the page faults on each fresh allocation dominate, and all three are the same.
   
   The end-to-end effect is small. I found this profiling DataFusion Parquet scans through Apache DataFusion Comet, where `read_range` was 1% to 4% of CPU samples, so it would save a fraction of a percent of a scan. It is a cheap change to the hottest local read path, though.
   
   **Describe the solution you'd like**
   
   Read into the `Vec`'s spare capacity without initializing it first, the way std's `File::read_buf` does for `read_to_end`. std has no stable positioned read into uninitialized memory yet, so that means either a small `unsafe` block that calls `pread` with a pointer into `spare_capacity_mut()` and then `set_len`s the bytes read, or waiting for the `BorrowedBuf`-based positioned reads to stabilize.
   
   Note that the `read_at`-into-uninitialized arm in the benchmark hands `read_at` a `&mut [u8]` over uninitialized memory. That is fine for measuring, but a real implementation should pass the raw pointer to `pread` instead of making that slice.
   
   **Describe alternatives you've considered**
   
   Keep `read_at` with the zero-fill for small ranges and go back to `seek` + `read_to_end` above a threshold of around 32 KiB. That needs no `unsafe`, and gets most of the gain from 64 KiB up (the middle column above).
   
   **Additional context**
   
   Benchmark source, built with `rustc -C opt-level=3 readbench.rs` and run as `./readbench <file> <range_bytes> <rounds>`:
   
   <details>
   <summary>readbench.rs</summary>
   
   ```rust
   // Compares object_store LocalFileSystem's read_range strategies on a page-cache-resident file.
   //
   //   zero_fill_read_at: current main (since apache/arrow-rs-object-store#628):
   //                      Vec::with_capacity + resize(n, 0) + read_at loop
   //   seek_read_to_end:  before #628: seek + take(n).read_to_end into an uninitialized Vec
   //   pread_uninit:      read_at straight into the Vec's spare capacity (reference only)
   //
   // Each range gets a fresh Vec, as read_range does. Usage: readbench <file> <range_bytes> <rounds>
   use std::fs::File;
   use std::io::{Read, Seek, SeekFrom};
   use std::os::unix::fs::FileExt;
   use std::time::Instant;
   
   fn zero_fill_read_at(file: &File, off: u64, n: usize) -> Vec<u8> {
       let mut buf = Vec::with_capacity(n);
       buf.resize(n, 0u8);
       let mut slice = &mut buf[..];
       let mut offset = off;
       while !slice.is_empty() {
           let r = file.read_at(slice, offset).unwrap();
           if r == 0 {
               break;
           }
           let tmp = slice;
           slice = &mut tmp[r..];
           offset += r as u64;
       }
       buf
   }
   
   fn seek_read_to_end(file: &mut File, off: u64, n: usize) -> Vec<u8> {
       file.seek(SeekFrom::Start(off)).unwrap();
       let mut buf = Vec::with_capacity(n);
       (&mut *file).take(n as u64).read_to_end(&mut buf).unwrap();
       buf
   }
   
   fn pread_uninit(file: &File, off: u64, n: usize) -> Vec<u8> {
       let mut buf: Vec<u8> = Vec::with_capacity(n);
       let mut filled = 0usize;
       while filled < n {
           // Reference only: hands read_at a slice over uninitialized capacity.
           let dst = unsafe { std::slice::from_raw_parts_mut(buf.as_mut_ptr().add(filled), n - filled) };
           let r = file.read_at(dst, off + filled as u64).unwrap();
           if r == 0 {
               break;
           }
           filled += r;
       }
       unsafe { buf.set_len(filled) };
       buf
   }
   
   fn main() {
       let args: Vec<String> = std::env::args().collect();
       let path = &args[1];
       let range: usize = args[2].parse().unwrap();
       let rounds: usize = args[3].parse().unwrap();
       let mut file = File::open(path).unwrap();
       let len = file.metadata().unwrap().len();
       let ranges: Vec<u64> = (0..len / range as u64).map(|i| i * range as u64).collect();
       let total_bytes = ranges.len() as f64 * range as f64;
   
       // Warm the page cache.
       let mut sink = 0u64;
       for &o in &ranges {
           sink += seek_read_to_end(&mut file, o, range)[0] as u64;
       }
   
       let names = ["zero_fill_read_at", "seek_read_to_end", "pread_uninit"];
       let mut results: Vec<Vec<f64>> = vec![Vec::new(); 3];
       for _ in 0..rounds {
           for (arm, res) in results.iter_mut().enumerate() {
               let t = Instant::now();
               for &o in &ranges {
                   let b = match arm {
                       0 => zero_fill_read_at(&file, o, range),
                       1 => seek_read_to_end(&mut file, o, range),
                       _ => pread_uninit(&file, o, range),
                   };
                   sink = sink.wrapping_add(b[b.len() / 2] as u64);
               }
               res.push(total_bytes / t.elapsed().as_secs_f64() / 1e9);
           }
       }
       for (name, mut r) in names.iter().zip(results) {
           r.sort_by(|a, b| a.partial_cmp(b).unwrap());
           println!(
               "{:>18} range={:>8} median={:6.2} GB/s min={:6.2} max={:6.2}",
               name,
               range,
               r[r.len() / 2],
               r[0],
               r[r.len() - 1]
           );
       }
       eprintln!("sink {sink}");
   }
   ```
   
   </details>
   


-- 
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: github-unsubscribe@arrow.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org