upd

core/actors/collector/_data.py

@@ -12,6 +12,7 @@ def __init__(self):
         self._producers = []
         self._consumers = []
         self._tasks = set()
+        self._stop_event = asyncio.Event()
 
     async def start(self, msg: Event):
         for producer in self._producers:
@@ -27,6 +28,8 @@ async def start(self, msg: Event):
             task.add_done_callback(self._tasks.discard)
 
     async def stop(self):
+        self._stop_event.set()
+
         await self._queue.put(STOP)
         await self._queue.join()
 
@@ -53,7 +56,7 @@ async def _run_producer(self, producer, msg):
         await self._queue.put(STOP)
 
     async def _run_consumer(self, consumer):
-        while True:
+        while not self._stop_event.is_set():
             data = await self._queue.get()
 
             if data is STOP:

reef/_actor.py

@@ -32,6 +32,7 @@ def __init__(
         self._datasource_factory = datasource_factory
         self._tasks = set()
         self.order_config = config_service.get("order")
+        self._stop_event = asyncio.Event()
 
     def on_start(self):
         worker_task = asyncio.create_task(self._process_orders())
@@ -43,6 +44,8 @@ def on_start(self):
         self._tasks.add(poll_task)
 
     def on_stop(self):
+        self._stop_event.set()
+
         for task in list(self._tasks):
             task.cancel()
 
@@ -68,7 +71,7 @@ async def _process_orders(self):
         monitor_interval = self.order_config.get("monitor_interval", 10)
 
         try:
-            while True:
+            while not self._stop_event.is_set():
                 pq_order = await self._order_queue.get()
                 current_time = time.time()