proxystore.endpoint.protocol¶

proxystore/endpoint/protocol.py

Wire protocol of endpoints.

Clients communicate with their local endpoint over a TCP connection using length-prefixed binary messages, and endpoints send the same messages to their peers over iroh streams. This module only contains the encoding and decoding logic. The MessageReader parses messages without performing I/O so the same parser is used by the blocking client, the asyncio server, and the iroh streams of peers.

A connection starts with a handshake:

The client sends a preamble (MAGIC and the newest protocol version it supports) followed by a HELLO message with a random nonce and the client's versions.
The endpoint replies with a preamble containing the negotiated protocol version (see negotiate_version()) and a message with its own nonce and a proof that it knows the endpoint's token.
The client verifies the endpoint's proof then sends an AUTH message with its own proof.
The endpoint verifies the client's proof and replies with its information (ID, name, and versions).

The formats of the preamble and the Hello message must never change (fields may only be added to Hello) because they are sent before the version is negotiated. If the client and endpoint support no common version, the endpoint replies with only its preamble, containing the newest version it supports, and closes the connection.

After the handshake, each request and response is a message consisting of a fixed-size Header, JSON-encoded metadata (e.g., the key), and a raw data payload. The endpoint echoes the request ID of each request in its response.

Peers do not use the handshake because iroh authenticates each endpoint. Each request to a peer is sent on its own stream so the request ID is unused. Peers exchange the same messages as clients and endpoints, and the protocol version is negotiated as part of the ALPN of the connection (see supported_alpns()).

MAGIC `module-attribute` ¶

MAGIC = b'PSEP'

Bytes that start every connection.

PROTOCOL_VERSION `module-attribute` ¶

PROTOCOL_VERSION = 1

Newest version of the protocol supported by this version of ProxyStore.

Increment on any incompatible change to the handshake or message formats.

MIN_PROTOCOL_VERSION `module-attribute` ¶

MIN_PROTOCOL_VERSION = 1

Oldest version of the protocol supported by this version of ProxyStore.

Clients and endpoints use the newest version both support, so keeping support for older versions lets clients and endpoints be upgraded separately. Only increase this when dropping support for an old version.

NONCE_SIZE `module-attribute` ¶

NONCE_SIZE = 32

Size in bytes of the random nonces exchanged in the handshake.

MAX_META_SIZE `module-attribute` ¶

MAX_META_SIZE = 64 * 1024

Maximum size in bytes of the metadata in a message.

VERSION_DOCS_URL `module-attribute` ¶

VERSION_DOCS_URL = 'https://docs.proxystore.dev/latest/guides/endpoints/#version-compatibility'

Documentation on version compatibility between clients and endpoints.

MessageData `module-attribute` ¶

MessageData: TypeAlias = bytes | bytearray

Data of a message.

Unlike BytesLike, views are excluded because endpoints store the data after handling the message.

STATUS_ERRORS `module-attribute` ¶

STATUS_ERRORS: dict[Status, type[EndpointError]] = {
    Status.ERROR: EndpointRequestError,
    Status.UNAUTHORIZED: EndpointAuthError,
    Status.BAD_REQUEST: EndpointProtocolError,
    Status.TOO_LARGE: ObjectSizeExceededError,
    Status.PEERING_DISABLED: PeeringDisabledError,
    Status.PEER_NOT_ALLOWED: PeerNotAllowedError,
    Status.PEER_UNAVAILABLE: PeerUnavailableError,
    Status.STORAGE_FULL: StorageFullError,
}

Exception type corresponding to each error status.

Nonce `module-attribute` ¶

Nonce = Annotated[
    bytes,
    Field(min_length=NONCE_SIZE, max_length=NONCE_SIZE),
]

Random nonce exchanged in the handshake.

Op ¶

Bases: IntEnum

Operation codes of requests sent by a client or a peer endpoint.

HELLO `class-attribute` `instance-attribute` ¶

HELLO = 1

First message of the handshake.

AUTH `class-attribute` `instance-attribute` ¶

AUTH = 2

Second message of the handshake with the client's proof.

GET `class-attribute` `instance-attribute` ¶

GET = 3

Get the object associated with a key.

SET `class-attribute` `instance-attribute` ¶

SET = 4

Set the object associated with a key.

EXISTS `class-attribute` `instance-attribute` ¶

EXISTS = 5

Check if an object associated with a key exists.

EVICT `class-attribute` `instance-attribute` ¶

EVICT = 6

Evict the object associated with a key.

PING `class-attribute` `instance-attribute` ¶

PING = 7

Measure the latency of and path to an endpoint.

Status ¶

Bases: IntEnum

Status codes of messages sent by an endpoint.

The metadata of a response with an error status contains the error message. Each error status corresponds to one exception type (see raise_for_status()).

Statuses added in later versions of the protocol must be errors so that older clients can handle them as generic errors.

OK `class-attribute` `instance-attribute` ¶

OK = 0

Request succeeded.

NOT_FOUND `class-attribute` `instance-attribute` ¶

NOT_FOUND = 1

No object is associated with the key.

ERROR `class-attribute` `instance-attribute` ¶

ERROR = 2

Request failed for an unexpected reason.

UNAUTHORIZED `class-attribute` `instance-attribute` ¶

UNAUTHORIZED = 3

Client failed authentication.

BAD_REQUEST `class-attribute` `instance-attribute` ¶

BAD_REQUEST = 4

Request was malformed.

TOO_LARGE `class-attribute` `instance-attribute` ¶

TOO_LARGE = 5

Request data exceeds the maximum object size of the endpoint.

PEERING_DISABLED `class-attribute` `instance-attribute` ¶

PEERING_DISABLED = 6

Request targets a peer but peering is disabled on the endpoint.

PEER_NOT_ALLOWED `class-attribute` `instance-attribute` ¶

PEER_NOT_ALLOWED = 7

Request targets a peer which is not allowed to communicate.

PEER_UNAVAILABLE `class-attribute` `instance-attribute` ¶

PEER_UNAVAILABLE = 8

Request targets a peer which cannot be reached.

STORAGE_FULL `class-attribute` `instance-attribute` ¶

STORAGE_FULL = 9

Request data does not fit in the storage of the endpoint.

Preamble `dataclass` ¶

Preamble(version: int = PROTOCOL_VERSION)

Preamble that starts every connection.

The format of the preamble must never change so that clients and endpoints using different protocol versions can detect the mismatch.

Attributes:

version (int) –

Protocol version of the sender.

FORMAT `class-attribute` ¶

FORMAT: Struct = struct.Struct('!4sH')

Format of the magic bytes and protocol version.

SIZE `class-attribute` ¶

SIZE: int = FORMAT.size

Size in bytes of the preamble.

pack ¶

pack() -> bytes

Pack the preamble.

Source code in proxystore/endpoint/protocol.py

def pack(self) -> bytes:
    """Pack the preamble."""
    return self.FORMAT.pack(MAGIC, self.version)

unpack `classmethod` ¶

unpack(buffer: BytesLike) -> Self

Unpack a preamble.

Raises:

EndpointProtocolError –

If the preamble does not start with MAGIC.

Source code in proxystore/endpoint/protocol.py

@classmethod
def unpack(cls, buffer: BytesLike) -> Self:
    """Unpack a preamble.

    Raises:
        EndpointProtocolError: If the preamble does not start with
            [`MAGIC`][proxystore.endpoint.protocol.MAGIC].
    """
    magic, version = cls.FORMAT.unpack(buffer)
    if magic != MAGIC:
        raise EndpointProtocolError(
            f'Expected connection to start with {MAGIC!r} but got '
            f'{magic!r}.',
        )
    return cls(version)

Header `dataclass` ¶

Header(
    code: int,
    flags: int,
    request_id: int,
    meta_len: int,
    data_len: int,
)

Message header.

Attributes:

code (int) –

Op of a request or Status of a response.
flags (int) –

Reserved for future use.
request_id (int) –

ID chosen by the client for a request and echoed in the response. IDs allow requests to be pipelined in the future. Handshake messages and messages between peers use 0.
meta_len (int) –

Length in bytes of the metadata.
data_len (int) –

Length in bytes of the data.

FORMAT `class-attribute` ¶

FORMAT: Struct = struct.Struct('!BBIIQ')

Format of the code, flags, request ID, metadata length, and data length.

SIZE `class-attribute` ¶

SIZE: int = FORMAT.size

Size in bytes of the header.

pack ¶

pack() -> bytes

Pack the header.

Source code in proxystore/endpoint/protocol.py

def pack(self) -> bytes:
    """Pack the header."""
    return self.FORMAT.pack(
        self.code,
        self.flags,
        self.request_id,
        self.meta_len,
        self.data_len,
    )

unpack `classmethod` ¶

unpack(buffer: BytesLike) -> Self

Unpack a header.

Raises:

EndpointProtocolError –

If the metadata length exceeds MAX_META_SIZE.

Source code in proxystore/endpoint/protocol.py

@classmethod
def unpack(cls, buffer: BytesLike) -> Self:
    """Unpack a header.

    Raises:
        EndpointProtocolError: If the metadata length exceeds
            [`MAX_META_SIZE`][proxystore.endpoint.protocol.MAX_META_SIZE].
    """
    header = cls(*cls.FORMAT.unpack(buffer))
    if header.meta_len > MAX_META_SIZE:
        raise EndpointProtocolError(
            f'Message metadata length ({header.meta_len} bytes) exceeds '
            f'the maximum of {MAX_META_SIZE} bytes.',
        )
    return header

Meta ¶

Bases: BaseModel

Base model of the metadata of a message.

Metadata is encoded as a JSON object. Unknown fields are ignored so that fields can be added to a message without breaking older readers, and bytes fields are hex-encoded.

Example

meta = Request(key='abc').encode()
assert Request.decode(meta) == Request(key='abc')

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

Versions ¶

Bases: Meta

ProxyStore and Python versions of a client or endpoint.

Attributes:

proxystore (str) –

ProxyStore version.
python (str) –

Python version.

current `classmethod` ¶

current() -> Self

Get the ProxyStore and Python versions of this process.

Source code in proxystore/endpoint/protocol.py

@classmethod
def current(cls) -> Self:
    """Get the ProxyStore and Python versions of this process."""
    return cls(
        proxystore=proxystore.__version__,
        python=platform.python_version(),
    )

mismatches ¶

mismatches(endpoint: Versions) -> list[str]

Find differences between this client's and the endpoint's versions.

The ProxyStore versions must have the same major version because proxies are compatible between versions with the same major version (see Versioning). The Python versions are not compared because endpoints do not deserialize objects.

Parameters:

endpoint (Versions) –

Versions of the endpoint.

Returns:

list[str] –

Human-readable descriptions of each mismatch. Empty if the versions are compatible.

Source code in proxystore/endpoint/protocol.py

def mismatches(self, endpoint: Versions) -> list[str]:
    """Find differences between this client's and the endpoint's versions.

    The ProxyStore versions must have the same major version because
    proxies are compatible between versions with the same major version
    (see [Versioning](../../versioning.md)). The Python versions are not
    compared because endpoints do not deserialize objects.

    Args:
        endpoint: Versions of the endpoint.

    Returns:
        Human-readable descriptions of each mismatch. Empty if the \
        versions are compatible.
    """
    mismatches = []
    if _major(self.proxystore) != _major(endpoint.proxystore):
        mismatches.append(
            f'ProxyStore {self.proxystore} (client) vs. '
            f'{endpoint.proxystore} (endpoint)',
        )
    return mismatches

mismatch_warning ¶

mismatch_warning(endpoint: Versions) -> str | None

Describe the version mismatches between a client and an endpoint.

Parameters:

endpoint (Versions) –

Versions of the endpoint.

Returns:

str | None –

A warning describing the
str | None –

mismatches()
str | None –

and their consequences or None if the versions are compatible.

Source code in proxystore/endpoint/protocol.py

def mismatch_warning(self, endpoint: Versions) -> str | None:
    """Describe the version mismatches between a client and an endpoint.

    Args:
        endpoint: Versions of the endpoint.

    Returns:
        A warning describing the
        [`mismatches()`][proxystore.endpoint.protocol.Versions.mismatches]
        and their consequences or `None` if the versions are compatible.
    """
    mismatches = self.mismatches(endpoint)
    if len(mismatches) == 0:
        return None
    return (
        f'{"; ".join(mismatches)}. Objects serialized in one environment '
        'may fail to deserialize in another. See '
        f'{VERSION_DOCS_URL} for details.'
    )

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

Hello ¶

Bases: Meta

First message of the handshake sent by the client.

Attributes:

nonce (Nonce) –

Random nonce chosen by the client.
versions (Versions) –

Versions of the client.

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

Challenge ¶

Bases: Meta

Reply of the endpoint to Hello.

Attributes:

nonce (Nonce) –

Random nonce chosen by the endpoint.
proof (bytes) –

Proof that the endpoint knows the token.

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

Auth ¶

Bases: Meta

Second message of the handshake with the proof of the client.

Attributes:

proof (bytes) –

Proof that the client knows the token.

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

EndpointInfo ¶

Bases: Meta

Information about an endpoint sent at the end of the handshake.

Attributes:

id (EndpointId) –

ID of the endpoint.
name (str) –

Name of the endpoint.
versions (Versions) –

Versions of the endpoint.
max_object_size (int | None) –

Maximum size in bytes of objects that can be set on the endpoint or None if there is no limit.

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

Request ¶

Bases: Meta

Metadata of a request.

Requests are sent by clients to their local endpoint and by endpoints to their peers.

Attributes:

key (str | None) –

Key of the object, or None for operations that do not operate on an object (e.g., PING).
target (EndpointId | None) –

ID of the endpoint to forward the request to or None for the endpoint receiving the request.

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

ExistsResult ¶

Bases: Meta

Result of an EXISTS request.

Attributes:

exists (bool) –

If an object associated with the key exists.

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

PingResult ¶

Bases: Meta

Result of a PING request.

Attributes:

peer_rtt_ms (float | None) –

Time in milliseconds for the local endpoint to send a request to the peer and receive the response or None if the local endpoint was pinged. The time of the first ping to a peer includes establishing the connection.
relayed (bool | None) –

If the connection to the peer is relayed, or None if the path is unknown or the local endpoint was pinged.
remote_addr (str | None) –

Address of the peer (or relay) on the path.
path_rtt_ms (int | None) –

Round-trip time in milliseconds of the path estimated by QUIC.

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

ErrorInfo ¶

Bases: Meta

Metadata of a response with an error status.

Attributes:

error (str) –

Message describing the error.

encode ¶

encode() -> bytes

Encode as message metadata.

Source code in proxystore/endpoint/protocol.py

def encode(self) -> bytes:
    """Encode as message metadata."""
    return self.__pydantic_serializer__.to_json(self)

decode `classmethod` ¶

decode(meta: bytes | bytearray) -> Self

Decode from message metadata.

Empty metadata is decoded as an empty JSON object.

Raises:

EndpointProtocolError –

If the metadata is malformed.

Source code in proxystore/endpoint/protocol.py

@classmethod
def decode(cls, meta: bytes | bytearray) -> Self:
    """Decode from message metadata.

    Empty metadata is decoded as an empty JSON object.

    Raises:
        EndpointProtocolError: If the metadata is malformed.
    """
    try:
        return cls.model_validate_json(meta if len(meta) > 0 else b'{}')
    except ValidationError as e:
        error = e.errors()[0]
        if len(error['loc']) == 0:
            reason = error['msg']
        else:
            field = '.'.join(str(part) for part in error['loc'])
            reason = f'missing or invalid {field!r} field ({error["msg"]})'
        raise EndpointProtocolError(
            f'Malformed {cls.__name__} message: {reason}.',
        ) from None

Message `dataclass` ¶

Message(
    code: int,
    meta: bytes = b'',
    data: MessageData = b'',
    request_id: int = 0,
)

A request or response message.

Attributes:

code (int) –

Op of a request or Status of a response.
meta (bytes) –

JSON-encoded metadata of the message. Each operation or response defines the model of its metadata (e.g., Request).
data (MessageData) –

Data of the message.
request_id (int) –

ID of the request (see Header).

error_message `property` ¶

error_message: str

Error message of a response with an error status.

This never raises so a malformed error response can still be reported.

error `classmethod` ¶

error(status: Status, message: str) -> Message

Create a response with an error message.

Source code in proxystore/endpoint/protocol.py

@classmethod
def error(cls, status: Status, message: str) -> Message:
    """Create a response with an error message."""
    return cls(status, ErrorInfo(error=message).encode())

from_error `classmethod` ¶

from_error(error: BaseException) -> Message

Create a response for a request which failed with an error.

The status is determined by error_status().

Source code in proxystore/endpoint/protocol.py

@classmethod
def from_error(cls, error: BaseException) -> Message:
    """Create a response for a request which failed with an error.

    The status is determined by
    [`error_status()`][proxystore.endpoint.protocol.error_status].
    """
    return cls.error(error_status(error), str(error))

pack_head ¶

pack_head(data_len: int | None = None) -> bytes

Pack the header and metadata of the message.

The data is not included so it can be sent separately without being copied.

Parameters:

data_len (int | None, default: None ) –

Length in bytes of the data sent after the head. Defaults to the length of data. Set this when the data is not stored in the message (e.g., a view of a buffer sent by a client).

Source code in proxystore/endpoint/protocol.py

def pack_head(self, data_len: int | None = None) -> bytes:
    """Pack the header and metadata of the message.

    The data is not included so it can be sent separately without being
    copied.

    Args:
        data_len: Length in bytes of the data sent after the head.
            Defaults to the length of
            [`data`][proxystore.endpoint.protocol.Message]. Set this when
            the data is not stored in the message (e.g., a view of a
            buffer sent by a client).
    """
    data_len = len(self.data) if data_len is None else data_len
    header = Header(
        self.code,
        0,
        self.request_id,
        len(self.meta),
        data_len,
    )
    return header.pack() + self.meta

MessageReader ¶

MessageReader(*, max_data_size: int | None = None)

Parser of a message which does not perform I/O.

The caller reads exactly size bytes from the connection and passes them to feed() until the message is done.

Example

reader = MessageReader(max_data_size=1000)
while not reader.done:
    reader.feed(await connection.readexactly(reader.size))
message = reader.message

Parameters:

max_data_size (int | None, default: None ) –

Maximum size in bytes of the data of the message or None for no limit. The size is checked once the header and metadata are read, before the data is read.

Source code in proxystore/endpoint/protocol.py

def __init__(self, *, max_data_size: int | None = None) -> None:
    self._max_data_size = max_data_size
    self._header: Header | None = None
    self._meta: bytes | None = None
    self._message: Message | None = None
    self._size = Header.SIZE

done `property` ¶

done: bool

The message has been read.

size `property` ¶

size: int

Size in bytes of the next part of the message to feed.

This is always greater than zero until the message is done.

header `property` ¶

header: Header

Header of the message.

Raises:

RuntimeError –

If the header has not been read.

message `property` ¶

message: Message

The message.

Raises:

RuntimeError –

If the message is not done.

feed ¶

feed(buffer: bytes | bytearray) -> None

Feed the next part of the message.

Parameters:

buffer (bytes | bytearray) –

Exactly size bytes read from the connection.

Raises:

RuntimeError –

If the message is already done.
ValueError –

If buffer is not the expected size.
EndpointProtocolError –

If the header is malformed (the metadata is decoded by the receiver of the message).
ObjectSizeExceededError –

If the size of the data exceeds the maximum. The header can be used to respond to the message.

Source code in proxystore/endpoint/protocol.py

def feed(self, buffer: bytes | bytearray) -> None:
    """Feed the next part of the message.

    Args:
        buffer: Exactly
            [`size`][proxystore.endpoint.protocol.MessageReader.size]
            bytes read from the connection.

    Raises:
        RuntimeError: If the message is already done.
        ValueError: If `buffer` is not the expected size.
        EndpointProtocolError: If the header is malformed (the metadata
            is decoded by the receiver of the message).
        ObjectSizeExceededError: If the size of the data exceeds the
            maximum. The
            [`header`][proxystore.endpoint.protocol.MessageReader.header]
            can be used to respond to the message.
    """
    if self.done:
        raise RuntimeError('The message has already been read.')
    if len(buffer) != self._size:
        raise ValueError(
            f'Expected {self._size} bytes but got {len(buffer)} bytes.',
        )

    if self._header is None:
        self._header = Header.unpack(buffer)
        if self._header.meta_len > 0:
            self._size = self._header.meta_len
            return
        buffer = b''
    if self._meta is None:
        self._meta = bytes(buffer)
        self._read_meta()
        return
    self._message = Message(
        self._header.code,
        self._meta,
        buffer,
        self._header.request_id,
    )
    self._size = 0

HandshakeReader ¶

HandshakeReader()

Bases: MessageReader

Parser of a handshake message which does not perform I/O.

Handshake messages never contain data, so a message with data is rejected before its data is read (see MessageReader).

Source code in proxystore/endpoint/protocol.py

def __init__(self) -> None:
    super().__init__(max_data_size=0)

done `property` ¶

done: bool

The message has been read.

size `property` ¶

size: int

Size in bytes of the next part of the message to feed.

This is always greater than zero until the message is done.

header `property` ¶

header: Header

Header of the message.

Raises:

RuntimeError –

If the header has not been read.

message `property` ¶

message: Message

The message.

Raises:

RuntimeError –

If the message is not done.

feed ¶

feed(buffer: bytes | bytearray) -> None

Feed the next part of the message.

Raises:

EndpointProtocolError –

If the message is malformed or contains data.

Source code in proxystore/endpoint/protocol.py

def feed(self, buffer: bytes | bytearray) -> None:
    """Feed the next part of the message.

    Raises:
        EndpointProtocolError: If the message is malformed or contains
            data.
    """
    try:
        super().feed(buffer)
    except ObjectSizeExceededError:
        raise EndpointProtocolError(
            'Handshake message with code '
            f'{self.header.code} contains data.',
        ) from None

supports_version ¶

supports_version(version: int) -> bool

Check if this version of ProxyStore supports a protocol version.

Source code in proxystore/endpoint/protocol.py

def supports_version(version: int) -> bool:
    """Check if this version of ProxyStore supports a protocol version."""
    return MIN_PROTOCOL_VERSION <= version <= PROTOCOL_VERSION

negotiate_version ¶

negotiate_version(newest: int) -> int | None

Negotiate the protocol version used with another client or endpoint.

Parameters:

newest (int) –

Newest version supported by the other side.

Returns:

int | None –

The newest version supported by both sides or None if there is no common version.

Source code in proxystore/endpoint/protocol.py

def negotiate_version(newest: int) -> int | None:
    """Negotiate the protocol version used with another client or endpoint.

    Args:
        newest: Newest version supported by the other side.

    Returns:
        The newest version supported by both sides or `None` if there is \
        no common version.
    """
    version = min(newest, PROTOCOL_VERSION)
    return version if supports_version(version) else None

alpn ¶

alpn(version: int) -> bytes

Get the application protocol of a version of the peer protocol.

Source code in proxystore/endpoint/protocol.py

def alpn(version: int) -> bytes:
    """Get the application protocol of a version of the peer protocol."""
    return f'proxystore/{version}'.encode()

alpn_version ¶

alpn_version(protocol: bytes) -> int

Get the version of the peer protocol from its application protocol.

Raises:

ValueError –

If protocol is not an application protocol of ProxyStore.

Source code in proxystore/endpoint/protocol.py

def alpn_version(protocol: bytes) -> int:
    """Get the version of the peer protocol from its application protocol.

    Raises:
        ValueError: If `protocol` is not an application protocol of
            ProxyStore.
    """
    prefix = b'proxystore/'
    if not protocol.startswith(prefix):
        raise ValueError(f'Unknown application protocol {protocol!r}.')
    return int(protocol[len(prefix) :])

supported_alpns ¶

supported_alpns() -> list[bytes]

Get the application protocols of the supported peer protocols.

Endpoints accept connections with any of these protocols and try each protocol when connecting to a peer, so the newest version supported by both peers is used.

Returns:

list[bytes] –

The application protocol of each supported version, newest first.

Source code in proxystore/endpoint/protocol.py

def supported_alpns() -> list[bytes]:
    """Get the application protocols of the supported peer protocols.

    Endpoints accept connections with any of these protocols and try each
    protocol when connecting to a peer, so the newest version supported by
    both peers is used.

    Returns:
        The application protocol of each supported version, newest first.
    """
    return [
        alpn(version)
        for version in range(PROTOCOL_VERSION, MIN_PROTOCOL_VERSION - 1, -1)
    ]

error_status ¶

error_status(error: BaseException) -> Status

Get the status of a response to a request which failed with an error.

Parameters:

error (BaseException) –

Error raised while handling the request.

Returns:

Status –

The status of the most specific type of error in STATUS_ERRORS or Status.ERROR if there is none.

Source code in proxystore/endpoint/protocol.py

def error_status(error: BaseException) -> Status:
    """Get the status of a response to a request which failed with an error.

    Args:
        error: Error raised while handling the request.

    Returns:
        The status of the most specific type of `error` in \
        [`STATUS_ERRORS`][proxystore.endpoint.protocol.STATUS_ERRORS] or \
        [`Status.ERROR`][proxystore.endpoint.protocol.Status.ERROR] if there \
        is none.
    """
    for kind in type(error).__mro__:
        if kind in _ERROR_STATUSES:
            return _ERROR_STATUSES[kind]
    return Status.ERROR

raise_for_status ¶

raise_for_status(
    response: Message, op: Op, *, source: str = 'Endpoint'
) -> Status

Raise the exception corresponding to the status of a response.

Parameters:

response (Message) –

Response message.
op (Op) –

Operation of the request.
source (str, default: 'Endpoint' ) –

Description of the sender of the response for error messages.

Returns:

Status –

Status.OK or Status.NOT_FOUND.

Raises:

EndpointRequestError –

If the status is unknown (e.g., an error status added in a later version of the protocol).
EndpointError –

The type in STATUS_ERRORS for any other status.

Source code in proxystore/endpoint/protocol.py

def raise_for_status(
    response: Message,
    op: Op,
    *,
    source: str = 'Endpoint',
) -> Status:
    """Raise the exception corresponding to the status of a response.

    Args:
        response: Response message.
        op: Operation of the request.
        source: Description of the sender of the response for error
            messages.

    Returns:
        [`Status.OK`][proxystore.endpoint.protocol.Status.OK] or \
        [`Status.NOT_FOUND`][proxystore.endpoint.protocol.Status.NOT_FOUND].

    Raises:
        EndpointRequestError: If the status is unknown (e.g., an error
            status added in a later version of the protocol).
        EndpointError: The type in
            [`STATUS_ERRORS`][proxystore.endpoint.protocol.STATUS_ERRORS]
            for any other status.
    """
    try:
        status = Status(response.code)
    except ValueError:
        raise EndpointRequestError(
            f'{source} returned unknown status code {response.code} for '
            f'{op.name} request: {response.error_message}',
        ) from None
    if status in (Status.OK, Status.NOT_FOUND):
        return status
    raise STATUS_ERRORS[status](
        f'{source} returned {status.name} for {op.name} request: '
        f'{response.error_message}',
    )

proxystore.endpoint.protocol¶

MAGIC module-attribute ¶

PROTOCOL_VERSION module-attribute ¶

MIN_PROTOCOL_VERSION module-attribute ¶

NONCE_SIZE module-attribute ¶

MAX_META_SIZE module-attribute ¶

VERSION_DOCS_URL module-attribute ¶

MessageData module-attribute ¶

STATUS_ERRORS module-attribute ¶

Nonce module-attribute ¶

Op ¶

HELLO class-attribute instance-attribute ¶

AUTH class-attribute instance-attribute ¶

GET class-attribute instance-attribute ¶

SET class-attribute instance-attribute ¶

EXISTS class-attribute instance-attribute ¶

EVICT class-attribute instance-attribute ¶

PING class-attribute instance-attribute ¶

Status ¶

OK class-attribute instance-attribute ¶

NOT_FOUND class-attribute instance-attribute ¶

ERROR class-attribute instance-attribute ¶

UNAUTHORIZED class-attribute instance-attribute ¶

BAD_REQUEST class-attribute instance-attribute ¶

TOO_LARGE class-attribute instance-attribute ¶

PEERING_DISABLED class-attribute instance-attribute ¶

PEER_NOT_ALLOWED class-attribute instance-attribute ¶

PEER_UNAVAILABLE class-attribute instance-attribute ¶

STORAGE_FULL class-attribute instance-attribute ¶

Preamble dataclass ¶

FORMAT class-attribute ¶

SIZE class-attribute ¶

pack ¶

unpack classmethod ¶

Header dataclass ¶

FORMAT class-attribute ¶

SIZE class-attribute ¶

pack ¶

unpack classmethod ¶

Meta ¶

encode ¶

decode classmethod ¶

Versions ¶

current classmethod ¶

mismatches ¶

mismatch_warning ¶

encode ¶

decode classmethod ¶

Hello ¶

encode ¶

decode classmethod ¶

Challenge ¶

encode ¶

decode classmethod ¶

Auth ¶

encode ¶

decode classmethod ¶

EndpointInfo ¶

encode ¶

decode classmethod ¶

Request ¶

encode ¶

decode classmethod ¶

ExistsResult ¶

encode ¶

decode classmethod ¶

PingResult ¶

encode ¶

decode classmethod ¶

ErrorInfo ¶

encode ¶

decode classmethod ¶

Message dataclass ¶

error_message property ¶

error classmethod ¶

from_error classmethod ¶

pack_head ¶

MessageReader ¶

done property ¶

size property ¶

MAGIC `module-attribute` ¶

PROTOCOL_VERSION `module-attribute` ¶

MIN_PROTOCOL_VERSION `module-attribute` ¶

NONCE_SIZE `module-attribute` ¶

MAX_META_SIZE `module-attribute` ¶

VERSION_DOCS_URL `module-attribute` ¶

MessageData `module-attribute` ¶

STATUS_ERRORS `module-attribute` ¶

Nonce `module-attribute` ¶

HELLO `class-attribute` `instance-attribute` ¶

AUTH `class-attribute` `instance-attribute` ¶

GET `class-attribute` `instance-attribute` ¶

SET `class-attribute` `instance-attribute` ¶

EXISTS `class-attribute` `instance-attribute` ¶

EVICT `class-attribute` `instance-attribute` ¶

PING `class-attribute` `instance-attribute` ¶

OK `class-attribute` `instance-attribute` ¶

NOT_FOUND `class-attribute` `instance-attribute` ¶

ERROR `class-attribute` `instance-attribute` ¶

UNAUTHORIZED `class-attribute` `instance-attribute` ¶

BAD_REQUEST `class-attribute` `instance-attribute` ¶

TOO_LARGE `class-attribute` `instance-attribute` ¶

PEERING_DISABLED `class-attribute` `instance-attribute` ¶

PEER_NOT_ALLOWED `class-attribute` `instance-attribute` ¶

PEER_UNAVAILABLE `class-attribute` `instance-attribute` ¶

STORAGE_FULL `class-attribute` `instance-attribute` ¶

Preamble `dataclass` ¶

FORMAT `class-attribute` ¶

SIZE `class-attribute` ¶

unpack `classmethod` ¶

Header `dataclass` ¶

FORMAT `class-attribute` ¶

SIZE `class-attribute` ¶

unpack `classmethod` ¶

decode `classmethod` ¶

current `classmethod` ¶

decode `classmethod` ¶

decode `classmethod` ¶

decode `classmethod` ¶

decode `classmethod` ¶

decode `classmethod` ¶

decode `classmethod` ¶

decode `classmethod` ¶

decode `classmethod` ¶

decode `classmethod` ¶

Message `dataclass` ¶

error_message `property` ¶

error `classmethod` ¶

from_error `classmethod` ¶

done `property` ¶

size `property` ¶

header `property` ¶

message `property` ¶

done `property` ¶

size `property` ¶

header `property` ¶

message `property` ¶