Complete worker settings initialization

Author: AlanCoding

dispatcher/factories.py

@@ -53,7 +53,7 @@ def from_settings(settings: LazySettings = global_settings) -> DispatcherMain:
     between the service, publisher, and any other interacting processes.
     """
     producers = producers_from_settings(settings=settings)
-    return DispatcherMain(settings.service, producers)
+    return DispatcherMain(settings.service, producers, settings=settings)
 
 
 # ---- Publisher objects ----

dispatcher/main.py

@@ -7,6 +7,7 @@
 
 from dispatcher.pool import WorkerPool
 from dispatcher.producers import BaseProducer
+from dispatcher.config import settings as global_settings, LazySettings
 
 logger = logging.getLogger(__name__)
 
@@ -76,7 +77,7 @@ def __init__(self) -> None:
 
 
 class DispatcherMain:
-    def __init__(self, service_config: dict, producers: Iterable[BaseProducer]):
+    def __init__(self, service_config: dict, producers: Iterable[BaseProducer], settings: LazySettings = global_settings):
         self.delayed_messages: list[SimpleNamespace] = []
         self.received_count = 0
         self.control_count = 0
@@ -85,7 +86,7 @@ def __init__(self, service_config: dict, producers: Iterable[BaseProducer]):
         # Lock for file descriptor mgmnt - hold lock when forking or connecting, to avoid DNS hangs
         # psycopg is well-behaved IFF you do not connect while forking, compare to AWX __clean_on_fork__
         self.fd_lock = asyncio.Lock()
-        self.pool = WorkerPool(fd_lock=self.fd_lock, **service_config)
+        self.pool = WorkerPool(fd_lock=self.fd_lock, settings=settings, **service_config)
 
         # Set all the producers, this should still not start anything, just establishes objects
         self.producers = producers

dispatcher/pool.py

@@ -12,13 +12,12 @@
 
 
 class PoolWorker:
-    def __init__(self, worker_id: int, process: ProcessProxy, settings: LazySettings = global_settings) -> None:
+    def __init__(self, worker_id: int, process: ProcessProxy) -> None:
         self.worker_id = worker_id
         self.process = process
         self.current_task: Optional[dict] = None
         self.started_at: Optional[int] = None
         self.is_active_cancel: bool = False
-        self.settings_stash: dict = settings.serialize()
 
         # Tracking information for worker
         self.finished_count = 0
@@ -96,9 +95,10 @@ def __init__(self) -> None:
 
 
 class WorkerPool:
-    def __init__(self, max_workers: int, fd_lock: Optional[asyncio.Lock] = None):
+    def __init__(self, max_workers: int, fd_lock: Optional[asyncio.Lock] = None, settings: LazySettings = global_settings):
         self.max_workers = max_workers
         self.workers: dict[int, PoolWorker] = {}
+        self.settings_stash: dict = settings.serialize()  # These are passed to the workers to initialize dispatcher settings
         self.next_worker_id = 0
         self.process_manager = ProcessManager()
         self.queued_messages: list[dict] = []  # TODO: use deque, invent new kinds of logging anxiety

tests/integration/test_main.py

@@ -45,7 +45,7 @@ async def test_run_decorated_function(apg_dispatcher, test_settings):
 @pytest.mark.asyncio
 async def test_submit_with_global_settings(apg_dispatcher, test_settings):
     clearing_task = asyncio.create_task(apg_dispatcher.pool.events.work_cleared.wait())
-    with temporary_settings(test_settings):
+    with temporary_settings(test_settings.serialize()):
         test_methods.print_hello.delay()  # settings are inferred from global context
     await asyncio.wait_for(clearing_task, timeout=3)