Worker

A Worker executes tasks delegated by a Manager. Workers are responsible for performing blocking, CPU-intensive, or long-running operations outside of the Manager’s main thread.

A developer must explicitly choose the worker type depending on the workload.

---

Worker types

All Workers inherit from the abstract Worker base class. There are two concrete worker types:

• WorkerThread

• WorkerProcess

There is no single "default" Worker you should use without thinking. You must decide which execution model fits your task.

WorkerThread

• Runs in the same process as the Manager.

• Shares memory with the Manager.

• Subject to Python’s GIL.

• Suitable for:

  • I/O-bound tasks

  • Short-lived operations

  • Lightweight background work

Because memory is shared, you technically can share objects — but this should be done carefully.

WorkerThread receives the same event instance that was passed to workers.start(). EDRI does not copy the event, positional arguments, or keyword arguments. This is also true when max_workers is reached: the pending queue keeps a reference to the event rather than taking a snapshot.

Do not assume that mutations performed by a thread worker are isolated from the caller. If isolation is required, create a copy or, preferably, pass a dedicated event containing only the data the worker needs (for example an ORM entity ID rather than an ORM instance).

WorkerProcess

• Runs in a separate process.

• Has isolated memory.

• True parallel execution (no GIL limitation).

• Suitable for:

  • CPU-bound tasks

  • Heavy computation

  • Long-running processing

  • Large file generation

Running a WorkerProcess in a separate Python environment

workers.start accepts an optional environment argument — a Path to a Python virtual environment. When given, that WorkerProcess is launched under <environment>/bin/python instead of the host interpreter:

from pathlib import Path

self.workers.start(event, PdfRenderWorker, environment=Path("/opt/venvs/pdf"))

How it works and what to keep in mind:

• The Manager forces the spawn start method for that worker and calls set_executable(...) so the child process boots the chosen interpreter.

• It applies only to WorkerProcess. For a WorkerThread the argument is ignored — threads share the host interpreter and cannot use a different environment.

• The target environment must be able to import EDRI and the event/worker classes involved: the spawned process unpickles the event and imports your WorkerProcess subclass, so the code has to be importable there.

Use this to isolate a worker's dependencies — for example a task that needs a conflicting library version, or a heavy native dependency you do not want in the main application environment.

---

Worker lifecycle

The lifecycle of a Worker is deterministic:

1. A Manager calls worker_id = self.workers.start(event, WorkerClass).

2. A lifecycle record is inserted immediately in self.workers as PENDING.

3. When capacity is available, the runtime is created (STARTING) and receives the event as self.event.

4. After bootstrap it emits WorkerStarted, enters RUNNING, and executes setup() followed by do().

5. The Worker may:

   • call self.event_send(event)

   • emit additional events

6. When execution finishes, WorkerQuit reports whether the runtime returned, raised, or cooperatively stopped. The Manager then removes the record.

The controller is a read-only mapping. Use self.workers[worker_id], get(), items(), and the live pending, running, and stopping views for introspection. Registry mutation is intentionally not exposed.

Cancellation and timeout policy

Cancel one execution with self.workers.cancel(worker_id), or start cancellation for all active executions with self.workers.cancel_all(). Thread workers must cooperate:

while not self.cancel_requested:
    do_one_piece()
    self.raise_if_cancelled()

# Interruptible alternative to time.sleep(5)
if self.wait_cancelled(5):
    return

WorkerThread cannot be safely force-terminated. cancel() is a request to stop; a thread that ignores it remains in STOPPING and is logged as stuck. For a WorkerProcess, EDRI may internally escalate from cooperative cancellation to terminate() and then kill() after the configured grace periods.

Use an immutable policy for queue, bootstrap, execution, or total lifetime limits:

from edri.abstract import WorkerPolicy

worker_id = self.workers.start(
    event,
    ReportWorker,
    policy=WorkerPolicy(run_timeout=30, cancel_grace=5, terminate_grace=2),
)

Timeouts use a monotonic clock. run_timeout starts at WorkerStarted, so it includes both setup() and do(). ManagerBase.start_worker() remains as a temporary deprecated shim, emits DeprecationWarning, and always returns None.

---

Communication model

⚠ Important difference from Manager behavior

Inside a Worker, you must explicitly send the response event back using self.event_send(...).

Unlike Manager handlers, where setting event.response fields or status is enough (the event is sent automatically after the handler finishes), a Worker does not automatically forward the modified event.

If you modify event.response inside a Worker and do not call self.event_send(event), the Manager will never receive it.

Workers always communicate explicitly.

Workers communicate only via Events.

Sending events

self.event_send(event)

This sends the event back to the Manager, which receives it in resolve_callback_worker — the single entry point for every event emitted by any worker. By default the Manager forwards it to the Router (which then routes it to a @response handler or the waiting client); a Manager may override the callback to modify, aggregate, drop, or re-emit worker output first.

See Receiving events back from a Worker for the full behavior, the worker argument, and override examples.

---

State and isolation

Workers must not mutate Manager state directly.

• WorkerProcess is fully isolated (separate memory).

• WorkerThread shares the event and argument objects supplied to workers.start(); shared mutation should be avoided unless explicitly designed.

All coordination should happen through events.

---

SharedMemoryPipe in Workers

SharedMemoryPipe is a bounded byte channel for moving bulk binary data between a Worker and the API layer at high throughput, in either direction — large file downloads and generated archives on the way out, streamed file uploads on the way in. Payload bytes go straight into shared memory, bypassing pickling and the Manager/Router hop, and two semaphores give backpressure for free: a slow client throttles the producing Worker automatically, so you can stream a file of any size in constant memory. For the full mechanism (ring buffer layout, safety guarantees, performance characteristics), see SharedMemoryPipe.

Download: Worker → API

The pipe's writer (created by SharedMemoryPipe()) cannot be pickled and stays in the Worker. Its reader (writer.reader()) is picklable — attach it to event.response.file and send the event; the API process reattaches to the same segment by name and reads, unlinking the shared memory once exhausted.

class ReportWorker(WorkerProcess[GenerateReport]):
    def do(self) -> None:
        with SharedMemoryPipe() as smp:          # writer end
            self.event.response.file = File(
                file_name="report.csv",
                mime_type="text/csv",
                path=smp.reader(),               # ship the reader, not the writer
            )
            self.event_send(self.event)          # API can start downloading now

            for chunk in produce_rows():         # blocks here if the client is slow
                smp.write(chunk)
        # leaving the `with` block calls close(), which sends EOF to the reader

On the consuming side the reader is iterable; iteration stops automatically at EOF:

for chunk in pipe_reader:   # EOF is signalled by write(..., close=True) or close()
    sink.write(chunk)

The other direction: reading an upload as it arrives

The same pipe also carries uploads into a Worker, so you can process a file while it is still being received instead of waiting for the whole body.

This kicks in automatically for a file-only event — an @api event whose only input field (after Header/Cookie/Scope fields are excluded) is typed File:

@api(url="/import", method=HTTPMethod.POST)
class ImportCsv(Event):
    file: File
    response: ImportResponse

When such a file is uploaded over HTTP with a filename header and the client sends Expect: 100-continue, the API layer:

1. creates the SharedMemoryPipe (the API is the writer this time),

2. sets event.file = File(..., path=<reader>) and dispatches the event immediately — before the upload finishes,

3. streams the incoming HTTP body into the pipe as chunks arrive.

Your Worker therefore receives the event right away and reads the body as a stream, with the same backpressure (reading slowly throttles the HTTP upload):

class ImportWorker(WorkerProcess[ImportCsv]):
    def do(self) -> None:
        source = self.event.file.path     # a SharedMemoryPipe reader
        for chunk in source:
            process(chunk)

This is especially powerful as a pass-through to object storage: feed the chunks straight into an S3/MinIO multipart upload and the file goes client → Worker → bucket without ever being fully held in memory or staged on disk — and because backpressure propagates end to end, a slow bucket throttles the HTTP upload itself. See the worked example in Streaming an upload straight into MinIO / S3. The download direction is the mirror image: stream an object out of the bucket into the pipe as the client downloads.

Without Expect: 100-continue, the body is buffered to a temporary file first and File.path is a plain Path instead — the Worker only sees the event once the upload is complete. Handle both: stream if path is a SharedMemoryPipe, open the file if it is a Path.

Worker → Worker

The same split (unpicklable writer, picklable reader) also lets one Worker stream bulk bytes to another Worker on the same host, without routing the payload through the Router. The key constraint: a Worker cannot start another Worker directly, and the writer cannot be pickled — so the producer keeps the writer and ships the reader to the consumer through its Manager, which starts the consumer with it.

The flow is:

1. The producer Worker creates the pipe, takes pipe.reader(), and sends it to its Manager inside an event.

2. The Manager handles that event and calls self.workers.start(event, ConsumerWorker) — the reader rides along on the event.

3. The consumer Worker reads from the reader while the producer writes. Backpressure applies as usual, so they run concurrently in constant memory.

@event
class ShmHandoff(Event):
    reader: SharedMemoryPipe   # picklable; travels producer -> manager -> consumer
    # ... plus whatever metadata the consumer needs (size, checksum, ...)


class ProducerWorker(WorkerProcess[StartEvent]):
    def do(self) -> None:
        pipe = SharedMemoryPipe()
        # Hand the reader off *before* streaming, so the consumer can drain
        # concurrently while we write.
        self.event_send(ShmHandoff(reader=pipe.reader()))
        pipe.write(big_data, close=True)   # blocks as the consumer drains, then EOF


class Relay(ManagerBase):
    @request
    def handoff(self, event: ShmHandoff) -> None:
        self.workers.start(event, ConsumerWorker)


class ConsumerWorker(WorkerProcess[ShmHandoff]):
    def do(self) -> None:
        for chunk in self.event.reader:    # iterates to EOF, then unlinks
            consume(chunk)

A few things to keep in mind:

• The producer must stay alive while it streams — its do() blocks on write() until the consumer drains, which is exactly what you want. After write(..., close=True) (or close()) it may finish.

• The reader transits the Manager (and the Router) as a normal event, so it is briefly re-attached in those processes; only the consumer actually reads from it, and it unlinks the shared memory at EOF.

• All the Constraints below still apply — same host, and strictly one writer / one reader (you cannot fan a single pipe out to several consumers).

Tuning

The defaults are fine for most cases, but the constructor exposes the ring geometry:

Parameter	Default	Purpose
max_slots	64	Number of slots in the ring. More slots = more buffering and less writer/reader contention, at the cost of more shared memory.
slot_size	EDRI_CHUNK_SIZE	Size of each slot in bytes (header + payload). Acts as a lower bound when total_size is given.
total_size	None	Expected total payload size. When set, the slot size is auto-computed (and 64-byte aligned) so the whole payload fits in at most max_slots slots — fewer chunks, less semaphore traffic.

Constraints

Single host only. SharedMemoryPipe relies on POSIX shared memory, so the writer and reader must live on the same machine. It does not work across the Switch / between distributed nodes — a shareable event whose response carries a pipe reader cannot be resolved by a remote node. Keep pipe-producing Workers and the API on the same host.

• One writer, one reader — it is not a fan-out/broadcast channel.

• Range/resume downloads are not possible while streaming, because the total size is generally unknown up front.

See Downloading files and the worked example in SharedMemoryPipe: streaming a ZIP of a folder.

---

Error handling

If an exception occurs inside do():

• The exception is logged.

• The Worker terminates.

• WorkerQuit is still emitted.

Workers should handle expected errors explicitly and emit meaningful response events when necessary.

Avoid swallowing exceptions silently.

---

Generic typing

Workers can declare a type parameter:

class MyWorker(WorkerProcess[MyEvent]):
    ...

This is type-hinting only.

It is not enforced at runtime.

---

When NOT to use a Worker

Do not create a Worker when:

• The task is trivial and fast.

• The logic does not block.

• The operation can safely run inside the Manager method.

Workers are appropriate when:

• The task blocks (I/O, network, filesystem).

• The task is CPU-heavy.

• The task is long-running.

• You need isolation.

Creating unnecessary workers increases complexity and overhead.

---

Minimal example

from edri.abstract.worker import WorkerProcess

class MyWorker(WorkerProcess[MyEvent]):
    def do(self) -> None:
        # Access original event
        data = self.event.data

        # Perform work
        result = heavy_compute(data)

        # Emit result
        self.event.response.result = result
        self.event_send(self.event)

---

Workers provide controlled concurrency in EDRI. Choose the correct worker type carefully and communicate exclusively through events.