Fix numpy reader returning unread sample data - #6482
Conversation
NumpyLoader::ReadSample defers reading the sample data whenever the stream cannot be memory mapped, leaving it to NumpyReaderCPU::Prefetch. The two disagreed on when that happens: the loader decides from the stream itself (`!opts.use_mmap || !current_file->CanMemoryMap()`), while Prefetch returned early unless the user passed dont_use_mmap. When the two disagreed the deferred read never happened and the sample was handed out as a freshly allocated, never-written buffer - zeros, or stale heap contents - with the correct shape and dtype and no error. This affects every s3:// path, because S3FileStream is never memory mappable, and local files whenever the mmap reservation cannot be satisfied. Drop the early return. The loop below already skips samples that need no deferred read, and in the memory mapped case that is all of them, so the behaviour there is unchanged. use_o_direct is unaffected; it already requires dont_use_mmap. Add a regression test that exhausts the mmap reservation locally by asking for an initial_fill above the max_map_count / 2 limit.
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved critical and moderate findings remain.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Fixes the NumPy reader returning unread sample data when memory mapping is unavailable.
Changes:
- Performs deferred reads for non-mappable samples.
- Adds a regression test for mmap fallback behavior.
File summaries
| File | Review summary |
|---|---|
dali/operators/reader/numpy_reader_op.cc |
Critical: Mixed mapped and non-mappable batches can still leave deferred samples unread due to the existing break. |
dali/test/python/reader/test_numpy.py |
Moderate: The test may create excessive file handles by scaling samples with the host max_map_count. |
Review details
- Files reviewed: 2/2 changed files
- Comments generated: 2
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| // Read the samples that the loader could not provide by memory mapping. Note that this is not | ||
| // the same as `dont_use_mmap_`: NumpyLoader::ReadSample decides based on the actual stream | ||
| // (`!opts.use_mmap || !current_file->CanMemoryMap()`), so a sample can need a deferred read even | ||
| // when the user did not ask for `dont_use_mmap`, e.g. for remote storage, which is never | ||
| // mappable, or when the mmap reservation could not be satisfied. Samples that the loader already |
| # FileLoader reserves initial_buffer_fill_ mappings up front and falls back to copying reads | ||
| # when that fails; the limit is max_map_count / 2 (mmaped_file.cc). initial_fill only feeds | ||
| # initial_buffer_fill_ when random_shuffle is on. | ||
| initial_fill = max_map_count // 2 + 1 |
 |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| // Read the samples that the loader could not provide by memory mapping. Note that this is not | ||
| // the same as `dont_use_mmap_`: NumpyLoader::ReadSample decides based on the actual stream | ||
| // (`!opts.use_mmap || !current_file->CanMemoryMap()`), so a sample can need a deferred read even | ||
| // when the user did not ask for `dont_use_mmap`, e.g. for remote storage, which is never | ||
| // mappable, or when the mmap reservation could not be satisfied. Samples that the loader already | ||
| // provided leave `current_file` empty and are skipped below. |
There was a problem hiding this comment.
When mmap is enabled, a batch can contain a mapped local sample followed by a deferred, non-mappable sample if files mixes local and S3 paths. The loop exits at the first mapped sample because its current_file is empty, so later remote samples are reset without being read and can produce empty or invalid output. Skip mapped samples individually while preserving the separate break for duplicate padding samples.
| pipe = Pipeline(batch_size=1, num_threads=1, device_id=None) | ||
| with pipe: | ||
| pipe.set_outputs( | ||
| fn.readers.numpy( | ||
| file_root=test_data_root, random_shuffle=True, initial_fill=initial_fill | ||
| ) | ||
| ) |
There was a problem hiding this comment.
This new test manually constructs a Pipeline and uses with pipe and set_outputs. The repository requires new tests to define pipelines with @pipeline_def, so this requirement must be satisfied before merging.
| pipe = Pipeline(batch_size=1, num_threads=1, device_id=None) | |
| with pipe: | |
| pipe.set_outputs( | |
| fn.readers.numpy( | |
| file_root=test_data_root, random_shuffle=True, initial_fill=initial_fill | |
| ) | |
| ) | |
| @pipeline_def(batch_size=1, num_threads=1, device_id=None) | |
| def make_pipe(): | |
| return fn.readers.numpy( | |
| file_root=test_data_root, random_shuffle=True, initial_fill=initial_fill | |
| ) | |
| pipe = make_pipe() |
Rule Used: New tests should use @pipeline_def and DALI_extra paths (via dali_extra_path() / DALI_EXTRA_PATH). Don't use the legacy ops. API in new tests, and don't hardcode personal data paths. (source)
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
| # FileLoader reserves initial_buffer_fill_ mappings up front and falls back to copying reads | ||
| # when that fails; the limit is max_map_count / 2 (mmaped_file.cc). initial_fill only feeds | ||
| # initial_buffer_fill_ when random_shuffle is on. | ||
| initial_fill = max_map_count // 2 + 1 |
There was a problem hiding this comment.
The test derives initial_fill from the host's max_map_count, and loader initialization performs that many sample reads while repeatedly opening and parsing the only file. This means tens of thousands of operations on common hosts and potentially hundreds of thousands on hosts with a larger limit, making test duration depend heavily on system configuration. Use a bounded way to trigger the reservation failure instead.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
|
!build |
|
CI MESSAGE: [67423401]: BUILD STARTED |
|
CI MESSAGE: [67423401]: BUILD FAILED |
Category:
Bug fix (non-breaking change which fixes an issue)
Description:
fn.readers.numpycould hand out sample buffers that were never read. No error was raised: theoperator reported the correct shape and dtype, and the read was reported as full-length, but the
contents were whatever happened to be in the freshly allocated buffer - zeros, or stale heap data.
Two places decide whether the sample data still needs reading, and they disagreed.
NumpyLoader::ReadSampledefers the read based on the actual stream:NumpyReaderCPU::Prefetch, which performs that deferred read, gated on the user-facing argumentinstead:
So when the stream was not memory mappable but
dont_use_mmapwas not requested, the loaderdeferred the read and the operator never performed it.
This affects every
s3://path, becauseS3FileStreamdoes not overrideCanMemoryMap()and so isnever mappable, and local files whenever the mmap reservation cannot be satisfied -
copy_read_data_is
dont_use_mmap_ || !mmap_reserver_.CanShareMappedData().The fix drops the early return. The loop below already skips samples that need no deferred read, and
in the memory mapped case that is all of them, so behaviour there is unchanged.
use_o_directisunaffected; it already requires
dont_use_mmap.Additional information:
Affected modules and functionalities:
dali/operators/reader/numpy_reader_op.cc:NumpyReaderCPU::Prefetchno longer returns early.dali/test/python/reader/test_numpy.py: new regression test.Key points relevant for the review:
breakon!target->current_fileis the behaviour we want for a batch thatmixes mappable and non-mappable sources - it is pre-existing, and reachable today via
files=with a mix of local and
s3://paths, but this change does not alter it.initial_fillabove the
max_map_count / 2limit, which makes the mmap reservation fail.initial_fillonlyfeeds
initial_buffer_fill_whenrandom_shuffleis on. It uses a single tiny sample so theshuffle buffer fill stays cheap (about 1 s) and shuffling cannot change which sample comes back.
test passes, with the file reverted it fails with an all-zero array, and with the fix restored it
passes again.
Tests:
test_numpy.py: test_sample_data_is_read_when_mmap_is_unavailableChecklist
Documentation
DALI team only
Requirements
REQ IDs: N/A
JIRA TASK: DALI-4887