Delay initialization of asyncio.Queue

This is needed for Python 3.8 because the initialization is using
the current event loop, for which we have multiple given multiple
threads in Ert. Thus we need to initialize as late as possible

src/ert/scheduler/driver.py

@@ -19,10 +19,12 @@ class JobEvent(Enum):
 class Driver(ABC):
     """Adapter for the HPC cluster."""
 
-    event_queue: asyncio.Queue[Tuple[int, JobEvent]]
-
     def __init__(self) -> None:
-        self.event_queue = asyncio.Queue()
+        self.event_queue: Optional[asyncio.Queue[Tuple[int, JobEvent]]] = None
+
+    async def ainit(self) -> None:
+        if self.event_queue is None:
+            self.event_queue = asyncio.Queue()
 
     @abstractmethod
     async def submit(self, iens: int, executable: str, /, *args: str, cwd: str) -> None:
@@ -49,5 +51,4 @@ def create_poll_task(self) -> Optional[asyncio.Task[None]]:
         Returns:
           `asyncio.Task`, or None if polling is not applicable (eg. for LocalDriver)
         """
-
         return None

src/ert/scheduler/job.py

@@ -109,4 +109,7 @@ async def _send(self, state: State) -> None:
                 "queue_event_type": status,
             },
         )
+        if self._scheduler._events is None:
+            await self._scheduler.ainit()
+        assert self._scheduler._events is not None
         await self._scheduler._events.put(to_json(event))

src/ert/scheduler/local_driver.py

@@ -32,6 +32,11 @@ async def _wait_until_finish(
             cwd=cwd,
             preexec_fn=os.setpgrp,
         )
+
+        if self.event_queue is None:
+            await self.ainit()
+        assert self.event_queue is not None
+
         await self.event_queue.put((iens, JobEvent.STARTED))
         try:
             if await proc.wait() == 0:

src/ert/scheduler/scheduler.py

@@ -6,7 +6,6 @@
 import os
 import ssl
 import threading
-from asyncio.queues import Queue
 from dataclasses import asdict
 from typing import (
     TYPE_CHECKING,
@@ -50,13 +49,19 @@ def __init__(self, driver: Optional[Driver] = None) -> None:
         self.driver = driver
         self._jobs: MutableMapping[int, Job] = {}
         self._tasks: MutableMapping[int, asyncio.Task[None]] = {}
-        self._events: Queue[Any] = Queue()
+
+        self._events: Optional[asyncio.Queue[Any]] = None
 
         self._ee_uri = ""
         self._ens_id = ""
         self._ee_cert: Optional[str] = None
         self._ee_token: Optional[str] = None
 
+    async def ainit(self) -> None:
+        # While supporting Python 3.8, this statement must be delayed.
+        if self._events is None:
+            self._events = asyncio.Queue()
+
     def add_realization(
         self, real: Realization, callback_timeout: Callable[[int], None]
     ) -> None:
@@ -86,6 +91,10 @@ async def _publisher(self) -> None:
         if self._ee_token:
             headers["token"] = self._ee_token
 
+        if self._events is None:
+            await self.ainit()
+        assert self._events is not None
+
         async with connect(
             self._ee_uri,
             ssl=tls,
@@ -132,6 +141,10 @@ async def execute(
         return EVTYPE_ENSEMBLE_STOPPED
 
     async def _process_event_queue(self) -> None:
+        if self.driver.event_queue is None:
+            await self.driver.ainit()
+        assert self.driver.event_queue is not None
+
         while True:
             iens, event = await self.driver.event_queue.get()
             if event == JobEvent.STARTED: