SharedMemoryPipe¶

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.

# 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

# 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 WorkerProcess. It generates data and streams it through shared memory directly to the HTTP client.

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 SharedMemoryPipeFile (or iterates the reader directly):

for chunk in pipe_reader:
    response_body.write(chunk)

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

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 RuntimeError; pickling the writer raises 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 Performance & scaling ("Free-threaded CPython").