From 7f05ab27f66d93edce0d20c1065f9a15cb324d69 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 03:35:44 +0000 Subject: [PATCH 1/2] fix(python-node): seed the pool-id counter randomly so restarts don't collide MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Memory-pool shared-memory names are "dora_pool_{dataflow_id}_{node_id}_{counter}", where the counter comes from a process-local static seeded at 0. Both dataflow_id and node_id are stable across a crash-restart, so a restarted node re-derives the exact same name as its previous incarnation. When the old pool is still live — the sender crashes while its receiver keeps reading, so #2881's reclaim deliberately retains it — ShmemConf::create() fails on the leftover segment and the daemon also rejects the registration as a duplicate. Under restart_policy: Always this is a crash-restart loop that never recovers. Seed PINNED_COUNTER with a random u64 (via std's OS-seeded RandomState, no new dependency) so each incarnation gets a distinct name. The counter stays a plain u64, so both existing name parsers (the writer fast path and try_doradma_read) keep working unchanged. Increment via wrapping_add to guard the now-possible overflow from a near-max seed. Fixes #3015 Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01KwYrfJpHX6PSwoewwKyA9z --- apis/python/node/src/lib.rs | 66 +++++++++++++++++++++++++++++++++++-- 1 file changed, 63 insertions(+), 3 deletions(-) diff --git a/apis/python/node/src/lib.rs b/apis/python/node/src/lib.rs index eb47631f28..24fe01253b 100644 --- a/apis/python/node/src/lib.rs +++ b/apis/python/node/src/lib.rs @@ -134,8 +134,32 @@ def basicConfig(*pargs, **kwargs): static CUDA_HELPERS: LazyLock>>> = LazyLock::new(|| std::sync::Mutex::new(None)); +/// A random `u64` seed derived from the standard library's `RandomState`, +/// which is seeded from the OS once per process. Used to make process-local +/// counters unique across process restarts without pulling in a new +/// dependency (dora-rs/dora#3015). +fn random_u64_seed() -> u64 { + use std::hash::{BuildHasher, Hasher}; + // `RandomState::new()` picks fresh random keys from the OS; hashing no + // input and finishing yields a value derived from those keys, so the + // result differs from process to process. + std::hash::RandomState::new().build_hasher().finish() +} + /// Counter to make pinned memory buffer IDs unique across registrations. -static PINNED_COUNTER: LazyLock> = LazyLock::new(|| std::sync::Mutex::new(0)); +/// +/// Seeded with a random value per process rather than `0` (dora-rs/dora#3015). +/// The counter is combined with the `(dataflow_id, node_id)` pair — both of +/// which are stable across a crash-restart — into the pool's shared-memory +/// name, so a deterministic `0` seed makes a restarted node re-derive the +/// exact name its previous incarnation used. If the old segment is still live +/// (its receiver is still reading it, so #2881's reclaim deliberately kept it), +/// `ShmemConf::create()` collides and the restarted node can never register its +/// pool — a crash-restart loop that never recovers. A random per-incarnation +/// seed keeps the id unique across restarts while the component stays a plain +/// `u64`, so both existing name parsers keep working unchanged. +static PINNED_COUNTER: LazyLock> = + LazyLock::new(|| std::sync::Mutex::new(random_u64_seed())); /// Tracks freed pool buffer IDs so the DORADMA fast path can detect /// read-after-free. Entries are inserted on free_memory_pool and never @@ -301,6 +325,40 @@ mod pin_tests { } } +#[cfg(test)] +mod pool_id_tests { + use super::*; + + /// dora-rs/dora#3015: the per-process pool counter is now seeded from + /// `random_u64_seed()` so a restarted node does not re-derive its previous + /// incarnation's shared-memory name and collide on a still-live segment. + /// The seed must vary from call to call (a proxy for varying from process + /// to process — `RandomState` reseeds on each `new()`). + #[test] + fn random_seed_is_not_constant() { + let seeds: std::collections::HashSet = (0..8).map(|_| random_u64_seed()).collect(); + assert!( + seeds.len() > 1, + "random_u64_seed() must not return a constant value" + ); + } + + /// A pool name built from a large, random-seeded counter must still be + /// recovered by the reader fast path, which takes the last `_`-separated + /// component as a `u64`. This is what lets #3015's random seed keep the + /// on-wire id format unchanged — including for a `node_id` that itself + /// contains underscores. + #[test] + fn large_counter_round_trips_through_the_name_parser() { + let counter = u64::MAX - 3; + let shmem_name = format!("dora_pool_{}_{}_{}", "dataflow-uuid", "my_node", counter); + let parsed = shmem_name + .rsplit_once('_') + .and_then(|(_, c)| c.parse::().ok()); + assert_eq!(parsed, Some(counter)); + } +} + // --------------------------------------------------------------------------- // GPU transport-path classification — pure decision logic extractable // from CUDA-runtime-embedded code so the full matrix can be exercised in @@ -1969,10 +2027,12 @@ impl Node { ); } - // Generate unique pool counter for this registration + // Generate unique pool counter for this registration. `wrapping_add` + // guards the (astronomically unlikely) overflow now that the counter + // starts from a random seed rather than 0 (dora-rs/dora#3015). let pool_counter = { let mut c = PINNED_COUNTER.lock().unwrap_or_else(|e| e.into_inner()); - *c += 1; + *c = c.wrapping_add(1); *c }; let shmem_name = format!( From 0029b8e64a7eee9bed34e633b8e4e57e0cb107e3 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 11 Aug 2026 13:06:44 +0000 Subject: [PATCH 2/2] refactor(python-node): centralize pool-id format; strengthen and document Address review on #3056: - Extract pool_shmem_name / parse_pool_counter as the single source of truth for the on-wire pool-id format, replacing ~7 duplicated inline format!/rsplit_once+parse sites. The round-trip test now exercises the real production functions (can go RED if the format regresses) and covers node ids containing underscores plus a non-numeric tail. - Document the leak trade-off: with the collision gone, a pathological crash-loop whose receiver never frees now leaks a pool per restart up to the registry cap, instead of failing fast. The complete fix is an owner-death reclaim, tracked as follow-up. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01KwYrfJpHX6PSwoewwKyA9z --- apis/python/node/src/lib.rs | 148 ++++++++++++++++++++++-------------- 1 file changed, 91 insertions(+), 57 deletions(-) diff --git a/apis/python/node/src/lib.rs b/apis/python/node/src/lib.rs index 24fe01253b..1b4ed528f0 100644 --- a/apis/python/node/src/lib.rs +++ b/apis/python/node/src/lib.rs @@ -146,6 +146,34 @@ fn random_u64_seed() -> u64 { std::hash::RandomState::new().build_hasher().finish() } +/// Build a memory pool's shared-memory name from the +/// `(dataflow_id, node_id, counter)` triple. +/// +/// This is the single source of truth for the on-wire pool-id format +/// (`dora_pool_{dataflow_id}_{node_id}_{counter}`); [`parse_pool_counter`] +/// recovers the counter from it. Keeping the two together means a change to +/// the format can only be made in one place (dora-rs/dora#3015). +fn pool_shmem_name( + dataflow_id: impl std::fmt::Display, + node_id: impl std::fmt::Display, + counter: u64, +) -> String { + format!("dora_pool_{dataflow_id}_{node_id}_{counter}") +} + +/// Recover the pool counter from a pool id of the form +/// `dora_pool_{dataflow_id}_{node_id}_{counter}` or `pool_{node_id}_{counter}`. +/// +/// The counter is always the final `_`-separated component — node ids may +/// themselves contain `_` (legal in dora node ids) — so the last segment is +/// taken and parsed as a `u64`. Returns `None` when the last segment is not a +/// `u64`. Inverse of [`pool_shmem_name`]'s counter component. +fn parse_pool_counter(pool_id: &str) -> Option { + pool_id + .rsplit_once('_') + .and_then(|(_, counter)| counter.parse::().ok()) +} + /// Counter to make pinned memory buffer IDs unique across registrations. /// /// Seeded with a random value per process rather than `0` (dora-rs/dora#3015). @@ -158,6 +186,15 @@ fn random_u64_seed() -> u64 { /// pool — a crash-restart loop that never recovers. A random per-incarnation /// seed keeps the id unique across restarts while the component stays a plain /// `u64`, so both existing name parsers keep working unchanged. +/// +/// Trade-off (dora-rs/dora#3015 review): with the collision gone, a repeatedly +/// crashing node whose receiver never calls `free_memory_pool` now leaks the +/// previous incarnation's registry entry and `/dev/shm` segment on each +/// restart, instead of failing fast on the first. `cleanup_orphans` only runs +/// at dataflow spawn, and the daemon registry is capped, so a pathological +/// crash loop can eventually exhaust it. The complete fix is an owner-death +/// pool reclaim (tracked as follow-up); this change trades fail-fast for +/// recover-on-transient-crash, which is the common case. static PINNED_COUNTER: LazyLock> = LazyLock::new(|| std::sync::Mutex::new(random_u64_seed())); @@ -343,19 +380,41 @@ mod pool_id_tests { ); } - /// A pool name built from a large, random-seeded counter must still be - /// recovered by the reader fast path, which takes the last `_`-separated - /// component as a `u64`. This is what lets #3015's random seed keep the - /// on-wire id format unchanged — including for a `node_id` that itself - /// contains underscores. + /// The on-wire pool-id format is a contract shared with the daemon + /// (`MemoryPoolManager` derives the same name). Pin the exact literal so a + /// change to `pool_shmem_name` that both in-process halves would still + /// agree on, yet breaks that cross-process contract, is caught here. + #[test] + fn pool_shmem_name_has_the_expected_wire_format() { + assert_eq!( + pool_shmem_name("dataflow-uuid", "cam", 7), + "dora_pool_dataflow-uuid_cam_7" + ); + } + + /// A pool name built from a large, random-seeded counter must round-trip + /// back through the *production* parser — including for a `node_id` that + /// itself contains underscores, where only taking the last segment as the + /// counter is correct. Exercising the real `pool_shmem_name` / + /// `parse_pool_counter` pair means a regression in either can turn this + /// RED (dora-rs/dora#3015). + #[test] + fn large_counter_round_trips_through_the_production_parser() { + for (node_id, counter) in [("my_node", u64::MAX - 3), ("cam_left", 1), ("plain", 0)] { + let shmem_name = pool_shmem_name("dataflow-uuid", node_id, counter); + assert_eq!( + parse_pool_counter(&shmem_name), + Some(counter), + "counter must round-trip for node id {node_id:?}" + ); + } + } + + /// A pool id whose final segment is not a `u64` yields `None` rather than a + /// wrong counter — the fall-back-to-daemon path relies on this. #[test] - fn large_counter_round_trips_through_the_name_parser() { - let counter = u64::MAX - 3; - let shmem_name = format!("dora_pool_{}_{}_{}", "dataflow-uuid", "my_node", counter); - let parsed = shmem_name - .rsplit_once('_') - .and_then(|(_, c)| c.parse::().ok()); - assert_eq!(parsed, Some(counter)); + fn parse_pool_counter_rejects_a_non_numeric_tail() { + assert_eq!(parse_pool_counter("dora_pool_df_node_notanumber"), None); } } @@ -2035,10 +2094,7 @@ impl Node { *c = c.wrapping_add(1); *c }; - let shmem_name = format!( - "dora_pool_{}_{}_{}", - self.dataflow_id, self.node_id, pool_counter - ); + let shmem_name = pool_shmem_name(&self.dataflow_id, &self.node_id, pool_counter); let header_meta = PyDict::new(py); header_meta.set_item("size", size)?; @@ -2535,9 +2591,7 @@ impl Node { if buffer_id.starts_with("pool_") { // Extract counter from the last underscore segment — node_id // may legitimately contain underscores. - if let Some((_, counter_str)) = buffer_id.rsplit_once('_') - && let Ok(counter) = counter_str.parse::() - { + if let Some(counter) = parse_pool_counter(&buffer_id) { // Try PINNED_POOL cache first to avoid per-iteration mmap/munmap. // register_memory_pool already stored the Shmem here; taking it // prevents munmap, and storing it back keeps the mapping alive. @@ -2563,10 +2617,7 @@ impl Node { } else { // Cache miss: open via ShmemConf, wrap immediately // so the mapping stays alive until post-write re-insert. - let shmem_name = format!( - "dora_pool_{}_{}_{}", - self.dataflow_id, self.node_id, counter - ); + let shmem_name = pool_shmem_name(&self.dataflow_id, &self.node_id, counter); match ShmemConf::new().os_id(&shmem_name).open() { Ok(shmem) => { let cap = shmem.len(); @@ -2890,9 +2941,7 @@ impl Node { if ipc_present == 1 && !is_cuda { // Extract counter for the DMA slot from buffer_id. - let slow_counter = buffer_id - .rsplit_once('_') - .and_then(|(_, c)| c.parse::().ok()); + let slow_counter = parse_pool_counter(&buffer_id); let gen_ptr = unsafe { shmem_ptr.add(96) as *mut u64 }; let pre_write_gen = unsafe { seqlock_begin_if_even(gen_ptr) }; let mut copy_ok = true; @@ -2937,19 +2986,17 @@ impl Node { let bound = helpers.bind(py); // Slow path transit look-up: PINNED_POOL // (contrast fast path which uses store_back). - let (transit_ptr, pool_device) = if let Some((_, counter_str)) = - buffer_id.rsplit_once('_') - && let Ok(c) = counter_str.parse::() - { - PINNED_POOL - .lock() - .unwrap_or_else(|e| e.into_inner()) - .get(&c) - .map(|s| (s.transit_ptr, s.pool_device)) - .unwrap_or((0, 0)) - } else { - (0, 0) - }; + let (transit_ptr, pool_device) = + if let Some(c) = parse_pool_counter(&buffer_id) { + PINNED_POOL + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(&c) + .map(|s| (s.transit_ptr, s.pool_device)) + .unwrap_or((0, 0)) + } else { + (0, 0) + }; let write_path = classify_write_path( ipc_present, /*is_cuda=*/ true, @@ -2966,10 +3013,7 @@ impl Node { .call_method1( "_transit_copy_gpu_buf", ( - buffer_id - .rsplit_once('_') - .and_then(|(_, cs)| cs.parse::().ok()) - .unwrap_or(0), + parse_pool_counter(&buffer_id).unwrap_or(0), ptr_val, sender_dev, transit_ptr, @@ -2983,10 +3027,7 @@ impl Node { .call_method1( "_cuda_memcpy_gpu_buf", ( - buffer_id - .rsplit_once('_') - .and_then(|(_, cs)| cs.parse::().ok()) - .unwrap_or(0), + parse_pool_counter(&buffer_id).unwrap_or(0), ptr_val, size, ), @@ -3622,12 +3663,8 @@ impl Node { // Format: "pool_{node_id}_{counter}". // Use rsplit to extract the counter from the end — the node_id // portion may itself contain underscores (legal in dora node ids). - let counter: u64 = match buffer_id.rsplit_once('_') { - Some((_, c)) => match c.parse() { - Ok(c) => c, - Err(_) => return Ok(None), - }, - None => return Ok(None), + let Some(counter) = parse_pool_counter(buffer_id) else { + return Ok(None); }; let pool_node_id = buffer_id .strip_prefix("pool_") @@ -3647,10 +3684,7 @@ impl Node { } } - let shmem_name = format!( - "dora_pool_{}_{}_{}", - self.dataflow_id, pool_node_id, counter - ); + let shmem_name = pool_shmem_name(&self.dataflow_id, pool_node_id, counter); // Open shared memory let shmem = match ShmemConf::new().os_id(&shmem_name).open() {