uhttp-workers 1.7.0

Multi-process worker dispatcher built on uhttp-server

uhttp-workers

Multi-process worker dispatcher built on uhttp-server.

Single dispatcher process handles all connections, N worker processes handle business logic in parallel. Communication via multiprocessing.Queue with efficient select() integration (POSIX only).

Architecture

                                    ┌─────────────┐
                                    │  Worker 0   │
                                    │  Worker 1   │
                 request_queue A    │  Worker 2   │
              ┌────────────────►    │  Worker 3   │  ComputeWorker
              │                     └──────┬──────┘
              │                            │
┌─────────────┤                            │
│             │                     ┌──────┴──────┐
│ Dispatcher  │  request_queue B    │  Worker 0   │
│  (main)     ├────────────────►    │  Worker 1   │  StorageWorker
│             │                     └──────┬──────┘
│  - static   │                            │
│  - sync     │    response_queue          │
│  - auth     │◄───────────────────────────┘
│             │    (shared)
└─────────────┘

Key design decisions:

• Sockets never leave the dispatcher — only serializable data goes through queues
• Each worker pool has its own request queue, all pools share one response queue
• Workers send heartbeats via response queue — dispatcher detects stuck/dead workers
• Each worker has a private control queue for config updates and stop signals

Installation

pip install uhttp-workers

Quick Start

import uhttp.workers as _workers


class ItemWorker(_workers.Worker):
    def setup(self):
        self.items = {}
        self.next_id = 1

    @_workers.api('/api/items', 'GET')
    def list_items(self, request):
        return {'items': list(self.items.values())}

    @_workers.api('/api/item/{id:int}', 'GET')
    def get_item(self, request):
        item = self.items.get(request.path_params['id'])
        if not item:
            return {'error': 'Not found'}, 404
        return item


class MyDispatcher(_workers.Dispatcher):
    def do_check(self, client):
        api_key = client.headers.get('x-api-key')
        if api_key not in VALID_KEYS:
            client.respond({'error': 'unauthorized'}, status=401)
            raise _workers.RejectRequest()

    @_workers.sync('/health')
    def health(self, client, path_params):
        client.respond({'status': 'ok'})


def main():
    dispatcher = MyDispatcher(
        port=8080,
        pools=[
            _workers.WorkerPool(
                ItemWorker, num_workers=4,
                routes=['/api/**'],
                timeout=30,
            ),
        ],
    )
    dispatcher.run()


if __name__ == '__main__':
    main()

Multiple Worker Pools

Route different endpoints to different worker pools with independent scaling:

dispatcher = MyDispatcher(
    port=8080,
    pools=[
        _workers.WorkerPool(
            ComputeWorker, num_workers=4,
            routes=['/api/compute/**'],
            timeout=60,
        ),
        _workers.WorkerPool(
            StorageWorker, num_workers=2,
            routes=['/api/items/**', '/api/item/**'],
            timeout=10,
        ),
        _workers.WorkerPool(
            GeneralWorker, num_workers=1,
        ),  # no routes = default/fallback pool
    ],
)

Request routing order:

1. Static files — served directly by dispatcher
2. Sync handlers — run in dispatcher process
3. Worker pools — first pool with matching route prefix, or fallback pool
4. 404 — no match

API Handlers

Group related endpoints under a common URL prefix using ApiHandler:

import uhttp.workers as _workers

class UserHandler(_workers.ApiHandler):
    PATTERN = '/api/user'

    @_workers.api('', 'GET')
    def list_users(self, request):
        return {'users': self.worker.db.list_users()}

    @_workers.api('/{id:int}', 'GET')
    def get_user(self, request):
        return self.worker.db.get_user(request.path_params['id'])

    @_workers.api('/{id:int}', 'DELETE')
    def delete_user(self, request):
        self.worker.db.delete_user(request.path_params['id'])
        return {'deleted': request.path_params['id']}

class OrderHandler(_workers.ApiHandler):
    PATTERN = '/api/order'

    @_workers.api('', 'GET')
    def list_orders(self, request):
        return {'orders': []}

    @_workers.api('/{id:int}', 'GET')
    def get_order(self, request):
        return {'id': request.path_params['id']}

class MyWorker(_workers.Worker):
    HANDLERS = [UserHandler, OrderHandler]

    def setup(self):
        self.db = Database(self.kwargs['db_login'])

@api patterns on handlers are automatically prefixed with the handler's PATTERN. Handlers access the worker instance via self.worker.

You can also define @api methods directly on the worker class — useful for simple workers that don't need handler grouping.

Handlers support inheritance — a subclass inherits all routes from its parent, using the subclass PATTERN as prefix.

Static Files

dispatcher = Dispatcher(
    port=8080,
    static_routes={
        '/static/': './static/',
        '/images/': '/var/data/images/',
    },
)

Static files are served directly by the dispatcher process with path traversal protection. Directory requests automatically serve index.html if present.

Sync Handlers

Lightweight handlers that run directly in the dispatcher process — no queue overhead. Define them as methods on the dispatcher class with the @sync decorator:

import uhttp.workers as _workers

class MyDispatcher(_workers.Dispatcher):
    @_workers.sync('/health')
    def health(self, client, path_params):
        client.respond({
            'status': 'ok',
            'pools': [pool.status() for pool in self._pools],
        })

    @_workers.sync('/version')
    def version(self, client, path_params):
        client.respond({'version': '1.0.0'})

Use sync handlers for fast, non-blocking responses only — long operations block the entire dispatcher.

Worker Lifecycle

Setup

setup() is called once when a worker process starts. Use it to initialize resources that cannot be shared across processes (database connections, models, etc.):

class MyWorker(_workers.Worker):
    def setup(self):
        self.db = Database(self.kwargs['db_login'])

Extra keyword arguments from WorkerPool(...) are available as self.kwargs.

Teardown

teardown() is called once when the worker process is stopping — use it to release resources:

class MyWorker(_workers.Worker):
    def setup(self):
        self.db = connect_db()

    def teardown(self):
        self.db.close()

Called after the run loop exits (clean stop, orphan detection, or pipe close), before the process terminates. Exceptions are logged but do not block shutdown.

Configuration Updates

Dispatcher can send configuration to workers at runtime via per-worker control queues:

# dispatcher side
for pool in dispatcher._pools:
    pool.send_config({'rate_limit': 100})

# worker side
class MyWorker(_workers.Worker):
    def on_config(self, config):
        self.rate_limit = config['rate_limit']

Health Monitoring

Workers send heartbeats automatically via the shared response queue. When a worker takes a request, it reports which request_id it is working on. If a worker stops responding:

• Dead worker (segfault, crash) — detected via is_alive(), restarted immediately
• Stuck worker (infinite loop in C code) — detected via heartbeat timeout, killed and restarted
• Too many restarts — pool marked as degraded, returns 503

Request Handling

@_workers.api('/process/{id:int}', 'POST')
def process(self, request):
    # request.request_id     — internal ID for dispatcher pairing
    # request.method          — 'POST'
    # request.path            — '/process/42'
    # request.path_params     — {'id': 42}
    # request.query           — {'page': '1'} or None
    # request.data            — dict (JSON), bytes (binary), or None
    # request.headers         — dict
    # request.cookies         — dict (lazy-parsed from Cookie header)
    # request.content_type    — 'application/json'
    # request.remote_address  — 'host:port' string (honors X-Forwarded-For
    #                            from trusted proxies; configured on the
    #                            HTTP server side)

    # return data (status 200)
    return {'result': 'ok'}

    # return data with status
    return {'error': 'not found'}, 404

    # return data with status and headers
    return {'ok': True}, 200, {'X-Custom': 'value'}

    # return Response object — full control (headers, cookies)
    return _workers.Response(
        None,  # request_id is set automatically
        data={'ok': True},
        headers={'X-Custom': 'value'},
        cookies={'session': 'abc123'})

    # defer response — worker continues accepting requests
    return _workers.DEFERRED

request.respond() (for deferred responses) also accepts headers and cookies:

request.respond(
    data={'result': 'done'},
    headers={'X-Job-Id': '42'},
    cookies={'session': 'abc'})

Deferred Responses

Return DEFERRED to skip immediate response. The worker stays in the select loop, accepts new requests, and sends the response later via request.respond():

class MyWorker(_workers.Worker):
    def setup(self):
        self._jobs = {}

    @_workers.api('/process', 'POST')
    def process(self, request):
        job_id = start_background_work(request.data)
        self._jobs[job_id] = request
        return _workers.DEFERRED

    def on_work_done(self, job_id, result):
        request = self._jobs.pop(job_id)
        request.respond(data={'result': result})

Note: deferred requests are still subject to dispatcher timeout — call self.keep_alive() periodically to prevent 504.

Keep Alive

Call self.keep_alive() during long operations to reset both the request timeout and stuck worker detection:

@_workers.api('/export', 'POST')
def export(self, request):
    for chunk in generate_large_export():
        self.keep_alive()
    return {'status': 'done'}

Server-Sent Events (SSE)

Stream events to clients using the same API as uhttp.server.Client:

class MyWorker(_workers.Worker):
    def setup(self):
        self._subscribers = {}

    @_workers.api('/events', 'GET')
    def events(self, request):
        request.response_stream()  # sends headers, keeps connection open
        self._subscribers[request.request_id] = request
        return _workers.DEFERRED

    def notify(self, data):
        for req in self._subscribers.values():
            req.send_event(data=data, event='update')

    def on_disconnect(self, request_id):
        self._subscribers.pop(request_id, None)

Available streaming methods on Request:

Method	Description
response_stream(content_type, headers, cookies)	Start streaming (default: text/event-stream)
send_event(data, event, event_id, retry)	Send SSE event
send_chunk(data)	Send raw data chunk
response_stream_end()	End stream and close connection

Streaming requests are excluded from dispatcher timeout expiration. When the client disconnects, the dispatcher notifies the worker via on_disconnect(request_id).

NDJSON Streaming

Stream JSON objects line-by-line (application/x-ndjson) — one JSON value per line, terminated by \n. Useful for incremental APIs that aren't event-shaped (long lists, log tails, progress updates):

class MyWorker(_workers.Worker):
    @_workers.api('/devices/scan', 'GET')
    def scan(self, request):
        request.response_ndjson()
        for device in self.discover_devices():
            request.send_ndjson({'id': device.id, 'name': device.name})
        request.response_stream_end()
        return _workers.DEFERRED

NDJSON methods on Request:

Method	Description
response_ndjson(headers, cookies)	Start NDJSON stream (wrapper over response_stream with application/x-ndjson)
send_ndjson(obj)	Send one JSON-serializable value as a line
response_stream_end()	End stream and close connection (shared with SSE)

Same lifecycle as SSE: excluded from timeout expiration, client disconnect triggers on_disconnect(request_id).

Flow Control

Workers can stop accepting new requests when busy. Requests stay in the shared pool queue for other workers to pick up:

class MyWorker(_workers.Worker):
    @_workers.api('/events', 'GET')
    def subscribe(self, request):
        request.response_stream()
        self._subscribers[request.request_id] = request
        if len(self._subscribers) >= 100:
            self.pause()
        return _workers.DEFERRED

    def on_disconnect(self, request_id):
        self._subscribers.pop(request_id, None)
        if not self._accepting and len(self._subscribers) < 100:
            self.resume()

pause() excludes the request queue from select() — the worker continues processing control messages, custom fd events, and on_idle(). resume() re-enables request acceptance.

URL Patterns

Dispatcher uses prefix matching to route requests to pools:

_workers.WorkerPool(MyWorker, routes=['/api/users/**'])  # matches /api/users/anything
_workers.WorkerPool(MyWorker, routes=['/api/status'])     # exact match only
_workers.WorkerPool(MyWorker)                              # fallback — catches everything else

Workers use full pattern matching with type conversion:

@_workers.api('/user/{id:int}', 'GET')        # id converted to int
@_workers.api('/price/{amount:float}', 'GET') # amount converted to float
@_workers.api('/tag/{name}')                   # name as str, all methods

Authentication

Override do_check() on the dispatcher — runs before any request is queued:

class MyDispatcher(_workers.Dispatcher):
    def __init__(self, valid_keys, **kwargs):
        super().__init__(**kwargs)
        self.valid_keys = valid_keys

    def do_check(self, client):
        api_key = client.headers.get('x-api-key')
        if api_key not in self.valid_keys:
            client.respond({'error': 'unauthorized'}, status=401)
            raise _workers.RejectRequest()

do_check() is only called for requests going to worker pools — static files and sync handlers are not affected.

Worker-Level Validation

Override do_check() on the worker — runs before routing to handler:

class MyWorker(_workers.Worker):
    def do_check(self, request):
        token = request.cookies.get('session')
        if not token:
            return {'error': 'unauthorized'}, 401

Return (data, status) tuple to reject, or None to continue. You can also raise RejectRequest:

    def do_check(self, request):
        if not self.validate_token(request.cookies.get('session')):
            raise _workers.RejectRequest(
                data={'error': 'forbidden'}, status=403)

RejectRequest accepts optional data (default: {'error': 'Rejected'}) and status (default: 403).

RejectRequest can also be raised from request handlers — useful for access control within individual endpoints:

    @_workers.api('/admin/users', 'GET')
    def admin_users(self, request):
        if not self.is_admin(request):
            raise _workers.RejectRequest(
                data={'error': 'admin only'}, status=403)
        return {'users': self.list_users()}

Error Handling

Override on_request_error() on the worker to customize error handling when a request handler raises an exception:

class MyWorker(_workers.Worker):
    def on_request_error(self, request, err):
        if isinstance(err, DatabaseError):
            self.db.reconnect()
        return super().on_request_error(request, err)

Default behavior logs the error with traceback and returns 500.

Post-Response Hook

Override on_response() on the dispatcher to post-process after a response is sent to the client — e.g., forward data to another worker pool:

class MyDispatcher(_workers.Dispatcher):
    def on_response(self, response, pending):
        if response.status == 200 and pending.pool.name == 'LprWorker':
            storage_pool = self._find_pool('/internal/storage')
            storage_pool.request_queue.put(_workers.Request(
                request_id=-1,
                method='POST',
                path='/internal/storage/save',
                data={
                    'image': pending.client.data,
                    'result': response.data,
                }))

pending is a _PendingRequest with client (original connection) and pool (source pool). Requests with request_id=-1 are ignored by the dispatcher when the worker responds.

on_response() fires only for the happy path (handler returned a response). For lifecycle cleanup that must run regardless of outcome — timeouts, client disconnects, shutdown — use on_pending_removed() (see below).

Request Lifecycle Hook

Override on_pending_removed(request_id, pending, reason) on the dispatcher when you keep side-state keyed by request_id and need it cleaned up exactly once, no matter how the request ended:

class MyDispatcher(_workers.Dispatcher):
    def on_pending_removed(self, request_id, pending, reason):
        self._side_state.pop(request_id, None)
        if reason == _workers.PENDING_TIMEOUT:
            self._metrics.timeouts += 1

Reason is one of:

Constant	When
PENDING_COMPLETED	Handler returned a response, client got it. on_response() runs first.
PENDING_TIMEOUT	Request exceeded pool.timeout; client got 504. Worker may still be processing.
PENDING_DISCONNECTED	Client disconnected mid-stream; worker was notified via control queue (race possible).
PENDING_STREAM_CLOSED	Worker ended the SSE stream cleanly.
PENDING_SHUTDOWN	Dispatcher is shutting down; client got 503.
PENDING_WORKER_DIED	Worker process died/was killed while owning the request; client got 500. on_worker_died() runs first.

The hook is invoked after the client-facing action (respond / disconnect / control queue put) so dispatcher state is finalized when it runs. Exceptions raised by the hook are logged at LOG_ERROR and swallowed — they will not crash the dispatcher loop.

Override on_response() if you only care about the happy path (e.g. cross-pool forwarding). Override on_pending_removed() if you need exactly-once cleanup. Overriding both is allowed but discouraged — for the PENDING_COMPLETED reason, on_response() is called immediately before on_pending_removed().

Worker Death Hook

Workers die — segfault in a C extension, OOM-kill, or the dispatcher kills them after stuck_timeout. Override on_worker_died() to capture which requests they had in-flight (useful for forensics when a malformed payload reproduces a crash):

class MyDispatcher(_workers.Dispatcher):
    def on_worker_died(
            self, pool, worker_id, reason, exitcode, victims):
        # `victims` is a list of (request_id, _PendingRequest) for all
        # requests this worker had claimed but not completed.
        # `exitcode` is None for stuck workers, otherwise the process exit
        # code (negative = signal: -9 OOM, -11 SIGSEGV).
        for rid, pending in victims:
            c = pending.client
            self._crash_queue.append({
                'reason': reason,
                'exitcode': exitcode,
                'method': c.method,
                'path': c.path,
                'address': c.address,
                'body': c.body,  # raw bytes — replay this to reproduce
            })
        # Default impl responds 500 to victims and fires
        # on_pending_removed(PENDING_WORKER_DIED). Call it after capture.
        super().on_worker_died(
            pool, worker_id, reason, exitcode, victims)

What's a victim: any request the worker had claimed via MSG_HEARTBEAT (pending.worker_id == worker_id). Requests still in the queue (worker_id is None) are not victims — other workers in the pool will pick them up after restart.

Default behavior (if you don't override) is to log the death + each victim, respond 500 (or close the stream for SSE/NDJSON), and fire on_pending_removed for each. Override only if you want to persist payloads or customize the response status/body.

500 vs 503: a victim of a crashed worker gets 500 (processing started, then the server failed). A new request arriving while the pool has zero alive workers gets 503 + Retry-After: 1 (rejected before processing — try again shortly). A request to a pool that has exceeded max_restarts in restart_window gets 503 permanently (pool.is_degraded). pool.alive_count is exposed for monitoring and also appears in pool.status().

Dispatcher Idle Hook

Override on_idle() on the dispatcher for periodic background tasks — called on each select() timeout (every SELECT_TIMEOUT seconds, default 1s):

class MyDispatcher(_workers.Dispatcher):
    def on_idle(self):
        # periodic cleanup, monitoring, etc.
        pass

Workers have their own on_idle() hook, called on each heartbeat_interval timeout.

Graceful Shutdown

On SIGTERM or SIGINT:

1. Stop accepting new connections
2. Wait for pending responses (up to shutdown_timeout)
3. Respond 503 to remaining pending requests
4. Send stop signal to all workers via control queues
5. Wait for workers to finish, kill after timeout

Monitoring

class MyDispatcher(_workers.Dispatcher):
    @_workers.sync('/monitor')
    def monitor(self, client, path_params):
        client.respond({
            'pools': [pool.status() for pool in self._pools],
            'pending': len(self._pending),
        })

Pool status includes per-worker info: alive, last seen, current request ID, queue size.

Logging

Workers have a built-in Logger accessible via self.log:

class MyWorker(_workers.Worker):
    @_workers.api('/item/{id:int}', 'GET')
    def get_item(self, request):
        item_id = request.path_params['id']
        # %-style (Python logging compatible)
        self.log.info("Getting item %d", item_id)
        # {}-style (kwargs)
        self.log.info("Getting item {id}", id=item_id)
        return {'id': item_id}

Log messages are sent to the dispatcher via the shared response queue and printed in the dispatcher process — no interleaved output from multiple processes.

The dispatcher itself also has a self.log Logger that writes directly via on_log() (no queue), so it can be used at any time — including before workers are started:

class MyDispatcher(_workers.Dispatcher):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.log.info("Dispatcher starting on port %d", self._port)

    def on_idle(self):
        self.log.debug("idle tick")

Set dispatcher log level via constructor: Dispatcher(log_level=LOG_DEBUG, ...) (default LOG_INFO).

Log levels: LOG_DEBUG (10), LOG_INFO (20), LOG_WARNING (30), LOG_ERROR (40), LOG_CRITICAL (50)

Check current level with is_* properties to skip expensive formatting:

if self.log.is_debug:
    self.log.debug("Details: %s", expensive_computation())

Available: is_debug, is_info, is_warning, is_error.

Set minimum level per pool:

_workers.WorkerPool(
    MyWorker, num_workers=4,
    log_level=_workers.LOG_INFO,  # default: LOG_WARNING
)

Output format — auto-detected at dispatcher init:

• Terminal: ANSI colors — bold red (critical), red (error), yellow (warning), dim (debug)
• systemd: Syslog priority prefixes (<3>, <4>, etc.) — journalctl colors by priority

Override Dispatcher.on_log(name, level, message) to customize output or forward to a logging framework.

Errors are logged automatically:

• Handler exceptions → ERROR with full traceback (returns 500 to client)
• setup() crash → CRITICAL with traceback (worker exits and restarts)
• Worker restart → ERROR with reason (died/stuck)
• Request timeout → WARNING with request ID and timeout value

Configuration

Dispatcher

Parameter	Default	Description
port	8080	Listen port
address	'0.0.0.0'	Listen address
pools	[]	List of WorkerPool instances
static_routes	{}	URL prefix → filesystem path
shutdown_timeout	10	Seconds to wait on shutdown
max_pending	1000	Max pending requests (503 when exceeded)
ssl_context	None	ssl.SSLContext for HTTPS

WorkerPool

Parameter	Default	Description
worker_class	—	Worker subclass
num_workers	1	Number of worker processes
routes	None	Prefix patterns (None = fallback pool)
timeout	30	Request timeout in seconds (504)
stuck_timeout	60	Heartbeat timeout before kill
heartbeat_interval	1	Seconds between worker heartbeats
log_level	LOG_WARNING	Minimum log level for worker loggers
max_restarts	10	Max restarts per restart_window
restart_window	300	Time window for restart counting (seconds)
queue_warning	100	Log warning when queue size exceeds this (0 = disable)

Extra **kwargs on WorkerPool are passed to worker constructor (accessible as self.kwargs).

Requirements

• Python >= 3.10
• POSIX system (Linux, macOS) — uses select() with queue._reader
• uhttp-server