Run main tests with forkserver

Author: AlanCoding

dispatcher/main.py

@@ -10,6 +10,7 @@
 from dispatcher.producers.brokered import BrokeredProducer
 from dispatcher.producers.scheduled import ScheduledProducer
 from dispatcher.utils import MODULE_METHOD_DELIMITER
+from dispatcher.process import ForkServerManager
 
 logger = logging.getLogger(__name__)
 
@@ -79,7 +80,7 @@ def __init__(self) -> None:
 
 
 class DispatcherMain:
-    def __init__(self, config: dict):
+    def __init__(self, config: dict, manager_cls: type = ForkServerManager):
         self.delayed_messages: list[SimpleNamespace] = []
         self.received_count = 0
         self.control_count = 0
@@ -88,7 +89,7 @@ def __init__(self, config: dict):
         # 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(config.get('pool', {}).get('max_workers', 3), self.fd_lock)
+        self.pool = WorkerPool(config.get('pool', {}).get('max_workers', 3), fd_lock=self.fd_lock, manager_cls=manager_cls)
 
         # Initialize all the producers, this should not start anything, just establishes objects
         self.producers: list[Union[ScheduledProducer, BrokeredProducer]] = []

dispatcher/pool.py

@@ -94,11 +94,11 @@ def __init__(self) -> None:
 
 
 class WorkerPool:
-    def __init__(self, num_workers: int, fd_lock: Optional[asyncio.Lock] = None):
+    def __init__(self, num_workers: int, fd_lock: Optional[asyncio.Lock] = None, manager_cls: type = ForkServerManager):
         self.num_workers = num_workers
         self.workers: dict[int, PoolWorker] = {}
         self.next_worker_id = 0
-        self.process_manager = ForkServerManager()
+        self.process_manager = manager_cls()
         self.queued_messages: list[dict] = []  # TODO: use deque, invent new kinds of logging anxiety
         self.read_results_task: Optional[Task] = None
         self.start_worker_task: Optional[Task] = None

tests/conftest.py

@@ -12,6 +12,7 @@
 from dispatcher.control import Control
 
 from dispatcher.brokers.pg_notify import apublish_message, aget_connection, get_connection
+from dispatcher.process import ProcessManager, ForkServerManager
 
 
 # List of channels to listen on
@@ -28,10 +29,15 @@ def pg_dispatcher() -> DispatcherMain:
     return DispatcherMain(BASIC_CONFIG)
 
 
-@pytest_asyncio.fixture(loop_scope="function", scope="function")
+@pytest_asyncio.fixture(
+    loop_scope="function",
+    scope="function",
+    params=[ProcessManager, ForkServerManager],
+    ids=["fork", "forkserver"],
+)
 async def apg_dispatcher(request) -> AsyncIterator[DispatcherMain]:
     try:
-        dispatcher = DispatcherMain(BASIC_CONFIG)
+        dispatcher = DispatcherMain(BASIC_CONFIG, manager_cls=request.param)
 
         await dispatcher.connect_signals()
         await dispatcher.start_working()