Add cache eviction, byte-range fetch, and worker-pool concurrency

.gitignore

@@ -131,6 +131,10 @@ docs/demos/SCA26/cpp_prod
 tests/integration/dlio_benchmark/logs
 scripts/checkpoints
 scripts/logs
+tests/pydyad_spsc_range/logs
+tests/pydyad_spsc_lazy_range/logs
 tests/integration/dlio_benchmark/perf_analysis/.ipynb_checkpoints
 
 /CLAUDE.md
+/deps
+/source-me.sh

docs/_fragments/design_cache.rst

@@ -0,0 +1,96 @@
+.. _dyad_cache_eviction:
+
+Node-Local Cache Eviction
+#########################
+
+By default, DYAD's producer- and consumer-managed directories (DMDs) grow
+without bound: every file produced or consumed accumulates on node-local
+storage forever. For workloads that repeatedly touch a shifting working set
+larger than any single node's capacity — for example the shuffled-sample-file
+training pattern described above, where a `DistributedSampler` shuffles
+across all shard files every epoch — this means a node's DMD eventually
+approaches the size of the entire dataset rather than just the fraction it
+actually needs at any one time.
+
+DYAD addresses this with an optional, pluggable cache-eviction subsystem,
+disabled by default (``DYAD_CACHE_CAPACITY`` unset or ``0``), so existing
+deployments see no change in behavior unless they opt in.
+
+Pluggable policy architecture
+==============================
+
+Cache eviction follows the same struct-of-function-pointers pattern as the
+:ref:`Data Transport Layer <dyad_dtl>`: a policy handle selected by a mode
+enum (:code:`DYAD_CACHE_NONE`, :code:`DYAD_CACHE_LRU`, :code:`DYAD_CACHE_FIFO`),
+with the active policy's implementation reached through a single function
+pointer, ``get_recency_key()``, rather than DTL's larger transport-specific
+interface. Unlike the UCX/Margo DTL backends, LRU and FIFO have no external
+dependencies, so both are always compiled in — no CMake feature flag is
+needed to enable them.
+
+Recency tracking without a persisted index
+=============================================
+
+Rather than maintaining a separate database of access times (which would
+itself need cross-process synchronization to stay consistent), each policy
+derives its ranking directly from filesystem metadata already maintained by
+the kernel:
+
+- **LRU** ranks by ``st_atime`` — updated on every read, with no DYAD code
+  changes needed on the hot read path.
+- **FIFO** ranks by ``st_mtime`` — files under a DYAD-managed directory are
+  effectively write-once, so modification time approximates insertion order.
+
+This works correctly across independent processes sharing one managed
+directory (for example, PyTorch ``DataLoader`` workers with ``num_workers >
+0``, which run as separate OS processes, each with its own DYAD context) with
+no shared memory or IPC: ``stat()`` is a kernel-level, cross-process-consistent
+syscall, so every process observes the same recency signal for a given file.
+
+Locking and eviction procedure
+=================================
+
+The only cross-process hazard is the *scan-and-evict decision* itself — two
+processes could otherwise race to decide a file needs evicting and both act
+on it. This is serialized with a dedicated per-directory advisory lock file
+(``.dyad_cache.lock``), reusing the same ``dyad_excl_flock()``/
+``dyad_release_flock()`` primitives already described in
+:ref:`Local file access coordination via locking <dyad_local-file-access>` for
+per-file coordination — just applied to a directory-level sentinel instead of
+a data file.
+
+``dyad_produce()`` and ``dyad_consume()``/``dyad_consume_w_metadata()`` each
+call ``dyad_cache_maybe_evict()`` once they otherwise complete successfully.
+That function:
+
+1. Returns immediately if eviction is disabled (``DYAD_CACHE_CAPACITY`` is
+   ``0``, the default) — a single branch, no filesystem access at all.
+2. Scans the managed directory's current on-disk usage. If already at or
+   under capacity, returns without taking any lock.
+3. Otherwise acquires the directory lock, re-checks usage (another process
+   may have already evicted enough while this one waited), and if still over
+   capacity, ranks every candidate file by the active policy's recency key.
+4. Walks candidates oldest-first, skipping any file accessed within
+   ``DYAD_CACHE_GRACE_PERIOD`` seconds or currently locked by another
+   in-flight ``dyad_produce()``/``dyad_consume()`` call (checked via a
+   non-blocking ``dyad_try_excl_flock()``, so the evictor never blocks
+   waiting on a file that's mid-transfer), removing files until usage drops
+   to ``DYAD_CACHE_LOW_WATERMARK`` (a fraction of capacity, default 0.8, so a
+   single eviction pass doesn't need to run again on the very next call).
+5. Releases the directory lock.
+
+Eviction failures on individual files are logged and skipped; this mechanism
+never turns an otherwise-successful ``dyad_produce()``/``dyad_consume()``
+call into a reported error.
+
+Known limitation: stale metadata on producer-side eviction
+===============================================================
+
+Evicting a file from the local DMD does not retract its published Flux KVS
+metadata. :ref:`Case 2 <dyad_metadata_lookup>` of the Hierarchical File
+Locator already tolerates a *consumer-side* cache miss by falling back to the
+global HFL hierarchy — but if the *producer's* origin copy is evicted, a
+later consumer resolving that file's metadata could still be pointed at a
+producer that no longer has it locally, causing the subsequent fetch to fail.
+Addressing this — for example with a metadata retraction/unpublish primitive,
+or a staleness check at fetch time — is left as future work.

docs/design.rst

@@ -56,5 +56,7 @@ representing PyTorch dataloader :ref:`Devarajan et al. 2024 <paper-sbacpad-2024>
 
 .. include:: _fragments/design_local_access.rst
 
+.. include:: _fragments/design_cache.rst
+
 .. include:: _fragments/design_init.rst
 

docs/runtime_configuration.rst

@@ -43,7 +43,18 @@ variables is shown below.
 |                                    |                 |              |          |                                                                 |
 |                                    |                 |              |          | hierarchical KVS within DYAD's namespace.                       |
 +------------------------------------+-----------------+--------------+----------+-----------------------------------------------------------------+
-
+| :code:`DYAD_CACHE_CAPACITY`        | Integer (bytes) | No           | 0        | Maximum bytes DYAD may use in a managed directory before        |
+|                                    |                 |              |          | evicting files. 0 (default) disables eviction entirely.         |
++------------------------------------+-----------------+--------------+----------+-----------------------------------------------------------------+
+| :code:`DYAD_CACHE_POLICY`          | String          | No           | LRU      | Eviction policy: :code:`LRU`, :code:`FIFO`, or :code:`NONE`.    |
+|                                    |                 |              |          | Ignored if :code:`DYAD_CACHE_CAPACITY` is 0.                    |
++------------------------------------+-----------------+--------------+----------+-----------------------------------------------------------------+
+| :code:`DYAD_CACHE_LOW_WATERMARK`   | Float (0..1)    | No           | 0.8      | Fraction of :code:`DYAD_CACHE_CAPACITY` to evict down to once   |
+|                                    |                 |              |          | eviction triggers.                                              |
++------------------------------------+-----------------+--------------+----------+-----------------------------------------------------------------+
+| :code:`DYAD_CACHE_GRACE_PERIOD`    | Integer (sec)   | No           | 5        | Skip eviction candidates accessed more recently than this many  |
+|                                    |                 |              |          | seconds.                                                        |
++------------------------------------+-----------------+--------------+----------+-----------------------------------------------------------------+
 .. [#two] For DYAD to do anything, at least one of :code:`DYAD_PATH_PRODUCER` or :code:`DYAD_PATH_CONSUMER` must be provided.
    Applications will still work if neither are provided, but DYAD will not do anything.
 

include/dyad/client/dyad_client.h

@@ -261,6 +261,59 @@ DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_consume (dyad_ctx_t *ctx, con
 DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t
 dyad_consume_w_metadata (dyad_ctx_t *ctx, const char *fname, const dyad_metadata_t *mdata);
 
+/**
+ * @brief Fetches a byte range of a file without materializing a local copy.
+ *
+ * @details
+ * Unlike @c dyad_consume()/@c dyad_consume_w_metadata(), which always
+ * materialize a complete local file (required for GOTCHA @c open()
+ * interception compatibility), this function returns the requested
+ * @p [offset, offset + length) range of @p fname directly to the caller as
+ * an in-memory buffer -- no local file is created or written. Intended for
+ * callers (e.g. the pydyad HDF/flat-cache binding) that only need raw bytes,
+ * not a file descriptor.
+ *
+ * The file must already have been published via @c dyad_produce() by some
+ * rank (its full contents staged there beforehand -- this function does not
+ * stage or copy files, only fetches a sub-range of one that's already fully
+ * present somewhere). Behavior:
+ *  1. Resolves @p fname to a path relative to the consumer-managed
+ *     directory, same as @c dyad_consume().
+ *  2. Calls the existing @c dyad_fetch_metadata() unchanged: if the file is
+ *     already present on this node (KVS lookup returns @c NULL), reads
+ *     @p [offset, length) directly from the local copy via @c pread() --
+ *     no RPC. Otherwise, issues a byte-range RPC to the owning rank
+ *     returned in the metadata.
+ *  3. Only implemented for @c DYAD_DTL_FLUX_RPC and @c DYAD_DTL_MARGO --
+ *     returns @c DYAD_RC_BADDTLMODE immediately under @c DYAD_DTL_UCX.
+ *
+ * @param[in]  ctx      Pointer to the DYAD context. Must not be @c NULL.
+ * @param[in]  fname    Path to the file to fetch from. Must not be @c NULL.
+ * @param[in]  offset   Starting byte offset of the requested range.
+ * @param[in]  length   Number of bytes requested.
+ * @param[out] data     Set to a newly allocated buffer containing the
+ *                       requested bytes. The caller must @c free() it.
+ * @param[out] data_len Set to the number of bytes actually returned in
+ *                       @p data (normally equal to @p length).
+ *
+ * @return @c dyad_rc_t return code indicating the outcome:
+ * @retval DYAD_RC_OK           The requested range was retrieved successfully.
+ * @retval DYAD_RC_NOCTX        The context @p ctx or its Flux handle is @c NULL.
+ * @retval DYAD_RC_BADMANAGEDPATH The consumer-managed path in the context is @c NULL.
+ * @retval DYAD_RC_BADDTLMODE   The active DTL mode is @c DYAD_DTL_UCX, which
+ *                              does not implement byte-range fetch.
+ * @retval DYAD_RC_BADFIO       A local @c pread() failed.
+ * @retval DYAD_RC_*            Any error code propagated from
+ *                              @c dyad_fetch_metadata() or the internal
+ *                              RPC/DTL path.
+ */
+DYAD_PFA_ANNOTATE DYAD_DLL_EXPORTED dyad_rc_t dyad_consume_range (dyad_ctx_t *ctx,
+                                                                  const char *fname,
+                                                                  size_t offset,
+                                                                  size_t length,
+                                                                  void **data,
+                                                                  size_t *data_len);
+
 #ifdef __cplusplus
 }
 #endif

include/dyad/common/dyad_cache.h

@@ -0,0 +1,90 @@
+/**
+ * @file dyad_cache.h
+ * @brief Cache-eviction policy interface definitions for DYAD.
+ *
+ * @details
+ * Defines the enumeration describing the available node-local cache
+ * eviction policies for DYAD's producer/consumer managed directories
+ * (DMDs). The eviction subsystem abstracts the choice of *which* cached
+ * file to remove when a managed directory exceeds its configured
+ * capacity, so the rest of DYAD does not need to know which policy is
+ * active. This mirrors the Data Transport Layer (DTL) abstraction in
+ * @c dyad_dtl.h.
+ */
+
+#ifndef DYAD_COMMON_DYAD_CACHE_H
+#define DYAD_COMMON_DYAD_CACHE_H
+
+#if defined(DYAD_HAS_CONFIG)
+#include <dyad/dyad_config.hpp>
+#else
+#error "no config"
+#endif
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * @brief Available cache-eviction policies for DYAD-managed directories.
+ *
+ * @details
+ * Selects how DYAD picks which cached file(s) to remove when a managed
+ * directory exceeds @c DYAD_CACHE_CAPACITY. Each policy ranks candidate
+ * files by a "recency key" derived from filesystem metadata (no
+ * separate persisted index is kept):
+ *
+ * - @c DYAD_CACHE_NONE disables eviction entirely (the default). DYAD's
+ *   managed directories grow unboundedly, matching all prior behavior.
+ * - @c DYAD_CACHE_LRU evicts the least-recently-*accessed* file first,
+ *   ranked by @c st_atime.
+ * - @c DYAD_CACHE_FIFO evicts the oldest file first, ranked by
+ *   @c st_mtime (files under DYAD's managed directories are effectively
+ *   write-once, so mtime approximates insertion order).
+ * - @c DYAD_CACHE_DEFAULT aliases @c DYAD_CACHE_LRU — the default
+ *   *policy* once eviction is enabled (i.e. once
+ *   @c DYAD_CACHE_CAPACITY is set to a nonzero value).
+ * - @c DYAD_CACHE_END is a sentinel marking the end of valid values.
+ *   It is used to size @c dyad_cache_policy_name and for bounds checking.
+ */
+enum dyad_cache_policy_mode {
+    DYAD_CACHE_NONE = 0,     ///< Eviction disabled (default).
+    DYAD_CACHE_LRU = 1,      ///< Evict least-recently-accessed file first (by st_atime).
+    DYAD_CACHE_FIFO = 2,     ///< Evict oldest file first (by st_mtime).
+    DYAD_CACHE_DEFAULT = 1,  ///< Default policy once enabled (alias for DYAD_CACHE_LRU).
+    DYAD_CACHE_END = 3       ///< Sentinel — number of valid cache policy modes.
+};
+typedef enum dyad_cache_policy_mode dyad_cache_policy_mode_t;
+
+/**
+ * @brief Human-readable names for each cache-eviction policy mode.
+ *
+ * @details
+ * Indexed by @c dyad_cache_policy_mode_t. The extra entry at index
+ * @c DYAD_CACHE_END holds @c "CACHE_UNKNOWN" for out-of-range values.
+ * Marked @c unused to suppress warnings when included in translation
+ * units that do not reference it directly.
+ */
+static const char *dyad_cache_policy_name[DYAD_CACHE_END + 1]
+    __attribute__ ((unused)) = {"NONE", "LRU", "FIFO", "CACHE_UNKNOWN"};
+
+/**
+ * @brief Opaque cache-eviction policy handle.
+ *
+ * @details
+ * Forward declaration of the policy implementation struct. The full
+ * definition is provided by each policy (LRU, FIFO) in its own source
+ * file. Callers interact with the policy exclusively through the
+ * function pointer embedded in the struct, allowing the policy to be
+ * swapped at initialization time without changing call sites.
+ *
+ * @see dyad_cache_policy_mode_t for available policies.
+ * @see dyad_ctx_t for the context that owns the cache policy handle.
+ */
+struct dyad_cache_policy;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* DYAD_COMMON_DYAD_CACHE_H */

include/dyad/common/dyad_dtl.h

@@ -61,7 +61,7 @@ typedef enum dyad_dtl_mode dyad_dtl_mode_t;
  * Marked @c unused to suppress warnings when included in translation
  * units that do not reference it directly.
  */
-static const char* dyad_dtl_mode_name[DYAD_DTL_END + 1]
+static const char *dyad_dtl_mode_name[DYAD_DTL_END + 1]
     __attribute__ ((unused)) = {"UCX", "MARGO", "FLUX_RPC", "DTL_UNKNOWN"};
 
 /**
@@ -93,6 +93,16 @@ typedef enum dyad_dtl_comm_mode dyad_dtl_comm_mode_t;
  */
 #define DYAD_DTL_RPC_NAME "dyad.fetch"
 
+/**
+ * @brief Flux RPC topic name for DYAD byte-range fetch requests.
+ *
+ * @details
+ * Like @c DYAD_DTL_RPC_NAME, but for @c dyad_consume_range() requests
+ * (FLUX_RPC and MARGO DTL modes only). Registered as a separate topic so
+ * the existing whole-file @c DYAD_DTL_RPC_NAME handler is untouched.
+ */
+#define DYAD_DTL_RPC_RANGE_NAME "dyad.fetch_range"
+
 /**
  * @brief Opaque DTL handle.
  *

include/dyad/common/dyad_envs.h

@@ -132,4 +132,72 @@
  */
 #define DYAD_MARGO_PROTO_ENV "DYAD_MARGO_PROTO"
 
+/**
+ * @brief Maximum bytes DYAD may use in a managed directory before evicting
+ *        cached files.
+ *
+ * @details
+ * @c 0 (the default) disables eviction entirely, preserving prior
+ * behavior for deployments that don't opt in.
+ */
+#define DYAD_CACHE_CAPACITY_ENV "DYAD_CACHE_CAPACITY"
+
+/**
+ * @brief Cache-eviction policy to use once @c DYAD_CACHE_CAPACITY is set.
+ *
+ * @details
+ * Valid values: @c LRU (default), @c FIFO, @c NONE. Ignored if
+ * @c DYAD_CACHE_CAPACITY is @c 0.
+ */
+#define DYAD_CACHE_POLICY_ENV "DYAD_CACHE_POLICY"
+
+/**
+ * @brief Fraction (0..1) of @c DYAD_CACHE_CAPACITY to evict down to once
+ *        eviction triggers.
+ *
+ * @details
+ * Evicting to a low-water mark rather than the exact capacity limit
+ * avoids evicting on every single produce/consume call while usage
+ * hovers at the boundary. Default: @c 0.8.
+ */
+#define DYAD_CACHE_LOW_WATERMARK_ENV "DYAD_CACHE_LOW_WATERMARK"
+
+/**
+ * @brief Skip eviction candidates accessed more recently than this many
+ *        seconds.
+ *
+ * @details
+ * Guards against evicting a file that may still be mid-access by another
+ * process. Default: @c 5.
+ */
+#define DYAD_CACHE_GRACE_PERIOD_ENV "DYAD_CACHE_GRACE_PERIOD"
+
+/**
+ * @brief Fallback source path (e.g. on the parallel file system) used by
+ *        @c dyad_range_cache_ensure() to lazily fill missing spans of a
+ *        DYAD-managed file on demand.
+ *
+ * @details
+ * Unset (the default) preserves prior behavior: managed files must already
+ * be fully present locally (e.g. pre-staged), and byte-range fetch never
+ * consults an origin.
+ */
+#define DYAD_PATH_ORIGIN_ENV "DYAD_PATH_ORIGIN"
+
+/**
+ * @brief Number of worker threads the DYAD Flux module uses to service
+ *        @c dyad.fetch_range RPCs (@c dyad_fetch_range_request_cb()).
+ *
+ * @details
+ * The module's single Flux reactor thread hands each byte-range fetch's
+ * blocking file I/O (and, on the Margo DTL, the RDMA send itself) off to
+ * this worker pool instead of doing it inline, so concurrent requests from
+ * different ranks/nodes don't serialize behind whichever one happened to
+ * arrive first. Read directly via @c getenv() in @c mod_main(), not
+ * through @c dyad_init_env(), since it configures the module's own
+ * internal servicing, not context state shared with client processes.
+ * Default: @c 8.
+ */
+#define DYAD_FETCH_WORKER_THREADS_ENV "DYAD_FETCH_WORKER_THREADS"
+
 #endif  // DYAD_COMMON_DYAD_ENVS_H

include/dyad/common/dyad_rc.h

@@ -66,6 +66,9 @@ enum dyad_core_return_codes {
     DYAD_RC_BAD_CLI_ARG_DEF = -1008,  ///< Trying to define a CLI argument failed
     DYAD_RC_BAD_CLI_PARSE = -1009,    ///< Trying to parse CLI arguments failed
     DYAD_RC_BADBUF = -1010,           ///< Invalid buffer/pointer passed to function
+    DYAD_RC_BUSY = -1011,             ///< Resource (e.g. a file lock) is held by another
+                                      ///< process; not a hard failure, caller should skip/retry
+    DYAD_RC_BADCACHEMODE = -1012,     ///< Invalid DYAD cache-eviction policy mode provided
 
     // FLUX
     DYAD_RC_FLUXFAIL = -2000,      ///< Some Flux function failed