Make Process joining less likely to block

[Vizonex]

Description

In an attempt to make the asynchronous portions of this library less reliant on polling I decided to try seeing if there were other more fine tuned ways to go about checking to see if a process was closed. I plan to try making it so that a callback could be hooked in the future or see if there are any ways to hook in a listener for a pid to be closed but here is the best I could come up with. I'll try to see about finding other alternative solutions to more await asyncio.sleep(0.005) lines of code soon but this is a start with what I had in mind.

I may change this so threat it for right now as a draft or a DNM (Do not merge)

Edit: it's ready

[x42005e1f]

There are two things to consider here: performance and cancellability.

Suppose we have two ways to wait for something:

1. Use blocking_call(timeout).
2. Poll via a Boolean flag done + sleep(delay).

Obviously, we cannot use blocking calls directly in an asynchronous environment, because otherwise they will block the event loop. Therefore, they are usually run in separate threads:

await asyncio.to_thread(blocking_call, timeout)

At first glance, everything works fine, and the wait is as fast as possible (not counting the overhead due to multithreading). But there are two nuances here:

1. Threads cannot be killed (except for some dirty and platform-specific tricks), which means that the blocking call cannot actually be cancelled. This leads to thread leaks.
2. When no separate executor is used, the default one created by the event loop is used. And, of course, it has a limit on the number of worker threads - once that limit is reached, no further blocking calls will even be passed to any thread, which leads to hangs.

>>> import asyncio
>>> import os
>>> import sys
>>> import threading
>>> if sys.version_info >= (3, 13):
...     max_workers = min(32, (os.process_cpu_count() or 1) + 4)
... else:
...     max_workers = min(32, os.cpu_count() + 4)
...     
>>> max_workers
8
>>> threading.active_count()
2  # = initial_count
>>> event = threading.Event()
>>> for _ in range(max_workers):
...     try:
...         await asyncio.wait_for(asyncio.to_thread(event.wait, 60), 1 / max_workers)
...     except TimeoutError:
...         pass
...         
>>> threading.active_count()
10  # == 2 + 8 == initial_count + max_workers
>>> await asyncio.to_thread(print, "ok")  # hangs!

Now let us look at polling. It usually looks something like this:

while not done:
    await asyncio.sleep(delay)

It has no cancellability issues, since asyncio.sleep(delay) is a purely asynchronous call. However, its performance suffers:

1. When choosing delay, we are looking for a compromise between processor load and response speed. If the delay is too low, we will just waste processor cycles. But if it is too high, we will waste time.
2. The delay affects the number of operations per second. Even if we have a modern system capable of performing thousands process launches per second, our average number of operations per second will usually be limited by that delay (for 0.005, it is only 200 operations per second!).

>>> import asyncio
>>> import random
>>> import time
>>> delay = 0.005
>>> iterations = 100
>>> real_ops = 10_000
>>> ops = [None] * iterations
>>> for i in range(iterations):
...     task = asyncio.create_task(asyncio.sleep((1 / real_ops) * random.random()))
...     await asyncio.sleep(0)  # switch to the task
...     start = time.perf_counter()
...     while not task.done():
...         await asyncio.sleep(delay)
...     ops[i] = 1 / (time.perf_counter() - start)
...     
>>> ops[len(ops) // 2]
185.88300231336297  # ~= (1 / 0.005) < 10_000

The most viable option is to combine both ways:

try:
    if not done:
        async with asyncio.timeout(timeout):
            while not done:
                done = await asyncio.to_thread(blocking_call, delay)
except TimeoutError:
    # done is False
else:
    # done is True

Key benefits:

1. If there is no need to wait (done = self.exitcode is not None and this is True), neither threads nor anything else are used, which is very cheap and fast.
2. It responds well to cancellation if a sufficiently low delay is chosen, which mitigates the cancellability issue.
3. The number of operations per second is not limited by delay (but suffers from multithreading overhead).

However, even this approach still has problems. In particular, we still have the limit on the maximum number of worker threads. Because of this, we cannot actually wait for all processes at the same time, and the dependence of the waiting time on the number of processes will therefore be O(n) multiplied by delay. This is almost the same as if we were polling all processes manually in a worker thread loop.

We can rely on communication between processes via sockets, which is purely asynchronous (we can work with pipes and files in a truly asynchronous manner using APIs such as io_uring (Linux) or I/O rings (Windows 11), but unfortunately, they are still poorly supported, possibly due to security issues), but what about zombie processes and other nasties? We can also try to rely on the fact that as soon as a process is dead, the operating system closes the file descriptors (and thus sockets) opened by it, but how reliable is that?

If we take the combined approach as a basis, it can be improved by polling all those not in worker threads once every delay seconds (via self.exitcode or something else non-blocking). This way, we can avoid the performance drop caused by the call queue, but it is much more difficult to implement and still far from ideal (we are just reducing the worst case to the same polling).

In short, I believe that it is virtually impossible to solve the problem using threads and/or polling, unless we create a thread for each call/process (slow and heavy):

async def longpoll_in_thread(blocking_call, delay):
    loop = asyncio.get_running_loop()

    future = loop.create_future()

    # Note: exceptions from blocking_call are actually suppressed after the
    # task is cancelled or the event loop is closed. Ideally, they should be
    # handled (especially if it is KeyboardInterrupt) or at least logged, but
    # this will also complicate the code.

    def set_result(future, result):
        try:
            future.set_result(result)
        except asyncio.InvalidStateError:  # task is cancelled
            pass

    def set_exception(future, exception):
        try:
            future.set_exception(exception)
        except asyncio.InvalidStateError:  # task is cancelled
            pass

    def longpoll(future):
        while not future.cancelled():
            try:
                result = blocking_call(delay)
            except BaseException as exc:
                try:
                    loop.call_soon_threadsafe(set_exception, future, exc)
                except RuntimeError:  # event loop is closed
                    break
                finally:
                    del future
            else:
                if result:
                    try:
                        loop.call_soon_threadsafe(set_result, future, result)
                    except RuntimeError:  # event loop is closed
                        break
                    else:
                        break

    # Note: it would be much better to wait for the thread to complete with
    # shielding from cancellation, but the correct implementation of shielding
    # in asyncio is too complicated, and in this case it would be more correct
    # to give an example with aiologic.

    # Note: thread.start() is a blocking call (due to self._started.wait())!
    threading.Thread(target=longpoll, args=[future], daemon=True).start()

    return await future

try:
    if not done:
        done = await asyncio.wait_for(
            longpoll_in_thread(blocking_call, delay),
            timeout,
        )
except TimeoutError:
    # done is False
else:
    # done is True

For the last two examples, delay determines the response speed to cancellation. Since it only effectively affects the thread's shutdown time, it makes sense to make it large enough so that waiting for a large number of processes requires fewer resources. Unless, of course, waiting for the thread to complete is implemented.

[Vizonex]

@x42005e1f I see what your saying. I had an idea so that python versions that still do not have asyncio.timeout(...) can have backwards compatability

async def join(self, timeout: Optional[int] = None) -> None:
        """Wait for the process to finish execution without blocking the main thread."""
        if not self.is_alive() and self.exitcode is None:
            raise ValueError("must start process before joining it")
        
        # XXX: some versions of python do not come with asyncio.timeout() so this is kept as backwards comptability
        if timeout:
            return await asyncio.wait_for(self.join, timeout)
        
        while self.exitcode is None:
            await asyncio.to_thread(self.aio_process.join, 0.005)

[x42005e1f]

I had an idea so that python versions that still do not have asyncio.timeout(...) can have backwards compatability

There is also a backport: graingert/taskgroup. But yes, it is redundant when the method can be called recursively.

[x42005e1f]

These changes seem strange. asyncio.wait_for() expects an awaitable object, not a function. Also, this code will not work correctly for timeout=0. It is better to leave it unchanged here.

[Vizonex]

@x42005e1f I think I'm going to try and make it so that the backend of aiomultiprocess can use asynchronous processes instead of synchronous ones so that way asyncio does a better job at cleaning up processes too since there isn't very many cases of needing aiomultiprocess outside of an eventloop by default.

[x42005e1f]

I am curious what exactly you mean by asynchronous processes, as there may be some differences in interpretation here. But I agree that it is better to stick to the main ecosystem unless there is a reason to do otherwise.

[Vizonex]

I am curious what exactly you mean by asynchronous processes.

Basically instead of using Popen for creating the core subprocesses like the one in the multiprocess implementation (which uses synchronous processes) My idea involves using the eventloop instead for making the subprocesses that are involved in making another python processes so that the processes can be easily awaited and cleaned up.

Know that this is a rought draft of the idea and is mainly used in order to visualize what I am after.

"""Windows support for asynchronous Popen processses"""
from .abc import AbstractPopen
import asyncio
from asyncio import AbstractEventLoop, get_event_loop
from asyncio.subprocess import SubprocessStreamProtocol
from multiprocessing import spawn
from multiprocessing import reduction
from multiprocessing.context import get_spawning_popen, set_spawning_popen
from multiprocessing.util import Finalize
import sys
import os
import _winapi, msvcrt
from typing import Callable
from functools import partial
# We need these from popen_spawn_win32.py a submodule of multiprocess

WINEXE = (sys.platform == 'win32' and getattr(sys, 'frozen', False))
WINSERVICE = sys.executable.lower().endswith("pythonservice.exe")

def _path_eq(p1, p2):
    return p1 == p2 or os.path.normcase(p1) == os.path.normcase(p2)

WINENV = not _path_eq(sys.executable, sys._base_executable)



class WindowsSubprocessProtocol(SubprocessStreamProtocol):
    """impliments a slight modification to enable callbacks on closing a process up"""
    def __init__(self, limit, loop, on_close:Callable[[], None]):
        super().__init__(limit, loop)
        self.on_close_cb = on_close
    
    def process_exited(self):
        super().process_exited()
        # Call function to terminate all other listening handles... 
        self.on_close_cb()
    


async def win32_create_subprocess_exec(on_close:Callable[[], None], program, *args, stdin=None, stdout=None,
                                 stderr=None, limit=65536,
                                 **kwds):
    """Implements slight modification to close lingering handles"""
    loop = asyncio.get_running_loop()
    protocol_factory = lambda: WindowsSubprocessProtocol(limit=limit,
                                                        loop=loop)
    transport, protocol = await loop.subprocess_exec(
        protocol_factory,
        program, *args,
        stdin=stdin, stdout=stdout,
        stderr=stderr, **kwds)
    return asyncio.subprocess.Process(transport, protocol, loop)


class Popen(AbstractPopen):
    method = "spawn"

    async def dump_async(self, obj, file) -> None:
        """Performs dumping reduction asynchronously via different thread"""
        return await asyncio.to_thread(reduction.dump, obj, file)


    def __init__(self, process_obj, loop:AbstractEventLoop | None = None):
    
        self._loop = loop or get_event_loop()
        self._process = process_obj
        self._subprocess = None
       
        self._python_exe = spawn.get_executable()
        
        if WINENV and _path_eq(self._python_exe, sys.executable):
            self._python_exe = sys._base_executable
            self._env = os.environ.copy()
            self._env["__PYVENV_LAUNCHER__"] = sys.executable
        else:
            self._env = None

    async def _launch(self, process_obj):
        # We need to perform and same setup as popen_spawn_win32.py from this point...
        rhandle, whandle = _winapi.CreatePipe(None, 0)
        wfd = msvcrt.open_osfhandle(whandle, 0)
        prep_data = spawn.get_preparation_data(process_obj.aio_process.name)

        cmd = ['"%s"' % x for x in spawn.get_command_line(
            parent_pid=os.getpid(),
            pipe_handle=rhandle
        )]

        with open(wfd, 'wb', closefd=True) as to_child:
            # Took some ideas after seeing how popen_spawn_win32 works...
            # this way a Finalize class isn't nessesary...
            await win32_create_subprocess_exec(
                partial(_winapi.CloseHandle, int(rhandle)),
                self._python_exe, *cmd,
            )
            set_spawning_popen(self)
            try:
                await self.dump_async(prep_data, to_child)
                await self.dump_async(process_obj, to_child)
            finally:
                set_spawning_popen(None)
        

    def duplicate_for_child(self, handle:int):
        assert self is get_spawning_popen()
        return reduction.duplicate(handle, self._process.pid)



    async def wait(self, timeout:int | None = None) -> int | None:
        try:
            return await asyncio.wait_for(self._subprocess.wait(), timeout)
        finally:
            return self._subprocess.returncode

    async def poll(self) -> int | None:
        """Attempts to wait a single eventloop cycle to see if the 
        windows subprocess completed"""
        return await self.wait(0)

    def terminate(self):
        if self._subprocess.returncode is None:
            self._subprocess.terminate()

    kill = terminate

    def close():
        # TODO:
        return

   

# abc.py
from __future__ import annotations
from asyncio import AbstractEventLoop
from abc import ABC, abstractmethod

from ..core import Process





class AbstractPopen(ABC):
    """
    Base Object for starting a subprocess to run code on.
    """
    @abstractmethod
    def __init__(self, process_obj: Process, loop:AbstractEventLoop | None = None) -> None:...
    
    @abstractmethod
    def duplicate_for_child(self, fd:int) -> int:...

    @abstractmethod
    async def poll(self) -> int | None:...

    @abstractmethod
    async def wait(self, timeout:int | None = None) -> int | None:...

    @abstractmethod
    def terminate(self) -> None:...

    @abstractmethod
    def kill(self) -> None:...

    @abstractmethod
    async def _launch(self, process_obj: Process) -> None:...

    @abstractmethod
    def close(self) -> None:...

[x42005e1f]

Basically instead of using Popen for creating the core subprocesses like the one in the multiprocess implementation (which uses synchronous processes) My idea involves using the eventloop instead for making the subprocesses that are involved in making another python processes so that the processes can be easily awaited and cleaned up.

The key drawback of this approach is that it essentially implements the spawn start method, but not the fork/forkserver start method (if implemented in a mixed way, would it not complicate the code base too much?). In some cases, spawn is too slow (and complicates certain memory-related things), so this approach will have limited application. As for asynchrony, yes, it will give better results on some platforms (but not all, see _ThreadedChildWatcher), so it makes sense.