Fix - subprocess was not ready for testing

Author: AlanCoding

tests/benchmark/conftest.py

@@ -31,6 +31,7 @@ async def run_pool(cls, queue_in, queue_out, workers, function='lambda: __import
         dispatcher = DispatcherMain({"producers": {"brokers": {}}, "pool": {"max_workers": workers}})
         pool = dispatcher.pool
         await pool.start_working(dispatcher)
+        queue_out.put('ready')
 
         print('waiting for message to start test')
         loop = asyncio.get_event_loop()
@@ -85,6 +86,9 @@ def start_server(self, workers, **kwargs):
     @contextlib.contextmanager
     def with_server(self, *args, **kwargs):
         process = self.start_server(*args, **kwargs)
+        msg = self.queue_out.get()
+        if msg != 'ready':
+            raise RuntimeError('never got ready message from subprocess')
         try:
             yield self
         finally:
@@ -128,6 +132,10 @@ async def run_pool(cls, queue_in, queue_out, workers):
             {"producers": {"brokers": {"pg_notify": {"conninfo": CONNECTION_STRING}, "channels": CHANNELS}}, "pool": {"max_workers": workers}}
         )
         await dispatcher.start_working()
+        # Make sure the dispatcher is listening before starting the tests which will submit messages
+        for producer in dispatcher.producers:
+            await producer.events.ready_event.wait()
+        queue_out.put('ready')
 
         print('waiting for message to start test')
         loop = asyncio.get_event_loop()