elog/mod_custom: Spawn processes in background

Since elog_process is typically called while the event loop is
running, hold references to spawned processes and wait for them
asynchronously, ultimately waiting for them if necessary when
the AsyncioEventLoop _close_main method calls _async_finalize.

ConfigProtectTestCase is useful for exercising this code, and
this little make.globals patch can be used to test failure
during finalize with this error message:

    !!! PORTAGE_ELOG_COMMAND failed with exitcode 1

--- a/cnf/make.globals
+++ b/cnf/make.globals
@@ -144 +144,2 @@ PORTAGE_ELOG_CLASSES="log warn error"
-PORTAGE_ELOG_SYSTEM="save_summary:log,warn,error,qa echo"
+PORTAGE_ELOG_SYSTEM="save_summary:log,warn,error,qa echo custom"
+PORTAGE_ELOG_COMMAND="/bin/false"

Bug: https://bugs.gentoo.org/925907
Signed-off-by: Zac Medico <zmedico@gentoo.org>

lib/portage/elog/mod_custom.py

@@ -1,10 +1,31 @@
 # elog/mod_custom.py - elog dispatch module
-# Copyright 2006-2020 Gentoo Authors
+# Copyright 2006-2024 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
+import types
+
+import portage
 import portage.elog.mod_save
 import portage.exception
 import portage.process
+from portage.util.futures import asyncio
+
+# Since elog_process is typically called while the event loop is
+# running, hold references to spawned processes and wait for them
+# asynchronously, ultimately waiting for them if necessary when
+# the AsyncioEventLoop _close_main method calls _async_finalize.
+_proc_refs = None
+
+
+def _get_procs() -> list[tuple[portage.process.MultiprocessingProcess, asyncio.Future]]:
+    """
+    Return list of (proc, asyncio.ensure_future(proc.wait())) which is not
+    inherited from the parent after fork.
+    """
+    global _proc_refs
+    if _proc_refs is None or _proc_refs.pid != portage.getpid():
+        _proc_refs = types.SimpleNamespace(pid=portage.getpid(), procs=[])
+    return _proc_refs.procs
 
 
 def process(mysettings, key, logentries, fulltext):
@@ -18,8 +39,49 @@ def process(mysettings, key, logentries, fulltext):
         mylogcmd = mysettings["PORTAGE_ELOG_COMMAND"]
         mylogcmd = mylogcmd.replace("${LOGFILE}", elogfilename)
         mylogcmd = mylogcmd.replace("${PACKAGE}", key)
-        retval = portage.process.spawn_bash(mylogcmd)
-        if retval != 0:
+        loop = asyncio.get_event_loop()
+        proc = portage.process.spawn_bash(mylogcmd, returnproc=True)
+        procs = _get_procs()
+        procs.append((proc, asyncio.ensure_future(proc.wait(), loop=loop)))
+        for index, (proc, waiter) in reversed(list(enumerate(procs))):
+            if not waiter.done():
+                continue
+            del procs[index]
+            if waiter.result() != 0:
+                raise portage.exception.PortageException(
+                    f"!!! PORTAGE_ELOG_COMMAND failed with exitcode {waiter.result()}"
+                )
+
+
+async def _async_finalize():
+    """
+    Async finalize is preferred, since we can wait for process exit status.
+    """
+    procs = _get_procs()
+    while procs:
+        proc, waiter = procs.pop()
+        if (await waiter) != 0:
+            raise portage.exception.PortageException(
+                f"!!! PORTAGE_ELOG_COMMAND failed with exitcode {waiter.result()}"
+            )
+
+
+def finalize():
+    """
+    NOTE: This raises PortageException if there are any processes
+    still running, so it's better to use _async_finalize instead
+    (inside the AsyncioEventLoop _close_main method).
+    """
+    procs = _get_procs()
+    while procs:
+        proc, waiter = procs.pop()
+        if not waiter.done():
+            waiter.cancel()
+            proc.terminate()
+            raise portage.exception.PortageException(
+                f"!!! PORTAGE_ELOG_COMMAND was killed after it was found running in the background (pid {proc.pid})"
+            )
+        elif waiter.result() != 0:
             raise portage.exception.PortageException(
-                "!!! PORTAGE_ELOG_COMMAND failed with exitcode %d" % retval
+                f"!!! PORTAGE_ELOG_COMMAND failed with exitcode {waiter.result()}"
             )

lib/portage/util/_eventloop/asyncio_event_loop.py

@@ -1,11 +1,11 @@
 # Copyright 2018-2024 Gentoo Authors
 # Distributed under the terms of the GNU General Public License v2
 
+import asyncio
 import os
 import signal
 import threading
 
-import asyncio as _real_asyncio
 from asyncio.events import AbstractEventLoop as _AbstractEventLoop
 from asyncio.unix_events import ThreadedChildWatcher
 
@@ -15,6 +15,11 @@
     PidfdChildWatcher = None
 
 import portage
+
+portage.proxy.lazyimport.lazyimport(
+    globals(),
+    "portage.elog:mod_custom",
+)
 from portage.util import socks5
 
 
@@ -25,7 +30,7 @@ class AsyncioEventLoop(_AbstractEventLoop):
     """
 
     def __init__(self, loop=None):
-        loop = loop or _real_asyncio.get_event_loop()
+        loop = loop or asyncio.get_event_loop()
         self._loop = loop
         self.run_until_complete = self._run_until_complete
         self.call_soon = loop.call_soon
@@ -75,12 +80,17 @@ def _close(self):
             self._closing = False
 
     async def _close_main(self):
+        tasks = []
+        tasks.append(asyncio.create_task(mod_custom._async_finalize()))
+
         # Even though this has an exit hook, invoke it here so that
         # we can properly wait for it and avoid messages like this:
         # [ERROR] Task was destroyed but it is pending!
         if socks5.proxy.is_running():
-            await socks5.proxy.stop()
+            tasks.append(asyncio.create_task(socks5.proxy.stop()))
 
+        for task in tasks:
+            await task
         portage.process.run_exitfuncs()
 
     @staticmethod
@@ -107,7 +117,7 @@ def _create_future(self):
         """
         Provide AbstractEventLoop.create_future() for python3.4.
         """
-        return _real_asyncio.Future(loop=self._loop)
+        return asyncio.Future(loop=self._loop)
 
     @property
     def _asyncio_child_watcher(self):