Refactor: drop slot ring, make DistTaskSlotState storage dynamic

[ChaoWao]

Summary

Internal cleanup PR, follow-up to #560. Replaces the fixed-size slot pool (DIST_TASK_WINDOW_SIZE = 128 + DistTaskSlotState[] inside DistWorker) with dynamic storage owned by DistRing. Slot state lives entirely in the parent process's heap at L3 — Orchestrator and Scheduler read it directly; child workers only receive payloads through a mailbox, never the slot state — so the shmem-backed ring index L2 uses is unnecessary at this level.

Only the heap still needs a pre-sized region (MAP_SHARED | MAP_ANONYMOUS inherited across fork). Heap back-pressure behaviour is unchanged.

No user-visible change: heap_ring_size=… still works, OUTPUT auto-alloc / INOUT WaW semantics unchanged, back-pressure timeout still throws std::runtime_error.

What changed

• DistRing owns three correlated per-task resources now:
  • a monotonic int32_t task id (no window, no modulo wrap),
  • the shared-memory heap slab (unchanged from Refactor: back orch.alloc with merged slot+heap DistRing + fork hygiene #560),
  • the per-slot state as std::deque<std::unique_ptr<DistTaskSlotState>>. std::deque::push_back never invalidates existing pointers, so ring.slot_state(id) hands out a pointer that stays valid for the slot's lifetime without keeping the mutex held past the lookup.
• init(heap_bytes, timeout_ms) drops the window_size parameter; the slot ring is gone entirely.
• reset_to_empty(): new method called by DistOrchestrator::drain() right after active_tasks_ hits 0. Drops all slot states and zeroes task-id / heap cursors so each Worker.run() starts from task_id = 0 with bounded memory (per-run peak, not cumulative).
• DistOrchestrator::init drops slots / num_slots. slot_state(id) delegates to ring.slot_state(id) with a nullptr-check.
• DistScheduler::Config drops slots / num_slots, takes DistRing *ring instead. Every cfg_.slots[id] access becomes *cfg_.ring->slot_state(id).
• DistWorker drops the std::unique_ptr<DistTaskSlotState[]> slots_ member; slot state lives inside the allocator now. DistWorker::init() is a straight pass-through.
• DIST_TASK_WINDOW_SIZE constant removed from dist_types.h.

Tests

• test_dist_ring rewritten:
  • drops window-size test cases
  • adds SlotAllocGrowsPastLegacyWindow — 2048 allocs past the old 128 cap, verifies no "window full" error
  • adds SlotStateIsPointerStable — grabs a ptr to slot 0, allocs 1000 more slots, verifies ptr identity
  • adds ResetToEmptyRequiresAllReleased + ResetToEmptyResetsCounters
  • BackPressureThenReleaseUnblocks / TimeoutThrowsRuntimeError / ShutdownUnblocksAlloc now exercise heap back-pressure directly (the only source of back-pressure remaining)
• test_dist_orchestrator / test_dist_scheduler fixtures drop the std::unique_ptr<DistTaskSlotState[]> slots member and access slot state via a short S(id) fixture helper that wraps ring.slot_state(id).

Test plan

• cpput: 7/7 targets pass locally (cmake --build /tmp/dist_ut_build && ctest)
• pyut: 100 passed, 3 deselected (torch-only; dev box has no torch). test_alloc_dep_wires_via_tensormap and friends unchanged — same INOUT-driven WaW path.
• Linux CI

Plan reference

Follow-up of #560 (PR-H). Tracked as PR-I in the hierarchical-runtime plan under "Follow-up PR chain" / Allowed Exception #6 in the L2 Consistency Audit.

[ChaoWao]

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors the task management system to use monotonic task IDs and dynamic slot state storage within DistRing, removing the fixed-size window limitation and allowing memory reclamation via a new reset_to_empty method. Feedback identifies a critical exception safety issue in DistRing::alloc where partial state updates could leave the allocator inconsistent, and suggests documentation improvements for type clarity in std::deque declarations.

[gemini-code-assist]

The sequence of state updates to released_, slot_heap_end_, and slot_states_ is not exception-safe. If a push_back or emplace_back call throws an exception (e.g., std::bad_alloc), the DistRing will be left in an inconsistent state. To ensure strong exception safety, the state modifications should be transactional, rolling back partial updates if a subsequent update fails. Additionally, ensure consistent type casting when assigning to int32_t to avoid narrowing conversion warnings.

    int32_t task_id = static_cast<int32_t>(next_task_id_);
    released_.push_back(0);
    try {
        slot_heap_end_.push_back(heap_end);
        try {
            slot_states_.emplace_back(std::make_unique<DistTaskSlotState>());
        } catch (...) {
            slot_heap_end_.pop_back();
            throw;
        }
    } catch (...) {
        released_.pop_back();
        throw;
    }
    next_task_id_++;
    return DistAllocResult{task_id, heap_ptr, heap_end};

References

1. Always check the return values of resource allocation functions and handle potential failures gracefully, for example, by rolling back partial allocations.
2. Be consistent in type casting when assigning size_t values to int32_t variables across similar methods to avoid compiler warnings about narrowing conversions.

[gemini-code-assist]

For clarity and completeness in the documentation, it's better to specify the full type in the std::deque declaration instead of using ....

Suggested change

	`std::deque<std::unique_ptr<...>>`. `std::deque::push_back` never
	std::deque<std::unique_ptr<DistTaskSlotState>>. std::deque::push_back never

[gemini-code-assist]

For improved clarity in the documentation, it would be better to use the full type name instead of ... in the std::deque declaration.

Suggested change

	* `std::deque<std::unique_ptr<...>>` so push_back never invalidates
	* 3. The per-task scheduling state (DistTaskSlotState). Stored in a
	* std::deque<std::unique_ptr<DistTaskSlotState>> so push_back never invalidates

[gemini-code-assist]

Code Review

This pull request refactors the distributed task management system by replacing the fixed-size task slot ring with a monotonic task ID system and dynamic state storage in DistRing. This change removes the legacy window size constraint and provides stable pointers to task states. However, a critical race condition was identified in DistOrchestrator::drain(), where reset_to_empty() is called before the scheduler thread has fully finished accessing task state, potentially leading to a use-after-free.

[gemini-code-assist]

There appears to be a critical race condition here. The drain() function, running on the user's thread, calls allocator_->reset_to_empty() as soon as active_tasks_ becomes zero. However, the scheduler thread might still be processing tasks and accessing slot state data.

Specifically, active_tasks_ is decremented in on_consumed, which is called from the scheduler thread. The scheduler thread might continue executing code within on_task_complete (e.g., accessing consumers of a completed task) after active_tasks_ has reached zero and unblocked drain().

When reset_to_empty() is called, it deallocates all DistTaskSlotState objects. If the scheduler thread is concurrently trying to access one of these objects via a pointer obtained from ring->slot_state(), it will result in a use-after-free.

To fix this, you need to ensure the scheduler is quiescent (i.e., not in the middle of processing any task logic) before reset_to_empty() is called. This might involve more sophisticated synchronization between the orchestrator's drain() and the scheduler thread.

References

1. If a component is accessed by multiple threads (e.g., Orchestrator and Scheduler), it requires a mutex to protect its operations to prevent data races.
2. A producer should only be skipped or resources released if it is in the CONSUMED state, not just COMPLETED, to prevent use-after-free issues.