# 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 `/bin/python` instead of the host interpreter: ```python 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: ```python 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: ```python 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 ```python 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 {ref}`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)= # 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 {doc}`/shared_memory_pipe`. ## 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. ```python 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: ```python 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 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`: ```python @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=)` 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): ```python 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 {doc}`/examples/worker/stream_upload_minio`. 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. ```python @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](#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 {doc}`/api_download` and the worked example in {doc}`/examples/worker/shared_memory_pipe`. --- # 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: ```python 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 ```python 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.