diff --git a/CHANGELOG.md b/CHANGELOG.md index 477b3b5..dfccb9f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,30 @@ # Changelog +## 0.5.1 + +### Changed + +- Under the `mmap` feature, the buffer-reuse read paths — + `H5Dataset::read_raw_into` and `read_slice_into`, and the reader's + `read_dataset_raw_into` / `read_slice_into` — serve contiguous reads + of any size from the file's memory map. The map's 8 KiB read ceiling + prices the cold case, where faulting a fresh allocation in during the + copy costs more than `pread` past that size; a caller who keeps the + destination buffer has already paid that fault, and for a kept buffer + the map wins at every size, so the into-reads take it with no ceiling. + Rereading a 128 MiB contiguous dataset into a kept buffer goes from + 1.04x libhdf5 to 0.39x, and covering it in 128 KiB slices from 0.92x + to 0.48x. The allocating reads (`read_raw`, `read_slice`) keep the + ceiling: their destination is fresh by construction. + +- The same for an unfiltered chunked dataset's into-reads: the chunk + runs that land straight in the caller's buffer (never materializing a + chunk image) carry the same destination fact, so a kept buffer serves + them from the map at any size too. A full 128 MiB chunked reread into + a kept buffer goes from 0.98x libhdf5 to 0.62x, and 1000 random + 64 KiB slices from 0.98x to 0.50x. Whole-chunk reads (a fresh image + by construction, and every filtered chunk) are unchanged. + ## 0.5.0 ### Added diff --git a/Cargo.toml b/Cargo.toml index ec7d608..cd7d6ef 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "rust-hdf5" description = "Pure Rust HDF5 library with full read/write and SWMR support" -version = "0.5.0" +version = "0.5.1" edition = "2021" rust-version = "1.89" license = "MIT" diff --git a/perf/probe.c b/perf/probe.c index d8405dc..7e0ca1e 100644 --- a/perf/probe.c +++ b/perf/probe.c @@ -188,6 +188,55 @@ int main(int argc, char **argv) { write_contig(path, data, CONTIG_N); free(data); TIMED({ sink = view_and_sum(path, CONTIG_N); }); + } else if (strcmp(wl, "into-read") == 0) { + /* Buffer reuse: open and the first read are setup, so every timed + * read lands in an already-faulted buffer. */ + double *data = ramp(CONTIG_N); + snprintf(path, sizeof path, "%s/c-intoread-in.h5", workdir); + write_contig(path, data, CONTIG_N); + free(data); + double *buf = malloc(CONTIG_N * sizeof(double)); + hid_t f = H5Fopen(path, H5F_ACC_RDONLY, H5P_DEFAULT); + hid_t d = H5Dopen2(f, "data", H5P_DEFAULT); + CHECK(H5Dread(d, H5T_NATIVE_DOUBLE, H5S_ALL, H5S_ALL, H5P_DEFAULT, buf)); + TIMED({ + CHECK(H5Dread(d, H5T_NATIVE_DOUBLE, H5S_ALL, H5S_ALL, H5P_DEFAULT, buf)); + if (buf[CONTIG_N - 1] != (double)(CONTIG_N - 1)) abort(); + }); + H5Dclose(d); + H5Fclose(f); + free(buf); + } else if (strcmp(wl, "into-slice") == 0) { + /* The same reuse per piece: sequential 128 KiB slices into one + * buffer, covering the dataset once per rep. */ + const int PIECE = 16 * 1024; + double *data = ramp(CONTIG_N); + snprintf(path, sizeof path, "%s/c-intoslice-in.h5", workdir); + write_contig(path, data, CONTIG_N); + free(data); + double *buf = malloc(PIECE * sizeof(double)); + hid_t f = H5Fopen(path, H5F_ACC_RDONLY, H5P_DEFAULT); + hid_t d = H5Dopen2(f, "data", H5P_DEFAULT); + hid_t fsp = H5Dget_space(d); + hsize_t count = PIECE; + hid_t msp = H5Screate_simple(1, &count, NULL); + hsize_t off0 = 0; + CHECK(H5Sselect_hyperslab(fsp, H5S_SELECT_SET, &off0, NULL, &count, NULL)); + CHECK(H5Dread(d, H5T_NATIVE_DOUBLE, msp, fsp, H5P_DEFAULT, buf)); + TIMED({ + for (int k = 0; k < CONTIG_N / PIECE; k++) { + hsize_t off = (hsize_t)k * PIECE; + CHECK(H5Sselect_hyperslab(fsp, H5S_SELECT_SET, &off, NULL, + &count, NULL)); + CHECK(H5Dread(d, H5T_NATIVE_DOUBLE, msp, fsp, H5P_DEFAULT, buf)); + } + if (buf[PIECE - 1] != (double)(CONTIG_N - 1)) abort(); + }); + H5Sclose(msp); + H5Sclose(fsp); + H5Dclose(d); + H5Fclose(f); + free(buf); } else if (strcmp(wl, "chunked-write") == 0) { double *data = ramp(CONTIG_N); snprintf(path, sizeof path, "%s/c-chunked.h5", workdir); @@ -200,6 +249,56 @@ int main(int argc, char **argv) { snprintf(path, sizeof path, "%s/c-chunked-in.h5", workdir); write_chunked(path, data, CONTIG_N, 0, CHUNK_ELEMS); TIMED({ free(read_full(path, CONTIG_N)); }); + } else if (strcmp(wl, "chunked-into-read") == 0) { + /* The buffer-reuse pair again, on an unfiltered chunked dataset. */ + double *data = ramp(CONTIG_N); + snprintf(path, sizeof path, "%s/c-chunkintoread-in.h5", workdir); + write_chunked(path, data, CONTIG_N, 0, CHUNK_ELEMS); + free(data); + double *buf = malloc(CONTIG_N * sizeof(double)); + hid_t f = H5Fopen(path, H5F_ACC_RDONLY, H5P_DEFAULT); + hid_t d = H5Dopen2(f, "data", H5P_DEFAULT); + CHECK(H5Dread(d, H5T_NATIVE_DOUBLE, H5S_ALL, H5S_ALL, H5P_DEFAULT, buf)); + TIMED({ + CHECK(H5Dread(d, H5T_NATIVE_DOUBLE, H5S_ALL, H5S_ALL, H5P_DEFAULT, buf)); + if (buf[CONTIG_N - 1] != (double)(CONTIG_N - 1)) abort(); + }); + H5Dclose(d); + H5Fclose(f); + free(buf); + } else if (strcmp(wl, "chunked-into-slice") == 0) { + /* slice-read's random 64 KiB selections, into a kept buffer. */ + double *data = ramp(CONTIG_N); + snprintf(path, sizeof path, "%s/c-chunkintoslice-in.h5", workdir); + write_chunked(path, data, CONTIG_N, 0, CHUNK_ELEMS); + free(data); + double *buf = malloc(SLICE_ELEMS * sizeof(double)); + hid_t f = H5Fopen(path, H5F_ACC_RDONLY, H5P_DEFAULT); + hid_t d = H5Dopen2(f, "data", H5P_DEFAULT); + hid_t fsp = H5Dget_space(d); + hsize_t count = SLICE_ELEMS; + hid_t msp = H5Screate_simple(1, &count, NULL); + hsize_t off0 = 0; + CHECK(H5Sselect_hyperslab(fsp, H5S_SELECT_SET, &off0, NULL, &count, NULL)); + CHECK(H5Dread(d, H5T_NATIVE_DOUBLE, msp, fsp, H5P_DEFAULT, buf)); + TIMED({ + lcg_state = 1; + hsize_t last = 0; + for (int k = 0; k < SLICE_READS; k++) { + hsize_t off = lcg_next() % (CONTIG_N - SLICE_ELEMS); + CHECK(H5Sselect_hyperslab(fsp, H5S_SELECT_SET, &off, NULL, + &count, NULL)); + CHECK(H5Dread(d, H5T_NATIVE_DOUBLE, msp, fsp, H5P_DEFAULT, + buf)); + last = off; + } + if (buf[0] != (double)last) abort(); + }); + H5Sclose(msp); + H5Sclose(fsp); + H5Dclose(d); + H5Fclose(f); + free(buf); } else if (strcmp(wl, "deflate-write") == 0) { double *data = compressible(DEFLATE_N); snprintf(path, sizeof path, "%s/c-deflate.h5", workdir); diff --git a/perf/run.py b/perf/run.py index d1610ed..1d8efb7 100644 --- a/perf/run.py +++ b/perf/run.py @@ -29,8 +29,16 @@ # workload is "obtain the data and sum it", and the view is what changes # between builds. See src/bin/perf_probe.rs. ("contig-view", 5), + # Buffer-reuse steady state: open and the first read are setup, timed + # reads land in an already-faulted buffer. + ("into-read", 5), + ("into-slice", 5), ("chunked-write", 5), ("chunked-read", 5), + # The buffer-reuse pair on an unfiltered chunked dataset: full rereads + # and slice-read's random selections into a kept buffer. + ("chunked-into-read", 5), + ("chunked-into-slice", 5), ("deflate-write", 3), ("deflate-read", 5), ("deflate-slice", 5), diff --git a/src/bin/perf_probe.rs b/src/bin/perf_probe.rs index 40cffda..ce76ea5 100644 --- a/src/bin/perf_probe.rs +++ b/src/bin/perf_probe.rs @@ -137,6 +137,41 @@ fn main() { std::hint::black_box(total); }); } + // The buffer-reuse pattern: the destination is faulted in once, so + // every timed read pays only for moving bytes, not for growing a + // fresh allocation. Open and the first (untimed) read happen in + // setup — this times the steady state of a reader that keeps its + // buffer. + "into-read" => { + let path = p("rs-intoread-in.h5"); + write_contig(&path, &ramp(CONTIG_N)); + let file = H5File::open(&path).unwrap(); + let ds = file.dataset("data").unwrap(); + let mut buf = vec![0f64; CONTIG_N]; + ds.read_raw_into(&mut buf).unwrap(); + timed(&workload, reps, || { + ds.read_raw_into(&mut buf).unwrap(); + assert_eq!(buf[CONTIG_N - 1], (CONTIG_N - 1) as f64); + }); + } + // The same reuse pattern per piece: sequential 128 KiB slices into + // one buffer, covering the dataset once per rep. + "into-slice" => { + const PIECE: usize = 16 * 1024; // f64 -> 128 KiB + let path = p("rs-intoslice-in.h5"); + write_contig(&path, &ramp(CONTIG_N)); + let file = H5File::open(&path).unwrap(); + let ds = file.dataset("data").unwrap(); + let mut buf = vec![0f64; PIECE]; + ds.read_slice_into(&mut buf, &[0], &[PIECE]).unwrap(); + timed(&workload, reps, || { + for k in 0..CONTIG_N / PIECE { + ds.read_slice_into(&mut buf, &[k * PIECE], &[PIECE]) + .unwrap(); + } + assert_eq!(buf[PIECE - 1], (CONTIG_N - 1) as f64); + }); + } "chunked-write" => { let data = ramp(CONTIG_N); let path = p("rs-chunked.h5"); @@ -155,6 +190,41 @@ fn main() { assert_eq!(v.len(), CONTIG_N); }); } + // The buffer-reuse pair again, on an unfiltered chunked dataset: + // the chunk runs land straight in the kept buffer, never + // materializing a chunk image. + "chunked-into-read" => { + let path = p("rs-chunkintoread-in.h5"); + write_chunked(&path, &ramp(CONTIG_N), false, CHUNK_ELEMS); + let file = H5File::open(&path).unwrap(); + let ds = file.dataset("data").unwrap(); + let mut buf = vec![0f64; CONTIG_N]; + ds.read_raw_into(&mut buf).unwrap(); + timed(&workload, reps, || { + ds.read_raw_into(&mut buf).unwrap(); + assert_eq!(buf[CONTIG_N - 1], (CONTIG_N - 1) as f64); + }); + } + // slice-read's random 64 KiB selections, into a kept buffer. + "chunked-into-slice" => { + let path = p("rs-chunkintoslice-in.h5"); + write_chunked(&path, &ramp(CONTIG_N), false, CHUNK_ELEMS); + let file = H5File::open(&path).unwrap(); + let ds = file.dataset("data").unwrap(); + let mut buf = vec![0f64; SLICE_ELEMS]; + ds.read_slice_into(&mut buf, &[0], &[SLICE_ELEMS]).unwrap(); + timed(&workload, reps, || { + let mut rng = Lcg(1); + let mut last = 0usize; + for _ in 0..SLICE_READS { + let off = (rng.next() % (CONTIG_N - SLICE_ELEMS) as u64) as usize; + ds.read_slice_into(&mut buf, &[off], &[SLICE_ELEMS]) + .unwrap(); + last = off; + } + assert_eq!(buf[0], last as f64); + }); + } "deflate-write" => { let data = compressible(DEFLATE_N); let path = p("rs-deflate.h5"); diff --git a/src/dataset.rs b/src/dataset.rs index 9692e7a..c05d687 100644 --- a/src/dataset.rs +++ b/src/dataset.rs @@ -15,6 +15,7 @@ use crate::format::messages::virtual_mapping::VirtualMapping; use crate::format::reference::{Reference, ReferenceTarget}; use crate::format::selection::Selection; use crate::format::storage_kind::AttributeStorage; +use crate::io::file_handle::ReadDst; use crate::io::reader::{read_image_into_new, ExternalFileSegment}; use crate::io::writer::ChunkIndexKind; use crate::types::H5Type; @@ -3811,7 +3812,13 @@ impl H5Dataset { // are touched once instead of being read into a byte buffer and // copied into a second one of the same size. read_image_into_new(count, |image| { - reader.read_slice_into(name, &starts_u64, &counts_u64, image)?; + reader.read_slice_into_dst( + name, + &starts_u64, + &counts_u64, + image, + ReadDst::Fresh, + )?; to_host_byte_order(image, &datatype, T::element_size()) }) } @@ -3928,7 +3935,7 @@ impl H5Dataset { return Err(Hdf5Error::InvalidState("file is not in read mode".into())); }; read_image_into_new(points_u64.len(), |image| { - reader.read_points_into(name, &points_u64, image)?; + reader.read_points_into(name, &points_u64, image, ReadDst::Fresh)?; to_host_byte_order(image, &datatype, T::element_size()) }) } @@ -4419,7 +4426,7 @@ impl H5Dataset { // bytes are touched once rather than being zeroed, read, and // then copied into a second buffer of the same size. read_image_into_new(total / T::element_size(), |image| { - reader.read_dataset_raw_into(name, image)?; + reader.read_dataset_raw_into_dst(name, image, ReadDst::Fresh)?; to_host_byte_order(image, &datatype, T::element_size()) }) } diff --git a/src/io/file_handle.rs b/src/io/file_handle.rs index a2bf1b3..1e1e722 100644 --- a/src/io/file_handle.rs +++ b/src/io/file_handle.rs @@ -143,9 +143,28 @@ enum ReadSource { /// 64 KiB and a whole chunk is megabytes; those stay on `pread`, and it /// matters that they do: at a 64 KiB ceiling the 1000-random-slice read /// workload lost 7.7%, which this ceiling gives back. +/// +/// The ceiling only prices the cold row of the table. A read whose +/// destination the caller keeps — [`ReadDst::Reused`], asserted by the +/// public `*_into` dataset reads — is the warm row, where the map wins at +/// every size, so it is served from the map with no ceiling at all. #[cfg(feature = "mmap")] const MAP_MAX_READ: usize = 8 << 10; +/// Where a read's bytes land, as far as the fault cost of the destination +/// is concerned. The table on [`MAP_MAX_READ`] is priced by this fact: +/// `Fresh` is its cold row (the read's own allocation, faulted in as it +/// fills), `Reused` its warm row (a buffer the caller holds across calls, +/// already faulted). Callers state the fact; [`ReadSource::map_for`] alone +/// turns it into a choice. +#[derive(Clone, Copy, PartialEq, Eq)] +pub enum ReadDst { + /// The destination was allocated for this read. + Fresh, + /// The destination is a buffer the caller reuses across reads. + Reused, +} + impl ReadSource { /// The best source a read-only handle can have: a whole-file map when one /// can be had, positioned reads otherwise. @@ -180,12 +199,13 @@ impl ReadSource { /// The map to serve a read of `len` bytes from, or `None` when this read /// goes to the descriptor — either because the handle has no map, or /// because the read is too big to be worth taking out of one - /// ([`MAP_MAX_READ`]). The one place that decides, so that no read entry - /// point can pick for itself and none can be added that skips the choice. + /// ([`MAP_MAX_READ`]) for the destination it lands in ([`ReadDst`]). + /// The one place that decides, so that no read entry point can pick for + /// itself and none can be added that skips the choice. #[cfg(feature = "mmap")] - fn map_for(&self, len: usize) -> Option<&memmap2::Mmap> { + fn map_for(&self, len: usize, dst: ReadDst) -> Option<&memmap2::Mmap> { match self { - ReadSource::Mapped(map) if len <= MAP_MAX_READ => Some(map), + ReadSource::Mapped(map) if dst == ReadDst::Reused || len <= MAP_MAX_READ => Some(map), _ => None, } } @@ -206,7 +226,7 @@ impl ReadSource { /// a buffer that is about to be overwritten is a second pass over it. fn read_vec(&self, file: &File, at: u64, len: usize) -> std::io::Result> { #[cfg(feature = "mmap")] - if let Some(map) = self.map_for(len) { + if let Some(map) = self.map_for(len, ReadDst::Fresh) { return Ok(mapped_range(map, at, len)?.to_vec()); } let mut buf = vec![0u8; len]; @@ -218,7 +238,7 @@ impl ReadSource { /// what the source holds. fn read_vec_upto(&self, file: &File, at: u64, max_len: usize) -> std::io::Result> { #[cfg(feature = "mmap")] - if let Some(map) = self.map_for(max_len) { + if let Some(map) = self.map_for(max_len, ReadDst::Fresh) { let avail = (map.len() as u64).saturating_sub(at) as usize; return Ok(mapped_range(map, at, max_len.min(avail))?.to_vec()); } @@ -239,12 +259,19 @@ impl ReadSource { /// Exactly `buf.len()` bytes at absolute offset `at`, straight into /// `buf`. Fails with `UnexpectedEof` when the source is too short, which /// is what a short `pread` reports. - fn read_exact_into(&self, file: &File, at: u64, buf: &mut [u8]) -> std::io::Result<()> { + fn read_exact_into( + &self, + file: &File, + at: u64, + buf: &mut [u8], + dst: ReadDst, + ) -> std::io::Result<()> { #[cfg(feature = "mmap")] - if let Some(map) = self.map_for(buf.len()) { + if let Some(map) = self.map_for(buf.len(), dst) { buf.copy_from_slice(mapped_range(map, at, buf.len())?); return Ok(()); } + let _ = dst; pread_exact(file, at, buf) } } @@ -572,7 +599,8 @@ impl FileHandle { let mut addr = 0u64; loop { if addr + HDF5_SIGNATURE.len() as u64 <= file_len { - self.source.read_exact_into(&self.file, addr, &mut buf)?; + self.source + .read_exact_into(&self.file, addr, &mut buf, ReadDst::Fresh)?; if buf == HDF5_SIGNATURE { return Ok(Some(addr)); } @@ -700,10 +728,20 @@ impl FileHandle { /// no allocation to bound. A read past EOF still fails — the positioned /// read returns `UnexpectedEof` when it cannot fill `buf` — but without /// paying a per-call `fstat` on the hot coalesced-read path. - pub fn read_exact_at_into(&self, offset: u64, buf: &mut [u8]) -> std::io::Result<()> { + /// + /// `dst` is the caller's word on where `buf` came from ([`ReadDst`]): + /// a reused buffer lets a mapped handle serve the read from its map at + /// any size. The fact is a required argument, not a defaulted wrapper, + /// so no call site can leave it unstated. + pub fn read_exact_at_into( + &self, + offset: u64, + buf: &mut [u8], + dst: ReadDst, + ) -> std::io::Result<()> { self.flush()?; self.source - .read_exact_into(&self.file, self.abs(offset)?, buf) + .read_exact_into(&self.file, self.abs(offset)?, buf, dst) } /// Read up to `max_len` bytes starting at the given byte offset. @@ -1224,7 +1262,9 @@ mod read_source_tests { assert_eq!(handle.read_at_most(1000, 64).unwrap(), Vec::::new()); assert_eq!(handle.read_at_most(990, 64).unwrap(), bytes[990..].to_vec()); let mut out = [0u8; 10]; - handle.read_exact_at_into(990, &mut out).unwrap(); + handle + .read_exact_at_into(990, &mut out, ReadDst::Fresh) + .unwrap(); assert_eq!(&out, &bytes[990..]); let _ = std::fs::remove_dir_all(&d); } @@ -1247,7 +1287,9 @@ mod read_source_tests { "read_at({offset}, {len})" ); let mut out = vec![0u8; len]; - let err = handle.read_exact_at_into(offset, &mut out).unwrap_err(); + let err = handle + .read_exact_at_into(offset, &mut out, ReadDst::Fresh) + .unwrap_err(); assert_eq!( err.kind(), std::io::ErrorKind::UnexpectedEof, @@ -1259,7 +1301,9 @@ mod read_source_tests { // reports `InvalidInput` where the map reports `UnexpectedEof`. What // the two owe each other is that neither reads. assert!(handle.read_at(u64::MAX, 8).is_err()); - assert!(handle.read_exact_at_into(u64::MAX, &mut [0u8; 8]).is_err()); + assert!(handle + .read_exact_at_into(u64::MAX, &mut [0u8; 8], ReadDst::Fresh) + .is_err()); // `read_at_most` is the one that reports a short read by returning // what there was, so past the end it returns nothing. @@ -1387,7 +1431,9 @@ mod read_source_tests { "read_at_most" ); let mut out = vec![0u8; len]; - handle.read_exact_at_into(offset, &mut out).unwrap(); + handle + .read_exact_at_into(offset, &mut out, ReadDst::Fresh) + .unwrap(); assert_eq!(out, want, "read_exact_at_into"); } let _ = std::fs::remove_dir_all(&d); diff --git a/src/io/reader.rs b/src/io/reader.rs index 2018c2b..d8413d9 100644 --- a/src/io/reader.rs +++ b/src/io/reader.rs @@ -54,7 +54,7 @@ use crate::format::superblock::{ use crate::format::symbol_table::SymbolTableNode; use crate::format::{BlockReader, FormatContext, UNDEF_ADDR}; -use crate::io::file_handle::FileHandle; +use crate::io::file_handle::{FileHandle, ReadDst}; use crate::io::hyperslab::{compute_strides, for_each_contiguous_run}; use crate::io::locking::FileLocking; use crate::io::{FileMeta, IoResult}; @@ -120,15 +120,19 @@ impl<'a> ChunkTarget<'a> { /// reading: the filter pipeline the chunks were stored through, the box of the /// dataset it fills, and the fill value every byte no chunk covers takes. /// -/// Carried as one value because all three are constant across a read and are +/// Carried as one value because all four are constant across a read and are /// threaded unchanged from the entry point through each index type down to /// [`place_chunk_jobs`] — so a new index reader cannot pick up the target and -/// forget the fill. +/// forget the fill or the destination fact. #[derive(Clone, Copy)] struct ChunkReadRequest<'a> { pipeline: Option<&'a FilterPipeline>, target: ChunkTarget<'a>, fill_value: Option<&'a [u8]>, + /// The caller's destination fact ([`ReadDst`]) for the output buffer, + /// which prices the per-run direct reads that land straight in it. + /// Whole-chunk reads into a fresh image are unaffected. + dst: ReadDst, } /// The dataset extent, chunk shape and element size a chunk-placement call @@ -2199,7 +2203,7 @@ fn push_skipped(skipped: &mut Vec<(usize, usize)>, out_len: usize, dst: usize, l /// own order, so every run of the intersection is one positioned read — the /// byte ranges libhdf5 reads for the same selection instead of the whole chunk /// (`H5D__chunk_read`, H5Dchunk.c). Adjacent runs coalesce into one read. -/// `image_len` is how many bytes of the chunk the whole-chunk read would have +/// `job.len` is how many bytes of the chunk the whole-chunk read would have /// held, so the runs placed here are exactly the runs /// [`copy_chunk_runs`] would have placed out of that image. /// @@ -2214,19 +2218,20 @@ fn push_skipped(skipped: &mut Vec<(usize, usize)>, out_len: usize, dst: usize, l /// output. fn read_chunk_runs_into( handle: &FileHandle, - addr: u64, - image_len: usize, + job: &ChunkReadJob, place: &ChunkPlacement, chunk_coords: &[u64], output: &mut [u8], skipped: &mut Vec<(usize, usize)>, + dst: ReadDst, ) -> bool { + let (addr, image_len) = (job.addr, job.len); // (file offset, offset in output, length), coalesced as they are planned. let mut runs: Vec<(u64, usize, usize)> = Vec::new(); let mut selected = 0u64; let out_len = output.len(); - for_each_chunk_run(place, chunk_coords, |src, dst, len| { - let (s, d) = (src as usize, dst as usize); + for_each_chunk_run(place, chunk_coords, |src, out_off, len| { + let (s, d) = (src as usize, out_off as usize); if s + len > image_len || d + len > out_len { push_skipped(skipped, out_len, d, len); return; @@ -2251,9 +2256,9 @@ fn read_chunk_runs_into( if (runs.len() as u64 - 1).saturating_mul(PREAD_COST_BYTES) > image_len as u64 + selected { return false; } - for (offset, dst, len) in runs { + for (offset, at, len) in runs { if handle - .read_exact_at_into(offset, &mut output[dst..dst + len]) + .read_exact_at_into(offset, &mut output[at..at + len], dst) .is_err() { return false; @@ -2303,6 +2308,7 @@ fn place_chunk_jobs( pipeline, target, fill_value, + dst, } = req; let rank = geo.dims.len(); let zeros = vec![0u64; rank]; @@ -2321,7 +2327,7 @@ fn place_chunk_jobs( for (i, job) in jobs.iter_mut().enumerate() { let Some(j) = job.as_ref() else { continue }; let mark = skipped.len(); - if read_chunk_runs_into(handle, j.addr, j.len, &place, at(i), output, &mut skipped) { + if read_chunk_runs_into(handle, j, &place, at(i), output, &mut skipped, dst) { *job = None; } else { // The whole-chunk path re-places this chunk and records what @@ -4837,7 +4843,7 @@ impl Hdf5Reader { } let (datatype, total) = self.raw_size_and_datatype(name)?; read_image_into_new(total as usize, |data| { - self.read_dataset_raw_into_unconverted(name, data)?; + self.read_dataset_raw_into_unconverted(name, data, ReadDst::Fresh)?; Self::apply_post_filter_conversion(data, &datatype) }) } @@ -4851,9 +4857,22 @@ impl Hdf5Reader { /// point for reading directly into a pinned/registered host buffer for an /// H2D transfer. pub fn read_dataset_raw_into(&mut self, name: &str, out: &mut [u8]) -> IoResult<()> { + self.read_dataset_raw_into_dst(name, out, ReadDst::Reused) + } + + /// [`read_dataset_raw_into`](Self::read_dataset_raw_into) with the + /// caller's destination fact made explicit, for internal callers whose + /// buffer is a fresh allocation rather than a kept one (the allocating + /// wrappers in `dataset.rs` / `swmr.rs`). + pub(crate) fn read_dataset_raw_into_dst( + &mut self, + name: &str, + out: &mut [u8], + dst: ReadDst, + ) -> IoResult<()> { if self.external_edge(name).is_some() { let (owner, path, _) = self.external_owner(name, MAX_EXTERNAL_HOPS)?; - return owner.read_dataset_raw_into(&path, out); + return owner.read_dataset_raw_into_dst(&path, out, dst); } let (datatype, total) = self.raw_size_and_datatype(name)?; if out.len() as u64 != total { @@ -4863,7 +4882,7 @@ impl Hdf5Reader { total ))); } - self.read_dataset_raw_into_unconverted(name, out)?; + self.read_dataset_raw_into_unconverted(name, out, dst)?; Self::apply_post_filter_conversion(out, &datatype)?; Ok(()) } @@ -4876,8 +4895,16 @@ impl Hdf5Reader { /// allocating `read_dataset_raw` and the zero-copy `read_dataset_raw_into` /// wrap it and apply the conversion exactly once. /// - /// `out.len()` must equal `product(dims) * element_size`. - fn read_dataset_raw_into_unconverted(&mut self, name: &str, out: &mut [u8]) -> IoResult<()> { + /// `out.len()` must equal `product(dims) * element_size`. `dst` is the + /// caller's destination fact ([`ReadDst`]): whether `out` is a fresh + /// allocation or a buffer the caller keeps across reads, which decides + /// whether a mapped contiguous read is priced at the cold or the warm row. + fn read_dataset_raw_into_unconverted( + &mut self, + name: &str, + out: &mut [u8], + dst: ReadDst, + ) -> IoResult<()> { let info = self .dataset_info_local(name) .ok_or_else(|| crate::io::IoError::NotFound(name.to_string()))?; @@ -4899,7 +4926,7 @@ impl Hdf5Reader { fill_tiled_into(out, fill_value.as_deref()); } else { // Read exactly the logical image straight into `out`. - self.handle.read_exact_at_into(*address, out)?; + self.handle.read_exact_at_into(*address, out, dst)?; } } DataLayoutMessage::Compact { data } => { @@ -4926,6 +4953,7 @@ impl Hdf5Reader { pipeline: pipeline.as_ref(), target: ChunkTarget::Full, fill_value: fill_value.as_deref(), + dst, }, out, )?; @@ -4952,6 +4980,7 @@ impl Hdf5Reader { pipeline: pipeline.as_ref(), target: ChunkTarget::Full, fill_value: fill_value.as_deref(), + dst, }, out, )?; @@ -5409,7 +5438,18 @@ impl Hdf5Reader { )) })?; copy_matched_selections( - |s, c, buf| self.read_slice_into_unconverted(source_name, s, c, buf, depth + 1), + |s, c, buf| { + // `buf` is `copy_matched_selections`' fresh per-box + // buffer, never the virtual dataset's own `out`. + self.read_slice_into_unconverted( + source_name, + s, + c, + buf, + depth + 1, + ReadDst::Fresh, + ) + }, &source_sel, &virtual_sel, element_size, @@ -5435,7 +5475,15 @@ impl Hdf5Reader { })?; copy_matched_selections( |s, c, buf| { - src_reader.read_slice_into_unconverted(source_name, s, c, buf, depth + 1) + // As above: `buf` is a fresh per-box buffer. + src_reader.read_slice_into_unconverted( + source_name, + s, + c, + buf, + depth + 1, + ReadDst::Fresh, + ) }, &source_sel, &virtual_sel, @@ -5586,9 +5634,7 @@ impl Hdf5Reader { output: &mut [u8], ) -> IoResult<()> { let ChunkReadRequest { - pipeline, - target, - fill_value: _, + pipeline, target, .. } = req; let ChunkIndexDesc { index_type, @@ -6894,7 +6940,7 @@ impl Hdf5Reader { // `read_slice_into_unconverted` defines every byte of the buffer it is // handed, so there is nothing to zero first and nothing to copy after. read_image_into_new::(out_bytes as usize, |image| { - self.read_slice_into_unconverted(name, starts, counts, image, 0)?; + self.read_slice_into_unconverted(name, starts, counts, image, 0, ReadDst::Fresh)?; Self::apply_post_filter_conversion(image, &datatype) }) } @@ -6912,10 +6958,25 @@ impl Hdf5Reader { starts: &[u64], counts: &[u64], out: &mut [u8], + ) -> IoResult<()> { + self.read_slice_into_dst(name, starts, counts, out, ReadDst::Reused) + } + + /// [`read_slice_into`](Self::read_slice_into) with the caller's + /// destination fact made explicit, for internal callers whose buffer is a + /// fresh allocation rather than a kept one (the allocating wrappers in + /// `dataset.rs`). + pub(crate) fn read_slice_into_dst( + &mut self, + name: &str, + starts: &[u64], + counts: &[u64], + out: &mut [u8], + dst: ReadDst, ) -> IoResult<()> { if self.external_edge(name).is_some() { let (owner, path, _) = self.external_owner(name, MAX_EXTERNAL_HOPS)?; - return owner.read_slice_into(&path, starts, counts, out); + return owner.read_slice_into_dst(&path, starts, counts, out, dst); } let (datatype, out_bytes) = self.slice_size_and_datatype(name, counts)?; if out.len() as u64 != out_bytes { @@ -6925,7 +6986,7 @@ impl Hdf5Reader { out_bytes ))); } - self.read_slice_into_unconverted(name, starts, counts, out, 0)?; + self.read_slice_into_unconverted(name, starts, counts, out, 0, dst)?; Self::apply_post_filter_conversion(out, &datatype)?; Ok(()) } @@ -6956,7 +7017,10 @@ impl Hdf5Reader { /// `out.len()` must equal `product(counts) * element_size`. `depth` /// counts virtual-dataset nesting for a caller reached through /// [`read_virtual_into`](Self::read_virtual_into); pass `0` for a - /// top-level call. + /// top-level call. `dst` is the caller's destination fact ([`ReadDst`]): + /// whether `out` is a fresh allocation or a buffer the caller keeps + /// across reads, which decides whether a mapped contiguous read is priced + /// at the cold or the warm row. fn read_slice_into_unconverted( &mut self, name: &str, @@ -6964,6 +7028,7 @@ impl Hdf5Reader { counts: &[u64], out: &mut [u8], depth: usize, + dst: ReadDst, ) -> IoResult<()> { let info = self .dataset_info_local(name) @@ -7041,6 +7106,7 @@ impl Hdf5Reader { .read_exact_at_into( base + src_off, &mut out[out_off..out_off + len], + dst, ) .map_err(Into::into) }, @@ -7080,6 +7146,7 @@ impl Hdf5Reader { pipeline: pipeline.as_ref(), target: ChunkTarget::Slice { starts, counts }, fill_value: fill_value.as_deref(), + dst, }, out, )?; @@ -7110,6 +7177,7 @@ impl Hdf5Reader { pipeline: pipeline.as_ref(), target: ChunkTarget::Slice { starts, counts }, fill_value: fill_value.as_deref(), + dst, }, out, )?; @@ -7248,7 +7316,11 @@ impl Hdf5Reader { // first, exactly as the allocating form always did. fill_tiled_into(out, None); copy_matched_selections( - |bstart, bcount, buf| self.read_slice_into_unconverted(name, bstart, bcount, buf, 0), + |bstart, bcount, buf| { + // `buf` is `copy_matched_selections`' fresh per-box buffer, + // not the caller's `out`. + self.read_slice_into_unconverted(name, bstart, bcount, buf, 0, ReadDst::Fresh) + }, &src_sel.resolve(&dims)?, &dst_sel.resolve(&out_dims)?, element_size, @@ -7276,15 +7348,19 @@ impl Hdf5Reader { /// One element-sized box per point covers the buffer exactly, so every /// byte of `out` is defined before it returns `Ok` and a typed caller /// reads straight into the vector it keeps. + /// `dst` is the caller's destination fact ([`ReadDst`]) for `out` — a + /// required argument rather than a defaulted wrapper, since unlike the + /// raw/slice pair this read has no kept-buffer caller to assert it for. pub fn read_points_into( &mut self, name: &str, points: &[Vec], out: &mut [u8], + dst: ReadDst, ) -> IoResult<()> { if self.external_edge(name).is_some() { let (owner, path, _) = self.external_owner(name, MAX_EXTERNAL_HOPS)?; - return owner.read_points_into(&path, points, out); + return owner.read_points_into(&path, points, out, dst); } let info = self .dataset_info_local(name) @@ -7326,6 +7402,7 @@ impl Hdf5Reader { bcount, &mut out[i * es..(i + 1) * es], 0, + dst, )?; } Self::apply_post_filter_conversion(out, &datatype) @@ -8183,6 +8260,7 @@ mod tests { pipeline: None, target: ChunkTarget::Full, fill_value: Some(&FILL), + dst: ReadDst::Fresh, }, &geo, None, diff --git a/src/swmr.rs b/src/swmr.rs index 459ec43..d0f34f1 100644 --- a/src/swmr.rs +++ b/src/swmr.rs @@ -639,7 +639,11 @@ impl SwmrFileReader { // As the non-SWMR full read: the image lands in the vector this // returns rather than in a byte buffer copied into it afterwards. crate::io::reader::read_image_into_new(total / es, |image| { - self.reader.read_dataset_raw_into(name, image)?; + self.reader.read_dataset_raw_into_dst( + name, + image, + crate::io::file_handle::ReadDst::Fresh, + )?; crate::dataset::to_host_byte_order(image, &datatype, es) }) }