Python#

pw_transport: Asynchronous datagram-oriented transport interface

The pw_transport Python package (//pw_transport/py:pw_transport) defines synchronous, thread-safe, datagram-oriented socket and listener interfaces matching the C++ pw::transport module:

pw_transport.Socket and pw_transport.ReliableSocket read and write whole datagrams over an established connection.
pw_transport.Listener and pw_transport.ReliableListener accept incoming connections.

Outgoing connections are opened with plain functions (such as pw_transport.framed_tcp.connect()) returning a Socket or ReliableSocket.

Usage#

Implementing a custom transport#

Protocol implementations (such as pw_rpc2) depend only on //pw_transport/py:pw_transport. Custom host transports subclass pw_transport.ReliableSocket (or pw_transport.Socket) and optionally provide a connect(...) function or Listener subclass:

import pw_transport


class CustomReliableSocket(pw_transport.ReliableSocket):
    @property
    def max_read_message_size(self) -> int | None:
        return 1024

    @property
    def max_write_message_size(self) -> int | None:
        return 512

    @property
    def closed(self) -> bool: ...

    def _read(self, deadline: pw_transport.Deadline) -> bytes:
        # Block until the next framed datagram arrives, or raise
        # TimeoutError / SocketClosed.
        ...

    def _write(self, data: bytes, deadline: pw_transport.Deadline) -> None:
        # Transmit one whole datagram, or raise TimeoutError /
        # SocketClosed.
        ...

    def close(self) -> None:
        # Close the link and unblock pending read() and write() calls.
        ...

Callers use the public read(), write(), and accept() methods, which validate their arguments and then call the _read(), _write(), and _accept() methods that implementations override. Implementations never see an invalid timeout or an outgoing datagram larger than max_write_message_size.

Timeouts are passed to implementations as a pw_transport.Deadline, which reports whether the operation should give up and how long to wait otherwise.

API reference#

pw_transport#

Datagram-oriented transport interfaces for Pigweed.

class pw_transport.Socket#

A bidirectional datagram-oriented connection to a peer.

This is the Python counterpart of pw::transport::Socket. Every datagram is delivered whole without merging or splitting; framing is the transport’s responsibility. Unlike ReliableSocket, a basic Socket may drop, duplicate, or reorder datagrams if the underlying link is unreliable.

Zero-length datagrams (b'') are valid in both directions and are distinct from closure, which raises SocketClosed.

Only one thread may call read() at a time, while any number of threads may call write() concurrently. close() may be called from any thread and unblocks pending read() and write() calls.

Implementations override _read() and _write() rather than read() and write(), which validate their arguments before calling into the implementation.

abstract _read(deadline: Deadline) → bytes#

Implements read(), which documents the required behavior.

Parameters:: deadline – When to give up and raise TimeoutError.

abstract _write(data: bytes, deadline: Deadline) → None#

Implements write(), which documents the required behavior.

Parameters:

data – Outgoing datagram payload, which is no larger than max_write_message_size.
deadline – When to give up and raise TimeoutError.

abstract close() → None#

Closes the socket and unblocks pending operations.

Calling close() on an already-closed socket is a no-op.

abstract property closed: bool#: True if the socket is closed.

property max_read_message_size: int | None#

Largest datagram that can be read from this socket, in bytes.

Returns None if there is no limit.

property max_write_message_size: int | None#

Largest datagram that can be written to this socket, in bytes.

Returns None if there is no limit. write() rejects larger datagrams with ValueError.

final read(timeout: float | None = None) → bytes#

Blocks until one whole datagram is received and returns it.

Only one thread may call read() at a time. If timeout expires, TimeoutError is raised without closing the socket or discarding any in-flight datagram.

Parameters:

timeout – Maximum time to wait in seconds, or None (or math.inf) to wait indefinitely.

Raises:

SocketClosed – The socket is closed.
TimeoutError – No complete datagram arrived before timeout.
RuntimeError – Another read() is already in progress.
ValueError – timeout is negative or NaN.

final write(data: bytes, timeout: float | None = None) → None#

Writes one whole datagram, blocking while backpressure applies.

Parameters:

data – Outgoing datagram payload.
timeout – Maximum time to wait in seconds while backpressure applies, or None (or math.inf) to wait indefinitely.

Raises:

SocketClosed – The socket is closed.
TimeoutError – The datagram could not be sent before timeout. Stream-backed transports that have already transmitted part of the frame close the socket before raising TimeoutError.
ValueError – data exceeds max_write_message_size, or timeout is negative or NaN.

class pw_transport.ReliableSocket#

A reliable, ordered, datagram-oriented connection to a peer.

This is the Python counterpart of pw::transport::ReliableSocket. Unless the connection fails and raises SocketClosed, datagrams written by one peer are delivered whole, in order, and without duplication.

__init__()#

abstract _read(deadline: Deadline) → bytes#

Implements read(), which documents the required behavior.

Parameters:: deadline – When to give up and raise TimeoutError.

abstract _write(data: bytes, deadline: Deadline) → None#

Implements write(), which documents the required behavior.

Parameters:

data – Outgoing datagram payload, which is no larger than max_write_message_size.
deadline – When to give up and raise TimeoutError.

abstract close() → None#

Closes the socket and unblocks pending operations.

Calling close() on an already-closed socket is a no-op.

abstract property closed: bool#: True if the socket is closed.

property max_read_message_size: int | None#

Largest datagram that can be read from this socket, in bytes.

Returns None if there is no limit.

property max_write_message_size: int | None#

Largest datagram that can be written to this socket, in bytes.

Returns None if there is no limit. write() rejects larger datagrams with ValueError.

read(timeout: float | None = None) → bytes#

Blocks until one whole datagram is received and returns it.

Only one thread may call read() at a time. If timeout expires, TimeoutError is raised without closing the socket or discarding any in-flight datagram.

Parameters:

timeout – Maximum time to wait in seconds, or None (or math.inf) to wait indefinitely.

Raises:

SocketClosed – The socket is closed.
TimeoutError – No complete datagram arrived before timeout.
RuntimeError – Another read() is already in progress.
ValueError – timeout is negative or NaN.

write(data: bytes, timeout: float | None = None) → None#

Writes one whole datagram, blocking while backpressure applies.

Parameters:

data – Outgoing datagram payload.
timeout – Maximum time to wait in seconds while backpressure applies, or None (or math.inf) to wait indefinitely.

Raises:

SocketClosed – The socket is closed.
TimeoutError – The datagram could not be sent before timeout. Stream-backed transports that have already transmitted part of the frame close the socket before raising TimeoutError.
ValueError – data exceeds max_write_message_size, or timeout is negative or NaN.

class pw_transport.Listener#

Accepts incoming Socket connections from peers.

This is the Python counterpart of pw::transport::Listener. Closing a listener stops accepting new connections without closing sockets that were already accepted.

Multiple threads may call accept() concurrently; when multiple calls are pending, connections are distributed in an unspecified order. close() may be called from any thread and unblocks pending accept() calls.

Implementations override _accept() rather than accept(), which validates its arguments before calling into the implementation.

abstract _accept(deadline: Deadline) → _SocketT_co#

Implements accept(), which documents the required behavior.

Parameters:: deadline – When to give up and raise TimeoutError.

final accept(timeout: float | None = None) → _SocketT_co#

Blocks until a peer connects and returns the accepted socket.

Multiple threads may call accept() concurrently; the order in which pending calls receive connections is unspecified.

Parameters:

timeout – Maximum time to wait in seconds, or None (or math.inf) to wait indefinitely.

Raises:

SocketClosed – The listener is closed.
TimeoutError – No connection arrived before timeout.
ValueError – timeout is negative or NaN.

abstract close() → None#

Stops listening and unblocks pending accept() calls.

Calling close() on an already-closed listener is a no-op.

abstract property closed: bool#: True if the listener is closed.

class pw_transport.ReliableListener#

Accepts incoming ReliableSocket connections from peers.

This is the Python counterpart of pw::transport::ReliableListener.

__init__()#

abstract _accept(deadline: Deadline) → _SocketT_co#

Implements accept(), which documents the required behavior.

Parameters:: deadline – When to give up and raise TimeoutError.

accept(timeout: float | None = None) → _SocketT_co#

Blocks until a peer connects and returns the accepted socket.

Multiple threads may call accept() concurrently; the order in which pending calls receive connections is unspecified.

Parameters:

timeout – Maximum time to wait in seconds, or None (or math.inf) to wait indefinitely.

Raises:

SocketClosed – The listener is closed.
TimeoutError – No connection arrived before timeout.
ValueError – timeout is negative or NaN.

abstract close() → None#

Stops listening and unblocks pending accept() calls.

Calling close() on an already-closed listener is a no-op.

abstract property closed: bool#: True if the listener is closed.

class pw_transport.Deadline(timeout: float | None = None)#

The point in time at which a blocking operation gives up.

Socket and Listener convert each caller-provided timeout to a Deadline before calling into the implementation, so implementations never handle unvalidated timeouts or compute their own deadlines.

Pass remaining() to blocking calls and raise TimeoutError once the deadline has expired():

while not ready():
    if deadline.expired():
        raise TimeoutError('Timed out')
    condition.wait(deadline.remaining())

__init__(timeout: float | None = None) → None#

Starts a deadline that expires timeout seconds from now.

Parameters:: timeout – Maximum time to wait in seconds, or None (or math.inf) to never expire.
Raises:: ValueError – timeout is negative or NaN.

expired() → bool#: True if the deadline has passed; the operation should give up.

remaining() → float | None#

Returns the time left to wait, for use as a timeout.

Returns:: None if the deadline never expires; otherwise the seconds left until it does, which is at least 0.

exception pw_transport.SocketClosed#: The socket or listener is closed; no more operations can complete.