broadcast results concurrently fixes #31

47163d22db1cdeb12e00213ed99542f9eb58c298 · 2024-07-18 16:09:08 -0700 · Alex Cheema

Files touched

M exo/orchestration/standard_node.py

Diff

commit 47163d22db1cdeb12e00213ed99542f9eb58c298
Author: Alex Cheema <alexcheema123@gmail.com>
Date:   Thu Jul 18 16:09:08 2024 -0700

    broadcast results concurrently fixes #31
---
 exo/orchestration/standard_node.py | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/exo/orchestration/standard_node.py b/exo/orchestration/standard_node.py
index 6e25829a..cf2f7e66 100644
--- a/exo/orchestration/standard_node.py
+++ b/exo/orchestration/standard_node.py
@@ -240,10 +240,14 @@ class StandardNode(Node):
         self.on_token.trigger_all(request_id, tokens, is_finished)
 
     async def broadcast_result(self, request_id: str, result: List[int], is_finished: bool) -> None:
-        for peer in self.peers:
+        async def send_result_to_peer(peer):
             try:
-                await peer.send_result(request_id, result, is_finished)
+                await asyncio.wait_for(peer.send_result(request_id, result, is_finished), timeout=15.0)
+            except asyncio.TimeoutError:
+                print(f"Timeout broadcasting result to {peer.id()}")
             except Exception as e:
+                print(f"Error broadcasting result to {peer.id()}: {e}")
                 import traceback
                 traceback.print_exc()
-                print(f"Error broadcasting result to {peer.id()}: {e}")
\ No newline at end of file
+
+        await asyncio.gather(*[send_result_to_peer(peer) for peer in self.peers], return_exceptions=True)
\ No newline at end of file

← 621a5f5d Add license badge · back to Exo · allow overriding inference_engine and separate flag for TINY 945f90f6 →