ouroboros: Rewrite flow/dev API and event loop

Rewrite the dev/flow wrapper for the new QoS and fccntl options
(timeouts, QoSSpec marshalling, errno-to-exception translation) and
update the event-loop layer (FEventQueue / FlowSet) to match.

2 files changed, 344 insertions, 437 deletions

diff --git a/ouroboros/dev.py b/ouroboros/dev.py
index d36a3ef..ffa59b6 100644
--- a/ouroboros/dev.py
+++ b/ouroboros/dev.py
@@ -19,476 +19,376 @@
 # Foundation, Inc., http://www.fsf.org/about/contact/.
 #
 
-import errno
+"""Flow allocation, acceptance and I/O for the Ouroboros dev API."""
+
+from __future__ import annotations
+
+import warnings
 from enum import IntFlag
-from math import modf
 from typing import Optional
 
 from _ouroboros_dev_cffi import ffi, lib
-from ouroboros.qos import *
-
-
-def _check_ouroboros_version():
-    ouro_major = lib.OUROBOROS_VERSION_MAJOR
-    ouro_minor = lib.OUROBOROS_VERSION_MINOR
-    try:
-        from importlib.metadata import version, PackageNotFoundError
-        try:
-            pyouro_parts = version('PyOuroboros').split('.')
-        except PackageNotFoundError:
-            return  # running from source, skip check
-        if ouro_major != int(pyouro_parts[0]) or \
-                ouro_minor != int(pyouro_parts[1]):
-            raise RuntimeError(
-                f"Ouroboros version mismatch: library is "
-                f"{ouro_major}.{ouro_minor}, "
-                f"pyouroboros is "
-                f"{pyouro_parts[0]}.{pyouro_parts[1]}"
-            )
-    except ImportError:
-        pass  # Python < 3.8
-
-
-_check_ouroboros_version()
-
-# Some constants
-MILLION = 1000 * 1000
-BILLION = 1000 * 1000 * 1000
-
-
-def _fl_to_timespec(timeo: float):
-    if timeo is None:
-        return ffi.NULL
-    elif timeo <= 0:
-        return ffi.new("struct timespec *", [0, 0])
-    else:
-        frac, whole = modf(timeo)
-        _timeo = ffi.new("struct timespec *")
-        _timeo.tv_sec = int(whole)
-        _timeo.tv_nsec = int(frac * BILLION)
-        return _timeo
-
-
-def _timespec_to_fl(_timeo) -> Optional[float]:
-    if _timeo is ffi.NULL:
-        return None
-    elif _timeo.tv_sec <= 0 and _timeo.tv_nsec == 0:
-        return 0
-    else:
-        return _timeo.tv_sec + _timeo.tv_nsec / BILLION
-
-
-# Intentionally duplicated, dev uses a separate FFI (ouroboros-dev).
-def _qos_to_qosspec(qos: QoSSpec):
-    if qos is None:
-        return ffi.NULL
-    else:
-        return ffi.new("qosspec_t *",
-                       [qos.delay,
-                        qos.bandwidth,
-                        qos.availability,
-                        qos.loss,
-                        qos.ber,
-                        qos.in_order,
-                        qos.max_gap,
-                        qos.timeout])
-
-
-def _qosspec_to_qos(_qos) -> Optional[QoSSpec]:
-    if _qos is ffi.NULL:
-        return None
-    else:
-        return QoSSpec(delay=_qos.delay,
-                       bandwidth=_qos.bandwidth,
-                       availability=_qos.availability,
-                       loss=_qos.loss,
-                       ber=_qos.ber,
-                       in_order=_qos.in_order,
-                       max_gap=_qos.max_gap,
-                       timeout=_qos.timeout)
-
-# FRCT flags
-FRCT_RETRANSMIT = 0o1
-FRCT_RESCNTL    = 0o2
-FRCT_LINGER     = 0o4
-
 
-# ouroboros exceptions
-class FlowAllocatedException(Exception):
-    pass
+from ouroboros._qosspec import qos_to_qosspec, qosspec_to_qos
+from ouroboros._timespec import fl_to_timespec, timespec_to_fl
+from ouroboros.errors import (
+    FlowAlreadyAllocatedError,
+    FlowDeallocWarning,
+    FlowNotAllocatedError,
+    check_ouroboros_version,
+    raise_errno,
+)
+from ouroboros.qos import QoSSpec
 
+check_ouroboros_version(lib.OUROBOROS_VERSION_MAJOR,
+                        lib.OUROBOROS_VERSION_MINOR)
 
-class FlowNotAllocatedException(Exception):
-    pass
 
+class FrctFlags(IntFlag):
+    """FRCT-level feature flags."""
+    RETRANSMIT = 0o1
+    RESCNTL    = 0o2
+    LINGER     = 0o4
 
-class FlowDownException(Exception):
-    pass
 
-
-class FlowPermissionException(Exception):
-    pass
-
-
-class FlowException(Exception):
-    pass
-
-
-class FlowDeallocWarning(Warning):
-    pass
+class FlowProperties(IntFlag):
+    """Flags describing flow properties and blocking behaviour."""
+    READ_ONLY          = 0o0
+    WRITE_ONLY         = 0o1
+    READ_WRITE         = 0o2
+    DOWN               = 0o4
+    NON_BLOCKING_READ  = 0o1000
+    NON_BLOCKING_WRITE = 0o2000
+    NON_BLOCKING       = NON_BLOCKING_READ | NON_BLOCKING_WRITE
+    NO_PARTIAL_READ    = 0o10000
+    NO_PARTIAL_WRITE   = 0o20000
 
 
-def _raise(e: int) -> None:
-    if e >= 0:
-        return
+class Flow:
+    """Represents an allocated Ouroboros flow."""
 
-    print("error: " + str(e))
-    if e == -errno.ETIMEDOUT:
-        raise TimeoutError()
-    if e == -errno.EINVAL:
-        raise ValueError()
-    if e == -errno.ENOMEM:
-        raise MemoryError()
-    else:
-        raise FlowException()
+    def __init__(self, fd: int = -1) -> None:
+        """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`.
+        """
+        self._fd: int = fd
 
-class FlowProperties(IntFlag):
-    ReadOnly = 0o0
-    WriteOnly = 0o1
-    ReadWrite = 0o2
-    Down = 0o4
-    NonBlockingRead = 0o1000
-    NonBlockingWrite = 0o2000
-    NonBlocking = NonBlockingRead | NonBlockingWrite
-    NoPartialRead = 0o10000
-    NoPartialWrite = 0o200000
+    def __enter__(self) -> Flow:
+        return self
 
+    def __exit__(self, exc_type, exc_value, tb) -> None:
+        self.dealloc()
 
-class Flow:
+    def __del__(self) -> None:
+        try:
+            self.dealloc()
+        except Exception:  # pylint: disable=broad-exception-caught
+            pass  # interpreter shutdown may have torn down lib already
 
-    def __init__(self):
-        self.__fd: int = -1
+    def __repr__(self) -> str:
+        return f"Flow(fd={self._fd})"
 
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_value, tb):
-        lib.flow_dealloc(self.__fd)
+    def fileno(self) -> int:
+        """Return the underlying ouroboros flow descriptor."""
+        return self._fd
 
     def alloc(self,
               dst: str,
-              qos: QoSSpec = None,
-              timeo: float = None) -> Optional[QoSSpec]:
-        """
-        Allocates a flow with a certain QoS to a destination
+              qos: Optional[QoSSpec] = None,
+              timeo: Optional[float] = None) -> Optional[QoSSpec]:
+        """Allocate a flow with a certain QoS to a destination.
 
-        :param dst:   The destination name (string)
-        :param qos:   The QoS for the requested flow (QoSSpec)
-        :param timeo: The timeout for the flow allocation (None -> forever, 0->async)
-        :return:      The QoS for the new flow
+        :param dst:   Destination name.
+        :param qos:   Requested QoS.
+        :param timeo: Allocation timeout (None blocks forever, 0 is async).
+        :return:      The QoS the IRM granted for the new flow.
         """
+        if self._fd >= 0:
+            raise FlowAlreadyAllocatedError()
 
-        if self.__fd >= 0:
-            raise FlowAllocatedException()
-
-        _qos = _qos_to_qosspec(qos)
-
-        _timeo = _fl_to_timespec(timeo)
+        _qos = qos_to_qosspec(ffi, qos)
+        _timeo = fl_to_timespec(ffi, timeo)
 
-        self.__fd = lib.flow_alloc(dst.encode(), _qos, _timeo)
+        rc = lib.flow_alloc(dst.encode(), _qos, _timeo)
+        raise_errno(rc)
+        self._fd = rc
 
-        _raise(self.__fd)
-
-        return _qosspec_to_qos(_qos)
+        return qosspec_to_qos(ffi, _qos)
 
     def accept(self,
-               timeo: float = None) -> QoSSpec:
-        """
-        Accepts new flows and returns the QoS
+               timeo: Optional[float] = None) -> Optional[QoSSpec]:
+        """Accept an incoming flow and return its QoS.
 
-        :param timeo: The timeout for the flow allocation (None -> forever, 0->async)
-        :return:      The QoS for the new flow
+        :param timeo: Accept timeout (None blocks forever, 0 is async).
+        :return:      The QoS of the accepted flow.
         """
-
-        if self.__fd >= 0:
-            raise FlowAllocatedException()
+        if self._fd >= 0:
+            raise FlowAlreadyAllocatedError()
 
         _qos = ffi.new("qosspec_t *")
+        _timeo = fl_to_timespec(ffi, timeo)
 
-        _timeo = _fl_to_timespec(timeo)
+        rc = lib.flow_accept(_qos, _timeo)
+        raise_errno(rc)
+        self._fd = rc
 
-        self.__fd = lib.flow_accept(_qos, _timeo)
-
-        _raise(self.__fd)
-
-        return _qosspec_to_qos(_qos)
+        return qosspec_to_qos(ffi, _qos)
 
     def join(self,
              dst: str,
-             timeo: float = None) -> None:
-        """
-        Join a broadcast layer
-
-        :param dst:   The destination broadcast layer name (string)
-        :param qos:   The QoS for the requested flow (QoSSpec)
-        :param timeo: The timeout for the flow allocation (None -> forever, 0->async)
-        :return:      The QoS for the flow
-        """
-
-        if self.__fd >= 0:
-            raise FlowAllocatedException()
+             timeo: Optional[float] = None) -> None:
+        """Join a broadcast layer.
 
-        _timeo = _fl_to_timespec(timeo)
-
-        self.__fd = lib.flow_join(dst.encode(), _timeo)
-
-        _raise(self.__fd)
-
-    def dealloc(self):
+        :param dst:   Destination broadcast layer name.
+        :param timeo: Join timeout (None blocks forever, 0 is async).
         """
-        Deallocate a flow
+        if self._fd >= 0:
+            raise FlowAlreadyAllocatedError()
 
-        """
+        _timeo = fl_to_timespec(ffi, timeo)
 
-        self.__fd = lib.flow_dealloc(self.__fd)
+        rc = lib.flow_join(dst.encode(), _timeo)
+        raise_errno(rc)
+        self._fd = rc
 
-        if self.__fd < 0:
-            raise FlowDeallocWarning
-
-        self.__fd = -1
+    def dealloc(self) -> None:
+        """Deallocate this flow. Idempotent: 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)
 
     def write(self,
               buf: bytes,
-              count: int = None) -> int:
-        """
-        Attempt to write <count> bytes to a flow
+              count: Optional[int] = None) -> int:
+        """Write up to *count* bytes to the flow.
 
-        :param buf:    Buffer to write from
-        :param count:  Number of bytes to write from the buffer
-        :return:       Number of bytes written
+        :param buf:   Buffer to write from.
+        :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.
         """
-
-        if self.__fd < 0:
-            raise FlowNotAllocatedException()
+        if self._fd < 0:
+            raise FlowNotAllocatedError()
 
         if count is None:
-            return lib.flow_write(self.__fd, ffi.from_buffer(buf), len(buf))
+            count = len(buf)
 
-        return lib.flow_write(self.__fd, ffi.from_buffer(buf), count)
+        rc = lib.flow_write(self._fd, ffi.from_buffer(buf), count)
+        return raise_errno(rc)
 
     def writeline(self,
                   ln: str) -> int:
-        """
-        Attempt to write a string to a flow
+        """Encode *ln* as UTF-8 and write it to the flow.
 
-        :param ln:  String to write
-        :return:    Number of bytes written
+        :param ln: String to write.
+        :return:   Number of bytes written.
         """
+        if self._fd < 0:
+            raise FlowNotAllocatedError()
 
-        if self.__fd < 0:
-            raise FlowNotAllocatedException()
-
-        return self.write(ln.encode(), len(ln))
+        data = ln.encode()
+        return self.write(data, len(data))
 
     def read(self,
-             count: int = None) -> bytes:
-        """
-        Attempt to read bytes from a flow
+             count: Optional[int] = None) -> bytes:
+        """Read up to *count* bytes from the flow.
 
-        :param count:   Maximum number of bytes to read
-        :return:        Bytes read
+        :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.
         """
-
-        if self.__fd < 0:
-            raise FlowNotAllocatedException()
+        if self._fd < 0:
+            raise FlowNotAllocatedError()
 
         if count is None:
             count = 2048
 
         _buf = ffi.new("char []", count)
 
-        result = lib.flow_read(self.__fd, _buf, count)
+        rc = lib.flow_read(self._fd, _buf, count)
+        raise_errno(rc)
 
-        return ffi.unpack(_buf, result)
+        return ffi.unpack(_buf, rc)
 
-    def readline(self):
-        """
-
-        :return: A string
-        """
-        if self.__fd < 0:
-            raise FlowNotAllocatedException()
+    def readline(self) -> str:
+        """Read from the flow and decode the result as UTF-8."""
+        if self._fd < 0:
+            raise FlowNotAllocatedError()
 
         return self.read().decode()
 
-    # flow manipulation
-    def set_snd_timeout(self,
-                        timeo: float):
-        """
-        Set the timeout for blocking writes
-        """
-        _timeo = _fl_to_timespec(timeo)
-
-        if lib.flow_set_snd_timeout(self.__fd, _timeo) != 0:
-            raise FlowPermissionException()
+    def set_snd_timeout(self, timeo: float) -> None:
+        """Set the timeout for blocking writes (seconds)."""
+        _timeo = fl_to_timespec(ffi, timeo)
+        raise_errno(lib.flow_set_snd_timeout(self._fd, _timeo))
 
-    def get_snd_timeout(self) -> float:
-        """
-        Get the timeout for blocking writes
-
-        :return: timeout for blocking writes
-        """
+    def get_snd_timeout(self) -> Optional[float]:
+        """Return the timeout for blocking writes (seconds)."""
         _timeo = ffi.new("struct timespec *")
+        raise_errno(lib.flow_get_snd_timeout(self._fd, _timeo))
+        return timespec_to_fl(ffi, _timeo)
 
-        if lib.flow_get_snd_timeout(self.__fd, _timeo) != 0:
-            raise FlowPermissionException()
-
-        return _timespec_to_fl(_timeo)
-
-    def set_rcv_timeout(self,
-                        timeo: float):
-        """
-        Set the timeout for blocking writes
-        """
-        _timeo = _fl_to_timespec(timeo)
-
-        if lib.flow_set_rcv_timeout(self.__fd, _timeo) != 0:
-            raise FlowPermissionException()
-
-    def get_rcv_timeout(self) -> float:
-        """
-        Get the timeout for blocking reads
+    def set_rcv_timeout(self, timeo: float) -> None:
+        """Set the timeout for blocking reads (seconds)."""
+        _timeo = fl_to_timespec(ffi, timeo)
+        raise_errno(lib.flow_set_rcv_timeout(self._fd, _timeo))
 
-        :return: timeout for blocking writes
-        """
+    def get_rcv_timeout(self) -> Optional[float]:
+        """Return the timeout for blocking reads (seconds)."""
         _timeo = ffi.new("struct timespec *")
+        raise_errno(lib.flow_get_rcv_timeout(self._fd, _timeo))
+        return timespec_to_fl(ffi, _timeo)
 
-        if lib.flow_get_rcv_timeout(self.__fd, _timeo) != 0:
-            raise FlowPermissionException()
-
-        return _timespec_to_fl(_timeo)
-
-    def get_qos(self) -> QoSSpec:
-        """
-
-        :return: Current QoS on the flow
-        """
+    def get_qos(self) -> Optional[QoSSpec]:
+        """Return the current QoS in effect on the flow."""
         _qos = ffi.new("qosspec_t *")
-
-        if lib.flow_get_qos(self.__fd, _qos) != 0:
-            raise FlowPermissionException()
-
-        return _qosspec_to_qos(_qos)
+        raise_errno(lib.flow_get_qos(self._fd, _qos))
+        return qosspec_to_qos(ffi, _qos)
 
     def get_rx_queue_len(self) -> int:
-        """
-
-        :return:
-        """
-
+        """Return the receive queue length (bytes)."""
         size = ffi.new("size_t *")
-
-        if lib.flow_get_rx_qlen(self.__fd, size) != 0:
-            raise FlowPermissionException()
-
-        return int(size)
+        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)."""
+        size = ffi.new("size_t *")
+        raise_errno(lib.flow_get_tx_qlen(self._fd, size))
+        return int(size[0])
 
-        :return:
+    def get_mtu(self) -> int:
+        """Return the per-packet MTU.
+
+        This is the maximum user payload that fits in one n-1 PDU,
+        after crypto headers. Returns 0 if unknown.
         """
+        mtu = ffi.new("size_t *")
+        raise_errno(lib.flow_get_mtu(self._fd, mtu))
+        return int(mtu[0])
 
-        size = ffi.new("size_t *")
+    def set_flags(self, flags: FlowProperties) -> None:
+        """Replace the full set of flags for this flow.
 
-        if lib.flow_get_tx_qlen(self.__fd, size) != 0:
-            raise FlowPermissionException()
+        .. 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`.
 
-        return int(size)
+            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.
 
-    def set_flags(self, flags: FlowProperties):
-        """
-        Set flags for this flow.
-        :param flags:
+        :param flags: Bitmask of :class:`FlowProperties` values.
         """
+        raise_errno(lib.flow_set_flags(self._fd, int(flags)))
 
-        if lib.flow_set_flags(self.__fd, int(flags)):
-            raise FlowPermissionException()
+    def add_flags(self, flags: FlowProperties) -> None:
+        """OR *flags* into the current ``oflags`` value.
 
-    def get_flags(self) -> FlowProperties:
-        """
-        Get the flags for this flow
+        Preserves the access mode and any other already-set bits.
         """
+        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.
 
-        flags = lib.flow_get_flags(self.__fd)
-        if flags < 0:
-            raise FlowPermissionException()
+        Preserves every other bit (including the access mode).
+        """
+        current = self.get_flags()
+        self.set_flags(current & ~FlowProperties(int(flags)))
 
+    def get_flags(self) -> FlowProperties:
+        """Return the current flags for this flow."""
+        flags = raise_errno(lib.flow_get_flags(self._fd))
         return FlowProperties(int(flags))
 
-    def set_frct_flags(self, flags: int):
-        """
-        Set FRCT flags for this flow.
-        :param flags: Bitmask of FRCT_RETRANSMIT, FRCT_RESCNTL, FRCT_LINGER
+    def set_frct_flags(self, flags: FrctFlags) -> None:
+        """Set FRCT flags for this flow.
+
+        :param flags: Bitmask of :class:`FrctFlags`.
         """
+        raise_errno(lib.flow_set_frct_flags(self._fd, int(flags)))
 
-        if lib.flow_set_frct_flags(self.__fd, flags):
-            raise FlowPermissionException()
+    def get_frct_flags(self) -> FrctFlags:
+        """Return the FRCT flags for this flow."""
+        flags = raise_errno(lib.flow_get_frct_flags(self._fd))
+        return FrctFlags(int(flags))
 
-    def get_frct_flags(self) -> int:
-        """
-        Get the FRCT flags for this flow
+    def set_frct_max_sdu(self, size: int) -> None:
+        """Set the maximum reassembly SDU size for FRCT (bytes)."""
+        raise_errno(lib.flow_set_frct_max_sdu(self._fd, size))
 
-        :return: Bitmask of FRCT flags
-        """
+    def get_frct_max_sdu(self) -> int:
+        """Return the maximum reassembly SDU size for FRCT (bytes)."""
+        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)."""
+        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)."""
+        size = ffi.new("size_t *")
+        raise_errno(lib.flow_get_frct_rcv_ring_sz(self._fd, size))
+        return int(size[0])
 
-        flags = lib.flow_get_frct_flags(self.__fd)
-        if flags < 0:
-            raise FlowPermissionException()
+    def set_frct_rto_min(self, rto_ns: int) -> None:
+        """Set the FRCT RTO floor (nanoseconds)."""
+        raise_errno(lib.flow_set_frct_rto_min(self._fd, rto_ns))
 
-        return int(flags)
+    def get_frct_rto_min(self) -> int:
+        """Return the FRCT RTO floor (nanoseconds)."""
+        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: QoSSpec = None,
-               timeo: float = None) -> Flow:
-    """
+               qos: Optional[QoSSpec] = None,
+               timeo: Optional[float] = None) -> Flow:
+    """Allocate a new flow and return the :class:`Flow` wrapper.
 
-    :param dst:    Destination name
-    :param qos:    Requested QoS
-    :param timeo:  Timeout to wait for the allocation
-    :return:       A new Flow()
+    :param dst:   Destination name.
+    :param qos:   Requested QoS.
+    :param timeo: Allocation timeout (None blocks forever, 0 is async).
     """
-
     f = Flow()
     f.alloc(dst, qos, timeo)
     return f
 
 
-def flow_accept(timeo: float = None) -> Flow:
-    """
+def flow_accept(timeo: Optional[float] = None) -> Flow:
+    """Accept an incoming flow and return the :class:`Flow` wrapper.
 
-    :param timeo:  Timeout to wait for the allocation
-    :return:       A new Flow()
+    :param timeo: Accept timeout (None blocks forever, 0 is async).
     """
-
     f = Flow()
     f.accept(timeo)
     return f
 
 
 def flow_join(dst: str,
-              qos: QoSSpec = None,
-              timeo: float = None) -> Flow:
-    """
+              timeo: Optional[float] = None) -> Flow:
+    """Join a broadcast layer and return the :class:`Flow` wrapper.
 
-    :param dst:    Broadcast layer name
-    :param qos:    Requested QoS
-    :param timeo:  Timeout to wait for the allocation
-    :return:       A new Flow()
+    :param dst:   Broadcast layer name.
+    :param timeo: Join timeout (None blocks forever, 0 is async).
     """
-
     f = Flow()
-    f.join(dst, qos, timeo)
+    f.join(dst, timeo)
     return f
diff --git a/ouroboros/event.py b/ouroboros/event.py
index ee0127e..aa03ad4 100644
--- a/ouroboros/event.py
+++ b/ouroboros/event.py
@@ -19,130 +19,137 @@
 # Foundation, Inc., http://www.fsf.org/about/contact/.
 #
 
-from ouroboros.dev import *
-from ouroboros.dev import _fl_to_timespec
+"""Asynchronous flow event monitoring for Ouroboros."""
 
+from __future__ import annotations
 
-# async API
-class FlowEventError(Exception):
-    pass
+from enum import IntFlag
+from typing import List, Optional, Tuple
+
+from _ouroboros_dev_cffi import ffi, lib
+
+from ouroboros._timespec import fl_to_timespec
+from ouroboros.dev import Flow
+from ouroboros.errors import FlowEventError, raise_errno
 
 
 class FEventType(IntFlag):
-    FlowPkt = lib.FLOW_PKT
-    FlowDown = lib.FLOW_DOWN
-    FlowUp = lib.FLOW_UP
-    FlowAlloc = lib.FLOW_ALLOC
-    FlowDealloc = lib.FLOW_DEALLOC
-    FlowPeer = lib.FLOW_PEER
+    """Types of flow events."""
+    FLOW_PKT     = lib.FLOW_PKT
+    FLOW_DOWN    = lib.FLOW_DOWN
+    FLOW_UP      = lib.FLOW_UP
+    FLOW_ALLOC   = lib.FLOW_ALLOC
+    FLOW_DEALLOC = lib.FLOW_DEALLOC
+    FLOW_PEER    = lib.FLOW_PEER
 
 
 class FEventQueue:
-    """
-    A queue of events waiting to be handled
-    """
+    """A queue of flow events waiting to be drained."""
 
-    def __init__(self):
-        self.__fq = lib.fqueue_create()
-        if self.__fq is ffi.NULL:
+    def __init__(self) -> None:
+        self._fq = lib.fqueue_create()
+        if self._fq == ffi.NULL:
             raise MemoryError("Failed to create FEventQueue")
 
-    def __del__(self):
-        lib.fqueue_destroy(self.__fq)
+    def __enter__(self) -> FEventQueue:
+        return self
 
-    def next(self):
-        """
-        Get the next event
-        :return: Flow and eventtype on that flow
-        """
-        f = Flow()
-        f._Flow__fd = lib.fqueue_next(self.__fq)
-        if f._Flow__fd < 0:
-            raise FlowEventError
+    def __exit__(self, exc_type, exc_value, tb) -> None:
+        self.destroy()
+
+    def __del__(self) -> None:
+        self.destroy()
+
+    def destroy(self) -> None:
+        """Destroy the underlying queue. Idempotent."""
+        if self._fq != ffi.NULL:
+            lib.fqueue_destroy(self._fq)
+            self._fq = ffi.NULL
 
-        _type = lib.fqueue_type(self.__fq)
-        if _type < 0:
-            raise FlowEventError
+    def next(self) -> Tuple[Flow, FEventType]:
+        """Return the next ``(flow, event_type)`` pair from the queue."""
+        fd = lib.fqueue_next(self._fq)
+        raise_errno(fd, default=FlowEventError)
 
-        return f, _type
+        _type = lib.fqueue_type(self._fq)
+        raise_errno(_type, default=FlowEventError)
+
+        return Flow(fd), FEventType(_type)
+
+    @property
+    def handle(self):
+        """Return the underlying ``fqueue_t *``.
+
+        Intended for use by :meth:`FlowSet.wait` within the same
+        package; not part of the user-facing API.
+        """
+        return self._fq
 
 
 class FlowSet:
-    """
-    A set of flows that can be monitored for events
-    """
-    def __init__(self,
-                 flows: [Flow] = None):
-
-        self.__set = lib.fset_create()
-        if self.__set is ffi.NULL:
+    """A set of flows that can be monitored for events."""
+
+    def __init__(self, flows: Optional[List[Flow]] = None) -> None:
+        self._set = lib.fset_create()
+        if self._set == ffi.NULL:
             raise MemoryError("Failed to create FlowSet")
 
         if flows is not None:
             for flow in flows:
-                if lib.fset_add(self.__set, flow._Flow__fd) != 0:
-                    lib.fset_destroy(self.__set)
-                    self.__set = ffi.NULL
-                    raise MemoryError("Failed to add flow " + str(flow._Flow__fd) + ".")
+                if lib.fset_add(self._set, flow.fileno()) != 0:
+                    lib.fset_destroy(self._set)
+                    self._set = ffi.NULL
+                    raise MemoryError(f"Failed to add flow {flow.fileno()}.")
 
-    def __enter__(self):
+    def __enter__(self) -> FlowSet:
         return self
 
-    def __exit__(self, exc_type, exc_value, tb):
-        lib.fset_destroy(self.__set)
+    def __exit__(self, exc_type, exc_value, tb) -> None:
+        self.destroy()
 
-    def add(self,
-            flow: Flow):
-        """
-        Add a Flow
+    def __del__(self) -> None:
+        self.destroy()
 
-        :param flow: The flow object to add
-        """
-
-        if self.__set is ffi.NULL:
-            raise ValueError
-
-        if lib.fset_add(self.__set, flow._Flow__fd) != 0:
-            raise MemoryError("Failed to add flow")
-
-    def zero(self):
-        """
-        Remove all Flows from this set
-        """
+    def destroy(self) -> None:
+        """Destroy the underlying set. Idempotent."""
+        if self._set != ffi.NULL:
+            lib.fset_destroy(self._set)
+            self._set = ffi.NULL
 
-        if self.__set is ffi.NULL:
-            raise ValueError
+    def add(self, flow: Flow) -> None:
+        """Add *flow* to this set."""
+        if self._set == ffi.NULL:
+            raise ValueError("FlowSet has been destroyed")
 
-        lib.fset_zero(self.__set)
+        if lib.fset_add(self._set, flow.fileno()) != 0:
+            raise MemoryError(f"Failed to add flow {flow.fileno()}")
 
-    def remove(self,
-               flow: Flow):
-        """
-        Remove a flow from a set
+    def zero(self) -> None:
+        """Remove all flows from this set."""
+        if self._set == ffi.NULL:
+            raise ValueError("FlowSet has been destroyed")
 
-        :param flow:
-        """
+        lib.fset_zero(self._set)
 
-        if self.__set is ffi.NULL:
-            raise ValueError
+    def remove(self, flow: Flow) -> None:
+        """Remove *flow* from this set."""
+        if self._set == ffi.NULL:
+            raise ValueError("FlowSet has been destroyed")
 
-        lib.fset_del(self.__set, flow._Flow__fd)
+        lib.fset_del(self._set, flow.fileno())
 
     def wait(self,
-             fq: FEventType,
-             timeo: float = None):
-        """
-        Wait for at least one event on one of the monitored flows
-        """
-
-        if self.__set is ffi.NULL:
-            raise ValueError
+             fq: FEventQueue,
+             timeo: Optional[float] = None) -> None:
+        """Wait for at least one event on one of the monitored flows.
 
-        _timeo = _fl_to_timespec(timeo)
+        :param fq:    The :class:`FEventQueue` to drain events into.
+        :param timeo: Wait timeout in seconds (None blocks forever).
+        """
+        if self._set == ffi.NULL:
+            raise ValueError("FlowSet has been destroyed")
 
-        ret = lib.fevent(self.__set, fq._FEventQueue__fq, _timeo)
-        if ret < 0:
-            raise FlowEventError
+        _timeo = fl_to_timespec(ffi, timeo)
 
-    def destroy(self):
-        lib.fset_destroy(self.__set)
+        rc = lib.fevent(self._set, fq.handle, _timeo)
+        raise_errno(rc, default=FlowEventError)