gisolate

Gevent has tormented me a thousand times, yet I keep coming back for more. This library is proof of that love.

Process isolation for gevent applications. Run any object in a clean subprocess, call its methods transparently via ZMQ IPC.

Why

gevent's monkey.patch_all() replaces stdlib modules globally. Some libraries (database drivers, native async frameworks, etc.) break under monkey-patching. gisolate spawns a clean child process — no monkey-patching — and proxies method calls over ZMQ, so incompatible code runs in isolation while your gevent app stays cooperative.

Install

pip install gisolate

Requires Python 3.12+.

Quick Start

ProcessProxy — persistent child process

Proxy method calls to an object living in an isolated subprocess:

import gevent.monkey
gevent.monkey.patch_all()

from gisolate import ProcessProxy

# Define a factory (must be importable / picklable)
def create_client():
    from some_native_lib import Client
    return Client(host="localhost")

# Option 1: inline
proxy = ProcessProxy.create(create_client, timeout=30)
result = proxy.query("SELECT 1")  # runs in child process
proxy.shutdown()

# Option 2: subclass
class ClientProxy(ProcessProxy):
    client_factory = staticmethod(create_client)
    timeout = 30

with ClientProxy() as proxy:
    result = proxy.query("SELECT 1")

serve / attach — one worker, many client processes

A ProcessProxy spawns a worker at a random address it never advertises, so N client processes pay for N copies of the isolated library. Where that matters (several gunicorn workers in one container, say), one process hosts the worker and the rest attach to it:

# Host process — becomes the worker; blocks.
import os
from gisolate import serve

os.makedirs("/run/myapp", exist_ok=True)
os.chmod("/run/myapp", 0o700)   # mode= on makedirs only applies when it creates
serve(create_client, "ipc:///run/myapp/client.sock")

# Every client process — no spawn, no factory.
from gisolate import ProcessProxy

proxy = ProcessProxy.attach("ipc:///run/myapp/client.sock", timeout=10)
result = proxy.query("SELECT 1")   # runs in the host
proxy.shutdown()                   # closes this client only; the host keeps serving

Put the socket in a directory you control. The wire is unauthenticated pickle and an empty frame stops the worker, so anyone who can connect can run code in the host or shut it down for every client. The socket file's own mode follows the host's umask — a world-writable one in a shared directory like /tmp is reachable by any local user — whereas a 0700 directory is not, whatever the umask. ipc:// only, and same host only: a call carries the caller's monotonic deadline, which means nothing on another machine's clock.

Attaching is asynchronous: it connects to an address nobody has bound yet just as happily, a restarted host is picked up by the DEALER's own reconnect, and an absent one surfaces as an RPC timeout. The proxy is process-local like any ZMQ socket — a forking server attaches after the fork, once per worker. serve()'s max_concurrency bounds the worker, so it is shared by every client.

run_in_subprocess — one-shot call

Run a single function in a subprocess and get the result:

from gisolate import run_in_subprocess

def heavy_compute(n):
    return sum(range(n))

result = run_in_subprocess(heavy_compute, args=(10_000_000,), timeout=60)

ProcessBridge — cross-process RPC

ZMQ-based RPC bridge for server/client architectures. Server side uses gevent, client side uses asyncio:

from gisolate import ProcessBridge

# Server (gevent side)
server = ProcessBridge("ipc:///tmp/rpc.sock", mode=ProcessBridge.Mode.SERVER)
server.start()

# Client (asyncio side)
import asyncio

async def main():
    client = ProcessBridge("ipc:///tmp/rpc.sock", mode=ProcessBridge.Mode.CLIENT)
    result = await client.call(lambda x, y: x + y, 3, 4, timeout=5)
    print(result)  # 7
    client.close()

asyncio.run(main())
server.close()

ProcessPublisher / ProcessSubscriber — one-way fan-out

ZMQ PUB/SUB for one-way data streaming (snapshots, signals, heartbeats). Use this when message loss is acceptable; use ProcessBridge when you need request/response with delivery guarantees.

Both ends take a runtime= kwarg (a PubSubRuntime enum, also accepts the strings "gevent" / "asyncio") selecting the concurrency backend:

The wire format is identical across runtimes, so a gevent publisher pairs with an asyncio subscriber (and vice versa) without any adapter.

# Producer (gevent side — default runtime)
from gisolate import ProcessPublisher

pub = ProcessPublisher("ipc:///tmp/stream.sock").start()
pub.publish("v1.snapshot.AAPL", {"price": 150.0})
pub.publish("v1.heartbeat.gevent", {"ts_ns": 1234567890})
pub.close()

# Consumer (asyncio side — default runtime)
import asyncio
from gisolate import ProcessSubscriber

async def main():
    sub = ProcessSubscriber("ipc:///tmp/stream.sock")

    async def on_snapshot(topic, payload):
        print(topic, payload)

    async def on_heartbeat(topic, payload):
        print("heartbeat", payload)

    sub.subscribe("v1.snapshot.", on_snapshot)
    sub.subscribe("v1.heartbeat.", on_heartbeat)
    sub.start()
    await asyncio.sleep(10)
    await sub.close()

asyncio.run(main())

Asyncio publisher / gevent subscriber — same wire format, just flip the runtime=:

# Producer (asyncio side)
from gisolate import ProcessPublisher, PubSubRuntime

async def producer():
    async with ProcessPublisher(addr, runtime=PubSubRuntime.ASYNC) as pub:
        await pub.publish("v1.tick.AAPL", {"price": 150.0})

# Consumer (gevent side) — handlers are sync
from gisolate import ProcessSubscriber, PubSubRuntime

def on_tick(topic, payload):  # sync, not async def
    print(topic, payload)

with ProcessSubscriber(addr, runtime=PubSubRuntime.GEVENT) as sub:
    sub.subscribe("v1.tick.", on_tick)
    gevent.sleep(10)

Notes:

• Runtime must match the host loop — start() requires a running asyncio loop in ASYNC mode and a greenlet context in GEVENT mode. Subsequent subscribe / unsubscribe / publish / close calls must stay on that same loop/hub; ZMQ sockets are not thread-safe.
• Handler signature follows the subscriber's runtime, not the publisher's — a gevent subscriber consuming from an asyncio publisher still uses sync handlers.
• Context managers — with for GEVENT, async with for ASYNC; using the wrong form raises RuntimeError. start() and close() are idempotent.
• Topic prefix matching — sub.subscribe("v1.snapshot.", h) receives every topic starting with that prefix. Multiple handlers may share a prefix; in ASYNC mode they run via asyncio.gather, in GEVENT mode each is spawned in its own greenlet. An exception in one handler is logged and does not kill the reader.
• close() from inside a handler is safe — the reader is not joined in that case (would self-deadlock); sibling handlers in the current dispatch are allowed to finish.
• Lossy by design — publish is non-blocking; messages are dropped when the send queue is full (slow subscriber). Tune via sndhwm= on the publisher.
• Late joiners miss history — PUB/SUB has no replay; a subscriber that connects after a message was published will not see it. Treat published state as a stream, not a store.
• IPC cleanup — close() unlinks the socket file for ipc:// addresses on the publisher side. Relying on __del__ is best-effort only; call close() (or use a context manager) for deterministic teardown.
• Pluggable serializer — defaults to SmartPickle (pickle, falling back to dill). Pass any object implementing the Serializer protocol (dumps/loads) to use msgpack, JSON, etc. Publisher and subscriber must agree.

ThreadLocalProxy — per-thread instances

Thread-local proxy using unpatched threading.local for true isolation in gevent.threadpool:

from gisolate import ThreadLocalProxy

proxy = ThreadLocalProxy(create_client)
proxy.query("SELECT 1")  # each real OS thread gets its own instance

Child Process Modes

# Child uses asyncio (default)
proxy = ProcessProxy.create(factory)

# Child uses gevent with selective patching
proxy = ProcessProxy.create(factory, patch_kwargs={"thread": False, "os": False})

API Reference

ProcessProxy

• ProcessProxy.create(factory, *, timeout=24, mp_context=None, patch_kwargs=None) — create a proxy without subclassing
• proxy.<method>(*args, **kwargs) — transparently call any method on the remote object
• ProcessProxy.attach(address, *, timeout=24) — proxy a worker hosted by serve(); never spawns, and shutdown() leaves the worker running
• proxy.restart_process() — kill and restart the child process; on an attached proxy there is no process to kill, so it rebuilds the local transport instead
• proxy.shutdown() — gracefully stop child process
• Supports context manager (with statement)
• Thread-safe: usable from greenlets and native threads

serve(factory, address, *, patch_kwargs=None, max_concurrency=None)

Turn this process into the worker, bound to address, for clients that ProcessProxy.attach(). Blocks until interrupted or terminated.

run_in_subprocess(target, args=(), kwargs=None, *, timeout=3600, mp_context=None)

Run a function in an isolated subprocess. Blocks with gevent-safe polling.

ProcessBridge(address, mode)

• bridge.start() — start the bridge (idempotent, returns self)
• bridge.address — IPC address
• await bridge.call(func, *args, timeout=60, **kwargs) — async RPC call (client mode)
• bridge.close() — cleanup resources

ProcessPublisher(address, *, runtime=PubSubRuntime.GEVENT, serializer=SmartPickle, sndhwm=1000)

• pub.start() — bind the PUB socket (idempotent, returns self). In ASYNC mode requires a running asyncio loop.
• pub.publish(topic, payload) — non-blocking publish; drops on slow consumers. Returns None in GEVENT mode, a coroutine in ASYNC mode (must await).
• pub.close() — cleanup (idempotent). Returns None in GEVENT mode, a coroutine in ASYNC mode.
• pub.address / pub.runtime — read-only properties.
• Context manager: with for GEVENT, async with for ASYNC. Using the wrong form raises RuntimeError.

ProcessSubscriber(address, *, runtime=PubSubRuntime.ASYNC, serializer=SmartPickle)

• sub.subscribe(topic_prefix, handler) — register a handler for a topic prefix. Handler must be sync (def) in GEVENT mode and async def (or returning an awaitable) in ASYNC mode. Safe to call before or after start().
• sub.unsubscribe(topic_prefix, handler=None) — remove a specific handler or all handlers for a prefix. When the last handler is removed, the ZMQ-level subscription is dropped.
• sub.start() — connect and spawn the reader (idempotent, returns self). In ASYNC mode requires a running asyncio loop; in GEVENT mode must be called from a greenlet context.
• sub.close() — tear down the socket and join the reader (idempotent). Returns None in GEVENT mode, a coroutine in ASYNC mode. Safe to call from inside a handler — the reader is not joined in that case to avoid self-deadlock.
• sub.address / sub.runtime — read-only properties.
• Context manager: with for GEVENT, async with for ASYNC.

PubSubRuntime (StrEnum)

• PubSubRuntime.GEVENT ("gevent") — bind to the gevent hub; sync APIs and sync handlers.
• PubSubRuntime.ASYNC ("asyncio") — bind to the running asyncio loop; awaitable APIs and async handlers.

Serializer (Protocol)

Anything with dumps(obj) -> bytes and loads(bytes) -> obj static methods can be used as a serializer for ProcessPublisher / ProcessSubscriber. Default is SmartPickle (pickle, falling back to dill). Publisher and subscriber must agree on the serializer.

ThreadLocalProxy(factory)

Transparent proxy delegating attribute access to a per-thread instance.

ensure_hub_started()

Pre-start the internal gevent hub loop on demand. Idempotent and thread-safe. Called automatically by ProcessProxy, but can be invoked explicitly to control initialization timing.

spawn_on_main_hub(func, *args, **kwargs)

Schedule a function on the main gevent hub without waiting. Thread-safe, fire-and-forget.

ProcessError

Raised when a child process dies or communication fails.

RemoteError

Wrapper for exceptions from the child process that can't be pickled. Preserves the original exception type name and message.

shutdown_hub()

Explicitly stop the internal gevent hub loop. Registered via atexit automatically.

set_default_mp_context(ctx) / get_default_mp_context()

Configure the default multiprocessing context for all proxies (default: "spawn").

Note on multiprocessing and __main__

multiprocessing spawn/forkserver children re-import the caller's __main__ module. If your main.py has top-level side effects (e.g. gevent.monkey.patch_all()), these will re-execute in the child — causing double-patching warnings or import errors.

Best practice: guard monkey-patching behind __name__ and defer heavy imports:

# main.py
if __name__ == "__main__":
    import gevent.monkey
    gevent.monkey.patch_all()

    import my_app
    my_app.run()

Spawn children re-import main.py but skip the __name__ block, avoiding side effects.

License

MIT