fix(event-plane): stop retired ZMQ subscriber pumps#11730
Conversation
Signed-off-by: zhongdaor <zhongdaor@nvidia.com>
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Enterprise Run ID: 📒 Files selected for processing (2)
WalkthroughChangesEvent-plane lifecycle handling
Estimated code review effort: 3 (Moderate) | ~20 minutes 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
Comment |
![dynamo-review-agent[bot]](https://avatars.githubusercontent.com/u/201626793?s=60&v=4){kind=small}
| // Pass cancellation through so the discovery backend can stop any | ||
| // task that it owns in addition to the consumer loop below. | ||
| let mut watch_stream = match discovery | ||
| .list_and_watch(query.clone(), Some(cancel_token.clone())) |
There was a problem hiding this comment.
Passing DynamicSubscriber's standalone token into KVStoreDiscovery replaces the backend's runtime-primary cancellation token, so runtime shutdown no longer stops the underlying KV watch task while the subscriber stream is still retained. Fix: pass a token linked to the runtime shutdown token, not an independent token.
🤖 AI Fix
Modify lib/runtime/src/transports/event_plane/dynamic_subscriber.rs DynamicSubscriber to add with_cancel_token(discovery: Arc<dyn Discovery>, query: DiscoveryQuery, topic: String, cancel_token: CancellationToken) -> Self, have new call it with CancellationToken::new(), and update lib/runtime/src/transports/event_plane/mod.rs EventSubscriber::new_internal direct ZMQ branch to call DynamicSubscriber::with_cancel_token(discovery, query, topic.clone(), drt.primary_token().child_token()).
|
Signed-off-by: zhongdaor <zhongdaor@nvidia.com>
2 participants
Summary
DynamicSubscribercancellation interrupt an idle discovery watch and propagate cancellation to backend-owned watch tasks.The root cause was an ordinary Tokio
JoinHandle: droppingZmqSubTransportdetached the pump while the pump continued to own the SUB socket. After the event plane moved to one shared ZMQ context, retained sockets accumulated in that finite context and eventually surfaced asEMFILE/ "Too many open files".Fixes DIS-2415.
Validation
cargo fmt --all -- --checkcargo clippy -p dynamo-runtime --lib --tests -- -D warningscargo test -p dynamo-runtime transports::event_plane --lib(25 passed)