Merge pull request #22 from jeromekelleher/more-ts-hardening

Diagnose mmap-read EAGAIN; mark mmap-read informational

Author: jeromekelleher

biofuse/access_log.py

@@ -94,8 +94,8 @@ def record_event(
         t_end: float | None = None,
     ) -> None:
         """Record a non-read lifecycle event (open / release / limiter_wait /
-        aclose). ``offset`` and ``size`` are zero for these events; the
-        ``[t_start, t_end]`` interval is what matters."""
+        limiter_timeout / aclose). ``offset`` and ``size`` are zero for
+        these events; the ``[t_start, t_end]`` interval is what matters."""
         if t_end is None:
             t_end = time.monotonic()
         self._write(

biofuse/encoder_ops.py

@@ -45,7 +45,9 @@
 # Maximum time a FUSE_OPEN may wait at the per-mount streaming-file
 # capacity limiter. On expiry the open returns ``EAGAIN`` to the kernel
 # rather than blocking forever — this guards against a leaked limiter
-# slot permanently wedging open().
+# slot permanently wedging open(). The timeout is also recorded in the
+# access log as a ``limiter_timeout`` event so a post-hoc trace can
+# tell EAGAIN-from-limiter from any other EAGAIN.
 _LIMITER_TIMEOUT_S = 30.0
 
 _STATIC_KIND = "static"
@@ -222,6 +224,9 @@ async def open(self, inode, flags, ctx=None):
                 await self._stream_limiter.acquire_on_behalf_of(fh)
             t_limiter_end = time.monotonic()
             if cs.cancelled_caught:
+                self._record_event(
+                    "limiter_timeout", name, fh, t_limiter_start, t_limiter_end
+                )
                 raise pyfuse3.FUSEError(errno.EAGAIN)
             self._record_event("limiter_wait", name, fh, t_limiter_start, t_limiter_end)
             try:

fs_tests/README.md

@@ -97,12 +97,22 @@ host (page cache, CPU, fixture size).
 each job in `harness/fio_runner.py`:
 
 - Gated streaming jobs against `.bed`: `seq-read`, `rand-read`,
-  `mmap-read`, `parallel-seq-read` (numjobs=4 sequential). Errors here
-  fail the runner.
-- Informational streaming job: `multithread` (numjobs=16, random) is
-  expected to time out under FUSE backpressure (the kernel returns
-  EAGAIN); its errors are recorded in the detail line but do not gate
-  the run. The concurrency-overlap check it drives is still gated.
+  `parallel-seq-read` (numjobs=4 sequential). Errors here fail the
+  runner.
+- Informational streaming jobs:
+  - `multithread` (numjobs=16 random) times out under FUSE
+    backpressure (EAGAIN).
+  - `mmap-read` (single-job mmap) triggers EAGAIN via a different
+    path: fio's mmap engine does a tight open/close storm (~5000/s on
+    a host filesystem), which on biofuse fills the streaming-fh
+    limiter because each FUSE_OPEN spins up a fresh encoder-server
+    connection. The limiter wait expires and surfaces as EAGAIN.
+    Bytes that fio did manage to read came back valid; the failure is
+    about the open rate, not data correctness. Triggers a
+    `limiter_timeout` event in the access log when it fires.
+  Both jobs' errors are recorded in the detail line but do not gate
+  the run. The concurrency-overlap checks they drive *are* gated —
+  they confirm the kernel is fanning out parallel readahead.
 - Gated static-file jobs: `static-stress-bim` / `static-stress-fam`
   hammer the in-memory cached `.bim` and `.fam` files with
   `numjobs=16` random reads. These must always pass.

fs_tests/harness/fio_runner.py

@@ -28,14 +28,28 @@ class _JobSpec:
 
 
 # ``gate=False`` jobs are informational: errors are recorded in the
-# detail line but don't fail the runner. ``fio-multithread.fio`` is the
-# 16-way parallel random read that reliably times out under FUSE
-# backpressure (EAGAIN); we keep it in the schedule because the
-# concurrency check it drives is still a useful regression signal.
+# detail line but don't fail the runner.
+#
+# - ``fio-multithread.fio`` is the 16-way parallel random read that
+#   reliably times out under FUSE backpressure (EAGAIN).
+# - ``fio-mmap-read.fio`` triggers EAGAIN via a different path: fio's
+#   mmap engine does a tight open/close storm (~5000/s on a host fs),
+#   which on biofuse fills the streaming-fh limiter because each
+#   FUSE_OPEN spins up a fresh encoder-server connection. The
+#   resulting EAGAIN is unrelated to data correctness — bytes already
+#   read came back valid. The job stays in the schedule because the
+#   ``:concurrent`` check it drives confirms the kernel is fanning out
+#   readahead on a single fh.
 JOBS: tuple[_JobSpec, ...] = (
     _JobSpec("seq-read", "fio-seq-read.fio", ".bed"),
     _JobSpec("rand-read", "fio-rand-read.fio", ".bed"),
-    _JobSpec("mmap-read", "fio-mmap-read.fio", ".bed", concurrent="single-fh"),
+    _JobSpec(
+        "mmap-read",
+        "fio-mmap-read.fio",
+        ".bed",
+        gate=False,
+        concurrent="single-fh",
+    ),
     _JobSpec(
         "parallel-seq-read",
         "fio-parallel-seq-read.fio",

tests/test_encoder_ops.py

@@ -484,6 +484,32 @@ async def test_open_returns_eagain_when_limiter_starved(
         finally:
             await ops.release(held.fh)
 
+    async def test_limiter_timeout_records_access_event(
+        self, fx_client, fx_spec, monkeypatch
+    ):
+        """A timed-out FUSE_OPEN must leave a ``limiter_timeout`` event
+        in the access log so post-hoc analysis can attribute an
+        ``EAGAIN`` to limiter starvation rather than some other path."""
+        timeout_s = 0.2
+        monkeypatch.setattr(encoder_ops, "_LIMITER_TIMEOUT_S", timeout_s)
+        log = access_log.AccessLogger()
+        ops = encoder_ops.EncoderOps(
+            fx_client, "small", fx_spec, max_open_stream=1, access_logger=log
+        )
+        stream_inode = ops._name_to_inode[f"small{fx_spec.streaming_suffix}"]
+        held = await ops.open(stream_inode, os.O_RDONLY)
+        try:
+            await _expect_fuse_error(ops.open(stream_inode, os.O_RDONLY), errno.EAGAIN)
+        finally:
+            await ops.release(held.fh)
+        timeouts = [r for r in log.records if r.kind == "limiter_timeout"]
+        assert len(timeouts) == 1
+        rec = timeouts[0]
+        # The held open ran first and got fh=1; the timed-out attempt is
+        # the next allocated fh.
+        assert rec.fh == held.fh + 1
+        assert rec.t_end - rec.t_start >= timeout_s
+
 
 class TestLifecycleEvents:
     """``open`` / ``release`` / ``aclose`` / ``limiter_wait`` events