import asyncio
import inspect
from functools import wraps
from typing import Any, Awaitable, Callable, Union
import pytest
from multiprocess import Pipe, Process
from multiprocess.connection import Connection
import multiprocess as mp
mp.set_start_method(method="fork", force=True)  # "spawn" works fine
class SubprocessError:
    def __init__(self, ex: Exception) -> None:
        self.exception = ex
def in_subprocess[T](func: Callable[..., Union[T, Awaitable[T]]]) -> Callable[..., Awaitable[T]]:
    @wraps(func)
    async def wrapper(*args: Any, **kwargs: Any) -> T:
        return await calculate_in_subprocess(func, *args, **kwargs)
    return wrapper
async def calculate_in_subprocess[T](func: Callable[..., Union[T, Awaitable[T]]], *args: Any, **kwargs: Any) -> T:
    rx, tx = Pipe(duplex=False)  # receiver & transmitter ; Pipe is one-way only
    process = Process(target=_inner, args=(tx, func, *args), kwargs=kwargs)
    process.start()
    event = asyncio.Event()
    loop = asyncio.get_event_loop()
    loop.add_reader(fd=rx.fileno(), callback=event.set)
    if not rx.poll():  # do not use process.is_alive() as condition here
        await event.wait()
    loop.remove_reader(fd=rx.fileno())
    event.clear()
    result = rx.recv()
    process.join()  # this blocks synchronously! make sure that process is terminated before you call join()
    rx.close()
    tx.close()
    if isinstance(result, SubprocessError):
        raise result.exception
    return result
def _inner[T](tx: Connection, fun: Callable[..., Union[T, Awaitable[T]]], *a, **kw_args) -> None:
    event_loop = None
    if inspect.iscoroutinefunction(fun):
        event_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(event_loop)
    try:
        if event_loop is not None:
            res = event_loop.run_until_complete(fun(*a, **kw_args))
        else:
            res = fun(*a, **kw_args)
    except Exception as ex:
        tx.send(SubprocessError(ex=ex))
    else:
        tx.send(res)
@pytest.mark.asyncio
async def test_in_subprocess_simple_async():
    @in_subprocess
    async def f() -> int:
        return 42
    assert await f() == 42

-------------------------------- live log call ---------------------------------
ERROR    asyncio:base_events.py:1875 Exception in callback <_asyncio.TaskStepMethWrapper object at 0x7e71ba729ff0>()
handle: <Handle <_asyncio.TaskStepMethWrapper object at 0x7e71ba729ff0>()>
Traceback (most recent call last):
  File "/usr/lib/python3.14/asyncio/events.py", line 94, in _run
    self._context.run(self._callback, *self._args)
    ~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
RuntimeError: Cannot enter into task <Task pending name='Task-2' coro=<test_in_subprocess_simple_async.<locals>.f() running at foo.py:73> cb=[_run_until_complete_cb() at /usr/lib/python3.14/asyncio/base_events.py:181]> while another task <Task pending name='Task-1' coro=<test_in_subprocess_simple_async() running at foo.py:77> cb=[_run_until_complete_cb() at /usr/lib/python3.14/asyncio/base_events.py:181]> is being executed.

certifi==2025.10.5
charset-normalizer==3.4.3
dill==0.4.0
docker==7.1.0
idna==3.10
iniconfig==2.1.0
multiprocess @ git+https://github.com/uqfoundation/multiprocess.git@02ea4bd36cac5013d70847815c92e1a736ef4a05
packaging==25.0
pluggy==1.6.0
Pygments==2.19.2
pytest==8.4.2
pytest-asyncio==1.2.0
pytest_docker_tools==3.1.9
requests==2.32.5
urllib3==2.5.0