Refactor: make worker endpoints uniformly progressable - #1728
Conversation
📝 WalkthroughWalkthroughWorker execution now uses explicit task frames and nonblocking progress for local and remote endpoints. WorkerThread tracks active and staged lanes with dispatch identity. Remote transports poll progress replies incrementally. Documentation and tests cover capacity-one execution, lifecycle handling, and shutdown. ChangesUnified Worker Progress Execution
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
Sequence Diagram(s)sequenceDiagram
participant WorkerThread
participant RemoteL3Endpoint
participant RemoteL3SocketTransport
participant RemoteWorker
WorkerThread->>RemoteL3Endpoint: submit_progress(ring, dispatch)
RemoteL3Endpoint->>RemoteL3SocketTransport: submit_progress_frame(frame)
RemoteL3SocketTransport->>RemoteWorker: send progress frame incrementally
WorkerThread->>RemoteL3Endpoint: poll_progress(progress)
RemoteL3Endpoint->>RemoteL3SocketTransport: poll_progress_reply(...)
RemoteL3SocketTransport-->>RemoteL3Endpoint: return completed reply
RemoteL3Endpoint-->>WorkerThread: report task completion
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (8)
src/common/hierarchical/remote_endpoint.cpp (3)
624-624: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueConsider avoiding the per-dispatch frame copy.
progress_write_ = frame;copies the whole encoded task frame on every dispatch. The caller at line 822 builds the frame as a temporary and discards it. Task payloads carry the full argument blob, so this is one allocation plus one copy per task on the dispatch path.Add a
std::vector<uint8_t>&&overload toRemoteL3Transport::submit_progress_frame, or change the parameter to by-value and move intoprogress_write_.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/remote_endpoint.cpp` at line 624, Update RemoteL3Transport::submit_progress_frame to accept the frame by value or add an rvalue-reference overload, then move it into progress_write_ instead of copying. Update the temporary-frame call site near the dispatch path to pass ownership while preserving existing lvalue callers.
873-887: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winAlign exception safety between the two stop paths.
request_progress_stopisnoexceptand wraps its body intry { ... } catch (...) {}.report_progress_errorperforms the same three operations without any guard.
WorkerThread::loop()calls the error reporter from its own exception handlers foractivate_progressandpoll_progress. An exception escapingreport_progress_errorthere propagates out of the worker loop thread. Wrap the body the same way, or state whytransport_->shutdown()cannot throw on this path.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/remote_endpoint.cpp` around lines 873 - 887, Update RemoteL3Endpoint::report_progress_error to match the exception-safety behavior of request_progress_stop: guard the mutex lock, progress state updates, and transport_->shutdown() call with a catch-all so exceptions cannot escape WorkerThread::loop()’s error-handling path.
799-803: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDocument the locking precondition of
finish_progress_command.
finish_progress_commandmutatespending_task_and notifiescommand_cv_without takingcommand_mu_. It requires the caller to hold that mutex.poll_progresssatisfies this today. The precondition is not stated, so a future caller can introduce a data race onpending_task_.Add a comment that states the precondition, or add an assertion in debug builds.
♻️ Proposed change
+// The caller must hold `command_mu_`. This function mutates `pending_task_` +// and notifies `command_cv_` without taking the mutex itself. void RemoteL3Endpoint::finish_progress_command(uint64_t sequence) {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/remote_endpoint.cpp` around lines 799 - 803, Document the locking precondition on RemoteL3Endpoint::finish_progress_command: state that callers must hold command_mu_ before invoking it, since the method updates pending_task_ and notifies command_cv_ without locking. Keep the existing behavior unchanged; poll_progress already satisfies this requirement.tests/ut/cpp/hierarchical/test_remote_endpoint.cpp (2)
387-407: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider asserting the progress failure path.
The test covers only
EndpointOutcome::SUCCESS.FakeRemoteTransportalready exposesnext_error_codeandnext_error_message. Set them and assert thatpoll_progressreportsEndpointOutcome::TASK_FAILUREwith the propagated message. That covers lines 858-863 ofRemoteL3Endpoint::poll_progress.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/ut/cpp/hierarchical/test_remote_endpoint.cpp` around lines 387 - 407, Add a failure-path case to TaskDispatchUsesProgressSubmissionAndPolling using FakeRemoteTransport::next_error_code and next_error_message before submission; then assert poll_progress returns EndpointOutcome::TASK_FAILURE and propagates the configured error message in the completion result, while preserving the existing success assertions.
280-283: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a not-ready mode to the fake transport.
poll_progress_replyalways returnstrue. The endpoint-level "reply not ready" path is therefore never exercised. That path isRemoteL3Endpoint::poll_progressline 852, which returnsfalseand leavespending_task_occupied for a later poll.That path is the core behavior this PR adds. The socket test at lines 620-647 covers the transport in isolation, not the endpoint.
Add a counter or flag that makes the fake return
falsefor the first N calls. Then assert thatpoll_progressreturnsfalse, thatpending_task_survives, and that a later poll completes the same dispatch.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/ut/cpp/hierarchical/test_remote_endpoint.cpp` around lines 280 - 283, Extend the fake transport used by poll_progress_reply with configurable not-ready behavior, returning false for the first N polls before delivering the reply. Add endpoint-level assertions that RemoteL3Endpoint::poll_progress returns false while pending_task_ remains occupied, then verify a later poll completes the same dispatch.tests/ut/cpp/hierarchical/test_scheduler.cpp (1)
1203-1232: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider covering the capacity-one failure path.
This test covers only the success path.
progressable()now returnstruefortask_frame_count=1, so capacity-one local endpoints newly use submit/poll. Add a case that sets the frame state toTASK_FAILEDwith a populated error region. Then assert thatpoll_progressreportsEndpointOutcome::TASK_FAILUREand propagates the error message.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/ut/cpp/hierarchical/test_scheduler.cpp` around lines 1203 - 1232, The test around CapacityOneMailboxUsesTheProgressTaskFrame covers only successful completion; add a capacity-one failure case using the same LocalMailboxEndpoint and progress submission flow. Set the task frame to TASK_FAILED with a populated error region, then assert poll_progress returns COMPLETED with EndpointOutcome::TASK_FAILURE and preserves the error message.src/common/hierarchical/remote_endpoint.h (1)
74-74: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReplace the literal
40with a shared header-size constant.
40is the wire frame-header size. It now appears as a bare literal here and atremote_endpoint.cpplines 626, 628, 671, and 689.remote_endpoint.cppline 539 already declaresstatic constexpr size_t HEADER_BYTES = 40;insideread_frame. A wire-format change must update six independent sites.Promote a single
remote_l3::FRAME_HEADER_BYTESconstant and use it at every site.♻️ Proposed change
- size_t progress_read_size_{40}; + size_t progress_read_size_{remote_l3::FRAME_HEADER_BYTES};🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/remote_endpoint.h` at line 74, Replace the duplicated frame-header size literals by promoting a shared remote_l3::FRAME_HEADER_BYTES constant in the common header, then use it for progress_read_size_ and every corresponding size calculation in read_frame and its related remote_endpoint.cpp sites. Remove the local HEADER_BYTES definition and ensure all six wire-header references use the shared constant.python/simpler/worker.py (1)
1808-1809: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueCheck mailbox size before slicing the task frame.
_run_mailbox_loopslicesbuf[MAILBOX_FRAME_SIZE : 2 * MAILBOX_FRAME_SIZE], butmemoryviewslices clamp when out of range. The C++ endpoint constants and Python shared-memory allocation useMAILBOX_SIZE(control frame plus two task frames), so the current callers have space. Keep the task-frame allocation invariant in C++/Python, or add an explicitlen(buf) >= 2 * MAILBOX_FRAME_SIZEcheck here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/simpler/worker.py` around lines 1808 - 1809, Update _run_mailbox_loop to validate that buf has at least 2 * MAILBOX_FRAME_SIZE bytes before slicing the task frame, while preserving the existing C++/Python mailbox allocation invariant. Ensure undersized buffers are rejected explicitly rather than allowing a clamped memoryview slice to proceed.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/common/hierarchical/remote_endpoint.cpp`:
- Around line 891-894: Update run_control at
src/common/hierarchical/remote_endpoint.cpp:891-894 to include
progress_stop_requested_ in the command_cv_ wait predicate and throw using
progress_stop_reason_ when woken by that state. In
src/common/hierarchical/remote_endpoint.cpp:1115-1120, notify command_cv_ after
setting the stop state. In request_progress_stop and report_progress_error at
src/common/hierarchical/remote_endpoint.cpp:873-887, also notify all waiters
after setting progress_stop_requested_ and progress_stop_reason_.
- Around line 826-829: Update the catch-all unwind block in the relevant run
flow to wrap command_lane_.finish_reply(sequence) in a nested try/catch that
suppresses any exception, then rethrow the original exception unchanged; match
the existing guarded pattern used by run() and run_control.
- Around line 639-679: Update poll_progress_reply to reset
progress_command_active_ on every exception path, including check_health(),
timeout, send/receive failures, peer closure, invalid payloads, and reply
validation errors. Use a scope guard that clears the progress state unless the
call completes successfully; ensure nonblocking return-false paths and the final
successful return mark the operation as not failed.
In `@tests/ut/cpp/hierarchical/test_remote_endpoint.cpp`:
- Around line 638-643: Reduce the retry-loop bound in the completion polling
block around poll_progress_reply from 200 to 100 iterations, preserving the 10
ms sleep and completion condition so the test remains within the transport
deadline while allowing the expected 500 ms reply.
---
Nitpick comments:
In `@python/simpler/worker.py`:
- Around line 1808-1809: Update _run_mailbox_loop to validate that buf has at
least 2 * MAILBOX_FRAME_SIZE bytes before slicing the task frame, while
preserving the existing C++/Python mailbox allocation invariant. Ensure
undersized buffers are rejected explicitly rather than allowing a clamped
memoryview slice to proceed.
In `@src/common/hierarchical/remote_endpoint.cpp`:
- Line 624: Update RemoteL3Transport::submit_progress_frame to accept the frame
by value or add an rvalue-reference overload, then move it into progress_write_
instead of copying. Update the temporary-frame call site near the dispatch path
to pass ownership while preserving existing lvalue callers.
- Around line 873-887: Update RemoteL3Endpoint::report_progress_error to match
the exception-safety behavior of request_progress_stop: guard the mutex lock,
progress state updates, and transport_->shutdown() call with a catch-all so
exceptions cannot escape WorkerThread::loop()’s error-handling path.
- Around line 799-803: Document the locking precondition on
RemoteL3Endpoint::finish_progress_command: state that callers must hold
command_mu_ before invoking it, since the method updates pending_task_ and
notifies command_cv_ without locking. Keep the existing behavior unchanged;
poll_progress already satisfies this requirement.
In `@src/common/hierarchical/remote_endpoint.h`:
- Line 74: Replace the duplicated frame-header size literals by promoting a
shared remote_l3::FRAME_HEADER_BYTES constant in the common header, then use it
for progress_read_size_ and every corresponding size calculation in read_frame
and its related remote_endpoint.cpp sites. Remove the local HEADER_BYTES
definition and ensure all six wire-header references use the shared constant.
In `@tests/ut/cpp/hierarchical/test_remote_endpoint.cpp`:
- Around line 387-407: Add a failure-path case to
TaskDispatchUsesProgressSubmissionAndPolling using
FakeRemoteTransport::next_error_code and next_error_message before submission;
then assert poll_progress returns EndpointOutcome::TASK_FAILURE and propagates
the configured error message in the completion result, while preserving the
existing success assertions.
- Around line 280-283: Extend the fake transport used by poll_progress_reply
with configurable not-ready behavior, returning false for the first N polls
before delivering the reply. Add endpoint-level assertions that
RemoteL3Endpoint::poll_progress returns false while pending_task_ remains
occupied, then verify a later poll completes the same dispatch.
In `@tests/ut/cpp/hierarchical/test_scheduler.cpp`:
- Around line 1203-1232: The test around
CapacityOneMailboxUsesTheProgressTaskFrame covers only successful completion;
add a capacity-one failure case using the same LocalMailboxEndpoint and progress
submission flow. Set the task frame to TASK_FAILED with a populated error
region, then assert poll_progress returns COMPLETED with
EndpointOutcome::TASK_FAILURE and preserves the error message.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: cb539989-1f47-4b80-9898-17aeb0647fd8
📒 Files selected for processing (8)
docs/worker-manager.mdpython/simpler/worker.pysrc/common/hierarchical/remote_endpoint.cppsrc/common/hierarchical/remote_endpoint.hsrc/common/hierarchical/worker_manager.cppsrc/common/hierarchical/worker_manager.htests/ut/cpp/hierarchical/test_remote_endpoint.cpptests/ut/cpp/hierarchical/test_scheduler.cpp
c5b9792 to
0685f5b
Compare
- Replace parallel active and staged identity fields with explicit lane records while keeping capacity accounting independent. - Drive capacity-one local and remote endpoints through submit/poll progress, mapping logical leases onto the available task frames. - Preserve single-frame child-loop compatibility and harden remote progress I/O, stop, error, and control-wait handling. - Cover single-frame, remote, SUB, L4, simulation, and hardware paths.
0685f5b to
4e7c027
Compare
2 participants
Summary
Remote control requests share the ordered command socket, so a control operation now waits for an in-flight remote task to complete or stop. Its worst-case added latency includes the task's remaining runtime, bounded by the progress deadline/stop path.
Testing
pre-commit run --files <changed files>(all hooks passed)ctest --test-dir tests/ut/cpp/build-pr1694-followup -LE requires_hardware --output-on-failure(90/90)pytest tests/ut -m "not requires_hardware" -q(1175 passed, 13 skipped, 14 deselected)test_host_worker.py,test_admission_fence.py, andtest_host_buffer_registration.py(4 passed)task-submit(task_20260806_201128_193519427714)