From 33f3c6fc4ca8810cee4c517594be725a26eec70c Mon Sep 17 00:00:00 2001 From: Dimitri Staessens Date: Sun, 26 Jul 2026 13:49:26 +0200 Subject: pyouroboros: Use consistent format docstrings This is basically a clean up pass to make sure all docstrings are aligned. The copyright banners are replaced by SPDX statements. Also did a general clean up on the comments. Passes pylint. No structural changes. Signed-off-by: Dimitri Staessens --- ouroboros/dev.py | 356 ++++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 272 insertions(+), 84 deletions(-) (limited to 'ouroboros/dev.py') diff --git a/ouroboros/dev.py b/ouroboros/dev.py index ffa59b6..1397bca 100644 --- a/ouroboros/dev.py +++ b/ouroboros/dev.py @@ -1,31 +1,17 @@ # -# Ouroboros - Copyright (C) 2016 - 2026 -# -# Python API for Ouroboros -# -# Dimitri Staessens -# -# This library is free software; you can redistribute it and/or -# modify it under the terms of the GNU Lesser General Public License -# version 2.1 as published by the Free Software Foundation. -# -# This library is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# Lesser General Public License for more details. -# -# You should have received a copy of the GNU Lesser General Public -# License along with this library; if not, write to the Free Software -# Foundation, Inc., http://www.fsf.org/about/contact/. +# SPDX-FileCopyrightText: 2016 - 2026 Dimitri Staessens +# SPDX-License-Identifier: LGPL-2.1-only # -"""Flow allocation, acceptance and I/O for the Ouroboros dev API.""" +""" +Flow allocation, acceptance and I/O for the Ouroboros dev API. +""" from __future__ import annotations import warnings from enum import IntFlag -from typing import Optional +from typing import TYPE_CHECKING, Optional, Type from _ouroboros_dev_cffi import ffi, lib @@ -40,19 +26,28 @@ from ouroboros.errors import ( ) from ouroboros.qos import QoSSpec +if TYPE_CHECKING: + from types import TracebackType + check_ouroboros_version(lib.OUROBOROS_VERSION_MAJOR, lib.OUROBOROS_VERSION_MINOR) class FrctFlags(IntFlag): - """FRCT-level feature flags.""" + """ + FRCT-level feature flags. + """ + RETRANSMIT = 0o1 RESCNTL = 0o2 LINGER = 0o4 class FlowProperties(IntFlag): - """Flags describing flow properties and blocking behaviour.""" + """ + Flags describing flow properties and blocking behaviour. + """ + READ_ONLY = 0o0 WRITE_ONLY = 0o1 READ_WRITE = 0o2 @@ -65,47 +60,68 @@ class FlowProperties(IntFlag): class Flow: - """Represents an allocated Ouroboros flow.""" + """ + Represents an allocated Ouroboros flow. + """ def __init__(self, fd: int = -1) -> None: - """Construct a Flow wrapper. + """ + Construct a Flow wrapper. - :param fd: Existing flow descriptor to wrap, or -1 for an - unallocated Flow that will be populated by - :meth:`alloc`, :meth:`accept` or :meth:`join`. + :param fd: Existing flow descriptor, or -1 for an unallocated + flow. """ + self._fd: int = fd def __enter__(self) -> Flow: return self - def __exit__(self, exc_type, exc_value, tb) -> None: + def __exit__(self, exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + tb: Optional[TracebackType]) -> None: self.dealloc() def __del__(self) -> None: + """ + Deallocate on collection; errors are swallowed as the library + may already be torn down at interpreter shutdown. + """ + try: self.dealloc() except Exception: # pylint: disable=broad-exception-caught - pass # interpreter shutdown may have torn down lib already + pass def __repr__(self) -> str: return f"Flow(fd={self._fd})" def fileno(self) -> int: - """Return the underlying ouroboros flow descriptor.""" + """ + Return the underlying ouroboros flow descriptor. + + :return: The ouroboros flow descriptor, or -1 if unallocated. + """ + return self._fd def alloc(self, dst: str, qos: Optional[QoSSpec] = None, timeo: Optional[float] = None) -> Optional[QoSSpec]: - """Allocate a flow with a certain QoS to a destination. + """ + Allocate a flow with a certain QoS to a destination. :param dst: Destination name. :param qos: Requested QoS. - :param timeo: Allocation timeout (None blocks forever, 0 is async). + :param timeo: Allocation timeout (None blocks forever, 0 is + async). :return: The QoS the IRM granted for the new flow. + :raises FlowAlreadyAllocatedError: If this Flow is already + allocated. + :raises FlowError: On a negative ouroboros return code. """ + if self._fd >= 0: raise FlowAlreadyAllocatedError() @@ -113,18 +129,25 @@ class Flow: _timeo = fl_to_timespec(ffi, timeo) rc = lib.flow_alloc(dst.encode(), _qos, _timeo) + raise_errno(rc) + self._fd = rc return qosspec_to_qos(ffi, _qos) def accept(self, timeo: Optional[float] = None) -> Optional[QoSSpec]: - """Accept an incoming flow and return its QoS. + """ + Accept an incoming flow and return its QoS. :param timeo: Accept timeout (None blocks forever, 0 is async). :return: The QoS of the accepted flow. + :raises FlowAlreadyAllocatedError: If this Flow is already + allocated. + :raises FlowError: On a negative ouroboros return code. """ + if self._fd >= 0: raise FlowAlreadyAllocatedError() @@ -132,7 +155,9 @@ class Flow: _timeo = fl_to_timespec(ffi, timeo) rc = lib.flow_accept(_qos, _timeo) + raise_errno(rc) + self._fd = rc return qosspec_to_qos(ffi, _qos) @@ -140,26 +165,38 @@ class Flow: def join(self, dst: str, timeo: Optional[float] = None) -> None: - """Join a broadcast layer. + """ + Join a broadcast layer. - :param dst: Destination broadcast layer name. + :param dst: Broadcast layer name. :param timeo: Join timeout (None blocks forever, 0 is async). + :raises FlowAlreadyAllocatedError: If this Flow is already + allocated. + :raises FlowError: On a negative ouroboros return code. """ + if self._fd >= 0: raise FlowAlreadyAllocatedError() _timeo = fl_to_timespec(ffi, timeo) rc = lib.flow_join(dst.encode(), _timeo) + raise_errno(rc) + self._fd = rc def dealloc(self) -> None: - """Deallocate this flow. Idempotent: a no-op on an unallocated flow.""" + """ + Deallocate this flow; a no-op on an unallocated flow. + """ + if self._fd < 0: return + rc = lib.flow_dealloc(self._fd) self._fd = -1 + if rc < 0: warnings.warn(f"flow_dealloc returned {rc}", FlowDeallocWarning, stacklevel=2) @@ -167,13 +204,17 @@ class Flow: def write(self, buf: bytes, count: Optional[int] = None) -> int: - """Write up to *count* bytes to the flow. + """ + Write up to *count* bytes to the flow. :param buf: Buffer to write from. - :param count: Number of bytes to write (defaults to ``len(buf)``). + :param count: Number of bytes to write (defaults to + ``len(buf)``). :return: Number of bytes written. - :raises FlowError: (or subclass) on negative ouroboros return codes. + :raises FlowError: On a negative ouroboros return code. + :raises FlowNotAllocatedError: If the flow is not allocated. """ + if self._fd < 0: raise FlowNotAllocatedError() @@ -181,30 +222,38 @@ class Flow: count = len(buf) rc = lib.flow_write(self._fd, ffi.from_buffer(buf), count) + return raise_errno(rc) def writeline(self, ln: str) -> int: - """Encode *ln* as UTF-8 and write it to the flow. + """ + Encode *ln* as UTF-8 and write it to the flow. :param ln: String to write. :return: Number of bytes written. + :raises FlowNotAllocatedError: If the flow is not allocated. """ + if self._fd < 0: raise FlowNotAllocatedError() data = ln.encode() + return self.write(data, len(data)) def read(self, count: Optional[int] = None) -> bytes: - """Read up to *count* bytes from the flow. + """ + Read up to *count* bytes from the flow. :param count: Maximum number of bytes to read (default 2048). - :return: Bytes read (may be empty when an SDU was just - fully consumed by a previous matching read). - :raises FlowError: (or subclass) on negative ouroboros return codes. + :return: Bytes read; empty when a previous read fully + consumed the SDU. + :raises FlowError: On a negative ouroboros return code. + :raises FlowNotAllocatedError: If the flow is not allocated. """ + if self._fd < 0: raise FlowNotAllocatedError() @@ -214,181 +263,320 @@ class Flow: _buf = ffi.new("char []", count) rc = lib.flow_read(self._fd, _buf, count) + raise_errno(rc) return ffi.unpack(_buf, rc) def readline(self) -> str: - """Read from the flow and decode the result as UTF-8.""" + """ + Read from the flow and decode the result as UTF-8. + + :return: Decoded UTF-8 string read from the flow. + :raises FlowNotAllocatedError: If the flow is not allocated. + """ + if self._fd < 0: raise FlowNotAllocatedError() return self.read().decode() def set_snd_timeout(self, timeo: float) -> None: - """Set the timeout for blocking writes (seconds).""" + """ + Set the timeout for blocking writes (seconds). + + :param timeo: Write timeout in seconds. + :raises FlowError: On a negative ouroboros return code. + """ + _timeo = fl_to_timespec(ffi, timeo) + raise_errno(lib.flow_set_snd_timeout(self._fd, _timeo)) def get_snd_timeout(self) -> Optional[float]: - """Return the timeout for blocking writes (seconds).""" + """ + Return the timeout for blocking writes (seconds). + + :return: Timeout for blocking writes, in seconds. + :raises FlowError: On a negative ouroboros return code. + """ + _timeo = ffi.new("struct timespec *") + raise_errno(lib.flow_get_snd_timeout(self._fd, _timeo)) + return timespec_to_fl(ffi, _timeo) def set_rcv_timeout(self, timeo: float) -> None: - """Set the timeout for blocking reads (seconds).""" + """ + Set the timeout for blocking reads (seconds). + + :param timeo: Read timeout in seconds. + :raises FlowError: On a negative ouroboros return code. + """ + _timeo = fl_to_timespec(ffi, timeo) + raise_errno(lib.flow_set_rcv_timeout(self._fd, _timeo)) def get_rcv_timeout(self) -> Optional[float]: - """Return the timeout for blocking reads (seconds).""" + """ + Return the timeout for blocking reads (seconds). + + :return: Timeout for blocking reads, in seconds. + :raises FlowError: On a negative ouroboros return code. + """ + _timeo = ffi.new("struct timespec *") + raise_errno(lib.flow_get_rcv_timeout(self._fd, _timeo)) + return timespec_to_fl(ffi, _timeo) def get_qos(self) -> Optional[QoSSpec]: - """Return the current QoS in effect on the flow.""" + """ + Return the current QoS in effect on the flow. + + :return: The QoS currently in effect on the flow. + :raises FlowError: On a negative ouroboros return code. + """ + _qos = ffi.new("qosspec_t *") + raise_errno(lib.flow_get_qos(self._fd, _qos)) + return qosspec_to_qos(ffi, _qos) def get_rx_queue_len(self) -> int: - """Return the receive queue length (bytes).""" + """ + Return the receive queue length (bytes). + + :return: Receive queue length in bytes. + :raises FlowError: On a negative ouroboros return code. + """ + size = ffi.new("size_t *") + raise_errno(lib.flow_get_rx_qlen(self._fd, size)) + return int(size[0]) def get_tx_queue_len(self) -> int: - """Return the transmit queue length (bytes).""" + """ + Return the transmit queue length (bytes). + + :return: Transmit queue length in bytes. + :raises FlowError: On a negative ouroboros return code. + """ + size = ffi.new("size_t *") + raise_errno(lib.flow_get_tx_qlen(self._fd, size)) + return int(size[0]) def get_mtu(self) -> int: - """Return the per-packet MTU. + """ + Return the maximum user payload that fits in one n-1 PDU after + crypto headers (bytes), or 0 if unknown. - This is the maximum user payload that fits in one n-1 PDU, - after crypto headers. Returns 0 if unknown. + :return: Maximum user payload in bytes, or 0 if unknown. + :raises FlowError: On a negative ouroboros return code. """ + mtu = ffi.new("size_t *") + raise_errno(lib.flow_get_mtu(self._fd, mtu)) + return int(mtu[0]) def set_flags(self, flags: FlowProperties) -> None: - """Replace the full set of flags for this flow. - - .. warning:: - This OVERWRITES the entire ``oflags`` value, including - the access-mode bits (``READ_ONLY`` / ``WRITE_ONLY`` / - ``READ_WRITE``). Passing only ``NO_PARTIAL_READ`` (for - example) silently downgrades the flow to read-only and - subsequent writes will fail with :class:`FlowPermissionError`. - - To preserve the existing access mode use - :meth:`add_flags` / :meth:`remove_flags`, or read the - current flags with :meth:`get_flags` and OR in the new - bits explicitly. + """ + Replace the full set of flags for this flow. :param flags: Bitmask of :class:`FlowProperties` values. + :raises FlowError: On a negative ouroboros return code. """ + raise_errno(lib.flow_set_flags(self._fd, int(flags))) def add_flags(self, flags: FlowProperties) -> None: - """OR *flags* into the current ``oflags`` value. + """ + OR *flags* into the current flags, leaving other bits set. - Preserves the access mode and any other already-set bits. + :param flags: :class:`FlowProperties` bits to set. + :raises FlowError: On a negative ouroboros return code. """ + current = self.get_flags() + self.set_flags(current | FlowProperties(int(flags))) def remove_flags(self, flags: FlowProperties) -> None: - """Clear *flags* from the current ``oflags`` value. + """ + Clear *flags* from the current flags, leaving other bits set. - Preserves every other bit (including the access mode). + :param flags: :class:`FlowProperties` bits to clear. + :raises FlowError: On a negative ouroboros return code. """ + current = self.get_flags() + self.set_flags(current & ~FlowProperties(int(flags))) def get_flags(self) -> FlowProperties: - """Return the current flags for this flow.""" + """ + Return the current flags for this flow. + + :return: The current :class:`FlowProperties` bitmask. + :raises FlowError: On a negative ouroboros return code. + """ + flags = raise_errno(lib.flow_get_flags(self._fd)) + return FlowProperties(int(flags)) def set_frct_flags(self, flags: FrctFlags) -> None: - """Set FRCT flags for this flow. + """ + Set FRCT flags for this flow. :param flags: Bitmask of :class:`FrctFlags`. + :raises FlowError: On a negative ouroboros return code. """ + raise_errno(lib.flow_set_frct_flags(self._fd, int(flags))) def get_frct_flags(self) -> FrctFlags: - """Return the FRCT flags for this flow.""" + """ + Return the FRCT flags for this flow. + + :return: The current :class:`FrctFlags` bitmask. + :raises FlowError: On a negative ouroboros return code. + """ + flags = raise_errno(lib.flow_get_frct_flags(self._fd)) + return FrctFlags(int(flags)) def set_frct_max_sdu(self, size: int) -> None: - """Set the maximum reassembly SDU size for FRCT (bytes).""" + """ + Set the maximum reassembly SDU size for FRCT (bytes). + + :param size: Maximum reassembly SDU size in bytes. + :raises FlowError: On a negative ouroboros return code. + """ + raise_errno(lib.flow_set_frct_max_sdu(self._fd, size)) def get_frct_max_sdu(self) -> int: - """Return the maximum reassembly SDU size for FRCT (bytes).""" + """ + Return the maximum reassembly SDU size for FRCT (bytes). + + :return: Maximum reassembly SDU size in bytes. + :raises FlowError: On a negative ouroboros return code. + """ + size = ffi.new("size_t *") + raise_errno(lib.flow_get_frct_max_sdu(self._fd, size)) + return int(size[0]) def set_frct_rcv_ring_size(self, size: int) -> None: - """Set the stream receive ring size (bytes, power of two).""" + """ + Set the stream receive ring size (bytes, power of two). + + :param size: Stream receive ring size in bytes (power of two). + :raises FlowError: On a negative ouroboros return code. + """ + raise_errno(lib.flow_set_frct_rcv_ring_sz(self._fd, size)) def get_frct_rcv_ring_size(self) -> int: - """Return the stream receive ring size (bytes).""" + """ + Return the stream receive ring size (bytes). + + :return: Stream receive ring size in bytes. + :raises FlowError: On a negative ouroboros return code. + """ + size = ffi.new("size_t *") + raise_errno(lib.flow_get_frct_rcv_ring_sz(self._fd, size)) + return int(size[0]) def set_frct_rto_min(self, rto_ns: int) -> None: - """Set the FRCT RTO floor (nanoseconds).""" + """ + Set the FRCT RTO floor (nanoseconds). + + :param rto_ns: FRCT RTO floor in nanoseconds. + :raises FlowError: On a negative ouroboros return code. + """ + raise_errno(lib.flow_set_frct_rto_min(self._fd, rto_ns)) def get_frct_rto_min(self) -> int: - """Return the FRCT RTO floor (nanoseconds).""" + """ + Return the FRCT RTO floor (nanoseconds). + + :return: FRCT RTO floor in nanoseconds. + :raises FlowError: On a negative ouroboros return code. + """ + rto = ffi.new("time_t *") + raise_errno(lib.flow_get_frct_rto_min(self._fd, rto)) + return int(rto[0]) def flow_alloc(dst: str, qos: Optional[QoSSpec] = None, timeo: Optional[float] = None) -> Flow: - """Allocate a new flow and return the :class:`Flow` wrapper. + """ + Allocate a new flow and return the :class:`Flow` wrapper. :param dst: Destination name. :param qos: Requested QoS. :param timeo: Allocation timeout (None blocks forever, 0 is async). + :return: The allocated flow. """ + f = Flow() + f.alloc(dst, qos, timeo) + return f def flow_accept(timeo: Optional[float] = None) -> Flow: - """Accept an incoming flow and return the :class:`Flow` wrapper. + """ + Accept an incoming flow and return the :class:`Flow` wrapper. :param timeo: Accept timeout (None blocks forever, 0 is async). + :return: The accepted flow. """ + f = Flow() + f.accept(timeo) + return f def flow_join(dst: str, timeo: Optional[float] = None) -> Flow: - """Join a broadcast layer and return the :class:`Flow` wrapper. + """ + Join a broadcast layer and return the :class:`Flow` wrapper. :param dst: Broadcast layer name. :param timeo: Join timeout (None blocks forever, 0 is async). + :return: The joined broadcast flow. """ + f = Flow() + f.join(dst, timeo) + return f -- cgit v1.2.3