uhttp-workers 1.8.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.

A mount value may also be a dict for per-mount options:

static_routes={
    '/assets/': {
        'path': '/var/www/assets',
        'headers': {'Cache-Control': 'public, max-age=31536000, immutable'},
        'authoritative': True,
    },
}

• headers — response headers attached to every file served from this mount.
• authoritative (default False) — when True, the mount owns its prefix: a missing file or blocked traversal returns 404 immediately instead of falling through to sync handlers / worker pools. Use this when a catch-all pool (routes=['/']) would otherwise swallow /assets/missing.png and answer only after a full worker round-trip.

Override on_static_served(client, file_path, status) for an access log — it fires on every 200, and on the 404 of an authoritative mount (status is 200 or 404). Exceptions raised by the hook are logged and swallowed.

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.

Memory Limit

Pass worker_memory_limit_mb=<MB> (a recognized worker kwarg) to cap each worker's address space via RLIMIT_AS:

WorkerPool(MyWorker, num_workers=4, worker_memory_limit_mb=512)

A runaway allocation then hits ENOMEM and the worker dies cleanly — the pool restarts the slot — instead of exhausting host RAM. Defense-in-depth for handling untrusted input. The cap is applied before setup(), so it also bounds any initialization done there. POSIX only (no-op on Windows); setrlimit failures are logged, not fatal.

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.

To gate sync routes, override do_check_sync() — the sync-handler counterpart, called after a sync route matches but before its handler runs:

class MyDispatcher(_workers.Dispatcher):
    def do_check_sync(self, client):
        if client.path == '/status' and not _authorized(client):
            client.respond({'error': 'unauthorized'}, status=401)
            raise _workers.RejectRequest()

Same contract as do_check: respond and raise RejectRequest to block the handler (the request is then considered handled — no fall-through). Default no-op. Keeping the two checks separate avoids silently gating sync routes for dispatchers that only override do_check.

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.

Dispatcher Hooks

Override these on a Dispatcher subclass to extend behavior — each is detailed in its own section below. None of them need super() except where noted.

Hook	When it fires	Typical use
do_check_sync(client)	a sync route matched, before its handler	auth/validate sync routes
do_check(client)	before dispatching to a worker pool	auth/validate API requests
on_request_accepted(request_id, client, pool)	just before the request is queued	attach request.context for the worker
on_response(response, pending)	after a worker response is sent	post-process / cross-pool work
on_forward(route, data)	a worker called forward()	route the payload to a sibling pool
on_static_served(client, file_path, status)	a static mount served 200 / authoritative 404	access logging
on_pending_removed(request_id, pending, reason)	a request leaves _pending (any outcome)	side-state cleanup
on_worker_died(pool, worker_id, reason, exitcode, victims)	a worker died/was parked	fail victims, post-mortem (call super())
on_idle()	each select() timeout with no events	periodic maintenance
on_log(name, level, message)	a log message is emitted	customize log output

Request flow and where each hook sits:

static:   _serve_static ──────────────────────────────────► on_static_served
sync:     match sync route ─► do_check_sync ─► handler
dispatch: do_check ─► on_request_accepted ─► queue ─► [worker]
                                                        ├─ forward()  ─► on_forward ─► sibling pool
                                                        └─ response   ─► on_response ─► on_pending_removed
on death: ──────────────────────────────────────────────► on_worker_died

do_check_sync, do_check, and sync handlers may raise RejectRequest to stop processing (respond first). Any other exception anywhere in this flow is logged and returned as 500 — it cannot break the dispatcher loop.

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 == 'ImageWorker':
            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).

Forwarding to a Sibling Pool

When the producer of the data is the worker itself (not the dispatcher), forward straight from the handler with self.forward(route, data) instead of post-processing in the dispatcher. The worker returns its client-facing response normally and hands off a separate payload to another pool — the dispatcher stays out of the data path:

class RecognizeWorker(_workers.Worker):
    @_workers.api('/recognize', 'POST')
    def recognize(self, request):
        result = run_recognition(request.data)
        # hand the raw result + image off to the storage pool
        self.forward('/_internal/store', {
            'image': request.data,
            'result': result,
        })
        return {'plate': result['plate']}   # narrow public response to client

forward() is fire-and-forget: the dispatcher routes it (via on_forward()) to the pool matching route as a request_id=-1 Request, and that worker's response is ignored. The target pool must be registered before any catch-all routes=['/'] pool, since the first prefix match wins.

Override on_forward(route, data) to change the routing or drop policy. The default routes to the matching pool with drop-on-full (a slow sink can't backpressure the emitting worker) and logs a warning on drop — never a silent loss.

Per-Request Context

Override on_request_accepted(request_id, client, pool) to attach dispatcher-only state to a request just before it is dispatched — the return value is serialized onto request.context for the worker to read. Use for state the worker can't derive itself (e.g. an auth role resolved during do_check):

class MyDispatcher(_workers.Dispatcher):
    def on_request_accepted(self, request_id, client, pool):
        return {'role': getattr(client, '_auth_role', None)}

class MyWorker(_workers.Worker):
    @_workers.api('/thing', 'GET')
    def thing(self, request):
        if request.context.get('role') == 'admin':
            ...

It is pure state-attach — it must not reject (that is do_check). Exceptions are logged and swallowed, leaving request.context as None. The value must be picklable.

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().

By default degraded is sticky — once set it stays until restart. Pass recovery_interval=<seconds> to let the pool auto-recover: that many seconds after entering degraded, the flag and restart counter are cleared and workers get a fresh chance. Useful for transient failures that resolve on their own. The dispatcher logs a WARNING when a pool enters degraded and an INFO when it recovers.

Permanent-Failure Parking

Some failures are not transient — a missing license/dongle, bad config, or an incompatible SDK makes a worker die the same way on every restart, burning the max_restarts budget for nothing. Pass permanent_failure_exitcode=<code> and have the worker exit(code) on such a fault:

WorkerPool(MyWorker, num_workers=4, permanent_failure_exitcode=42)

A worker that dies with that exact code is parked: the slot is not restarted and does not count toward max_restarts. Parked slots drop out of pool.alive_count and are flagged in pool.status() (parked per worker, parked_count on the pool). Any requests the worker had in flight are still failed with 500 via on_worker_died (reason parked exit=N). When every slot is parked the pool goes degraded — clients get a chronic 503 (no Retry-After, since retrying won't help), and recovery_interval will not clear it (there is nothing left to retry). Default None disables parking entirely.

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).

Logger Names

Each logger is named after its class — MyWorker[0] for worker 0, MyDispatcher for the dispatcher. Override the LOG_NAME class attribute to customize:

class MyWorker(_workers.Worker):
    LOG_NAME = "api-{pool_name}-{worker_id}"   # → "api-MyWorker-0"

class MyDispatcher(_workers.Dispatcher):
    LOG_NAME = "gateway-{pid}"                  # → "gateway-12345"

LOG_NAME is a format template. Available placeholders:

Placeholder	Worker	Dispatcher	Value
{cls}	✓	✓	class name
{worker_id}	✓	—	worker index in pool
{pool_name}	✓	—	owning pool name
{pid}	✓	✓	process PID

The defaults are themselves templates — '{cls}[{worker_id}]' for workers and '{cls}' for the dispatcher — so overriding LOG_NAME gives you the whole name. The worker index is only present if your template includes {worker_id}. An invalid placeholder logs a warning and falls back to the built-in default name.

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)
recovery_interval	None	Seconds before auto-recovering from degraded (None = sticky)
permanent_failure_exitcode	None	Exit code that parks a slot instead of restarting (None = disabled)

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