Add cache eviction, byte-range fetch, and worker-pool concurrency#183
Merged
Conversation
Signed-off-by: Chen Wang <wangvsa@gmail.com>
Adds two related opt-in features for training-style workloads with a working set larger than any single node's local storage: - A pluggable cache-eviction subsystem (LRU/FIFO) for DYAD-managed directories, so node-local storage doesn't grow unbounded (DYAD_CACHE_CAPACITY/_POLICY/_LOW_WATERMARK/_GRACE_PERIOD). - dyad_consume_range(): fetches a byte range of a file directly into memory (no local file materialization), with an optional PFS/origin fallback (DYAD_PATH_ORIGIN) that lazily fills only the requested span of a managed file on first touch via a new on-disk range cache (range_cache.c) tracking which byte spans are already present. Implemented for the FLUX_RPC and Margo DTL backends. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
DYAD's Flux broker module runs a single reactor thread, so every dyad.fetch_range request -- including its blocking file I/O and, for Margo, the RDMA send itself -- was serviced fully synchronously on that one thread. Concurrent requests from different ranks/nodes queued behind whichever one happened to arrive first, capping fetch throughput at 1/(mean per-request time) regardless of how many requests were actually independent. Adds a fixed-size worker-thread pool (DYAD_FETCH_WORKER_THREADS, default 8) that the reactor hands blocking work off to via a new detach/send-detached DTL vtable extension (rpc_detach_request / rpc_send_detached / rpc_abort_detached), implemented for the FLUX_RPC and Margo backends (UCX keeps the original synchronous path via a NULL rpc_detach_request). Completion is handed back to the reactor thread through a self-pipe + fd-watcher for backends that must not touch Flux state off the reactor thread. Two correctness hazards from adding concurrency, both fixed: - fcntl() locks only exclude other *processes*, not threads of the same process -- added a module-local pthread_mutex_t in range_cache.c alongside the existing fcntl locks. - Both backends stored per-request state (a message reference / resolved RDMA address) in a single shared slot, safe only under strictly serial servicing -- replaced with the detach/send-detached handoff so each in-flight request owns independent state. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Two bugs surfaced under sustained, full-scale byte-range fetch load (not visible in short test runs): - dyad_dtl_margo_send_to() and data_ready_rpc() both registered a bulk handle via margo_bulk_create() but only freed it on error paths, never on success -- a per-request RDMA memory-registration leak that exhausts the CXI NIC's finite MR table (fi_mr_enable() ENOSPC) after enough concurrent requests. - Every fetch resolved the consumer's Margo address via margo_addr_lookup() and freed it again after use, even though the same consumer is looked up repeatedly over a job's lifetime. Churning through resolve/free this often was found to exhaust or corrupt state in Mercury's CXI provider under sustained request volume, surfacing as indefinite address-lookup failures on addresses that had resolved successfully many times before. Added a per-consumer address cache (populated once, reused for the life of the DTL handle) matching the pattern the UCX backend already uses (ucx_ep_cache.cpp). Also checks previously-ignored return codes from margo_addr_lookup(), margo_addr_self(), and margo_addr_to_string() instead of silently proceeding on failure. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Dyad.finalize() guards against a second call by checking self.initialized, but never reset it back to False after finalizing -- so the guard never actually fired. A second call (e.g. an explicit finalize() followed by __del__() at interpreter exit) re-invoked dyad_finalize() on already-freed C-side state, more likely to crash at higher rank counts. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Signed-off-by: Chen Wang <wangvsa@gmail.com>
Signed-off-by: Chen Wang <wangvsa@gmail.com>
Signed-off-by: Chen Wang <wangvsa@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds opt-in node-local cache eviction and origin-backed byte-range fetch to
DYAD, makes the Flux module's fetch-RPC path concurrent (fixing a real
single-reactor-thread serialization bottleneck), and fixes two correctness
bugs found while hardening the Margo DTL backend and pydyad under load.
dyad_consume_range()):pluggable LRU/FIFO eviction for DYAD-managed directories
(
DYAD_CACHE_CAPACITY/_POLICY/_LOW_WATERMARK/_GRACE_PERIOD), plus abyte-range consume API that fetches directly into memory with an optional
PFS/origin fallback (
DYAD_PATH_ORIGIN) that lazily fills only therequested span of a managed file on first touch. Implemented for the
FLUX_RPC and Margo DTL backends.
thread previously serviced every
dyad.fetch_rangerequest fullysynchronously — including blocking file I/O and, for Margo, the RDMA send
itself — so concurrent requests from independent ranks/nodes queued
behind whichever one happened to arrive first. Adds a fixed-size
worker-thread pool (
DYAD_FETCH_WORKER_THREADS, default 8) via a newDTL vtable detach/send-detached extension, implemented for FLUX_RPC and
Margo (UCX keeps the original synchronous path).
(
margo_bulk_create()never freed on the success path, exhausting theCXI NIC's finite MR table under sustained load) and a per-consumer
address cache (repeated
margo_addr_lookup()/margo_addr_free()churnwas found to exhaust/corrupt Mercury's CXI provider state under
sustained request volume).
Dyad.finalize()never resetself.initialized, so itsguard against double-finalization never actually fired — a second call
(e.g. an explicit
finalize()followed by__del__()at interpreterexit) could call
dyad_finalize()twice on already-freed C-side state.Test plan
splitting into this commit sequence)
tests/pydyad_spsc_range/tests/pydyad_spsc_lazy_rangebyte-rangecorrectness (bytes match origin) under both
DYAD_DTL_MODE=FLUX_RPCand
DYAD_DTL_MODE=MARGOscale, both DTL backends, including 1000+-iteration runs with the
worker pool and Margo fixes active — zero errors
(
tests/unit/cache/cache_functions.cpp) — included, not yet run inCI here
🤖 Generated with Claude Code