feat: add remote Deadpool executor

.coveragerc

@@ -1,4 +1,12 @@
 [run]
 source = deadpool
 branch = True
-#concurrency=multiprocessing
+relative_files = True
+parallel = True
+concurrency =
+    multiprocessing
+    thread
+patch =
+    fork
+    subprocess
+    _exit

README.rst

@@ -763,6 +763,91 @@ when creating the instance:
    ):
        fut = exe.submit(...)
 
+Remote executor (experimental)
+==============================
+
+``deadpool.remote`` lets independent processes share one server-owned
+``Deadpool`` over a Unix-domain or IPv4 TCP socket. The client implements the
+standard ``Executor`` interface, including ordered ``map()`` results:
+
+.. code-block:: python
+
+   from functools import partial
+
+   import deadpool
+   from deadpool.remote import (
+       DeadpoolClient,
+       DeadpoolServer,
+       UnixAddress,
+       UnixListener,
+   )
+
+   server = DeadpoolServer(
+       partial(deadpool.Deadpool, max_workers=8),
+       listeners=[UnixListener("/run/example/deadpool.sock")],
+   )
+   server.start()
+
+   with DeadpoolClient(UnixAddress("/run/example/deadpool.sock")) as executor:
+       future = executor.submit(pow, 2, 10, deadpool_priority=5)
+       assert future.result() == 1024
+
+A server may also expose registered operations. Registered mode rejects unknown
+operation names before worker dispatch:
+
+.. code-block:: python
+
+   server = DeadpoolServer(
+       partial(deadpool.Deadpool, max_workers=8),
+       listeners=[UnixListener("/run/example/deadpool.sock")],
+       task_registry={"math.pow": pow},
+   )
+   future = executor.submit_task("math.pow", 2, 10)
+
+TCP uses the same protocol. Plaintext is restricted to an explicit loopback
+opt-in (``TcpListener("127.0.0.1", port, insecure=True)``); other deployments
+must supply configured server and client ``ssl.SSLContext`` objects. Mutual TLS
+is recommended.
+
+Client and server wire bounds for control data, frames, messages, metadata, and
+chunk counts must match. The handshake rejects asymmetric wire limits before
+accepting submissions.
+
+``RemoteFuture.add_done_callback()`` follows standard ``Future`` timing:
+callbacks registered after terminal completion run synchronously before the
+method returns, while callbacks registered earlier use the isolated callback
+executor. ``callback_queue_size`` bounds callback batches admitted to that
+executor, not application-owned registrations waiting for dispatch; callback
+backpressure never blocks transport or Future state updates.
+
+Remote cancellation requires an authoritative broker decision. Unlike a local
+``Future.cancel()``, cancelling submitted work may therefore block up to the
+client's ``control_timeout`` or raise ``RemoteCancellationOutcomeUnknown`` when
+the distributed outcome cannot be determined. Work still in the client's local
+submission stage cancels immediately.
+
+.. warning::
+
+   The default ``PickleSerializer`` requires mutually trusted clients and
+   servers in both callable and registered-task modes. TLS authenticates and
+   encrypts peers but does not sandbox deserialization. Registered operation
+   authorization controls callable selection; it is not a deserialization
+   sandbox because workers still unpickle invocation arguments. Clients also
+   unpickle result and exception payloads, establishing reverse trust in the
+   server.
+
+The current framed wire layout is private and experimental because the remote
+executor specification intentionally leaves stable frame octets and canonical
+schemas to a later wire reference. Matching Deadpool releases support bounded
+JSON control headers, opaque chunked payloads, request identity, typed failures,
+ordinary and hard cancellation, health/statistics requests, fair broker
+scheduling, and explicit shutdown behavior. The first implementation does not
+negotiate session resumption, compression, IPv6, protocol-5 out-of-band buffers,
+shared-memory/file result transports, durable state, progress events, or
+streaming results. It also does not advertise stronger parent-death cleanup than
+the local pool's existing process monitor. A lost unacknowledged submission is
+reported as ambiguous and is never silently retried with a new request ID.
+
 Developer Workflow
 ==================
 
@@ -837,7 +922,7 @@ I currently do:
 
 1. Ensure that ``main`` branch is fully up to date with all to
    be released, and all the tests succeed.
-2. Change the ``__version__`` field in ``deadpool.py``. Flit
+2. Change the ``__version__`` field in ``deadpool/__init__.py``. Flit
    uses this to stamp the version.
 3. Verify that ``flit build`` succeeds. This will produce a
    wheel in the ``dist/`` directory. You can inspect this

deadpool/__init__.py

@@ -0,0 +1,36 @@
+"""A resilient process-pool executor and its optional remote interface."""
+
+__version__ = "2026.6.1"
+
+from . import _pool
+from ._pool import (
+    CancelledError,
+    Deadpool,
+    Future,
+    PoolClosed,
+    ProcessError,
+    TimeoutError,
+    as_completed,
+)
+
+__all__ = [
+    "Deadpool",
+    "Future",
+    "CancelledError",
+    "TimeoutError",
+    "ProcessError",
+    "PoolClosed",
+    "as_completed",
+]
+
+
+def __getattr__(name: str):
+    """Preserve access to implementation helpers exposed by the old module."""
+    try:
+        return getattr(_pool, name)
+    except AttributeError:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}") from None
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(dir(_pool)))

deadpool.py → deadpool/_pool.py

@@ -21,6 +21,7 @@
 
 import concurrent.futures
 import ctypes
+import itertools
 import logging
 import multiprocessing as mp
 import os
@@ -32,7 +33,6 @@
 import typing
 import weakref
 import atexit
-import json
 from concurrent.futures import CancelledError, Executor, InvalidStateError, as_completed
 from dataclasses import dataclass, field
 from multiprocessing.connection import Connection
@@ -44,7 +44,8 @@
 import psutil
 from setproctitle import setproctitle
 
-__version__ = "2026.6.1"
+from . import __version__ as __version__
+
 __all__ = [
     "Deadpool",
     "Future",
@@ -117,6 +118,7 @@ def to_dict(self) -> dict[str, typing.Any]:
 class PrioritizedItem:
     priority: int
     item: typing.Any = field(compare=False)
+    sequence: int = 0
 
 
 @dataclass(init=False)
@@ -276,6 +278,11 @@ def pid(self, value):
 
     def add_pid_callback(self, fn):
         self.pid_callback = fn
+        if self.pid is not None:
+            try:
+                fn(self)
+            except Exception:  # pragma: no cover
+                logger.exception("Error calling pid_callback")
 
     def cancel_and_kill_if_running(self, sig=signal.SIGKILL):
         self.cancel()
@@ -376,6 +383,7 @@ def __init__(
             malloc_trim_rss_memory_threshold_bytes
         )
         self._statistics = Statistics()
+        self._submission_sequence = itertools.count()
 
         # TODO: overcommit
         self.workers: SimpleQueue[WorkerProcess] = SimpleQueue()
@@ -558,7 +566,20 @@ def run_task(self, fn, args, kwargs, timeout, fut: Future):
                 retry_count -= 1
                 worker: WorkerProcess = self.get_process()
                 try:
-                    worker.submit_job((fn, args, kwargs, timeout))
+                    before_submit = getattr(fut, "_before_submit", None)
+                    if before_submit is not None and not before_submit(worker.pid):
+                        self.done_with_process(worker)
+                        return
+                    try:
+                        worker.submit_job((fn, args, kwargs, timeout))
+                    except BaseException:
+                        submit_failed = getattr(fut, "_submit_failed", None)
+                        if submit_failed is not None:
+                            submit_failed(worker.pid)
+                        raise
+                    after_submit = getattr(fut, "_after_submit", None)
+                    if after_submit is not None:
+                        after_submit(worker.pid)
                     break
                 except (pickle.PicklingError, AttributeError) as e:
                     # If the user passed in a function or params that can't
@@ -690,11 +711,43 @@ def submit(
             raise PoolClosed("The pool is closed. No more tasks can be submitted.")
 
         fut = Future()
+        return self._submit_future(
+            fut,
+            fn,
+            args,
+            kwargs,
+            deadpool_timeout,
+            deadpool_priority,
+        )
+
+    def _submit_future(
+        self,
+        fut: Future,
+        fn: Callable,
+        args: tuple,
+        kwargs: dict,
+        timeout,
+        priority,
+        *,
+        block: bool = True,
+    ) -> Future:
+        """Submit using a framework-owned Future.
+
+        Remote brokerage uses this narrow private seam to arbitrate queued
+        cancellation against worker dispatch. Its broker lane may request
+        nonblocking admission so remote deadlines remain live when the local
+        backlog is full. Ordinary :meth:`submit` calls retain blocking
+        backpressure.
+        """
+        if self.closed:
+            raise PoolClosed("The pool is closed. No more tasks can be submitted.")
         self.submitted_jobs.put(
             PrioritizedItem(
-                priority=deadpool_priority,
-                item=(fn, args, kwargs, deadpool_timeout, fut),
-            )
+                priority=priority,
+                item=(fn, args, kwargs, timeout, fut),
+                sequence=next(self._submission_sequence),
+            ),
+            block=block,
         )
         self._statistics.tasks_received.increment()
         return fut
@@ -720,7 +773,11 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False) -> None:
 
         try:
             self.submitted_jobs.put(
-                PrioritizedItem(priority=shutdown_priority, item=None),
+                PrioritizedItem(
+                    priority=shutdown_priority,
+                    item=None,
+                    sequence=next(self._submission_sequence),
+                ),
                 timeout=2.0,
             )
         except TimeoutError:  # pragma: no cover

deadpool/remote/__init__.py

@@ -0,0 +1,95 @@
+"""Experimental socket-based access to a server-owned Deadpool.
+
+Security warning
+----------------
+The default PickleSerializer requires mutually trusted clients and servers in
+both callable and registered-task modes. TLS authenticates peers but does not
+sandbox deserialization. Registered operation authorization is not a
+serialization sandbox because workers still unpickle invocation arguments;
+clients likewise unpickle result and exception payloads, creating reverse trust.
+
+The current wire layout is private and versioned but not a stable interoperability
+contract. Session resumption, compression, IPv6, out-of-band pickle buffers,
+and durable request state are intentionally not negotiated by this release.
+"""
+
+from ._future import RemoteFuture, SubmissionState
+from .client import DeadpoolClient
+from .config import (
+    Authenticator,
+    Authorizer,
+    PeerInfo,
+    Principal,
+    RemoteLimits,
+    TcpAddress,
+    TcpListener,
+    UnixAddress,
+    UnixListener,
+)
+from .errors import (
+    AcceptanceCertainty,
+    ExecutionCertainty,
+    RemoteAuthenticationError,
+    RemoteCancellationOutcomeUnknown,
+    RemoteCancelledError,
+    RemoteCompatibilityError,
+    RemoteConnectionLost,
+    RemoteExecutorError,
+    RemoteExecutorUnavailable,
+    RemoteForkedProcessError,
+    RemoteProcessError,
+    RemoteProtocolError,
+    RemoteQueueFull,
+    RemoteQueueTimeout,
+    RemoteResultEncodingError,
+    RemoteResultLost,
+    RemoteResultTooLarge,
+    RemoteServerRestarted,
+    RemoteSubmissionTimeout,
+    RemoteTaskError,
+    SubmissionOutcomeUnknown,
+)
+from .serializer import PickleSerializer, SerializationLimitError, Serializer
+from .server import DeadpoolServer, RequestState, ServerState
+
+__all__ = [
+    "DeadpoolClient",
+    "DeadpoolServer",
+    "RemoteFuture",
+    "SubmissionState",
+    "RequestState",
+    "ServerState",
+    "UnixAddress",
+    "TcpAddress",
+    "UnixListener",
+    "TcpListener",
+    "RemoteLimits",
+    "PeerInfo",
+    "Principal",
+    "Authenticator",
+    "Authorizer",
+    "Serializer",
+    "PickleSerializer",
+    "SerializationLimitError",
+    "AcceptanceCertainty",
+    "ExecutionCertainty",
+    "RemoteExecutorError",
+    "RemoteExecutorUnavailable",
+    "RemoteAuthenticationError",
+    "RemoteCompatibilityError",
+    "RemoteQueueFull",
+    "RemoteSubmissionTimeout",
+    "SubmissionOutcomeUnknown",
+    "RemoteQueueTimeout",
+    "RemoteTaskError",
+    "RemoteResultEncodingError",
+    "RemoteResultTooLarge",
+    "RemoteCancellationOutcomeUnknown",
+    "RemoteForkedProcessError",
+    "RemoteProcessError",
+    "RemoteCancelledError",
+    "RemoteConnectionLost",
+    "RemoteServerRestarted",
+    "RemoteProtocolError",
+    "RemoteResultLost",
+]