How to attach metadata to a box or stream¶

A box or a stream carries values, but not what they mean, such as the unit of a weight or the calibration a reading was taken with. Metadata carries that, in the same segment, so every process that opens the box or stream reads it from there, in Python, C or C++. Each change adds a new metadata version and keeps the older ones for a while, so a reader can tell which version an item was sent under.

The metadata you pass to field is a different thing: a mapping kept in each process's class and never stored in the segment, from which sharedbox reads only its own keys, such as "shape".

Create a box with metadata¶

Give the type of the metadata as the type argument of SharedBox, and pass the first value as metadata when you create the box:

from typing import TypedDict

from sharedbox import SharedBox, SharedStream


class Calibration(TypedDict):
    unit: str
    offset: float


class Scale(SharedBox[Calibration]):
    weight: float = 0.0


scale = Scale.create("example-scale", metadata={"unit": "kg", "offset": 0.0})
print(scale.metadata_version)  # 1
print(scale.metadata)  # {'unit': 'kg', 'offset': 0.0}

Here the metadata is a TypedDict, so a type checker knows what metadata holds and checks every value you add. Without a type argument, as in class Scale(SharedBox), the metadata is a Mapping[str, Any]. metadata gives the value of the newest version, or None for a box that has none, and metadata_version gives its number, counting from 1.

A value is None, a bool, an int, a float, a str, bytes, a list or tuple, or a mapping with str keys, nested up to 32 levels; update lists the limits. The type argument must be a mapping type, but only a type checker holds you to that: at run time a value that is not a mapping is stored all the same. Calling the class takes the metadata keyword too, but a type checker accepts it only in create, where it does not check the value.

Add a version¶

Any process that has the box open adds a version by passing metadata to update. The call may also name fields, and then the new values and the new version go in as one write, so a reader that reads the fields and then metadata gets the version written with them, or a newer one. Without fields, the call only adds the version:

scale.update(weight=1.2, metadata={"unit": "kg", "offset": 0.25})
print(scale.weight, scale.metadata_version)  # 1.2 2
print(scale.metadata_version_of(1)["offset"])  # 0.0
print(scale.metadata_version_of(2)["offset"])  # 0.25
scale.update(metadata={"unit": "kg", "offset": 0.3})
print(scale.weight, scale.metadata_version)  # 1.2 3

The first call writes weight and version 2 together, and the second adds version 3. An update that raises, because a value is out of range or the box has no room for the version, writes neither the fields nor the version, and metadata=None, the default, adds no version. Every process that listens to the metadata signal of the box's events gets the new version's number, once for several versions added close together, and gets it before the signals of the fields written with it:

scale.events.metadata.connect(lambda version: print("calibration", version))

metadata_version_of reads an older version for as long as the box keeps it. The box keeps the last metadata_versions versions, 64 unless the class sets that keyword, or fewer when they no longer fit in its room for metadata. Reading a version it has let go raises MetadataGoneError.

Leave room for metadata added later¶

A box created with metadata gets 65536 bytes of room for its metadata, and one created without it gets none, so update(metadata=...) on it raises MetadataFullError. When the first version comes after the box is created, give the class a metadata_capacity:

class Display(SharedBox[Calibration], metadata_capacity=4096, metadata_versions=8):
    weight: float = 0.0


display = Display.create("example-display")
print(display.metadata)  # None
display.update(metadata={"unit": "g", "offset": 0.0})

The capacity is the number of bytes all the kept versions share, 0 or at least 24. One version can take up to a third of it, rounded down to a multiple of 8, and a longer value raises ValueError and adds nothing. A value is stored as CBOR, a compact binary form of JSON-like values, so {"unit": "kg", "offset": 0.0} takes 19 bytes. metadata_versions, 2 to 65535, limits how many versions the box keeps. A stream takes both as keywords of create.

Attach metadata to a stream¶

A stream takes its first version as metadata in create, and its sender adds each later one with the metadata of send. asend, send_nowait and the send of the slot that writing gives take the keyword too. A version applies from the item sent with it, and a reader's metadata and metadata_version describe the item it returned last, by a receive or a fetch, so the reader can tell which version each item was sent under. The metadata of this stream is a TypedDict too:

class Calibration(TypedDict):
    unit: str
    offset: float

with SharedStream.create(
    float,
    "example-scale:readings",
    capacity=8,
    metadata={"unit": "kg", "offset": 0.0},
    metadata_type=Calibration,
) as stream:
    reader = stream.reader()
    with stream.sender() as sender:
        sender.send(1.5)
        sender.send(1.75, metadata={"unit": "kg", "offset": 0.25})
    for weight in reader:
        calibration = reader.metadata
        if calibration is not None:
            print(reader.metadata_version, weight - calibration["offset"])
            # prints 1 1.5, then 2 1.5
SharedStream.unlink("example-scale:readings")

The reader prints version 1 for the first reading, which was sent before version 2 existed, and version 2 for the reading sent with it. Before its first item, a reader's metadata_version is 0 and its metadata is None, and the same holds for an item sent before the stream had any version. The stream's own metadata gives the newest version, as a box's does.

A type checker takes the stream's metadata type from metadata_type, given to create as above or to attach in another process, as in SharedStream.attach(float, name, metadata_type=Calibration). Without it the metadata is a Mapping[str, Any], whatever metadata you pass.

A stream keeps every version that an item still in its ring was sent under, and keeps version 1 while an item sent before it is in the ring, so that a reader can tell such an item from one whose version is gone. A send with metadata that would let such a version go raises MetadataFullError before it waits for room in the ring, so it sends nothing and releases no reader; the slot's send in a writing block raises it the same way and leaves the slot reserved. The same send succeeds once the sender has sent enough further items that none of those items is left in the ring. Once a reader's last item has left the ring, though, its version can go, and the reader's metadata then raises MetadataGoneError. A lossy or latest reader meets this first, because the sender may overwrite its item right after the reader receives it.

Change a value you read¶

A value you read is a read-only view, because every read of a version through the same box or stream shares one decoded copy. A mapping reads back as a types.MappingProxyType and a list as a tuple, nested the same way. A type checker still sees the metadata's type, though, so it accepts a change that fails when it runs. To change a key, build a new value and add it as the next version:

stored = scale.metadata_version_of(2)
try:
    stored["offset"] = 0.5
except TypeError as error:
    print(error)  # 'mappingproxy' object does not support item assignment
scale.update(metadata={**stored, "offset": 0.5})

Read metadata from C or C++¶

A C program reads a box's versions with sbx_metadata_head and sbx_read_metadata, and adds one with sbx_add_metadata. In C++, handle and stream read versions with metadata_head and read_metadata, handle and stream_sender add them with add_metadata, and stream_reader::metadata_version_at finds the version an item was sent under; streams have no C calls. Each version is one CBOR value, which your program decodes with a CBOR library of its own. Segment layout lists the calls and the CBOR a version may hold.

"""The script of the guide "How to attach metadata to a box or stream"."""

from typing import TypedDict

from sharedbox import SharedBox, SharedStream


class Calibration(TypedDict):
    unit: str
    offset: float


class Scale(SharedBox[Calibration]):
    weight: float = 0.0


scale = Scale.create("example-scale", metadata={"unit": "kg", "offset": 0.0})
print(scale.metadata_version)  # 1
print(scale.metadata)  # {'unit': 'kg', 'offset': 0.0}

scale.update(weight=1.2, metadata={"unit": "kg", "offset": 0.25})
print(scale.weight, scale.metadata_version)  # 1.2 2
print(scale.metadata_version_of(1)["offset"])  # 0.0
print(scale.metadata_version_of(2)["offset"])  # 0.25
scale.update(metadata={"unit": "kg", "offset": 0.3})
print(scale.weight, scale.metadata_version)  # 1.2 3


class Display(SharedBox[Calibration], metadata_capacity=4096, metadata_versions=8):
    weight: float = 0.0


display = Display.create("example-display")
print(display.metadata)  # None
display.update(metadata={"unit": "g", "offset": 0.0})

stored = scale.metadata_version_of(2)
try:
    stored["offset"] = 0.5
except TypeError as error:
    print(error)  # 'mappingproxy' object does not support item assignment
scale.update(metadata={**stored, "offset": 0.5})


with SharedStream.create(
    float,
    "example-scale:readings",
    capacity=8,
    metadata={"unit": "kg", "offset": 0.0},
    metadata_type=Calibration,
) as stream:
    reader = stream.reader()
    with stream.sender() as sender:
        sender.send(1.5)
        sender.send(1.75, metadata={"unit": "kg", "offset": 0.25})
    for weight in reader:
        calibration = reader.metadata
        if calibration is not None:
            print(reader.metadata_version, weight - calibration["offset"])
            # prints 1 1.5, then 2 1.5
SharedStream.unlink("example-scale:readings")

scale.close()
display.close()
Scale.unlink("example-scale")
Display.unlink("example-display")