# SharedMemoryPipe

```{eval-rst}
.. module:: edri.utility.shared_memory_pipe
   :no-index:
```

A bounded, single-producer / single-consumer byte channel built on **POSIX shared memory** and **POSIX named semaphores**. It moves bulk binary data between processes without pickling, copying, or routing through the event system.

---

## Problem

Normal EDRI events travel as pickled objects through `multiprocessing.Queue`, the Router, the Broker, and Managers. For hundreds of megabytes of binary data — file downloads, generated archives, streamed uploads — this path is too slow and consumes too much memory.

`SharedMemoryPipe` sidesteps all of it: payload bytes go straight into a shared memory segment. Only a small reader handle (a few integers and a name string) travels through the event system on `event.response.file` or `event.file.path`.

---

## Architecture

The shared memory region is divided into a ring of fixed-size **slots**. Each slot has a small binary header followed by a payload region.

```
┌──────────────────────────────────────────────────┐
│              SharedMemory segment                │
├─────────┬─────────┬─────────┬─────────┬─────────┤
│ Slot 0  │ Slot 1  │ Slot 2  │ Slot 3  │  ...    │
├─────────┴─────────┴─────────┴─────────┴─────────┤
│ Each slot: [ HEADER (8 B) | PAYLOAD (variable) ] │
│   HEADER = uint32 size + uint8 eof + 3B pad     │
└──────────────────────────────────────────────────┘
```

### Slot state cycle

```
  ┌──────────┐
  │  Empty   │ ← slots semaphore counts these
  └────┬─────┘
       │ writer acquires a free slot
       ▼
  ┌──────────┐
  │ Writing  │ ← writer fills header + payload
  └────┬─────┘
       │ items.release()
       ▼
  ┌──────────┐
  │  Ready   │ ← items semaphore counts these
  └────┬─────┘
       │ reader acquires a filled slot
       ▼
  ┌──────────┐
  │ Reading  │ ← reader copies payload out
  └────┬─────┘
       │ slots.release()
       ▼
  ┌──────────┐
  │  Empty   │
  └──────────┘
```

---

## Key parameters

| Parameter | Default | Purpose |
|---|---|---|
| `EDRI_CHUNK_SIZE` (`slot_size`) | 524288 (512 KiB) | Minimum slot size (including header). |
| `max_slots` | 64 | Number of slots in the ring. More slots = more buffering, less contention. |
| `total_size` | `None` | Expected total payload. When set, slot size is computed so the whole payload fits in ≤ `max_slots` slots, reducing semaphore traffic. |

Total shared memory consumed is `slot_size × max_slots`.

---

## Producer/consumer model

There are exactly two roles, and each pipe has exactly one endpoint of each role.

### Writer

Created by `SharedMemoryPipe()` — the constructor with no `name` argument. The writer:

- Creates the shared memory segment and two POSIX semaphores.
- Owns creation and **cannot be pickled**.
- Writes data in chunks via `write(data)` and signals EOF via `write(data, close=True)` or `close()`.

### Reader

Created by `writer.reader()`. The reader:

- Attaches to the existing shared memory and semaphores by name.
- **Can be pickled** — it travels through the event system as a handle.
- Reads chunks via `read()` or iteration; returns `None` / stops at EOF.
- Unlinks the shared memory when exhausted.

```python
# Writer side (stays in the producing process)
pipe = SharedMemoryPipe()                # creates SHM + semaphores
reader = pipe.reader()                   # picklable handle

# Send the reader to the consuming process via an event
event.response.file = File(path=reader)
self.event_send(event)

# Write data — blocks if the buffer is full (backpressure)
pipe.write(chunk1)
pipe.write(chunk2)
pipe.close()                             # sends EOF
```

```python
# Reader side (in the consuming process)
for chunk in reader:
    sink.write(chunk)                    # stops at EOF, unlinks SHM
```

### EOF signal

EOF is an empty slot whose header has `eof=1`. It is written by `close()` and ensures the reader observes every byte before stopping.

---

## Code example: Worker → SharedMemoryPipe → API download

The following producer runs inside a {class}`~edri.abstract.worker.WorkerProcess`. It generates data and streams it through shared memory directly to the HTTP client.

```python
from edri.abstract.worker import WorkerProcess
from edri.api.dataclass import File
from edri.utility.shared_memory_pipe import SharedMemoryPipe


class ReportWorker(WorkerProcess[GenerateReport]):
    def do(self) -> None:
        with SharedMemoryPipe() as smp:                     # writer
            self.event.response.file = File(
                file_name="report.csv",
                mime_type="text/csv",
                path=smp.reader(),                          # picklable handle
            )
            self.event_send(self.event)                     # API starts serving now

            for chunk in produce_rows():                    # blocks on backpressure
                smp.write(chunk)
```

The consuming side uses {class}`~edri.api.dataclass.file.SharedMemoryPipeFile` (or iterates the reader directly):

```python
for chunk in pipe_reader:
    response_body.write(chunk)
```

See the worked example in {doc}`/examples/worker/shared_memory_pipe`.

---

## Performance characteristics

- **Zero copy between processes** — the writer writes directly into the shared segment; the reader reads directly from it.
- **Constant memory** — the ring buffer has a fixed size regardless of payload, so a 10 GB stream uses the same memory as a 1 MB stream.
- **Automatic backpressure** — the `slots` and `items` semaphores block the producer when the consumer is slow and vice-versa.
- **No kernel involvement for data** after creation — no pipes, no sockets, no `sendmsg`/`recvmsg`.

## Safety guarantees

- **Exactly one writer, one reader.** Creating multiple readers from the same pipe raises {class}`RuntimeError`; pickling the writer raises {class}`RuntimeError`.
- **Idempotent `close()`.** Both ends can be closed multiple times safely.
- **Context manager support** — use `with SharedMemoryPipe() as smp:`.
- **Single host only.** POSIX shared memory does not work across machines.
- **Re-enables the GIL on a free-threaded build.** `posix_ipc` has not declared free-threading support; importing this module on a `3.13t`+ interpreter re-enables the GIL for the whole process. See {doc}`/performance` ("Free-threaded CPython").

---

## See also

- {doc}`/worker` — SharedMemoryPipe in context of the Worker model
- {doc}`/examples/worker/shared_memory_pipe` — streaming ZIP example
- {doc}`/api_download` — file download responses