Improve publisher pool reliability []

Author: wchilders-nvidia

dev_tools/pylibs/edgacknowledg/forge/publisher/pool.py

@@ -327,7 +327,25 @@ async def _replace(self, slot: _Slot) -> None:
 
   async def _start(self, slot: _Slot) -> None:
     '''Start ``slot``'s process and wait until it is ready; raise
-    ``PoolError`` (and delay the next try) if it isn't.
+    ``PoolError`` (and delay the next try) if it isn't, whatever the
+    reason (e.g. no process could be started at all).
+    '''
+    try:
+      await self._launch(slot)
+    except Exception as ex:
+      returncode = await self._stop(slot)
+      slot.process = None
+      slot.failed_starts += 1
+      delay = min(2.0 ** (slot.failed_starts - 1), _MAX_RESTART_DELAY_S)
+      slot.not_before = time.monotonic() + delay
+      _log.error('pool process %d failed to start (%s; exit status %s); '
+                 'trying again in %g s', slot.index, ex, returncode, delay)
+      raise PoolError(f"pool process {slot.index} failed to start") from ex
+    slot.failed_starts = 0
+
+  async def _launch(self, slot: _Slot) -> None:
+    '''Start ``slot``'s process and wait until it is ready; raise if it
+    isn't.
     '''
     parent_end, process_end = socket.socketpair()
     try:
@@ -360,15 +378,7 @@ async def _start(self, slot: _Slot) -> None:
       self._kill(slot)
       raise
     if ready.strip() != b'{"ready": true}':
-      returncode = await self._stop(slot)
-      slot.process = None
-      slot.failed_starts += 1
-      delay = min(2.0 ** (slot.failed_starts - 1), _MAX_RESTART_DELAY_S)
-      slot.not_before = time.monotonic() + delay
-      _log.error('pool process %d failed to start (%s); trying again in '
-                 '%g s', slot.index, returncode, delay)
-      raise PoolError(f"pool process {slot.index} failed to start")
-    slot.failed_starts = 0
+      raise PoolError('it never said it was ready')
     _log.info('started pool process %d (pid %d)', slot.index, process.pid)
 
   def _kill(self, slot: _Slot) -> None:

dev_tools/services/acknowledg-forge/tests/test_pool.py

@@ -8,6 +8,7 @@
 '''
 
 import asyncio
+import errno
 import os
 import signal
 import time
@@ -236,6 +237,35 @@ async def test_a_process_failing_to_start_backs_off(
     await pool.close()
 
 
+async def test_no_process_starting_at_all_backs_off(
+    env: _Env,
+    monkeypatch: pytest.MonkeyPatch) -> None:
+  create_subprocess_exec = asyncio.create_subprocess_exec
+  failing = True
+
+  async def out_of_processes(*args: Any, **kwargs: Any) -> Any:
+    if failing:
+      raise OSError(errno.EAGAIN, 'Resource temporarily unavailable')
+    return await create_subprocess_exec(*args, **kwargs)
+
+  monkeypatch.setattr(asyncio, 'create_subprocess_exec', out_of_processes)
+  pool = ProcessPool(tools_service_url = env.tools.url, processes = 1)
+  try:
+    # As when it starts but isn't ready.
+    with pytest.raises(PoolError, match = 'failed to start'):
+      await pool.start()
+    with pytest.raises(PoolError, match = 'restarting'):
+      await pool.answer(env.section, _show_source('suite/a.ifc'))
+
+    # Its dispatcher still tries again, once it may.
+    failing = False
+    await asyncio.sleep(1.5)
+    answer = await pool.answer(env.section, _show_source('suite/a.ifc'))
+    assert 'dumped a.ifc' in _content(answer)
+  finally:
+    await pool.close()
+
+
 class _Channel:
   '''One request in; the replies the stand-in handler relays, out.'''