Worker API

run_worker, the thread and prefork pools, graceful shutdown, and worker introspection.

A worker is the process that claims jobs and executes their handlers. Producers and workers share the same Queue object — you enqueue from one process and run run_worker() in another, both pointed at the same storage.

python

queue.run_worker(queues=["default", "emails"])   # blocks until interrupted

`queue.run_worker()`

python

queue.run_worker(
    queues: Sequence[str] | None = None,
    tags: list[str] | None = None,
    pool: str = "thread",
    app: str | None = None,
    mesh: MeshWorker | None = None,
) -> None

Blocks until interrupted. Before dispatching it registers periodic tasks, initializes worker resources, starts the resource health checker, and installs signal handlers.

Parameter	Type	Default	Description
`queues`	`Sequence[str] \| None`	`None`	Queue names to consume. `None` consumes the `default` queue.
`tags`	`list[str] \| None`	`None`	Tags for worker specialization / routing.
`pool`	`str`	`"thread"`	`"thread"` or `"prefork"`.
`app`	`str \| None`	`None`	Import path to the `Queue` (e.g. `"myapp:queue"`). Required when `pool="prefork"`.
`mesh`	`MeshWorker \| None`	`None`	Gossip-based discovery, affinity, and work stealing. Not in the PyPI wheel — see Mesh.

arun_worker() is the async twin — it runs the same blocking loop in a thread executor so it does not block the event loop, and takes the same arguments.

Pools

Pool	How it runs tasks	Use for
`"thread"` (default)	Worker threads in one process, sharing a GIL.	I/O-bound work — HTTP calls, database queries, file transfer.
`"prefork"`	Child processes with independent GILs.	CPU-bound work that needs real parallelism.

python

queue.run_worker(pool="prefork", app="myapp:queue")

app is how a prefork child re-imports your queue and its task registry, which is why it is mandatory there. Prefork is not supported on Windows — run_worker raises NotImplementedError rather than silently degrading.

Async tasks are driven concurrently within the pool, bounded by the queue's async_concurrency (default 100) rather than by workers.

Sizing

Concurrency per process comes from the Queue constructor, not from run_worker:

python

queue = Queue(workers=8)          # 0 (default) auto-detects the CPU count

See Concurrency for how this interacts with cluster-wide task caps and per-queue limits.

Shutdown

SIGINT and SIGTERM start a graceful drain: the scheduler stops dispatching new work and run_worker() returns once running jobs finish, bounded by the queue's drain_timeout (default 30 seconds).

python

queue = Queue(drain_timeout=60)

To trigger the same drain from code:

python

queue.shutdown()

shutdown() is non-blocking and safe from any thread — it is the programmatic equivalent of the signal, useful from a health endpoint or a supervisor thread. It is also safe to call while a worker is still starting: the request is held until the runs it was aimed at have finished, rather than discarded. The other side of that is a call made while nothing is running — the request stays armed, and the next run_worker() on that queue returns without processing anything.

Set drain_timeout above your longest task's timeout, or shutdown stops waiting on jobs that would have finished. drain_timeout bounds how long the drain waits — on the thread pool it does not interrupt a task that is still running, so a job left unsettled is re-claimed and re-run by another worker once the scheduler reaps it. The work is not lost, but it is repeated.

From the CLI

bash

flexiq worker --app myapp:queue
flexiq worker --app myapp:queue --queues emails,default --pool prefork
flexiq worker --app myapp:queue --drain-timeout 60

The CLI resolves --app, applies --drain-timeout if given, and calls run_worker() — identical behaviour to running it yourself. See the CLI reference.

Introspection

python

for worker in queue.workers():
    print(worker["worker_id"], worker["status"], worker["queues"])

Each entry describes a registered worker:

Key	Type	Description
`worker_id`	`str`	Unique worker id.
`hostname`	`str`	OS hostname.
`pid`	`int`	Process id.
`status`	`str`	`"active"` or `"draining"`.
`pool_type`	`str`	`"thread"`, `"prefork"`, or `"native-async"`.
`started_at`	`int`	Registration time (ms since epoch).
`last_heartbeat`	`int`	Last heartbeat (ms since epoch).
`queues`	`str`	Comma-separated queue names.
`threads`	`int`	Worker thread / process count.
`tags`	`str \| None`	Specialization tags.
`resources`	`str \| None`	Registered resource names (JSON).
`resource_health`	`str \| None`	Per-resource health (JSON).
`registry_fingerprint`	`str \| None`	Fingerprint of the tasks this worker has handlers for. Workers that disagree on it run different task sets.

aworkers() is the async variant. A worker whose last_heartbeat has gone stale has died without unregistering — that is the signal to look for when jobs appear stuck in running.

Attached executor

An attached executor runs tasks without opening storage: it connects to a scheduler that already has it, and the scheduler dispatches over that channel. flexiq executor --app myapp:queue --attach … builds one; the Executor object it drives is the same one the declarations list.

Constructing an Executor performs the handshake, so a bad token or an unreachable scheduler raises there rather than after the pool has been built. Jobs then run on the prefork pool, one child per slot.

Member	Description
`executor_id`	The identity this executor attached under. Either the one you asked for or a generated one.
`scheduler_id`	The identity the scheduler announced when it accepted the attach — the answer to "which scheduler did I actually reach".
`peer`	Peer label of the scheduler connection, for logs and diagnostics.
`is_running()`	Whether the session is still open. `False` after the scheduler drops it, which is how a supervisor notices without polling the socket.
`wait(timeout_ms)`	Block up to `timeout_ms`, returning whether the session has ended. Loop on this rather than blocking indefinitely — each return hands the GIL back, and that is the only moment a pending Python signal handler can run.
`stop()` / `shutdown()`	Stop accepting work and finish what is in flight (returns at once), versus drain, disconnect and join (idempotent).

The attach token comes from FLEXIQ_ATTACH_TOKEN, never a flag: a token in argv is visible in ps output and lands in shell history.

ConcurrencyPool size, cluster-wide caps, per-worker fairness, admission control.Prefork poolTrue parallelism for CPU-bound tasks, and its constraints.Queue — workersThe full worker-related Queue surface, including hooks and breakers.