Control card routing for managing subscription for decalarative tasks#2817
Open
mustafab0 wants to merge 5 commits into
Open
Control card routing for managing subscription for decalarative tasks#2817mustafab0 wants to merge 5 commits into
mustafab0 wants to merge 5 commits into
Conversation
Pins the wire-visible routing semantics of all five coordinator input streams (joint_command position/velocity precedence, frame_id-addressed cartesian and EEF twist, teleop buttons, the twist->virtual-joint BASE mapping and set_velocity_command fan-out, and subscription lifecycle) ahead of the card-routing refactor. Green against the pre-refactor coordinator; must stay green unchanged after it.
…ards The coordinator no longer knows task types by name for four of its five input streams. Tasks registered with a task_type get the stream routes their card declares (claim_overlap / by_task_name / broadcast), one generic _dispatch replaces the four per-stream handlers, and streams subscribe when their route table is non-empty. eef_twist gets its card and coordinator_ee_twist_command becomes card-routable. twist_command deliberately keeps its legacy path (BASE virtual-joint mapping + set_velocity_command fan-out, same subscription gate); only its internal joint fan-in now goes through _dispatch. A follow-up PR migrates it. Intentional deltas: teleop buttons narrow from all-tasks to card-bound tasks (only teleop_ik overrides on_buttons in-tree), and a task added via bare add_task() no longer receives stream routing — pass task_type to bind its card. Per-task handler delivery is now guarded by try/except instead of unwinding the dispatch loop. Characterization tests from the previous commit pass unchanged.
mustafab0
requested review from
leshy,
paul-nechifor and
spomichter
as code owners
Codecov Report❌ Patch coverage is
@@ Coverage Diff @@
## mustafa/control-coordinator-refactor-main #2817 +/- ##
============================================================================
Coverage ? 72.41%
============================================================================
Files ? 946
Lines ? 84841
Branches ? 7694
============================================================================
Hits ? 61434
Misses ? 21341
Partials ? 2066
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
Contributor
Greptile SummaryThis PR wires up the declarative card-routing system (introduced in the prior refactor) so the coordinator actually uses it. Per-stream subscriptions for
Confidence Score: 5/5The change is safe to merge. All three imperative subscription paths are cleanly replaced by the declarative route table, subscription lifecycle is correctly implemented, and handler-exception isolation prevents one failing task from blocking others. The refactor ships a thorough characterization test suite covering normal delivery, unmatched routing, subscription lifecycle, error isolation, and runtime card registration. No regressions in existing behavior were identified. No files require special attention. The two observations in coordinator.py are non-blocking: the stale snapshot race has a harmless worst-case outcome, and the BROADCAST+BY_TASK_NAME warning edge case cannot be triggered by any currently registered card. Important Files Changed
Flowchart%%{init: {'theme': 'neutral'}}%%
flowchart TD
A[add_task task, task_type] --> B{task_type provided?}
B -- Yes --> C[_register_routes\nlookup card bindings]
B -- No --> D[task registered\nno stream routing]
C --> E{bindings.consumes\nempty?}
E -- Yes + unknown type --> F[logger.warning\nunknown task_type]
E -- No --> G[populate _routes\nstream to task, handler, routing]
G --> H[_sync_stream_subscriptions]
D --> H
F --> H
H --> I{tick_loop running?}
I -- No --> J[no-op, deferred to start]
I -- Yes --> K{new active streams?}
K -- Yes --> L[subscribe stream]
L --> M[_stream_unsubs stream = unsub]
K -- No lost consumers? --> N[pop and call unsub]
M --> O[Incoming message on stream]
N --> O
O --> P[_dispatch stream, msg]
P --> Q{routing mode}
Q -- CLAIM_OVERLAP --> R{joints overlap?}
R -- Yes --> S[call handler]
R -- No --> T[skip task]
Q -- BY_TASK_NAME --> U{frame_id matches?}
U -- Yes --> V[call handler]
U -- No --> W[skip, warn if none matched]
Q -- BROADCAST --> S
S --> X{exception?}
X -- Yes --> Y[log, continue]
X -- No --> Z[done]
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
A[add_task task, task_type] --> B{task_type provided?}
B -- Yes --> C[_register_routes\nlookup card bindings]
B -- No --> D[task registered\nno stream routing]
C --> E{bindings.consumes\nempty?}
E -- Yes + unknown type --> F[logger.warning\nunknown task_type]
E -- No --> G[populate _routes\nstream to task, handler, routing]
G --> H[_sync_stream_subscriptions]
D --> H
F --> H
H --> I{tick_loop running?}
I -- No --> J[no-op, deferred to start]
I -- Yes --> K{new active streams?}
K -- Yes --> L[subscribe stream]
L --> M[_stream_unsubs stream = unsub]
K -- No lost consumers? --> N[pop and call unsub]
M --> O[Incoming message on stream]
N --> O
O --> P[_dispatch stream, msg]
P --> Q{routing mode}
Q -- CLAIM_OVERLAP --> R{joints overlap?}
R -- Yes --> S[call handler]
R -- No --> T[skip task]
Q -- BY_TASK_NAME --> U{frame_id matches?}
U -- Yes --> V[call handler]
U -- No --> W[skip, warn if none matched]
Q -- BROADCAST --> S
S --> X{exception?}
X -- Yes --> Y[log, continue]
X -- No --> Z[done]
Reviews (2): Last reviewed commit: "style(control): trim added docstrings to..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
…umer removal Addresses review feedback on the card-routing PR: - CLAIM_OVERLAP gate + empty-name early-return were only exercised through real servo/velocity tasks, which self-filter and mask over-delivery. Add a pure-recorder ProbeTask test that pins the dispatcher's own overlap gate. - The per-task try/except in _dispatch had no test; add one where a raising handler must not abort delivery to a sibling on the same stream nor propagate out of the port callback. - remove_task pruned routes but left the subscription live when the last consumer went away. _sync_stream_subscriptions now reconciles both directions (subscribe active streams, drop empty ones) and remove_task calls it; add a test that removing the last consumer unsubscribes the stream. All three gaps were mutation-verified: each new test fails when its target code is neutered.
A task_type with no registered card silently set up no routing, so a typo was indistinguishable from a legitimately card-less type. _register_routes now warns when the type is unknown to the registry entirely (no factory and no card), while staying quiet for real card-less types like trajectory. The config path already fails loudly at registry.create(); this covers the direct add_task(task, task_type=...) RPC path. Tests pin both the warning and the no-false-positive-on-trajectory behavior; both mutation-verified.
github-actions
Bot
added
the
ready-to-merge
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
1 participant
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.
Problem
Adding routing and dispatcher for control coordinator topics
Contributor License Agreement