ezmsg.tools.shmem.shmem#

It is possible to move data from ezmsg to non-ezmsg processes using shared memory. This module contains the ezmsg half of that communication. The non-ezmsg half is in the .shmem_mirror module. The same shmem_name must be passed to both the ShMemCircBuff and the EZShmMirror objects!

The ShMemCircBuff class is a sink node that receives AxisArray messages and writes them to a shared memory buffer.

Upon initialization, or upon receiving updated settings with a different shmem_name value, the node creates a shared memory object located at {shorten_shmem_name(shmem_name)} to hold the metadata initialized with placeholder values (e.g., srate = -1). Additionally, the node has a convenience handle to the metadata via self.STATE.meta_struct = ShmemArrMeta.from_buffer(shmem.buf).

Upon receiving a data message, its metadata is checked, and if it does not match the shmem metadata

(which will always be true for the first message) then the node first updates the metadata, then it (re-)creates a shared memory buffer to hold the data, located at shorten_shmem_name(f”{shmem_name}/buffer{buffer_generation}”), where buffer_generation is an integer that tracks how many times the buffer has been reset. This corresponds to the same integer stored in the metadata.

The other half must monitor the metadata shared memory to see if it changes, and if it does then it must recreate the data shared memory buffer reader at the new location.

The data segment holds two arrays: first one float64 timestamp per frame (the buffered axis’s value for that frame), then the frames themselves. Both are written before write_index advances, so a reader that takes frames from the ring takes their times from the same indices and the two always agree. See ring_views.

Finally, there is a third piece of shared memory carrying everything about the AxisArray that does not fit in the fixed-size metadata header: the non-buffered coordinate axes (e.g. a ch axis holding per-channel bank/elec/label), axis units, and the message attrs. It lives at shorten_shmem_name(f”{shmem_name}/meta{meta_generation}”) and is republished – under a fresh generation, following the same pattern as the data buffer – only when that metadata actually changes, which for a typical stream means once per session. See the .aux_meta module for the wire format.

Functions

ring_nbytes(n_frames, frame_shape, itemsize)[source]#

Size of the data segment described by ring_views().

Parameters:
Return type:

int

ring_views(buf, meta)[source]#

The data segment’s two arrays: per-frame timestamps (float64), then the frames.

Times lead so both arrays are aligned for any frame dtype. Writer and reader both call this, so the layout is stated once.

Parameters:
Return type:

Tuple[NDArray, NDArray]

shorten_shmem_name(long_name)[source]#

Convert a potentially long shared memory name to a shorter, fixed-length name.

Parameters:

long_name (str) – The original, potentially long shared memory name

Returns:

A shortened, deterministic name suitable for shared memory

Return type:

str

to_bytes(data)[source]#
Parameters:

data (Any)

Return type:

bytes

Classes

class ShMemCircBuff(*args, settings=None, **kwargs)[source]#

Bases: Unit

Parameters:

settings (Settings | None)

SETTINGS#

alias of ShMemCircBuffSettings

STATE#

alias of ShMemCircBuffState

INPUT_SIGNAL = InputStream:unlocated[AxisArray]()#
INPUT_SETTINGS = InputStream:unlocated[ShMemCircBuffSettings]()#
async initialize()[source]#

Runs when the Unit is instantiated.

This is called from within the same process this unit will live in. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

This method is where you should initialize your unit’s state and prepare for message processing.

Return type:

None

on_settings(msg)[source]#
Parameters:

msg (ShMemCircBuffSettings)

Return type:

None

async shutdown()[source]#

Runs when the Unit terminates.

This is called from within the same process this unit will live in. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

This method is where you should clean up resources and perform any necessary shutdown procedures.

Return type:

None

async check_continue()[source]#
async on_message(msg)[source]#
Parameters:

msg (AxisArray)

class ShMemCircBuffSettings(shmem_name: str | None, buf_dur: float, conn: Connection | None = None, axis: str | None = None)[source]#

Bases: Settings

Parameters:
shmem_name: str | None#
buf_dur: float#
conn: Connection | None = None#
axis: str | None = None#

Dimension to buffer along. None follows the message’s stream_dim.

The ring is a history of the stream, so this has to be the dimension messages accumulate along; buffering a static one would store the same elements over and over. Only the producer reliably knows which that is – it is time on a raw signal and win downstream of a windowing stage.

The old default of "time" was silently wrong for the latter. It is present in a (win, time, ch) message, so nothing rejected it: the window count ended up inside frame_shape – reallocating the buffer whenever the window count jittered – and the reported sample rate was the within-window rate, a 10x error in the viewer’s time base for a 10-sample window. Set explicitly only for a producer that declares no stream_dim.

__init__(shmem_name, buf_dur, conn=None, axis=None)#
Parameters:
Return type:

None

class ShMemCircBuffState[source]#

Bases: State

meta_shmem: SharedMemory | None = None#
meta_struct: ShmemArrMeta | None = None#
buffer_shmem: SharedMemory | None = None#
buffer_arr: NDArray | None = None#
times_arr: NDArray | None = None#
meta_hash: int = -1#
buff_axis: str | None = None#
aux_shmem: SharedMemory | None = None#
last_aux_src: tuple | None = None#
last_aux_blob: bytes | None = None#
warned_dropped_attrs: frozenset | None = None#
class ShmemArrMeta[source]#

Bases: Structure

Structure containing the metadata describing the separate shmem buffer.

The SharedMemory object is expected to have allocated enough memory for this header + the memory required for the buffer described by this header. i.e., meta_size = ctypes.sizeof(ShmemArrMeta) item_size = np.dtype(dtype).itemsize shmem_size = int(meta_size + np.prod(shape) * item_size) shmem = SharedMemory(name=”…”, create=True, size=shmem_size) meta = ShmemArrMeta.from_buffer(shmem) meta.dtype = dtype meta.ndim = len(shape) meta.shape[:meta.ndim] = shape circ_buff = np.ndarray(shape, dtype=dtype, buffer=shmem.buf[meta_size:])

property key: str#
aux_nbytes#

Structure/Union member

buffer_generation#

Structure/Union member

bvalid#

Structure/Union member

dtype#

Structure/Union member

magic#

Structure/Union member

meta_generation#

Structure/Union member

ndim#

Structure/Union member

shape#

Structure/Union member

srate#

Structure/Union member

struct_version#

Structure/Union member

wrap_counter#

Structure/Union member

write_index#

Structure/Union member

Exceptions

exception ShmemVersionError[source]#

Bases: RuntimeError

A shmem segment was written by an incompatible build.

Not recoverable and not transient: upgrade both ends together.

to_bytes(data)[source]#
Parameters:

data (Any)

Return type:

bytes

shorten_shmem_name(long_name)[source]#

Convert a potentially long shared memory name to a shorter, fixed-length name.

Parameters:

long_name (str) – The original, potentially long shared memory name

Returns:

A shortened, deterministic name suitable for shared memory

Return type:

str

exception ShmemVersionError[source]#

Bases: RuntimeError

A shmem segment was written by an incompatible build.

Not recoverable and not transient: upgrade both ends together.

class ShmemArrMeta[source]#

Bases: Structure

Structure containing the metadata describing the separate shmem buffer.

The SharedMemory object is expected to have allocated enough memory for this header + the memory required for the buffer described by this header. i.e., meta_size = ctypes.sizeof(ShmemArrMeta) item_size = np.dtype(dtype).itemsize shmem_size = int(meta_size + np.prod(shape) * item_size) shmem = SharedMemory(name=”…”, create=True, size=shmem_size) meta = ShmemArrMeta.from_buffer(shmem) meta.dtype = dtype meta.ndim = len(shape) meta.shape[:meta.ndim] = shape circ_buff = np.ndarray(shape, dtype=dtype, buffer=shmem.buf[meta_size:])

property key: str#
aux_nbytes#

Structure/Union member

buffer_generation#

Structure/Union member

bvalid#

Structure/Union member

dtype#

Structure/Union member

magic#

Structure/Union member

meta_generation#

Structure/Union member

ndim#

Structure/Union member

shape#

Structure/Union member

srate#

Structure/Union member

struct_version#

Structure/Union member

wrap_counter#

Structure/Union member

write_index#

Structure/Union member

ring_views(buf, meta)[source]#

The data segment’s two arrays: per-frame timestamps (float64), then the frames.

Times lead so both arrays are aligned for any frame dtype. Writer and reader both call this, so the layout is stated once.

Parameters:
Return type:

Tuple[NDArray, NDArray]

ring_nbytes(n_frames, frame_shape, itemsize)[source]#

Size of the data segment described by ring_views().

Parameters:
Return type:

int

class ShMemCircBuffSettings(shmem_name: str | None, buf_dur: float, conn: Connection | None = None, axis: str | None = None)[source]#

Bases: Settings

Parameters:
shmem_name: str | None#
buf_dur: float#
conn: Connection | None = None#
axis: str | None = None#

Dimension to buffer along. None follows the message’s stream_dim.

The ring is a history of the stream, so this has to be the dimension messages accumulate along; buffering a static one would store the same elements over and over. Only the producer reliably knows which that is – it is time on a raw signal and win downstream of a windowing stage.

The old default of "time" was silently wrong for the latter. It is present in a (win, time, ch) message, so nothing rejected it: the window count ended up inside frame_shape – reallocating the buffer whenever the window count jittered – and the reported sample rate was the within-window rate, a 10x error in the viewer’s time base for a 10-sample window. Set explicitly only for a producer that declares no stream_dim.

__init__(shmem_name, buf_dur, conn=None, axis=None)#
Parameters:
Return type:

None

class ShMemCircBuffState[source]#

Bases: State

meta_shmem: SharedMemory | None = None#
meta_struct: ShmemArrMeta | None = None#
buffer_shmem: SharedMemory | None = None#
buffer_arr: NDArray | None = None#
times_arr: NDArray | None = None#
meta_hash: int = -1#
buff_axis: str | None = None#
aux_shmem: SharedMemory | None = None#
last_aux_src: tuple | None = None#
last_aux_blob: bytes | None = None#
warned_dropped_attrs: frozenset | None = None#
class ShMemCircBuff(*args, settings=None, **kwargs)[source]#

Bases: Unit

Parameters:

settings (Settings | None)

SETTINGS#

alias of ShMemCircBuffSettings

STATE#

alias of ShMemCircBuffState

INPUT_SIGNAL = InputStream:unlocated[AxisArray]()#
INPUT_SETTINGS = InputStream:unlocated[ShMemCircBuffSettings]()#
async initialize()[source]#

Runs when the Unit is instantiated.

This is called from within the same process this unit will live in. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

This method is where you should initialize your unit’s state and prepare for message processing.

Return type:

None

on_settings(msg)[source]#
Parameters:

msg (ShMemCircBuffSettings)

Return type:

None

async shutdown()[source]#

Runs when the Unit terminates.

This is called from within the same process this unit will live in. This lifecycle hook can be overridden. It can be run as async functions by simply adding the async keyword when overriding.

This method is where you should clean up resources and perform any necessary shutdown procedures.

Return type:

None

async check_continue()[source]#
async on_message(msg)[source]#
Parameters:

msg (AxisArray)

On this page
Show Source