Use PipeChannels or FdStreams to make worker processes natively async

docs/source/reference-core.rst

@@ -1816,6 +1816,45 @@ to spawn a child thread, and then use a :ref:`memory channel
 
 .. literalinclude:: reference-core/from-thread-example.py
 
+.. _worker_processes:
+
+Worker Processes
+----------------
+
+Given that Python (and CPython in particular) has ongoing difficulties with
+CPU-bound work, Trio provides a method to dispatch synchronous function execution to
+special subprocesses known as "Worker Processes". By default, Trio will create as many
+workers as the system has CPUs (as reported by :func:`os.cpu_count`), allowing fair
+and truly parallel dispatch of CPU-bound work. As with Trio threads, these processes
+are cached in a process pool to minimize latency and resource usage. Despite this,
+executing a function in a process is at best an order of magnitude slower than in
+a thread, and possibly even slower when dealing with large arguments or a cold pool.
+Therefore, we recommend avoiding worker process dispatch for functions with a
+duration of less than about 10 ms.
+
+Unlike threads, subprocesses are strongly isolated from the parent process, which
+allows two important features that cannot be portably implemented in threads:
+
+  - Forceful cancellation: a deadlocked call or infinite loop can be cancelled
+    by completely terminating the process.
+  - Protection from errors: if a call segfaults or an extension module has an
+    unrecoverable error, the worker may die but Trio will raise
+    :exc:`BrokenWorkerError` and carry on.
+
+In both cases the workers die suddenly and violently, and at an unpredictable point
+in the execution of the dispatched function, so avoid using the cancellation feature
+if loss of intermediate results, writes to the filesystem, or shared memory writes
+may leave the larger system in an incoherent state.
+
+.. module:: trio.to_process
+.. currentmodule:: trio
+
+Putting CPU-bound functions in worker processes
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. autofunction:: trio.to_process.run_sync
+
+.. autofunction:: trio.to_process.current_default_process_limiter
 
 Exceptions and warnings
 -----------------------
@@ -1834,6 +1873,8 @@ Exceptions and warnings
 
 .. autoexception:: BrokenResourceError
 
+.. autoexception:: BrokenWorkerError
+
 .. autoexception:: RunFinishedError
 
 .. autoexception:: TrioInternalError

newsfragments/1781.headline.rst

@@ -0,0 +1,4 @@
+Trio now provides `multiprocessing` based worker processes for the delegation
+of cpu-bound work via :func:`trio.to_process.run_sync`. The API is comparable to
+:func:`trio.to_thread.run_sync` but true cancellation is achieved by killing the
+worker process.

trio/__init__.py

@@ -90,12 +90,15 @@
 
 from ._deprecate import TrioDeprecationWarning
 
+from ._worker_processes import BrokenWorkerError
+
 # Submodules imported by default
 from . import lowlevel
 from . import socket
 from . import abc
 from . import from_thread
 from . import to_thread
+from . import to_process
 
 # Not imported by default, but mentioned here so static analysis tools like
 # pylint will know that it exists.

trio/_core/_windows_cffi.py

@@ -179,6 +179,25 @@
   LPOVERLAPPED lpOverlapped
 );
 
+BOOL PeekNamedPipe(
+  HANDLE  hNamedPipe,
+  LPVOID  lpBuffer,
+  DWORD   nBufferSize,
+  LPDWORD lpBytesRead,
+  LPDWORD lpTotalBytesAvail,
+  LPDWORD lpBytesLeftThisMessage
+);
+
+BOOL GetNamedPipeHandleStateA(
+  HANDLE  hNamedPipe,
+  LPDWORD lpState,
+  LPDWORD lpCurInstances,
+  LPDWORD lpMaxCollectionCount,
+  LPDWORD lpCollectDataTimeout,
+  LPSTR   lpUserName,
+  DWORD   nMaxUserNameSize
+);
+
 // From https://github.com/piscisaureus/wepoll/blob/master/src/afd.h
 typedef struct _AFD_POLL_HANDLE_INFO {
   HANDLE Handle;
@@ -245,6 +264,7 @@ class ErrorCodes(enum.IntEnum):
     ERROR_INVALID_PARMETER = 87
     ERROR_NOT_FOUND = 1168
     ERROR_NOT_SOCKET = 10038
+    ERROR_MORE_DATA = 234
 
 
 class FileFlags(enum.IntEnum):
@@ -296,6 +316,13 @@ class IoControlCodes(enum.IntEnum):
     IOCTL_AFD_POLL = 0x00012024
 
 
+class PipeModes(enum.IntFlag):
+    PIPE_WAIT = 0x00000000
+    PIPE_NOWAIT = 0x00000001
+    PIPE_READMODE_BYTE = 0x00000000
+    PIPE_READMODE_MESSAGE = 0x00000002
+
+
 ################################################################
 # Generic helpers
 ################################################################
@@ -321,3 +348,21 @@ def raise_winerror(winerror=None, *, filename=None, filename2=None):
         _, msg = ffi.getwinerror(winerror)
     # https://docs.python.org/3/library/exceptions.html#OSError
     raise OSError(0, msg, filename, winerror, filename2)
+
+
+def get_pipe_state(handle):
+    lpState = ffi.new("LPDWORD")
+    if not kernel32.GetNamedPipeHandleStateA(
+        _handle(handle), lpState, ffi.NULL, ffi.NULL, ffi.NULL, ffi.NULL, 0
+    ):
+        raise_winerror()  # pragma: no cover
+    return lpState[0]
+
+
+def peek_pipe_message_left(handle):
+    left = ffi.new("LPDWORD")
+    if not kernel32.PeekNamedPipe(
+        _handle(handle), ffi.NULL, 0, ffi.NULL, ffi.NULL, left
+    ):
+        raise_winerror()  # pragma: no cover
+    return left[0]

trio/_windows_pipes.py

@@ -1,9 +1,16 @@
 import sys
 from typing import TYPE_CHECKING
 from . import _core
-from ._abc import SendStream, ReceiveStream
+from ._abc import SendStream, ReceiveStream, SendChannel, ReceiveChannel
 from ._util import ConflictDetector, Final
-from ._core._windows_cffi import _handle, raise_winerror, kernel32, ffi
+from ._core._windows_cffi import (
+    _handle,
+    raise_winerror,
+    kernel32,
+    ffi,
+    ErrorCodes,
+    peek_pipe_message_left,
+)
 
 assert sys.platform == "win32" or not TYPE_CHECKING
 
@@ -132,3 +139,70 @@ async def receive_some(self, max_bytes=None) -> bytes:
 
     async def aclose(self):
         await self._handle_holder.aclose()
+
+
+class PipeSendChannel(SendChannel[bytes]):
+    """Represents a message stream over a pipe object."""
+
+    def __init__(self, handle: int) -> None:
+        self._pss = PipeSendStream(handle)
+        # needed for "detach" via _handle_holder.handle = -1
+        self._handle_holder = self._pss._handle_holder
+
+    async def send(self, value: bytes):
+        # Works just fine if the pipe is message-oriented
+        await self._pss.send_all(value)
+
+    async def aclose(self):
+        await self._handle_holder.aclose()
+
+
+class PipeReceiveChannel(ReceiveChannel[bytes]):
+    """Represents a message stream over a pipe object."""
+
+    def __init__(self, handle: int) -> None:
+        self._handle_holder = _HandleHolder(handle)
+        self._conflict_detector = ConflictDetector(
+            "another task is currently using this pipe"
+        )
+
+    async def receive(self) -> bytes:
+        with self._conflict_detector:
+            buffer = bytearray(DEFAULT_RECEIVE_SIZE)
+            try:
+                received = await self._receive_some_into(buffer)
+            except OSError as e:
+                if e.winerror != ErrorCodes.ERROR_MORE_DATA:
+                    raise  # pragma: no cover
+                left = peek_pipe_message_left(self._handle_holder.handle)
+                # preallocate memory to avoid an extra copy of very large messages
+                newbuffer = bytearray(DEFAULT_RECEIVE_SIZE + left)
+                with memoryview(newbuffer) as view:
+                    view[:DEFAULT_RECEIVE_SIZE] = buffer
+                    await self._receive_some_into(view[DEFAULT_RECEIVE_SIZE:])
+                return newbuffer
+            else:
+                del buffer[received:]
+                return buffer
+
+    async def _receive_some_into(self, buffer) -> bytes:
+        if self._handle_holder.closed:
+            raise _core.ClosedResourceError("this pipe is already closed")
+        try:
+            return await _core.readinto_overlapped(self._handle_holder.handle, buffer)
+        except BrokenPipeError:
+            if self._handle_holder.closed:
+                raise _core.ClosedResourceError(
+                    "another task closed this pipe"
+                ) from None
+
+            # Windows raises BrokenPipeError on one end of a pipe
+            # whenever the other end closes, regardless of direction.
+            # Convert this to EndOfChannel.
+            #
+            # Do we have to checkpoint manually? We are raising an exception.
+            await _core.checkpoint()
+            raise _core.EndOfChannel
+
+    async def aclose(self):
+        await self._handle_holder.aclose()