Trim the PR: drop the byte budget, threading and test bloat

Four review passes (reuse, simplification, efficiency, altitude) against the
shape of #14216. Measured rather than assumed:

- byte-budget heuristic: never fires. n_per already caps a chunk near 10 MiB
  of source bytes, so temporaries stay under 50 MB of the 64 MB cap even for
  an adversarial 2-channel/6-hour file with reversed picks.
- n_read == 1 branch: 0/200 hits on the windowed benchmark, and the general
  branch produces the same values.
- threading: real (13-22% on preload) but reachable only via direct_output,
  and it is one of only two ThreadPoolExecutor sites in MNE. mne.parallel
  already has parallel_func(prefer='threads'). Deferred to its own PR with a
  benchmark.
- direct_output: kept, it measures 27-28% on full preload.

_read_segments_file mmap/threading and the BrainVision block sizing are
reverted to main and move to a follow-up: no MNE fixture is large enough to
reach the 64 MB threading threshold (largest .eeg is 3.7 MB), so it was
untested at any realistic scale.

Tests 549 -> 156 lines: four near-identical stride tests merged into one
parametrized test, redundant file_kind axes dropped, and a test asserting
x * 1.0 == x bit-exactly removed.

+884/-29 -> +330/-18. Output stays bit-identical to main: 77/77 array
snapshots and 56/56 annotation snapshots across 28 files.

Author: bruAristimunha

mne/_fiff/tests/test_utils.py

@@ -4,15 +4,7 @@
 # License: BSD-3-Clause
 # Copyright the MNE-Python contributors.
 
-import threading
-from types import SimpleNamespace
-
-import numpy as np
-import pytest
-from numpy.testing import assert_allclose, assert_array_equal
-
-from mne._fiff import utils as fiff_utils
-from mne._fiff.utils import _check_orig_units, _read_segments_file
+from mne._fiff.utils import _check_orig_units
 
 
 def test_check_orig_units():
@@ -24,150 +16,3 @@ def test_check_orig_units():
     assert orig_units["Pz"] == "µV"
     assert orig_units["greekMu"] == "µV"
     assert orig_units["microSign"] == "µV"
-
-
-@pytest.mark.parametrize("use_mult", (False, True))
-def test_read_segments_file_max_block_bytes(tmp_path, use_mult):
-    """Test reading in configurable complete channel frames."""
-    source = np.arange(20, dtype="<i2").reshape(2, 10, order="F")
-    data_fname = tmp_path / "interleaved.bin"
-    source.ravel(order="F").tofile(data_fname)
-    raw = SimpleNamespace(
-        filenames=[data_fname], _raw_extras=[dict(orig_nchan=source.shape[0])]
-    )
-    if use_mult:
-        data = np.empty((1, 7))
-        cals = None
-        mult = np.array([[0.5, -2.0]])
-        want = mult @ source[:, 1:8]
-    else:
-        data = np.empty((2, 7))
-        cals = np.array([0.5, -2.0])
-        mult = None
-        want = source[:, 1:8] * cals[:, np.newaxis]
-
-    _read_segments_file(
-        raw,
-        data,
-        slice(None),
-        0,
-        1,
-        8,
-        cals,
-        mult,
-        dtype=source.dtype,
-        max_block_bytes=1,
-    )
-
-    if use_mult:
-        assert_allclose(data, want, rtol=1e-15)
-    else:
-        assert_array_equal(data, want)
-
-
-def test_read_segments_file_mmap_threaded(tmp_path, monkeypatch):
-    """Test mapped blocks are calibrated on worker threads."""
-    source = np.arange(40, dtype="<i2").reshape(2, 20, order="F")
-    data_fname = tmp_path / "interleaved.bin"
-    source.ravel(order="F").tofile(data_fname)
-    raw = SimpleNamespace(filenames=[data_fname], _raw_extras=[dict(orig_nchan=2)])
-    data = np.empty(source.shape)
-    thread_ids = set()
-    calibrate = fiff_utils._mult_cal_one
-
-    def _record_thread(*args):
-        thread_ids.add(threading.get_ident())
-        return calibrate(*args)
-
-    def _fail_fromfile(*args, **kwargs):
-        raise AssertionError("mapped reads must not use np.fromfile")
-
-    monkeypatch.setattr(fiff_utils, "_READ_SEGMENTS_FILE_THREAD_MIN_BYTES", 0)
-    monkeypatch.setattr(fiff_utils, "_mult_cal_one", _record_thread)
-    monkeypatch.setattr(np, "fromfile", _fail_fromfile)
-    _read_segments_file(
-        raw,
-        data,
-        slice(None),
-        0,
-        0,
-        source.shape[1],
-        np.ones(source.shape[0]),
-        None,
-        dtype=source.dtype,
-        max_block_bytes=8,
-        use_mmap=True,
-        n_jobs=2,
-    )
-
-    assert threading.get_ident() not in thread_ids
-    assert_array_equal(data, source)
-
-
-def test_read_segments_file_mmap_fallback(tmp_path, monkeypatch):
-    """Test a failed source mapping falls back to ordinary file reads."""
-    source = np.arange(20, dtype="<i2").reshape(2, 10, order="F")
-    data_fname = tmp_path / "interleaved.bin"
-    source.ravel(order="F").tofile(data_fname)
-    raw = SimpleNamespace(filenames=[data_fname], _raw_extras=[dict(orig_nchan=2)])
-    data = np.empty(source.shape)
-
-    def _fail_mmap(*args, **kwargs):
-        raise OSError("mapping unavailable")
-
-    monkeypatch.setattr(fiff_utils.mmap, "mmap", _fail_mmap)
-    _read_segments_file(
-        raw,
-        data,
-        slice(None),
-        0,
-        0,
-        source.shape[1],
-        np.ones(source.shape[0]),
-        None,
-        dtype=source.dtype,
-        max_block_bytes=1,
-        use_mmap=True,
-        n_jobs=2,
-    )
-
-    assert_array_equal(data, source)
-
-
-def test_read_segments_file_mmap_worker_error(tmp_path, monkeypatch):
-    """Test worker errors do not prevent closing the source mapping."""
-    source = np.arange(20, dtype="<i2").reshape(2, 10, order="F")
-    data_fname = tmp_path / "interleaved.bin"
-    source.ravel(order="F").tofile(data_fname)
-    raw = SimpleNamespace(filenames=[data_fname], _raw_extras=[dict(orig_nchan=2)])
-    mappings = []
-    map_file = fiff_utils.mmap.mmap
-
-    def _record_mapping(*args, **kwargs):
-        mapping = map_file(*args, **kwargs)
-        mappings.append(mapping)
-        return mapping
-
-    def _fail_calibration(*args):
-        raise RuntimeError("expected worker failure")
-
-    monkeypatch.setattr(fiff_utils, "_READ_SEGMENTS_FILE_THREAD_MIN_BYTES", 0)
-    monkeypatch.setattr(fiff_utils, "_mult_cal_one", _fail_calibration)
-    monkeypatch.setattr(fiff_utils.mmap, "mmap", _record_mapping)
-    with pytest.raises(RuntimeError, match="expected worker failure"):
-        _read_segments_file(
-            raw,
-            np.empty(source.shape),
-            slice(None),
-            0,
-            0,
-            source.shape[1],
-            np.ones(source.shape[0]),
-            None,
-            dtype=source.dtype,
-            max_block_bytes=8,
-            use_mmap=True,
-            n_jobs=2,
-        )
-
-    assert mappings[0].closed

mne/_fiff/utils.py

@@ -2,20 +2,15 @@
 # License: BSD-3-Clause
 # Copyright the MNE-Python contributors.
 
-import mmap
 import os
 import os.path as op
-from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
-from traceback import clear_frames
 
 import numpy as np
 
 from .constants import FIFF
 from .meas_info import _get_valid_units
 
-_READ_SEGMENTS_FILE_THREAD_MIN_BYTES = 64 * 1024**2
-
 
 def _check_orig_units(orig_units):
     """Check original units from a raw file.
@@ -219,9 +214,6 @@ def _read_segments_file(
     n_channels=None,
     offset=0,
     trigger_ch=None,
-    max_block_bytes=int(100e6),
-    use_mmap=False,
-    n_jobs=1,
 ):
     """Read a chunk of raw data."""
     if n_channels is None:
@@ -233,90 +225,28 @@ def _read_segments_file(
     data_offset = n_channels * start * n_bytes + offset
     data_left = (stop - start) * n_channels
 
-    # block_size is in data samples and spans complete channel frames.
-    block_size = max(
-        n_channels,
-        ((max_block_bytes // n_bytes) // n_channels) * n_channels,
-    )
+    # Read up to 100 MB of data at a time, block_size is in data samples
+    block_size = ((int(100e6) // n_bytes) // n_channels) * n_channels
     block_size = min(data_left, block_size)
     with open(raw.filenames[fi], "rb", buffering=0) as fid:
-        mapped = None
-        if use_mmap and data_left * n_bytes >= max_block_bytes:
-            try:
-                mapped = mmap.mmap(fid.fileno(), 0, access=mmap.ACCESS_READ)
-            except (OSError, ValueError):
-                pass
         fid.seek(data_offset)
         # extract data in chunks
-        executor = None
-        pending = []
-        worker_error = None
-        if (
-            mapped is not None
-            and n_jobs > 1
-            and data.nbytes >= _READ_SEGMENTS_FILE_THREAD_MIN_BYTES
-        ):
-            executor = ThreadPoolExecutor(max_workers=n_jobs)
-        try:
-            for sample_start in np.arange(0, data_left, block_size) // n_channels:
-                count = min(block_size, data_left - sample_start * n_channels)
-                raw_block = None
-                block = None
-                try:
-                    if mapped is None:
-                        raw_block = np.fromfile(fid, dtype, count)
-                    else:
-                        byte_offset = data_offset + sample_start * n_channels * n_bytes
-                        if len(mapped) - byte_offset < count * n_bytes:
-                            raw_block = np.empty(0, dtype=dtype)
-                        else:
-                            raw_block = np.frombuffer(
-                                mapped, dtype, count=count, offset=byte_offset
-                            )
-                    if raw_block.size != count:
-                        raise RuntimeError(
-                            f"Incorrect number of samples ({raw_block.size} != "
-                            f"{count}), please report this error to MNE-Python "
-                            "developers"
-                        )
-                    block = raw_block.reshape(n_channels, -1, order="F")
-                    n_samples = block.shape[1]  # = count // n_channels
-                    sample_stop = sample_start + n_samples
-                    if trigger_ch is not None:
-                        stim_ch = trigger_ch[start:stop][sample_start:sample_stop]
-                        block = np.vstack((block, stim_ch))
-                    data_view = data[:, sample_start:sample_stop]
-                    if executor is None:
-                        _mult_cal_one(data_view, block, idx, cals, mult)
-                    else:
-                        future = executor.submit(
-                            _mult_cal_one, data_view, block, idx, cals, mult
-                        )
-                        pending.append((future, block, raw_block))
-                finally:
-                    del block, raw_block
-            for pending_item in pending:
-                error = pending_item[0].exception()
-                if worker_error is None and error is not None:
-                    worker_error = error
-            if pending:
-                del pending_item
-        finally:
-            if executor is not None:
-                executor.shutdown(wait=True)
-            # Worker tracebacks retain their mapped NumPy inputs. Clear frame locals
-            # before closing the mapping, but keep the traceback locations intact.
-            for pending_item in pending:
-                error = pending_item[0].exception()
-                if error is not None and error.__traceback__ is not None:
-                    clear_frames(error.__traceback__)
-            if pending:
-                del pending_item
-            pending.clear()
-            if mapped is not None:
-                mapped.close()
-        if worker_error is not None:
-            raise worker_error
+        for sample_start in np.arange(0, data_left, block_size) // n_channels:
+            count = min(block_size, data_left - sample_start * n_channels)
+            block = np.fromfile(fid, dtype, count)
+            if block.size != count:
+                raise RuntimeError(
+                    f"Incorrect number of samples ({block.size} != {count}), please "
+                    "report this error to MNE-Python developers"
+                )
+            block = block.reshape(n_channels, -1, order="F")
+            n_samples = block.shape[1]  # = count // n_channels
+            sample_stop = sample_start + n_samples
+            if trigger_ch is not None:
+                stim_ch = trigger_ch[start:stop][sample_start:sample_stop]
+                block = np.vstack((block, stim_ch))
+            data_view = data[:, sample_start:sample_stop]
+            _mult_cal_one(data_view, block, idx, cals, mult)
 
 
 def read_str(fid, count=1):

mne/io/brainvision/brainvision.py

@@ -35,9 +35,6 @@
 )
 from ..base import BaseRaw
 
-_BRAINVISION_BLOCK_BYTES = 8 * 1024**2
-_BRAINVISION_READ_WORKERS = 4
-
 
 @fill_doc
 class RawBrainVision(BaseRaw):
@@ -194,9 +191,6 @@ def _read_segment_file(self, data, idx, fi, start, stop, cals, mult):
                 mult,
                 dtype=dtype,
                 n_channels=n_data_ch,
-                max_block_bytes=_BRAINVISION_BLOCK_BYTES,
-                use_mmap=True,
-                n_jobs=_BRAINVISION_READ_WORKERS,
             )
         else:
             offsets = self._raw_extras[fi]["offsets"]