← back to Exo
fix: replace Master event_sender after EventRouter recreation (#1630) (#1637)
858dc808dffd940bd1412d119b3743b70bdfc309 · 2026-03-02 18:05:47 +0800 · wysie
## Motivation
Fix for https://github.com/exo-explore/exo/issues/1630.
When loading a model that spans 2 nodes as the first action after
startup, Exo crashes with anyio.BrokenResourceError in
Master._command_processor. The receive end of the Master's event_sender
channel is already closed (open_receive_channels=0 ) when the Master
tries to send an InstanceCreated event.
<!-- Why is this change needed? What problem does it solve? -->
<!-- If it fixes an open issue, please link to the issue here -->
## Changes
In _elect_loop, when is_new_master is True and the EventRouter is
replaced, the Master's event_sender is now swapped in-place with a fresh
sender from the new EventRouter. This mirrors the existing pattern where
the Worker and DownloadCoordinator are already recreated with new
channels from the new router.
<!-- Describe what you changed in detail -->
## Why It Works
When is_new_master fires (which happens on every startup when a second
node joins), the old EventRouter is shut down and a new one is created.
Shutting down the old router cancels its _ingest coroutine, which closes
the receive end of every channel created by event_router.sender(). The
Worker and DownloadCoordinator are already recreated with senders from
the new router, but the Master was not — its event_sender still pointed
to the dead channel from the old router.
This patch replaces self.master.event_sender with a new sender from the
new EventRouter, so events flow through a live channel. We swap in-place
rather than recreating the Master to preserve its DiskEventLog, internal
state, and running tasks.
The reason loading a single-node model first appeared to work around the
issue is a timing race: if the model loads fast enough, the Master sends
its events before the election fires and kills the old channel. With a
2-node model, placement and RDMA setup take longer, so the election
completes first and the channel is already dead by the time the Master
tries to send.
<!-- Explain why your approach solves the problem -->
## Test Plan
### Manual Testing
<!-- Hardware: (e.g., MacBook Pro M1 Max 32GB, Mac Mini M2 16GB,
connected via Thunderbolt 4) -->
2 x Mac Studio M4 Max 128GB, RDMA
<!-- What you did: -->
Loaded a model that requires 2 nodes and it works properly now.
<!-- - -->
### Automated Testing
<!-- Describe changes to automated tests, or how existing tests cover
this change -->
<!-- - -->
Co-authored-by: Wysie <wysie@users.noreply.github.com>
Files touched
Diff
commit 858dc808dffd940bd1412d119b3743b70bdfc309
Author: wysie <sohyuanchin@gmail.com>
Date: Mon Mar 2 18:05:47 2026 +0800
fix: replace Master event_sender after EventRouter recreation (#1630) (#1637)
## Motivation
Fix for https://github.com/exo-explore/exo/issues/1630.
When loading a model that spans 2 nodes as the first action after
startup, Exo crashes with anyio.BrokenResourceError in
Master._command_processor. The receive end of the Master's event_sender
channel is already closed (open_receive_channels=0 ) when the Master
tries to send an InstanceCreated event.
<!-- Why is this change needed? What problem does it solve? -->
<!-- If it fixes an open issue, please link to the issue here -->
## Changes
In _elect_loop, when is_new_master is True and the EventRouter is
replaced, the Master's event_sender is now swapped in-place with a fresh
sender from the new EventRouter. This mirrors the existing pattern where
the Worker and DownloadCoordinator are already recreated with new
channels from the new router.
<!-- Describe what you changed in detail -->
## Why It Works
When is_new_master fires (which happens on every startup when a second
node joins), the old EventRouter is shut down and a new one is created.
Shutting down the old router cancels its _ingest coroutine, which closes
the receive end of every channel created by event_router.sender(). The
Worker and DownloadCoordinator are already recreated with senders from
the new router, but the Master was not — its event_sender still pointed
to the dead channel from the old router.
This patch replaces self.master.event_sender with a new sender from the
new EventRouter, so events flow through a live channel. We swap in-place
rather than recreating the Master to preserve its DiskEventLog, internal
state, and running tasks.
The reason loading a single-node model first appeared to work around the
issue is a timing race: if the model loads fast enough, the Master sends
its events before the election fires and kills the old channel. With a
2-node model, placement and RDMA setup take longer, so the election
completes first and the channel is already dead by the time the Master
tries to send.
<!-- Explain why your approach solves the problem -->
## Test Plan
### Manual Testing
<!-- Hardware: (e.g., MacBook Pro M1 Max 32GB, Mac Mini M2 16GB,
connected via Thunderbolt 4) -->
2 x Mac Studio M4 Max 128GB, RDMA
<!-- What you did: -->
Loaded a model that requires 2 nodes and it works properly now.
<!-- - -->
### Automated Testing
<!-- Describe changes to automated tests, or how existing tests cover
this change -->
<!-- - -->
Co-authored-by: Wysie <wysie@users.noreply.github.com>
---
src/exo/main.py | 20 +++++++++++---------
1 file changed, 11 insertions(+), 9 deletions(-)
diff --git a/src/exo/main.py b/src/exo/main.py
index 166bf2b4..3b19dd3b 100644
--- a/src/exo/main.py
+++ b/src/exo/main.py
@@ -176,6 +176,17 @@ class Node:
# - Shutdown and re-create the worker
# - Shut down and re-create the API
+ if result.is_new_master:
+ await anyio.sleep(0)
+ self.event_router.shutdown()
+ self.event_router = EventRouter(
+ result.session_id,
+ self.router.sender(topics.COMMANDS),
+ self.router.receiver(topics.GLOBAL_EVENTS),
+ self.router.sender(topics.LOCAL_EVENTS),
+ )
+ self._tg.start_soon(self.event_router.run)
+
if (
result.session_id.master_node_id == self.node_id
and self.master is not None
@@ -212,15 +223,6 @@ class Node:
f"Node {result.session_id.master_node_id} elected master"
)
if result.is_new_master:
- await anyio.sleep(0)
- self.event_router.shutdown()
- self.event_router = EventRouter(
- result.session_id,
- self.router.sender(topics.COMMANDS),
- self.router.receiver(topics.GLOBAL_EVENTS),
- self.router.sender(topics.LOCAL_EVENTS),
- )
- self._tg.start_soon(self.event_router.run)
if self.download_coordinator:
self.download_coordinator.shutdown()
self.download_coordinator = DownloadCoordinator(
← 635118ef Support trace deletion in dashboard (#1628)
·
back to Exo
·
feat: add POST /v1/cancel/{command_id} endpoint (#1579) f0d4ccbe →