p4net.network

Orchestration layer: bring a Topology up end-to-end.

__all__ module-attribute

__all__ = ['Network', 'NetworkAlreadyRunningError', 'NetworkError', 'NetworkNotRunningError', 'NodeNotFoundError', 'RunningHost', 'RunningSwitch']

NetworkAlreadyRunningError

Bases: NetworkError

start() called on a Network that is already running.

Source code in src/p4net/network/exceptions.py

class NetworkAlreadyRunningError(NetworkError):
    """`start()` called on a Network that is already running."""

NetworkError

Bases: P4NetError

Base class for orchestration-layer failures.

Source code in src/p4net/network/exceptions.py

class NetworkError(P4NetError):
    """Base class for orchestration-layer failures."""

NetworkNotRunningError

Bases: NetworkError

An operation that requires a running network was called before start().

Source code in src/p4net/network/exceptions.py

class NetworkNotRunningError(NetworkError):
    """An operation that requires a running network was called before `start()`."""

NodeNotFoundError

Bases: NetworkError

No host or switch with the requested name exists in the network.

Source code in src/p4net/network/exceptions.py

class NodeNotFoundError(NetworkError):
    """No host or switch with the requested name exists in the network."""

RunningHost

A host that has been brought up and is ready to run commands.

Source code in src/p4net/network/nodes.py

class RunningHost:
    """A host that has been brought up and is ready to run commands."""

    def __init__(
        self,
        host: Host,
        namespace: NetworkNamespace,
        interfaces: Mapping[str, str | None],
        interfaces6: Mapping[str, str | None] | None = None,
    ) -> None:
        self._host = host
        self._namespace = namespace
        self._interfaces: dict[str, str | None] = dict(interfaces)
        self._interfaces6: dict[str, str | None] = dict(interfaces6 or {})

    @property
    def name(self) -> str:
        """The host's name as declared in the topology."""
        return self._host.name

    @property
    def descriptor(self) -> Host:
        """The :class:`Host` topology descriptor that produced this runtime."""
        return self._host

    @property
    def namespace(self) -> NetworkNamespace:
        """The Linux network namespace this host runs in."""
        return self._namespace

    @property
    def interfaces(self) -> Mapping[str, str | None]:
        """Map of interface name → IPv4 CIDR (or ``None``) for this host."""
        return self._interfaces

    @property
    def interfaces6(self) -> Mapping[str, str | None]:
        """Map of interface name → IPv6 CIDR (or ``None``) for this host."""
        return self._interfaces6

    @property
    def primary_ip(self) -> str | None:
        """Address of the first configured IPv4 interface (without /mask), or None."""
        for cidr in self._interfaces.values():
            if cidr:
                return _strip_mask(cidr)
        return None

    @property
    def primary_ip6(self) -> str | None:
        """Address of the first configured IPv6 interface (without /mask), or None."""
        for cidr in self._interfaces6.values():
            if cidr:
                return _strip_mask(cidr)
        return None

    def exec(
        self,
        argv: Sequence[str],
        *,
        timeout: float | None = None,
        check: bool = True,
        capture_output: bool = False,
        env: Mapping[str, str] | None = None,
    ) -> subprocess.CompletedProcess[bytes]:
        """Run ``argv`` synchronously inside the host's namespace.

        Args:
            argv: Command and arguments to execute.
            timeout: Per-call timeout in seconds.
            check: Raise :class:`subprocess.CalledProcessError` on rc != 0.
            capture_output: Capture stdout/stderr instead of inheriting.
            env: Optional environment variable overrides.

        Returns:
            The completed :class:`subprocess.CompletedProcess`.
        """
        return self._namespace.exec(
            argv,
            timeout=timeout,
            check=check,
            capture_output=capture_output,
            env=env,
        )

    def popen(
        self,
        argv: Sequence[str],
        *,
        env: Mapping[str, str] | None = None,
        stdout: int | IO[bytes] | None = None,
        stderr: int | IO[bytes] | None = None,
        stdin: int | IO[bytes] | None = None,
    ) -> NSProcess:
        """Spawn a long-running process inside the host's namespace.

        Returns an :class:`NSProcess` whose lifecycle the caller manages.
        """
        return self._namespace.popen(argv, env=env, stdout=stdout, stderr=stderr, stdin=stdin)

    def ping(
        self,
        dst: str | RunningHost,
        *,
        count: int = 1,
        timeout: float = 2.0,
        force_ipv6: bool = False,
    ) -> bool:
        """Send at most ``count`` echo requests; return True if any reply arrived.

        ``timeout`` is passed to ``ping -W`` in seconds, including fractions.
        A process timeout covers the default one-second send interval and stops
        ping with SIGINT so replies already received still count as success.

        Auto-selects IPv4 vs IPv6 based on the target string (``:`` → IPv6),
        or pass ``force_ipv6=True`` to force the IPv6 path. When ``dst`` is a
        :class:`RunningHost`, the IPv4 primary is preferred (least surprise
        for existing callers); pass ``force_ipv6=True`` plus a string target
        if you need IPv6 explicitly.
        """
        if isinstance(dst, RunningHost):
            target = dst.primary_ip if not force_ipv6 else dst.primary_ip6
            if target is None and not force_ipv6:
                target = dst.primary_ip6  # v4-less host falls through to v6
            if target is None:
                raise NetworkError(f"cannot ping host {dst.name!r}: no primary IP configured")
        else:
            target = dst
        is_v6 = force_ipv6 or ":" in target
        probes = int(count)
        deadline = math.ceil(max(probes * timeout, probes - 1 + timeout)) + 1
        with self._namespace.popen(
            [
                "ping",
                "-6" if is_v6 else "-4",
                "-c",
                str(probes),
                "-W",
                str(timeout).removesuffix(".0"),
                target,
            ],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        ) as process:
            try:
                process.communicate(timeout=deadline)
            except subprocess.TimeoutExpired:
                process.send_signal(signal.SIGINT)
                try:
                    process.communicate(timeout=1.0)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.communicate()
            return process.wait() == 0

    def __repr__(self) -> str:
        return f"RunningHost(name={self.name!r}, primary_ip={self.primary_ip!r})"

name property

name: str

The host's name as declared in the topology.

descriptor property

descriptor: Host

The :class:Host topology descriptor that produced this runtime.

namespace property

namespace: NetworkNamespace

The Linux network namespace this host runs in.

interfaces property

interfaces: Mapping[str, str | None]

Map of interface name → IPv4 CIDR (or None) for this host.

interfaces6 property

interfaces6: Mapping[str, str | None]

Map of interface name → IPv6 CIDR (or None) for this host.

primary_ip property

primary_ip: str | None

Address of the first configured IPv4 interface (without /mask), or None.

primary_ip6 property

primary_ip6: str | None

Address of the first configured IPv6 interface (without /mask), or None.

exec

exec(argv: Sequence[str], *, timeout: float | None = None, check: bool = True, capture_output: bool = False, env: Mapping[str, str] | None = None) -> subprocess.CompletedProcess[bytes]

Run argv synchronously inside the host's namespace.

Parameters:

Name	Type	Description	Default
argv	Sequence[str]	Command and arguments to execute.	required
timeout	float | None	Per-call timeout in seconds.	None
check	bool	Raise :class:subprocess.CalledProcessError on rc != 0.	True
capture_output	bool	Capture stdout/stderr instead of inheriting.	False
env	Mapping[str, str] | None	Optional environment variable overrides.	None

Returns:

Type	Description
CompletedProcess[bytes]	The completed :class:subprocess.CompletedProcess.

Source code in src/p4net/network/nodes.py

def exec(
    self,
    argv: Sequence[str],
    *,
    timeout: float | None = None,
    check: bool = True,
    capture_output: bool = False,
    env: Mapping[str, str] | None = None,
) -> subprocess.CompletedProcess[bytes]:
    """Run ``argv`` synchronously inside the host's namespace.

    Args:
        argv: Command and arguments to execute.
        timeout: Per-call timeout in seconds.
        check: Raise :class:`subprocess.CalledProcessError` on rc != 0.
        capture_output: Capture stdout/stderr instead of inheriting.
        env: Optional environment variable overrides.

    Returns:
        The completed :class:`subprocess.CompletedProcess`.
    """
    return self._namespace.exec(
        argv,
        timeout=timeout,
        check=check,
        capture_output=capture_output,
        env=env,
    )

popen

popen(argv: Sequence[str], *, env: Mapping[str, str] | None = None, stdout: int | IO[bytes] | None = None, stderr: int | IO[bytes] | None = None, stdin: int | IO[bytes] | None = None) -> NSProcess

Spawn a long-running process inside the host's namespace.

Returns an :class:NSProcess whose lifecycle the caller manages.

Source code in src/p4net/network/nodes.py

def popen(
    self,
    argv: Sequence[str],
    *,
    env: Mapping[str, str] | None = None,
    stdout: int | IO[bytes] | None = None,
    stderr: int | IO[bytes] | None = None,
    stdin: int | IO[bytes] | None = None,
) -> NSProcess:
    """Spawn a long-running process inside the host's namespace.

    Returns an :class:`NSProcess` whose lifecycle the caller manages.
    """
    return self._namespace.popen(argv, env=env, stdout=stdout, stderr=stderr, stdin=stdin)

ping

ping(dst: str | RunningHost, *, count: int = 1, timeout: float = 2.0, force_ipv6: bool = False) -> bool

Send at most count echo requests; return True if any reply arrived.

timeout is passed to ping -W in seconds, including fractions. A process timeout covers the default one-second send interval and stops ping with SIGINT so replies already received still count as success.

Auto-selects IPv4 vs IPv6 based on the target string (: → IPv6), or pass force_ipv6=True to force the IPv6 path. When dst is a :class:RunningHost, the IPv4 primary is preferred (least surprise for existing callers); pass force_ipv6=True plus a string target if you need IPv6 explicitly.

Source code in src/p4net/network/nodes.py

def ping(
    self,
    dst: str | RunningHost,
    *,
    count: int = 1,
    timeout: float = 2.0,
    force_ipv6: bool = False,
) -> bool:
    """Send at most ``count`` echo requests; return True if any reply arrived.

    ``timeout`` is passed to ``ping -W`` in seconds, including fractions.
    A process timeout covers the default one-second send interval and stops
    ping with SIGINT so replies already received still count as success.

    Auto-selects IPv4 vs IPv6 based on the target string (``:`` → IPv6),
    or pass ``force_ipv6=True`` to force the IPv6 path. When ``dst`` is a
    :class:`RunningHost`, the IPv4 primary is preferred (least surprise
    for existing callers); pass ``force_ipv6=True`` plus a string target
    if you need IPv6 explicitly.
    """
    if isinstance(dst, RunningHost):
        target = dst.primary_ip if not force_ipv6 else dst.primary_ip6
        if target is None and not force_ipv6:
            target = dst.primary_ip6  # v4-less host falls through to v6
        if target is None:
            raise NetworkError(f"cannot ping host {dst.name!r}: no primary IP configured")
    else:
        target = dst
    is_v6 = force_ipv6 or ":" in target
    probes = int(count)
    deadline = math.ceil(max(probes * timeout, probes - 1 + timeout)) + 1
    with self._namespace.popen(
        [
            "ping",
            "-6" if is_v6 else "-4",
            "-c",
            str(probes),
            "-W",
            str(timeout).removesuffix(".0"),
            target,
        ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    ) as process:
        try:
            process.communicate(timeout=deadline)
        except subprocess.TimeoutExpired:
            process.send_signal(signal.SIGINT)
            try:
                process.communicate(timeout=1.0)
            except subprocess.TimeoutExpired:
                process.kill()
                process.communicate()
        return process.wait() == 0

RunningSwitch

A switch whose BMv2 process is up and whose P4Runtime client is connected.

Source code in src/p4net/network/nodes.py

class RunningSwitch:
    """A switch whose BMv2 process is up and whose P4Runtime client is connected."""

    def __init__(
        self,
        switch: P4Switch,
        bmv2: BMv2Switch,
        client: P4RuntimeClient,
        compile_result: CompileResult,
    ) -> None:
        self._switch = switch
        self._bmv2 = bmv2
        self._client = client
        self._compile_result = compile_result
        self._async_client: AsyncP4RuntimeClient | None = None

    @property
    def name(self) -> str:
        """The switch's name as declared in the topology."""
        return self._switch.name

    @property
    def descriptor(self) -> P4Switch:
        """The :class:`P4Switch` topology descriptor that produced this runtime."""
        return self._switch

    @property
    def bmv2(self) -> BMv2Switch:
        """The wrapped BMv2 process (PID, gRPC address, log file)."""
        return self._bmv2

    @property
    def client(self) -> P4RuntimeClient:
        """The P4Runtime gRPC client connected to this switch."""
        return self._client

    @property
    def compile_result(self) -> CompileResult:
        """The compiler output (BMv2 JSON + P4Info paths) used by this switch."""
        return self._compile_result

    @property
    def log_file(self) -> Path:
        """Path to the BMv2 log file."""
        return self._bmv2.log_file

    @property
    def boot_timestamp_us(self) -> int:
        """Wall-clock microseconds since Unix epoch when this switch's BMv2
        process started.

        Combined with INT shim ``ingress_timestamp_us``, gives wall-clock
        arrival time::

            wall_clock_us = switch.boot_timestamp_us + shim.ingress_timestamp_us

        This is the alignment point for comparing timestamps across multiple
        switches — BMv2's per-process ``ingress_global_timestamp`` clock
        starts at zero on each process's boot.

        Raises:
            NetworkNotRunningError: if the underlying BMv2 process has not
                been started (or has already been stopped).
        """
        ts = self._bmv2.boot_timestamp_us
        if ts is None:
            raise NetworkNotRunningError(
                f"switch {self.name!r} has no boot timestamp; BMv2 is not running"
            )
        return ts

    @property
    def async_client(self) -> AsyncP4RuntimeClient:
        """Lazy-constructed async P4Runtime client for this switch.

        Returns an **unconnected** :class:`AsyncP4RuntimeClient`. Call
        ``await async_client.connect()`` to attach. The async client
        receives the sync client's current P4Info index when created.

        The async client generates its own timestamp-based election ID
        on first connection. Its primary status is independent of the
        sync client's. To choose an election ID explicitly, construct an
        :class:`AsyncP4RuntimeClient` yourself. A higher election ID can
        take primary status from an existing client; ``(0, 0)`` permits
        a backup connection for reads.

        The returned instance is cached; subsequent property accesses
        return the same object. ``Network.stop()`` and ``Network.astop()``
        close the cached client on its owning event loop and clear it.

        **Stable** in p4net 1.x since version 1.7.0 — see
        :class:`AsyncP4RuntimeClient`.
        """
        if self._async_client is None:
            self._async_client = AsyncP4RuntimeClient(
                grpc_address=("127.0.0.1", self._bmv2.grpc_port),
                device_id=self._bmv2.device_id,
                info_index=self._client._index,
                thrift_address=("127.0.0.1", self._bmv2.thrift_port),
            )
        return self._async_client

    def _reset_async_client(self) -> None:
        """Drop the cached async client. Called by ``Network.stop()``."""
        self._async_client = None

    def __repr__(self) -> str:
        return f"RunningSwitch(name={self.name!r}, grpc={self._bmv2.grpc_address!r})"

name property

name: str

The switch's name as declared in the topology.

descriptor property

descriptor: P4Switch

The :class:P4Switch topology descriptor that produced this runtime.

bmv2 property

bmv2: BMv2Switch

The wrapped BMv2 process (PID, gRPC address, log file).

client property

client: P4RuntimeClient

The P4Runtime gRPC client connected to this switch.

compile_result property

compile_result: CompileResult

The compiler output (BMv2 JSON + P4Info paths) used by this switch.

log_file property

log_file: Path

Path to the BMv2 log file.

boot_timestamp_us property

boot_timestamp_us: int

Wall-clock microseconds since Unix epoch when this switch's BMv2 process started.

Combined with INT shim ingress_timestamp_us, gives wall-clock arrival time::

wall_clock_us = switch.boot_timestamp_us + shim.ingress_timestamp_us

This is the alignment point for comparing timestamps across multiple switches — BMv2's per-process ingress_global_timestamp clock starts at zero on each process's boot.

Raises:

Type	Description
NetworkNotRunningError	if the underlying BMv2 process has not been started (or has already been stopped).

async_client property

async_client: AsyncP4RuntimeClient

Lazy-constructed async P4Runtime client for this switch.

Returns an unconnected :class:AsyncP4RuntimeClient. Call await async_client.connect() to attach. The async client receives the sync client's current P4Info index when created.

The async client generates its own timestamp-based election ID on first connection. Its primary status is independent of the sync client's. To choose an election ID explicitly, construct an :class:AsyncP4RuntimeClient yourself. A higher election ID can take primary status from an existing client; (0, 0) permits a backup connection for reads.

The returned instance is cached; subsequent property accesses return the same object. Network.stop() and Network.astop() close the cached client on its owning event loop and clear it.

Stable in p4net 1.x since version 1.7.0 — see :class:AsyncP4RuntimeClient.

Network

Brings up a P4 SDN topology end-to-end.

Source code in src/p4net/network/orchestrator.py

class Network:
    """Brings up a P4 SDN topology end-to-end."""

    def __init__(
        self,
        topology: Topology,
        *,
        compiler: P4Compiler | None = None,
        log_dir: Path | None = None,
        pcap_dir: Path | None = None,
        unsafe: bool = False,
        extra_compile_args: Sequence[str] = (),
    ) -> None:
        self._topology = topology
        self._compiler = compiler if compiler is not None else P4Compiler()
        self._log_dir_explicit = log_dir
        self._log_dir: Path | None = None
        self._pcap_dir = pcap_dir
        self._unsafe = unsafe
        self._extra_compile_args: tuple[str, ...] = tuple(extra_compile_args)
        self._running = False
        self._registered = False

        self._namespaces: dict[str, NetworkNamespace] = {}
        self._veth_pairs: list[VethPair] = []
        self._compile_results: dict[str, CompileResult] = {}
        self._bmv2_switches: dict[str, BMv2Switch] = {}
        self._clients: dict[str, P4RuntimeClient] = {}
        self._running_hosts: dict[str, RunningHost] = {}
        self._running_switches: dict[str, RunningSwitch] = {}
        self._host_iface_ip: dict[str, dict[str, str | None]] = {}
        self._host_iface_ip6: dict[str, dict[str, str | None]] = {}
        self._spawned_processes: list[NSProcess] = []
        self._async_disconnects: list[tuple[str, asyncio.AbstractEventLoop, Future[None]]] = []
        self._stop_lock = threading.RLock()
        self._start_future: Future[None] | None = None
        self._start_thread_id: int | None = None
        self._stop_future: Future[None] | None = None
        self._stop_thread_id: int | None = None

    # Read-only views ----------------------------------------------------

    @property
    def is_running(self) -> bool:
        """``True`` once :meth:`start` has succeeded and before :meth:`stop`."""
        return self._running

    @property
    def topology(self) -> Topology:
        """The :class:`Topology` description backing this network."""
        return self._topology

    @property
    def hosts(self) -> Mapping[str, RunningHost]:
        """Map of host name → :class:`RunningHost`. Empty until :meth:`start`."""
        return self._running_hosts

    @property
    def switches(self) -> Mapping[str, RunningSwitch]:
        """Map of switch name → :class:`RunningSwitch`. Empty until :meth:`start`."""
        return self._running_switches

    @property
    def log_dir(self) -> Path:
        """Directory where BMv2 log files are written.

        Raises:
            RuntimeError: if accessed before :meth:`start`.
        """
        if self._log_dir is None:
            raise RuntimeError("log_dir is not yet allocated; call start() first")
        return self._log_dir

    def host(self, name: str) -> RunningHost:
        """Return the :class:`RunningHost` named ``name``.

        Raises:
            NodeNotFoundError: if no such host is in this network.
        """
        rh = self._running_hosts.get(name)
        if rh is None:
            raise NodeNotFoundError(f"no running host named {name!r}")
        return rh

    def switch(self, name: str) -> RunningSwitch:
        """Return the :class:`RunningSwitch` named ``name``.

        Raises:
            NodeNotFoundError: if no such switch is in this network.
        """
        rs = self._running_switches.get(name)
        if rs is None:
            raise NodeNotFoundError(f"no running switch named {name!r}")
        return rs

    @property
    def boot_timestamps(self) -> dict[str, int]:
        """Mapping of switch name to wall-clock μs when its BMv2 started.

        Equivalent to ``{name: self.switch(name).boot_timestamp_us for name
        in self.topology.switches}``, but more concise and stays in sync
        with the running set.

        Returns:
            Fresh dict (callers may mutate it without affecting the
            network's internal state).

        Raises:
            NetworkNotRunningError: if the network has not been started or
                has been stopped.
        """
        if not self._running:
            raise NetworkNotRunningError(
                "Network is not running; call start() before boot_timestamps"
            )
        return {name: rs.boot_timestamp_us for name, rs in self._running_switches.items()}

    # Lifecycle ----------------------------------------------------------

    def start(self) -> None:
        """Bring the topology up end-to-end.

        Validates the topology (unless ``unsafe=True``), compiles each P4
        source, creates host namespaces and veth pairs, configures
        addresses and impairment, launches BMv2 processes, opens
        P4Runtime clients, and pushes pipeline configs. Waits up to five
        seconds for configured IPv6 addresses to finish duplicate address
        detection. On failure, rolls back via :meth:`_do_stop` before re-raising.

        Raises:
            NetworkAlreadyRunningError: if already running or starting.
            NetworkError: if shutdown interrupts startup, or configured IPv6
                addresses fail duplicate address detection or remain unready.
        """
        self._run_start(self._prepare_start())

    def _prepare_start(self, *, background: bool = False) -> Future[None]:
        with self._stop_lock:
            if self._running or self._start_future is not None:
                raise NetworkAlreadyRunningError("Network is already running or starting")
            if self._stop_future is not None and not self._stop_future.done():
                raise NetworkError("Network shutdown is still running")
            self._stop_future = None
            self._start_thread_id = None if background else threading.get_ident()
            self._start_future = Future()
            return self._start_future

    def _end_start(self, future: Future[None]) -> None:
        with self._stop_lock:
            future.set_result(None)
            self._start_future = None
            self._start_thread_id = None

    def _run_start(self, future: Future[None]) -> None:
        with self._stop_lock:
            self._start_thread_id = threading.get_ident()
        try:
            with self._stop_lock:
                if self._stop_future is not None:
                    raise NetworkError("Network startup was interrupted by shutdown")
            self._do_start()
            with self._stop_lock:
                if self._stop_future is not None:
                    raise NetworkError("Network startup was interrupted by shutdown")
                self._running = True
        except BaseException:
            self._do_stop()
            raise
        finally:
            self._end_start(future)

    def _register_cleanup(self) -> None:
        if not self._registered:
            install_handlers()
            register(self)
            self._registered = True

    async def astart(self) -> None:
        """Bring the topology up in a worker thread without blocking the event loop.

        Call from the main thread's event loop to install SIGINT/SIGTERM
        cleanup handlers. Cancellation waits for startup and then shuts
        the network down before propagating. Startup failures roll back
        partial resources and propagate their original exception.

        Raises:
            NetworkAlreadyRunningError: if already running or starting.
            NetworkError: if shutdown is in progress or interrupts startup.
        """
        future = self._prepare_start(background=True)
        try:
            self._register_cleanup()
            startup = asyncio.get_running_loop().run_in_executor(None, self._run_start, future)
        except BaseException:
            try:
                self._do_stop()
            finally:
                self._end_start(future)
            raise

        cancelled = False
        try:
            while not startup.done():
                try:
                    await asyncio.shield(startup)
                except asyncio.CancelledError:
                    cancelled = True
            startup.result()
        except BaseException:
            await wait_for_signal_cleanup()
            if cancelled and self._stop_future is not None:
                await self.astop()
                raise asyncio.CancelledError from None
            raise
        if cancelled:
            await self.astop()
            raise asyncio.CancelledError

    def stop(self) -> None:
        """Tear the network down. Repeated calls are safe.

        Async clients close on their owning event loops. Inside an event
        loop, shutdown runs in a background thread; use :meth:`astop` to
        wait for it. Outside an event loop, this method waits for cleanup.
        Startup already in progress finishes or rolls back before teardown.
        """
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            in_event_loop = False
        else:
            in_event_loop = True

        with self._stop_lock:
            if self._stop_future is None or (self._stop_future.done() and self._async_disconnects):
                self._stop_future = Future()
                initiate = True
            else:
                initiate = False
            future = self._stop_future
        if initiate:
            if in_event_loop:
                threading.Thread(
                    target=self._finish_stop, args=(future,), name="p4net-stop", daemon=True
                ).start()
            else:
                self._finish_stop(future)
        if not in_event_loop and threading.get_ident() not in (
            self._stop_thread_id,
            self._start_thread_id,
        ):
            future.result()

    def _finish_stop(self, future: Future[None]) -> None:
        self._stop_thread_id = threading.get_ident()
        try:
            with self._stop_lock:
                startup = self._start_future
                start_thread_id = self._start_thread_id
            if startup is not None and start_thread_id != threading.get_ident():
                startup.result()
            self._do_stop()
        except BaseException as exc:
            future.set_exception(exc)
        else:
            future.set_result(None)
        finally:
            self._stop_thread_id = None

    async def astop(self) -> None:
        """Wait for async clients and network resources to finish shutting down.

        Client event loops must remain running until shutdown completes.
        Cancellation is propagated after cleanup finishes.
        """
        with self._stop_lock:
            self.stop()
            future = self._stop_future
        assert future is not None
        shutdown = asyncio.wrap_future(future)
        cancelled = False
        while not shutdown.done():
            try:
                await asyncio.shield(shutdown)
            except asyncio.CancelledError:
                cancelled = True
        shutdown.result()
        if self._async_disconnects:
            raise NetworkError("async client shutdown did not finish")
        if cancelled:
            raise asyncio.CancelledError

    # Ping helpers -------------------------------------------------------

    def ping(
        self,
        src: str | RunningHost,
        dst: str | RunningHost,
        *,
        count: int = 1,
        timeout: float = 2.0,
    ) -> bool:
        """Send at most `count` echo requests from `src` to `dst`.

        `src` must resolve to a host in this network. `dst` may be a
        `RunningHost`, the name of a host in this network, or a literal IP
        address (anything else is passed verbatim to the underlying ping).
        Returns True iff at least one reply arrived.
        """
        src_host = src if isinstance(src, RunningHost) else self.host(src)
        if isinstance(dst, RunningHost):
            return src_host.ping(dst, count=count, timeout=timeout)
        # Resolve string `dst`: host name first, otherwise pass through as IP.
        target_host = self._running_hosts.get(dst)
        if target_host is not None:
            return src_host.ping(target_host, count=count, timeout=timeout)
        return src_host.ping(dst, count=count, timeout=timeout)

    def pingall(
        self,
        *,
        count: int = 1,
        timeout: float = 2.0,
    ) -> dict[tuple[str, str], bool]:
        """Run pings between every ordered pair of distinct hosts that have a primary IP."""
        eligible = {name: rh for name, rh in self._running_hosts.items() if rh.primary_ip}
        result: dict[tuple[str, str], bool] = {}
        for src_name, src_host in eligible.items():
            for dst_name, dst_host in eligible.items():
                if src_name == dst_name:
                    continue
                result[(src_name, dst_name)] = src_host.ping(dst_host, count=count, timeout=timeout)
        return result

    def pingall6(
        self,
        *,
        count: int = 1,
        timeout: float = 2.0,
    ) -> dict[tuple[str, str], bool]:
        """IPv6 equivalent of pingall over hosts that have ``primary_ip6`` set.

        Hosts without a primary IPv6 are skipped silently.
        """
        eligible = {name: rh for name, rh in self._running_hosts.items() if rh.primary_ip6}
        result: dict[tuple[str, str], bool] = {}
        for src_name, src_host in eligible.items():
            for dst_name, dst_host in eligible.items():
                if src_name == dst_name:
                    continue
                target_ip6 = dst_host.primary_ip6
                assert target_ip6 is not None  # guarded by `eligible` filter
                result[(src_name, dst_name)] = src_host.ping(
                    target_ip6,
                    count=count,
                    timeout=timeout,
                    force_ipv6=True,
                )
        return result

    # ----- xterm helper -------------------------------------------------

    def xterm(
        self,
        host: str | RunningHost,
        *,
        title: str | None = None,
        shell: str = "bash",
    ) -> NSProcess:
        """Spawn an ``xterm`` running ``shell`` inside ``host``'s namespace.

        Returns the :class:`NSProcess`; the orchestrator tracks it and
        terminates it on :meth:`stop`. Raises :class:`NetworkError` if
        ``$DISPLAY`` is unset (no X server reachable from the current
        process). This method is intended for interactive use; the test
        suite does not exercise it because CI has no X server.
        """
        target = host if isinstance(host, RunningHost) else self.host(host)
        if not os.environ.get("DISPLAY"):
            raise NetworkError(
                "cannot spawn xterm: $DISPLAY is unset (no X server reachable). "
                "Set DISPLAY (e.g. ':0') and ensure xhost permits this process."
            )
        argv = ["xterm", "-T", title or f"p4net: {target.name}", "-e", shell]
        proc = target.popen(argv)
        self._spawned_processes.append(proc)
        return proc

    # ----- Internal start/stop ------------------------------------------

    def _do_start(self) -> None:
        logger.info(
            "Network.start: %d hosts, %d switches, %d links",
            len(self._topology.hosts),
            len(self._topology.switches),
            len(self._topology.links),
        )
        # 1. validate topology
        if not self._unsafe:
            self._topology.validate()

        # 3. log/pcap dirs
        if self._log_dir_explicit is not None:
            self._log_dir = Path(self._log_dir_explicit)
            self._log_dir.mkdir(parents=True, exist_ok=True)
        else:
            self._log_dir = Path(tempfile.mkdtemp(prefix="p4net-"))
        if self._pcap_dir is not None:
            self._pcap_dir.mkdir(parents=True, exist_ok=True)

        # 4. compile each switch's P4 source
        for sw_name, sw in self._topology.switches.items():
            self._compile_results[sw_name] = self._compiler.compile(
                sw.p4_src,
                arch=sw.arch,
                extra_args=self._extra_compile_args,
            )

        # 5. cleanup hooks
        self._register_cleanup()

        # 6. host namespaces; bring lo up
        for h_name in self._topology.hosts:
            ns = NetworkNamespace(h_name)
            ns.create()
            self._namespaces[h_name] = ns
            ns.exec(["ip", "link", "set", "lo", "up"])

        # 7. veth pairs + addressing + impairment
        first_link_seen: set[str] = set()
        for h_name in self._topology.hosts:
            self._host_iface_ip[h_name] = {}
            self._host_iface_ip6[h_name] = {}
        for link in self._topology.links:
            self._wire_link(link, first_link_seen)

        # 8. default routes for hosts
        for h_name, host in self._topology.hosts.items():
            if host.default_route:
                self._namespaces[h_name].exec(
                    ["ip", "route", "add", "default", "via", host.default_route]
                )
            if host.default_route6:
                self._namespaces[h_name].exec(
                    ["ip", "-6", "route", "add", "default", "via", host.default_route6]
                )

        # 9. BMv2 switches
        for sw_name, sw in self._topology.switches.items():
            port_to_iface = self._port_to_iface_for(sw_name)
            compile_result = self._compile_results[sw_name]
            assert self._log_dir is not None
            bmv2 = BMv2Switch(
                sw_name,
                device_id=int(sw.device_id) if sw.device_id is not None else 0,
                grpc_port=int(sw.grpc_port) if sw.grpc_port is not None else 50051,
                thrift_port=int(sw.thrift_port) if sw.thrift_port is not None else 9090,
                bmv2_json=compile_result.bmv2_json,
                port_to_iface=port_to_iface,
                log_dir=self._log_dir,
                pcap_dir=self._pcap_dir if sw.pcap_enabled else None,
                cpu_port=sw.cpu_port,
                log_level=sw.log_level,
            )
            bmv2.start()
            self._bmv2_switches[sw_name] = bmv2
            bmv2.wait_until_ready()
            logger.info("BMv2 %r ready on %s", sw_name, bmv2.grpc_address)

        # 10. P4Runtime clients
        election_id = (int(time.time_ns() // 1_000_000), 0)
        for sw_name, bmv2 in self._bmv2_switches.items():
            sw = self._topology.switches[sw_name]
            client = P4RuntimeClient(
                bmv2.grpc_address,
                device_id=int(sw.device_id) if sw.device_id is not None else 0,
                election_id=election_id,
                thrift_address=("127.0.0.1", int(bmv2.thrift_port)),
            )
            client.connect()
            self._clients[sw_name] = client
            client.set_pipeline_config(
                bmv2_json=self._compile_results[sw_name].bmv2_json,
                p4info=self._compile_results[sw_name].p4info,
            )
            logger.info("P4Runtime client connected to %r as primary", sw_name)

        # 11. Configured IPv6 addresses must be usable before startup succeeds.
        self._wait_for_ipv6_addresses()

        # 12. RunningHost / RunningSwitch facades
        for h_name, host in self._topology.hosts.items():
            self._running_hosts[h_name] = RunningHost(
                host,
                self._namespaces[h_name],
                self._host_iface_ip[h_name],
                self._host_iface_ip6[h_name],
            )
        for sw_name, sw in self._topology.switches.items():
            self._running_switches[sw_name] = RunningSwitch(
                sw,
                self._bmv2_switches[sw_name],
                self._clients[sw_name],
                self._compile_results[sw_name],
            )
        logger.info("Network.start: ready")

    def _wait_for_ipv6_addresses(self) -> None:
        pending = {
            (host, iface): ipaddress.IPv6Interface(cidr).ip
            for host, interfaces in self._host_iface_ip6.items()
            for iface, cidr in interfaces.items()
            if cidr is not None
        }
        if not pending:
            return
        timeout = _IPV6_READY_TIMEOUT
        deadline = time.monotonic() + timeout

        def not_ready() -> NetworkError:
            addresses = ", ".join(
                f"{host}/{iface} ({address})" for (host, iface), address in pending.items()
            )
            return NetworkError(f"IPv6 addresses not ready after {timeout:g}s: {addresses}")

        while pending:
            with self._stop_lock:
                if self._stop_future is not None:
                    raise NetworkError("Network startup was interrupted by shutdown")
            for host in dict.fromkeys(name for name, _ in pending):
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    raise not_ready()
                try:
                    result = self._namespaces[host].exec(
                        ["ip", "-j", "-6", "addr", "show"],
                        capture_output=True,
                        timeout=remaining,
                    )
                except subprocess.TimeoutExpired as exc:
                    raise not_ready() from exc
                for interface in json.loads(result.stdout):
                    key = (host, interface["ifname"])
                    if key not in pending:
                        continue
                    for address in interface.get("addr_info", []):
                        if ipaddress.IPv6Address(address["local"]) != pending[key]:
                            continue
                        if address.get("dadfailed", False):
                            raise NetworkError(
                                f"IPv6 address {pending[key]} on {host}/{key[1]} "
                                "failed duplicate address detection"
                            )
                        if not address.get("tentative", False):
                            del pending[key]
                        break
            if pending:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    raise not_ready()
                time.sleep(min(0.05, remaining))

    def _wire_link(self, link: Link, first_link_seen: set[str]) -> None:
        name_a = link.a.iface_name
        name_b = link.b.iface_name
        assert name_a is not None and name_b is not None, (
            "link endpoints must have iface_name resolved by Topology"
        )
        node_a = self._topology.node(link.a.node)
        node_b = self._topology.node(link.b.node)
        veth = VethPair(
            name_a,
            name_b,
            namespace_a=self._namespaces[node_a.name] if isinstance(node_a, Host) else None,
            namespace_b=self._namespaces[node_b.name] if isinstance(node_b, Host) else None,
        )
        veth.create()
        self._veth_pairs.append(veth)

        # Configure addresses + MAC + MTU + state.
        # Host-side configuration goes through `ip` inside the host namespace
        # (avoids per-side pyroute2.NetNS churn that surfaces flakiness when
        # several veth pairs are wired in rapid succession). Switch-side
        # configuration stays on VethPair's root-ns netlink path.
        for side, ep in (("a", link.a), ("b", link.b)):
            node = self._topology.node(ep.node)
            assert ep.iface_name is not None
            ip_to_use: str | None = None
            ip6_to_use: str | None = None
            mac_to_use: str | None = None
            if isinstance(node, Host):
                ip_to_use, ip6_to_use = _effective_host_addresses(
                    node, ep, first_link=node.name not in first_link_seen
                )
                if ep.mac is not None:
                    mac_to_use = ep.mac
                elif node.mac is not None and node.name not in first_link_seen:
                    mac_to_use = node.mac
                first_link_seen.add(node.name)
                self._host_iface_ip[node.name][ep.iface_name] = ip_to_use
                self._host_iface_ip6[node.name][ep.iface_name] = ip6_to_use
                # Gate IPv6 BEFORE bringing the iface up so the kernel does not
                # auto-configure a link-local address we don't want.
                ns = self._namespaces[node.name]
                if ip6_to_use is not None:
                    enable_ipv6(ns, ep.iface_name)
                else:
                    disable_ipv6(ns, ep.iface_name)
                self._configure_host_iface(
                    ns,
                    ep.iface_name,
                    ip=ip_to_use,
                    ip6=ip6_to_use,
                    mac=mac_to_use,
                    mtu=link.mtu,
                )
            else:
                # Switch endpoint lives in the root namespace. Suppress its
                # IPv6 link-local so MLD chatter doesn't leak into PCAPs or
                # the CPU-punt stream.
                disable_ipv6(None, ep.iface_name)
                if ep.mac is not None:
                    mac_to_use = ep.mac
                if mac_to_use is not None:
                    veth.set_mac(side, mac_to_use)  # type: ignore[arg-type]
                if link.mtu is not None:
                    veth.set_mtu(side, link.mtu)  # type: ignore[arg-type]
                veth.set_up(side)  # type: ignore[arg-type]

        # Apply per-direction netem. The veth side at endpoint a shapes the
        # a→b direction (egress at a == arrives at b); same for b→a at b.
        a_rate, a_delay, a_jitter, a_loss = self._direction_params(link, "a_to_b")
        if any(v is not None for v in (a_rate, a_delay, a_jitter, a_loss)):
            node_a = self._topology.node(link.a.node)
            ns_a = self._namespaces[node_a.name] if isinstance(node_a, Host) else None
            assert link.a.iface_name is not None
            apply_netem(
                ns_a,
                link.a.iface_name,
                rate=a_rate,
                delay=a_delay,
                jitter=a_jitter,
                loss_pct=a_loss,
            )
        b_rate, b_delay, b_jitter, b_loss = self._direction_params(link, "b_to_a")
        if any(v is not None for v in (b_rate, b_delay, b_jitter, b_loss)):
            node_b = self._topology.node(link.b.node)
            ns_b = self._namespaces[node_b.name] if isinstance(node_b, Host) else None
            assert link.b.iface_name is not None
            apply_netem(
                ns_b,
                link.b.iface_name,
                rate=b_rate,
                delay=b_delay,
                jitter=b_jitter,
                loss_pct=b_loss,
            )

    @staticmethod
    def _direction_params(
        link: Link,
        direction: str,
    ) -> tuple[str | None, str | None, str | None, float | None]:
        """Return ``(rate, delay, jitter, loss_pct)`` netem args for one direction.

        Each element falls back to the symmetric value if the matching
        ``*_a_to_b`` / ``*_b_to_a`` field is unset. If the matching
        ``*_extra`` field is set, it is summed on top of the symmetric base.
        """
        suffix = "_a_to_b" if direction == "a_to_b" else "_b_to_a"
        rate: str | None = getattr(link, "bandwidth" + suffix) or link.bandwidth
        delay = _resolve_dir_str(
            link.delay, getattr(link, "delay" + suffix), getattr(link, "delay" + suffix + "_extra")
        )
        jitter = _resolve_dir_str(
            link.jitter,
            getattr(link, "jitter" + suffix),
            getattr(link, "jitter" + suffix + "_extra"),
        )
        loss = _resolve_dir_loss(
            link.loss_pct,
            getattr(link, "loss_pct" + suffix),
            getattr(link, "loss_pct" + suffix + "_extra"),
        )
        return rate, delay, jitter, loss

    @staticmethod
    def _configure_host_iface(
        ns: NetworkNamespace,
        iface: str,
        *,
        ip: str | None,
        ip6: str | None,
        mac: str | None,
        mtu: int | None,
    ) -> None:
        if mac is not None:
            ns.exec(["ip", "link", "set", iface, "address", mac])
        if mtu is not None:
            ns.exec(["ip", "link", "set", iface, "mtu", str(mtu)])
        if ip is not None:
            ns.exec(["ip", "addr", "add", ip, "dev", iface])
        if ip6 is not None:
            ns.exec(["ip", "-6", "addr", "add", ip6, "dev", iface])
        ns.exec(["ip", "link", "set", iface, "up"])

    def _port_to_iface_for(self, switch_name: str) -> dict[int, str]:
        result: dict[int, str] = {}
        for link in self._topology.links:
            for ep in (link.a, link.b):
                if ep.node == switch_name and ep.port is not None and ep.iface_name is not None:
                    result[int(ep.port)] = ep.iface_name
        return result

    def _do_stop(self) -> None:
        if self._running:
            logger.info("Network.stop: tearing down")
        # 0. user-spawned processes (xterm, etc.) — reap before namespaces vanish.
        for proc in list(self._spawned_processes):
            try:
                if proc.poll() is None:
                    proc.terminate()
                    try:
                        proc.wait(timeout=2.0)
                    except Exception:
                        proc.kill()
            except Exception as exc:
                logger.warning("spawned process reap (pid=%s): %r", getattr(proc, "pid", "?"), exc)
        self._spawned_processes.clear()

        # 1a. Submit disconnects to the loops that own the gRPC resources.
        for sw_name, rs in list(self._running_switches.items()):
            ac = rs._async_client
            if ac is None:
                continue
            try:
                loop = ac._loop
                if loop is not None:
                    if not loop.is_running():
                        raise RuntimeError("async client event loop is not running")
                    disconnect = ac.disconnect()
                    try:
                        future = asyncio.run_coroutine_threadsafe(disconnect, loop)
                    except BaseException:
                        disconnect.close()
                        raise
                    self._async_disconnects.append((sw_name, loop, future))
            except Exception as exc:
                logger.warning("async client disconnect %r: %r", sw_name, exc)
            rs._reset_async_client()

        try:
            current_loop = asyncio.get_running_loop()
        except RuntimeError:
            current_loop = None
        pending = []
        for sw_name, loop, future in self._async_disconnects:
            if loop is current_loop and not future.done():
                pending.append((sw_name, loop, future))
                continue
            try:
                future.result(timeout=5.0)
            except FutureTimeoutError:
                pending.append((sw_name, loop, future))
                logger.warning("async client disconnect %r timed out", sw_name)
            except Exception as exc:
                logger.warning("async client disconnect %r: %r", sw_name, exc)
        self._async_disconnects = pending

        # 1. P4Runtime clients
        for sw_name, client in list(self._clients.items()):
            try:
                client.disconnect()
            except Exception as exc:
                logger.warning("disconnect %r: %r", sw_name, exc)
        self._clients.clear()

        # 2. BMv2 switches
        for sw_name, bmv2 in list(self._bmv2_switches.items()):
            try:
                bmv2.stop()
            except Exception as exc:
                logger.warning("BMv2 stop %r: %r", sw_name, exc)
        self._bmv2_switches.clear()

        # 3. veth pairs
        for veth in list(self._veth_pairs):
            try:
                veth.destroy()
            except Exception as exc:
                logger.warning("veth destroy %r: %r", veth, exc)
        self._veth_pairs.clear()

        # 4. namespaces
        for h_name, ns in list(self._namespaces.items()):
            try:
                if ns.exists:
                    ns.destroy()
            except Exception as exc:
                logger.warning("ns destroy %r: %r", h_name, exc)
        self._namespaces.clear()

        # 5. unregister hooks
        if self._registered:
            with contextlib.suppress(Exception):
                unregister(self)
            self._registered = False

        # 6. clear facades + flag
        self._running_hosts.clear()
        self._running_switches.clear()
        self._compile_results.clear()
        self._host_iface_ip.clear()
        self._host_iface_ip6.clear()
        self._running = False

    # Context manager ----------------------------------------------------

    def __enter__(self) -> Network:
        self.start()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        self.stop()

is_running property

is_running: bool

True once :meth:start has succeeded and before :meth:stop.

topology property

topology: Topology

The :class:Topology description backing this network.

hosts property

hosts: Mapping[str, RunningHost]

Map of host name → :class:RunningHost. Empty until :meth:start.

switches property

switches: Mapping[str, RunningSwitch]

Map of switch name → :class:RunningSwitch. Empty until :meth:start.

log_dir property

log_dir: Path

Directory where BMv2 log files are written.

Raises:

Type	Description
RuntimeError	if accessed before :meth:start.

boot_timestamps property

boot_timestamps: dict[str, int]

Mapping of switch name to wall-clock μs when its BMv2 started.

Equivalent to {name: self.switch(name).boot_timestamp_us for name in self.topology.switches}, but more concise and stays in sync with the running set.

Returns:

Type	Description
dict[str, int]	Fresh dict (callers may mutate it without affecting the
dict[str, int]	network's internal state).

Raises:

Type	Description
NetworkNotRunningError	if the network has not been started or has been stopped.

host

host(name: str) -> RunningHost

Return the :class:RunningHost named name.

Raises:

Type	Description
NodeNotFoundError	if no such host is in this network.

Source code in src/p4net/network/orchestrator.py

def host(self, name: str) -> RunningHost:
    """Return the :class:`RunningHost` named ``name``.

    Raises:
        NodeNotFoundError: if no such host is in this network.
    """
    rh = self._running_hosts.get(name)
    if rh is None:
        raise NodeNotFoundError(f"no running host named {name!r}")
    return rh

switch

switch(name: str) -> RunningSwitch

Return the :class:RunningSwitch named name.

Raises:

Type	Description
NodeNotFoundError	if no such switch is in this network.

Source code in src/p4net/network/orchestrator.py

def switch(self, name: str) -> RunningSwitch:
    """Return the :class:`RunningSwitch` named ``name``.

    Raises:
        NodeNotFoundError: if no such switch is in this network.
    """
    rs = self._running_switches.get(name)
    if rs is None:
        raise NodeNotFoundError(f"no running switch named {name!r}")
    return rs

start

start() -> None

Bring the topology up end-to-end.

Validates the topology (unless unsafe=True), compiles each P4 source, creates host namespaces and veth pairs, configures addresses and impairment, launches BMv2 processes, opens P4Runtime clients, and pushes pipeline configs. Waits up to five seconds for configured IPv6 addresses to finish duplicate address detection. On failure, rolls back via :meth:_do_stop before re-raising.

Raises:

Type	Description
NetworkAlreadyRunningError	if already running or starting.
NetworkError	if shutdown interrupts startup, or configured IPv6 addresses fail duplicate address detection or remain unready.

Source code in src/p4net/network/orchestrator.py

def start(self) -> None:
    """Bring the topology up end-to-end.

    Validates the topology (unless ``unsafe=True``), compiles each P4
    source, creates host namespaces and veth pairs, configures
    addresses and impairment, launches BMv2 processes, opens
    P4Runtime clients, and pushes pipeline configs. Waits up to five
    seconds for configured IPv6 addresses to finish duplicate address
    detection. On failure, rolls back via :meth:`_do_stop` before re-raising.

    Raises:
        NetworkAlreadyRunningError: if already running or starting.
        NetworkError: if shutdown interrupts startup, or configured IPv6
            addresses fail duplicate address detection or remain unready.
    """
    self._run_start(self._prepare_start())

astart async

astart() -> None

Bring the topology up in a worker thread without blocking the event loop.

Call from the main thread's event loop to install SIGINT/SIGTERM cleanup handlers. Cancellation waits for startup and then shuts the network down before propagating. Startup failures roll back partial resources and propagate their original exception.

Raises:

Type	Description
NetworkAlreadyRunningError	if already running or starting.
NetworkError	if shutdown is in progress or interrupts startup.

Source code in src/p4net/network/orchestrator.py

async def astart(self) -> None:
    """Bring the topology up in a worker thread without blocking the event loop.

    Call from the main thread's event loop to install SIGINT/SIGTERM
    cleanup handlers. Cancellation waits for startup and then shuts
    the network down before propagating. Startup failures roll back
    partial resources and propagate their original exception.

    Raises:
        NetworkAlreadyRunningError: if already running or starting.
        NetworkError: if shutdown is in progress or interrupts startup.
    """
    future = self._prepare_start(background=True)
    try:
        self._register_cleanup()
        startup = asyncio.get_running_loop().run_in_executor(None, self._run_start, future)
    except BaseException:
        try:
            self._do_stop()
        finally:
            self._end_start(future)
        raise

    cancelled = False
    try:
        while not startup.done():
            try:
                await asyncio.shield(startup)
            except asyncio.CancelledError:
                cancelled = True
        startup.result()
    except BaseException:
        await wait_for_signal_cleanup()
        if cancelled and self._stop_future is not None:
            await self.astop()
            raise asyncio.CancelledError from None
        raise
    if cancelled:
        await self.astop()
        raise asyncio.CancelledError

stop

stop() -> None

Tear the network down. Repeated calls are safe.

Async clients close on their owning event loops. Inside an event loop, shutdown runs in a background thread; use :meth:astop to wait for it. Outside an event loop, this method waits for cleanup. Startup already in progress finishes or rolls back before teardown.

Source code in src/p4net/network/orchestrator.py

def stop(self) -> None:
    """Tear the network down. Repeated calls are safe.

    Async clients close on their owning event loops. Inside an event
    loop, shutdown runs in a background thread; use :meth:`astop` to
    wait for it. Outside an event loop, this method waits for cleanup.
    Startup already in progress finishes or rolls back before teardown.
    """
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        in_event_loop = False
    else:
        in_event_loop = True

    with self._stop_lock:
        if self._stop_future is None or (self._stop_future.done() and self._async_disconnects):
            self._stop_future = Future()
            initiate = True
        else:
            initiate = False
        future = self._stop_future
    if initiate:
        if in_event_loop:
            threading.Thread(
                target=self._finish_stop, args=(future,), name="p4net-stop", daemon=True
            ).start()
        else:
            self._finish_stop(future)
    if not in_event_loop and threading.get_ident() not in (
        self._stop_thread_id,
        self._start_thread_id,
    ):
        future.result()

astop async

astop() -> None

Wait for async clients and network resources to finish shutting down.

Client event loops must remain running until shutdown completes. Cancellation is propagated after cleanup finishes.

Source code in src/p4net/network/orchestrator.py

async def astop(self) -> None:
    """Wait for async clients and network resources to finish shutting down.

    Client event loops must remain running until shutdown completes.
    Cancellation is propagated after cleanup finishes.
    """
    with self._stop_lock:
        self.stop()
        future = self._stop_future
    assert future is not None
    shutdown = asyncio.wrap_future(future)
    cancelled = False
    while not shutdown.done():
        try:
            await asyncio.shield(shutdown)
        except asyncio.CancelledError:
            cancelled = True
    shutdown.result()
    if self._async_disconnects:
        raise NetworkError("async client shutdown did not finish")
    if cancelled:
        raise asyncio.CancelledError

ping

ping(src: str | RunningHost, dst: str | RunningHost, *, count: int = 1, timeout: float = 2.0) -> bool

Send at most count echo requests from src to dst.

src must resolve to a host in this network. dst may be a RunningHost, the name of a host in this network, or a literal IP address (anything else is passed verbatim to the underlying ping). Returns True iff at least one reply arrived.

Source code in src/p4net/network/orchestrator.py

def ping(
    self,
    src: str | RunningHost,
    dst: str | RunningHost,
    *,
    count: int = 1,
    timeout: float = 2.0,
) -> bool:
    """Send at most `count` echo requests from `src` to `dst`.

    `src` must resolve to a host in this network. `dst` may be a
    `RunningHost`, the name of a host in this network, or a literal IP
    address (anything else is passed verbatim to the underlying ping).
    Returns True iff at least one reply arrived.
    """
    src_host = src if isinstance(src, RunningHost) else self.host(src)
    if isinstance(dst, RunningHost):
        return src_host.ping(dst, count=count, timeout=timeout)
    # Resolve string `dst`: host name first, otherwise pass through as IP.
    target_host = self._running_hosts.get(dst)
    if target_host is not None:
        return src_host.ping(target_host, count=count, timeout=timeout)
    return src_host.ping(dst, count=count, timeout=timeout)

pingall

pingall(*, count: int = 1, timeout: float = 2.0) -> dict[tuple[str, str], bool]

Run pings between every ordered pair of distinct hosts that have a primary IP.

Source code in src/p4net/network/orchestrator.py

def pingall(
    self,
    *,
    count: int = 1,
    timeout: float = 2.0,
) -> dict[tuple[str, str], bool]:
    """Run pings between every ordered pair of distinct hosts that have a primary IP."""
    eligible = {name: rh for name, rh in self._running_hosts.items() if rh.primary_ip}
    result: dict[tuple[str, str], bool] = {}
    for src_name, src_host in eligible.items():
        for dst_name, dst_host in eligible.items():
            if src_name == dst_name:
                continue
            result[(src_name, dst_name)] = src_host.ping(dst_host, count=count, timeout=timeout)
    return result

pingall6

pingall6(*, count: int = 1, timeout: float = 2.0) -> dict[tuple[str, str], bool]

IPv6 equivalent of pingall over hosts that have primary_ip6 set.

Hosts without a primary IPv6 are skipped silently.

Source code in src/p4net/network/orchestrator.py

def pingall6(
    self,
    *,
    count: int = 1,
    timeout: float = 2.0,
) -> dict[tuple[str, str], bool]:
    """IPv6 equivalent of pingall over hosts that have ``primary_ip6`` set.

    Hosts without a primary IPv6 are skipped silently.
    """
    eligible = {name: rh for name, rh in self._running_hosts.items() if rh.primary_ip6}
    result: dict[tuple[str, str], bool] = {}
    for src_name, src_host in eligible.items():
        for dst_name, dst_host in eligible.items():
            if src_name == dst_name:
                continue
            target_ip6 = dst_host.primary_ip6
            assert target_ip6 is not None  # guarded by `eligible` filter
            result[(src_name, dst_name)] = src_host.ping(
                target_ip6,
                count=count,
                timeout=timeout,
                force_ipv6=True,
            )
    return result

xterm

xterm(host: str | RunningHost, *, title: str | None = None, shell: str = 'bash') -> NSProcess

Spawn an xterm running shell inside host's namespace.

Returns the :class:NSProcess; the orchestrator tracks it and terminates it on :meth:stop. Raises :class:NetworkError if $DISPLAY is unset (no X server reachable from the current process). This method is intended for interactive use; the test suite does not exercise it because CI has no X server.

Source code in src/p4net/network/orchestrator.py

def xterm(
    self,
    host: str | RunningHost,
    *,
    title: str | None = None,
    shell: str = "bash",
) -> NSProcess:
    """Spawn an ``xterm`` running ``shell`` inside ``host``'s namespace.

    Returns the :class:`NSProcess`; the orchestrator tracks it and
    terminates it on :meth:`stop`. Raises :class:`NetworkError` if
    ``$DISPLAY`` is unset (no X server reachable from the current
    process). This method is intended for interactive use; the test
    suite does not exercise it because CI has no X server.
    """
    target = host if isinstance(host, RunningHost) else self.host(host)
    if not os.environ.get("DISPLAY"):
        raise NetworkError(
            "cannot spawn xterm: $DISPLAY is unset (no X server reachable). "
            "Set DISPLAY (e.g. ':0') and ensure xhost permits this process."
        )
    argv = ["xterm", "-T", title or f"p4net: {target.name}", "-e", shell]
    proc = target.popen(argv)
    self._spawned_processes.append(proc)
    return proc