From 35c41c2f59dce48d5af93e8ab304ca395d2d59ca Mon Sep 17 00:00:00 2001 From: Jaeyoon Choi Date: Wed, 29 Jul 2026 15:57:20 +0900 Subject: [PATCH 1/5] fix: --device lookup read an unmapped key and matched nothing 0.3.10 taught --device to bypass the class filter. The check reads props.get("bdf"), but the scan loop stores the address under "slot", the key lspci prints. The rename to bdf happens later, in Device.from_dict. The lookup always came back empty. --device matched nothing at all, not even NVMe devices that used to match. The command still exited 0, so the failure was silent. Read the slot key, and add a regression test for the bypass. Signed-off-by: Jaeyoon Choi --- src/devbind/devbind.py | 2 +- tests/test_linux.py | 44 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) create mode 100644 tests/test_linux.py diff --git a/src/devbind/devbind.py b/src/devbind/devbind.py index c85ded3..15ca198 100644 --- a/src/devbind/devbind.py +++ b/src/devbind/devbind.py @@ -219,7 +219,7 @@ def device_scan(args): for line in proc.stdout.splitlines(): if not line: classcode = int(props.get("classcode", "0"), 16) - bdf = props.get("bdf", "") + bdf = props.get("slot", "") if args.device: matches = args.device == bdf else: diff --git a/tests/test_linux.py b/tests/test_linux.py new file mode 100644 index 0000000..01ead26 --- /dev/null +++ b/tests/test_linux.py @@ -0,0 +1,44 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) Jaeyoon Choi +"""Unit tests for the Linux backend (runnable on any platform).""" + +from types import SimpleNamespace + +from devbind import devbind + + +# Representative `lspci -Dvmmnk` output: an NVMe controller and a +# non-matching ethernet device. +LSPCI_DVMMNK = ( + "Slot:\t0000:01:00.0\n" + "Class:\t0108\n" + "Vendor:\t144d\n" + "Device:\ta808\n" + "SVendor:\t144d\n" + "SDevice:\ta801\n" + "Driver:\tnvme\n" + "Module:\tnvme\n" + "\n" + "Slot:\t0000:03:00.0\n" + "Class:\t0200\n" + "Vendor:\t8086\n" + "Device:\t10d3\n" + "\n" +) + + +def test_device_scan_named_device_bypasses_class_filter(monkeypatch): + monkeypatch.setattr( + devbind, + "run", + lambda cmd: SimpleNamespace(stdout=LSPCI_DVMMNK, stderr="", returncode=0), + ) + # Avoid touching the real /sys and /dev during probing + for name in ("probe_handles", "probe_usage", "probe_driver", "probe_iommugroup"): + monkeypatch.setattr(devbind.Device, name, lambda self: None) + + args = SimpleNamespace(device="0000:03:00.0", classcode=0x0108) + devices = list(devbind.device_scan(args)) + + # The ethernet device is found by BDF even though its class is 0x0200 + assert [d.bdf for d in devices] == ["0000:03:00.0"] From a7751d34f48c941820a4d18ba3ad862acfd472b6 Mon Sep 17 00:00:00 2001 From: Jaeyoon Choi Date: Wed, 29 Jul 2026 16:19:18 +0900 Subject: [PATCH 2/5] refactor: decouple scanning and binding from parsed arguments device_scan, bind, and unbind read the argparse namespace directly. That coupled them to the CLI. unbind never used args at all. bind took it only to pass it along. Tests could not call the helpers without building a fake args object. Take classcode, bdf, and driver_name as plain parameters instead. main() passes the parsed values at the call sites. This keeps the upcoming Backend interface free of argparse. Behavior is unchanged. Signed-off-by: Jaeyoon Choi --- src/devbind/devbind.py | 28 +++++++++++++--------------- tests/test_linux.py | 3 +-- 2 files changed, 14 insertions(+), 17 deletions(-) diff --git a/src/devbind/devbind.py b/src/devbind/devbind.py index 15ca198..76647b1 100644 --- a/src/devbind/devbind.py +++ b/src/devbind/devbind.py @@ -205,12 +205,12 @@ def probe_usage(self): self.is_used = bool(proc.stdout) -def device_scan(args): +def device_scan(classcode: int, bdf: Optional[str] = None): """Yields matching PCIe devices. - When args.device is set, the class filter is bypassed and the named device - is yielded regardless of its class. Otherwise devices whose class matches - args.classcode are yielded. + When bdf is set, the class filter is bypassed and the named device is + yielded regardless of its class. Otherwise devices whose class matches + classcode are yielded. """ proc = run("lspci -Dvmmnk") @@ -218,12 +218,10 @@ def device_scan(args): props = {} for line in proc.stdout.splitlines(): if not line: - classcode = int(props.get("classcode", "0"), 16) - bdf = props.get("slot", "") - if args.device: - matches = args.device == bdf + if bdf: + matches = bdf == props.get("slot", "") else: - matches = classcode == args.classcode + matches = int(props.get("classcode", "0"), 16) == classcode if matches: device = Device.from_dict(props) device.probe_handles() @@ -253,7 +251,7 @@ def print_props(args, device: Device): print(f" {key}: '{val}'") -def unbind(args, device: Device): +def unbind(device: Device): log.info(f"Unbinding({device.bdf}) from '{device.driver}'") driver_path = Path("/sys") / "bus" / "pci" / "devices" / device.bdf / "driver" @@ -266,10 +264,10 @@ def unbind(args, device: Device): sysfs_write(unbind, device.bdf) -def bind(args, device: Device, driver_name: str): +def bind(device: Device, driver_name: str): """Bind the driver named 'driver_name' with 'device'""" - unbind(args, device) + unbind(device) log.info(f"Binding({device.bdf}) to '{driver_name}'") @@ -372,7 +370,7 @@ def main(): if args.list: system.pp() - devices = list(device_scan(args)) + devices = list(device_scan(args.classcode, args.device)) try: for cur, device in enumerate(devices, 1): @@ -385,13 +383,13 @@ def main(): if device.is_used: log.info(f"Skipping unbind({device.driver}); device is in use.") else: - unbind(args, device) + unbind(device) if args.bind: if device.is_used: log.info(f"Skipping bind({args.bind}); device is in use.") else: - bind(args, device, args.bind) + bind(device, args.bind) except PermissionError as exc: log.error(str(exc)) log.error("Binding/unbinding PCIe devices requires root. Re-run with sudo.") diff --git a/tests/test_linux.py b/tests/test_linux.py index 01ead26..3a9e46d 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -37,8 +37,7 @@ def test_device_scan_named_device_bypasses_class_filter(monkeypatch): for name in ("probe_handles", "probe_usage", "probe_driver", "probe_iommugroup"): monkeypatch.setattr(devbind.Device, name, lambda self: None) - args = SimpleNamespace(device="0000:03:00.0", classcode=0x0108) - devices = list(devbind.device_scan(args)) + devices = list(devbind.device_scan(0x0108, "0000:03:00.0")) # The ethernet device is found by BDF even though its class is 0x0200 assert [d.bdf for d in devices] == ["0000:03:00.0"] From 328ef200ece1f6641dc6841e686fd2c9015c143d Mon Sep 17 00:00:00 2001 From: Jaeyoon Choi Date: Wed, 29 Jul 2026 17:00:58 +0900 Subject: [PATCH 3/5] refactor: move the Linux code onto a Backend class Prepare for adding other platforms. The module-level scan, bind, unbind, and probe helpers move onto a LinuxBackend class behind a small abstract Backend interface. get_backend() picks the implementation for the running platform. System.drivers/limits become instance state filled in from the backend. The memlock remediation text becomes a backend hint, assembled into the same warning. The parse-time driver vocabulary splits off as KNOWN_DRIVERS, and Device.MANDATORY_KEYS becomes _LSPCI_KEYS next to the lspci parser. The tool and sysfs notes from the file header move onto LinuxBackend as its docstring. Output and behavior on Linux are unchanged. scan_devices() keeps the --device bypass of 0.3.10. Add Linux backend unit tests: lspci parsing and filtering, the --device bypass, the driver_override -> bind -> setpci sequence, and backend selection. Signed-off-by: Jaeyoon Choi --- src/devbind/devbind.py | 427 +++++++++++++++++++++++------------------ tests/test_linux.py | 58 +++++- 2 files changed, 299 insertions(+), 186 deletions(-) diff --git a/src/devbind/devbind.py b/src/devbind/devbind.py index 76647b1..c4f4db1 100644 --- a/src/devbind/devbind.py +++ b/src/devbind/devbind.py @@ -4,30 +4,16 @@ # # Get info about and control driver associated with NVMe devices # -# This makes use of the following tools: +# Platform-specific operations (device enumeration, driver binding) are handled by +# a Backend selected at runtime via get_backend(): # -# * lspci -Dvmmnk -# * lsof {devhandle1, devhandle2, ... devhandleN} +# * Linux -- sysfs (/sys/bus/pci/...) + lspci / lsof / setpci # -# The following sysfs entries for driver bindings: -# -# * /sys/bus/pci/devices/{bdf}/driver -# * /sys/bus/pci/devices/{bdf}/driver_override -# * /sys/bus/pci/devices/{bdf}/driver/unbind -# * /sys/bus/pci/devices/{bdf}/iommu_group -# * /sys/bus/pci/devices/{bdf}/nvme/nvme* -# * /sys/bus/pci/devices/{bdf}/nvme/nvme*/ng* -# * /sys/bus/pci/devices/{bdf}/nvme/nvme*/nvme* -# * /sys/bus/pci/drivers/{driver_name}/bind -# -# The following could, but currently are not, be used for automatic detection based on -# class-code etc. -# -# * /sys/bus/pci/drivers/{driver_name}/new_id -# * /sys/bus/pci/drivers_probe +# Kept as a single, stdlib-only file so it can be installed by copying this script. # import sys import os +import abc import subprocess import argparse import errno @@ -35,7 +21,7 @@ import time import logging as log from itertools import chain -from typing import Optional +from typing import Iterable, Optional from pathlib import Path from dataclasses import dataclass, asdict, field @@ -43,6 +29,10 @@ PCIE_DEFAULT_CLASSCODE = 0x0108 # Mass Storage - NVM +# Driver-names recognized across platforms; used for argument parsing and +# completion. The active backend reports which are actually available. +KNOWN_DRIVERS = {"nvme", "vfio-pci", "vfio-noiommu", "uio_pci_generic"} + BASH_COMPLETION = r"""# bash completion for devbind _devbind() { local cur="${COMP_WORDS[COMP_CWORD]}" @@ -70,22 +60,102 @@ def run(cmd: str): return subprocess.run(cmd, capture_output=True, shell=True, text=True) +@dataclass +class Device: + """Encapsulation of a PCIe device""" + + bdf: str # PCI address of the device, e.g. "0000:02:00.0" + vendor: str # Vendor ID (hex), e.g. "144d" for Samsung + device: str # Device ID (hex), identifies the specific device model + classcode: str # PCI class code (hex), e.g. "0108" for NVMe controller + + driver: Optional[str] = None # Name of the driver bound to the device, e.g. "nvme" + iommugroup: Optional[int] = None # IOMMU group number the device belongs to + + is_used: bool = True # Whether or not the device is in use; assume it is + handles: list = field(default_factory=list) + + +class Backend(abc.ABC): + """Platform-specific PCI device-driver binding operations""" + + #: Drivers this platform knows how to bind devices to + DRIVERS: set = set() + + def driver_names(self) -> set: + """Return the set of driver-names this backend can bind to""" + return set(self.DRIVERS) + + @abc.abstractmethod + def probe_drivers(self) -> dict: + """Return ``{driver_name: {"available": bool}}`` for the known drivers""" + + @abc.abstractmethod + def scan_devices(self, classcode: int, bdf: Optional[str] = None) -> Iterable[Device]: + """Yield a fully-probed Device for each matching PCIe device + + A bdf (domain:bus:device.function, e.g. "0000:01:00.0") bypasses the class filter: + only that device is yielded, whatever its class. Otherwise devices + whose class matches classcode are yielded. + """ + + @abc.abstractmethod + def unbind(self, device: Device): + """Detach device from its current driver""" + + @abc.abstractmethod + def bind(self, device: Device, driver_name: str): + """Unbind if bound, then bind device to driver_name""" + + @abc.abstractmethod + def memlock_remediation_hint(self) -> str: + """Platform-specific guidance for raising RLIMIT_MEMLOCK""" + + +# --- Linux backend ----------------------------------------------------------- + +# Mapping from ``lspci -Dvmmnk`` record keys to Device fields +_LSPCI_KEYS = { + "slot": "bdf", + "vendor": "vendor", + "device": "device", + "classcode": "classcode", +} + + def sysfs_write(path: Path, text): log.info(f'{path} "{text}"') with os.fdopen(os.open(path, os.O_WRONLY), "w") as f: f.write(f"{text}\n") -class System: - DRIVERS = {"nvme", "vfio-pci", "vfio-noiommu", "uio_pci_generic"} +class LinuxBackend(Backend): + """Linux implementation, via sysfs and CLI tools - # DPDK/SPDK and xNVMe/uPCIe convention: VFIO_IOMMU_MAP_DMA pins user - # space pages against RLIMIT_MEMLOCK. Below 64 MiB the buffer-pool - # allocation fails outright. - MEMLOCK_MIN_BYTES = 64 * 1024 * 1024 + This makes use of the following tools: + + * lspci -Dvmmnk + * lsof {devhandle1, devhandle2, ... devhandleN} - drivers: dict = {} - limits: dict = {} + And the following sysfs entries for driver bindings: + + * /sys/bus/pci/devices/{bdf}/driver + * /sys/bus/pci/devices/{bdf}/driver_override + * /sys/bus/pci/devices/{bdf}/driver/unbind + * /sys/bus/pci/devices/{bdf}/iommu_group + * /sys/bus/pci/devices/{bdf}/nvme/nvme* + * /sys/bus/pci/devices/{bdf}/nvme/nvme*/ng* + * /sys/bus/pci/devices/{bdf}/nvme/nvme*/nvme* + * /sys/bus/pci/drivers/{driver_name}/bind + + The following could, but currently are not, be used for automatic + detection based on class-code etc. + + * /sys/bus/pci/drivers/{driver_name}/new_id + * /sys/bus/pci/drivers_probe + """ + + DRIVERS = {"nvme", "vfio-pci", "vfio-noiommu", "uio_pci_generic"} def probe_drivers(self): loaded = set( @@ -94,206 +164,197 @@ def probe_drivers(self): missing = self.DRIVERS - loaded + drivers = {} for driver_name in self.DRIVERS: - self.drivers[driver_name] = { + drivers[driver_name] = { "available": driver_name not in missing, } + return drivers - def probe_limits(self): - """Read process resource limits relevant to vfio-pci consumers""" - - soft, hard = resource.getrlimit(resource.RLIMIT_MEMLOCK) - self.limits["memlock_soft"] = soft - self.limits["memlock_hard"] = hard - - if soft != resource.RLIM_INFINITY and soft < self.MEMLOCK_MIN_BYTES: - log.warning( - f"memlock soft limit ({self._fmt_bytes(soft)}) is below " - f"{self._fmt_bytes(self.MEMLOCK_MIN_BYTES)}; " - "VFIO_IOMMU_MAP_DMA will fail for DPDK/SPDK and xNVMe/uPCIe. " - "Raise via /etc/security/limits.d/, prlimit, or systemd LimitMEMLOCK=" - ) + def memlock_remediation_hint(self): + return "Raise via /etc/security/limits.d/, prlimit, or systemd LimitMEMLOCK=" @staticmethod - def _fmt_bytes(n): - if n == resource.RLIM_INFINITY: - return "unlimited" - for unit in ("B", "kB", "MB", "GB", "TB"): - if n < 1024: - return f"{n} {unit}" - n //= 1024 - return f"{n} PB" - - def pp(self): - print("system:") - print(" drivers:") - for driver_name, props in self.drivers.items(): - print(f" - {driver_name}: {props}") - print(" limits:") - for name, val in self.limits.items(): - print(f" {name}: {self._fmt_bytes(val)}") - - -@dataclass -class Device: - """Encapsulation of a PCIe device""" - - MANDATORY_KEYS = { - "slot": "bdf", - "vendor": "vendor", - "device": "device", - "classcode": "classcode", - } - - bdf: str # PCI address of the device, e.g. "0000:02:00.0" - vendor: str # Vendor ID (hex), e.g. "144d" for Samsung - device: str # Device ID (hex), identifies the specific device model - classcode: str # PCI class code (hex), e.g. "0108" for NVMe controller - - driver: Optional[str] = None # Name of the driver bound to the device, e.g. "nvme" - iommugroup: Optional[int] = None # IOMMU group number the device belongs to - - is_used: bool = True # Whether or not the device is in use; assume it is - handles: list = field(default_factory=list) - - @classmethod - def from_dict(cls, data: dict) -> "Device": + def _device_from_dict(data: dict) -> Device: cdata = {} - for src, tgt in Device.MANDATORY_KEYS.items(): + for src, tgt in _LSPCI_KEYS.items(): cdata[tgt] = data.copy().get(src) - return cls(**cdata) + return Device(**cdata) - def probe_driver(self): + @staticmethod + def _probe_driver(device: Device): """Populate driver via sysfs; returns False if no driver is found""" - try: - self.driver = Path(f"/sys/bus/pci/devices/{self.bdf}/driver").resolve(strict=True).name + device.driver = ( + Path(f"/sys/bus/pci/devices/{device.bdf}/driver").resolve(strict=True).name + ) except FileNotFoundError: pass - return self.driver is not None + return device.driver is not None - def probe_iommugroup(self): + @staticmethod + def _probe_iommugroup(device: Device): """Populate iommugroup via sysfs; returns False if no iommugroup is found""" - try: - self.iommugroup = int( - Path(f"/sys/bus/pci/devices/{self.bdf}/iommu_group").resolve(strict=True).name + device.iommugroup = int( + Path(f"/sys/bus/pci/devices/{device.bdf}/iommu_group").resolve(strict=True).name ) except FileNotFoundError: - self.iommugroup = None - return self.iommugroup is not None + device.iommugroup = None + return device.iommugroup is not None - def probe_handles(self): + @staticmethod + def _probe_handles(device: Device): """Determine possible handles to the NVMe device""" - - for top in Path(f"/sys/bus/pci/devices/{self.bdf}/nvme").glob("nvme*"): + for top in Path(f"/sys/bus/pci/devices/{device.bdf}/nvme").glob("nvme*"): for bottom in chain(top.glob("ng*"), top.glob("nvme*")): for path in Path("/dev").glob(f"{bottom.name}*"): - self.handles.append(str(path)) + device.handles.append(str(path)) - def probe_usage(self): + @staticmethod + def _probe_usage(device: Device): """Attempt to determine whether the device is in use""" - - if not self.handles: - self.is_used = False + if not device.handles: + device.is_used = False return - - handles = " ".join(self.handles) + handles = " ".join(device.handles) proc = run(f"lsof {handles}") + device.is_used = bool(proc.stdout) - self.is_used = bool(proc.stdout) + def scan_devices(self, classcode: int, bdf: Optional[str] = None): + proc = run("lspci -Dvmmnk") + props = {} + for line in proc.stdout.splitlines(): + if not line: + if bdf: + matches = bdf == props.get("slot", "") + else: + matches = int(props.get("classcode", "0"), 16) == classcode + if matches: + device = self._device_from_dict(props) + self._probe_handles(device) + self._probe_usage(device) + self._probe_driver(device) + self._probe_iommugroup(device) + yield device -def device_scan(classcode: int, bdf: Optional[str] = None): - """Yields matching PCIe devices. + props = {} + continue - When bdf is set, the class filter is bypassed and the named device is - yielded regardless of its class. Otherwise devices whose class matches - classcode are yielded. - """ + key, val = [txt.strip().lower() for txt in str(line).split(":", 1)] + if key == "class": + key = "classcode" - proc = run("lspci -Dvmmnk") + props[key] = val - props = {} - for line in proc.stdout.splitlines(): - if not line: - if bdf: - matches = bdf == props.get("slot", "") - else: - matches = int(props.get("classcode", "0"), 16) == classcode - if matches: - device = Device.from_dict(props) - device.probe_handles() - device.probe_usage() - device.probe_driver() - device.probe_iommugroup() - yield device + def unbind(self, device: Device): + log.info(f"Unbinding({device.bdf}) from '{device.driver}'") - props = {} - continue + driver_path = Path("/sys") / "bus" / "pci" / "devices" / device.bdf / "driver" - key, val = [txt.strip().lower() for txt in str(line).split(":", 1)] - if key == "class": - key = "classcode" + unbind = driver_path / "unbind" + if not unbind.exists(): + log.info("Not bound; skipping unbind()") + return - props[key] = val + sysfs_write(unbind, device.bdf) + def bind(self, device: Device, driver_name: str): + """Bind the driver named 'driver_name' with 'device'""" -def print_props(args, device: Device): - """Pretty-print the properties of a device""" + self.unbind(device) - print("props:") - for key, val in asdict(device).items(): - if isinstance(val, int) or isinstance(val, list): - print(f" {key}: {val}") + log.info(f"Binding({device.bdf}) to '{driver_name}'") + + sysfs = Path("/sys") / "bus" / "pci" + + sysfs_write(sysfs / "devices" / device.bdf / "driver_override", driver_name) + + max_attempts = 10 + for attempt in range(1, max_attempts + 1): + try: + sysfs_write(sysfs / "drivers" / driver_name / "bind", device.bdf) + break + except OSError as exc: + if attempt == max_attempts or exc.errno != errno.EBUSY: + log.error(f"Could not bind despite {max_attempts} retries.") + raise + delay = attempt * 1 + log.info(f"Retrying in in {delay} second(s)") + time.sleep(delay) + + # Enable BUS-mastering (tell it that it can initiate DMA) + if driver_name == "uio_pci_generic": + log.info(f"Running setpci to enable bus-mastering; driver_name({driver_name})") + run(f"setpci -s {device.bdf} COMMAND=0x06") else: - print(f" {key}: '{val}'") + log.info(f"Not running setpci; driver_name({driver_name})") -def unbind(device: Device): - log.info(f"Unbinding({device.bdf}) from '{device.driver}'") +def get_backend() -> Backend: + """Return the Backend implementation for the running platform""" + if sys.platform.startswith("linux"): + return LinuxBackend() - driver_path = Path("/sys") / "bus" / "pci" / "devices" / device.bdf / "driver" + raise NotImplementedError(f"devbind has no backend for platform '{sys.platform}'") - unbind = driver_path / "unbind" - if not unbind.exists(): - log.info("Not bound; skipping unbind()") - return - sysfs_write(unbind, device.bdf) +class System: + # DPDK/SPDK and xNVMe/uPCIe convention: VFIO_IOMMU_MAP_DMA pins user + # space pages against RLIMIT_MEMLOCK. Below 64 MiB the buffer-pool + # allocation fails outright. + MEMLOCK_MIN_BYTES = 64 * 1024 * 1024 + + def __init__(self): + self.drivers: dict = {} + self.limits: dict = {} + def probe_limits(self, remediation_hint: str = ""): + """Read process resource limits relevant to vfio-pci consumers""" + + soft, hard = resource.getrlimit(resource.RLIMIT_MEMLOCK) + self.limits["memlock_soft"] = soft + self.limits["memlock_hard"] = hard -def bind(device: Device, driver_name: str): - """Bind the driver named 'driver_name' with 'device'""" + if soft != resource.RLIM_INFINITY and soft < self.MEMLOCK_MIN_BYTES: + log.warning( + f"memlock soft limit ({self._fmt_bytes(soft)}) is below " + f"{self._fmt_bytes(self.MEMLOCK_MIN_BYTES)}; " + "VFIO_IOMMU_MAP_DMA will fail for DPDK/SPDK and xNVMe/uPCIe. " + f"{remediation_hint}" + ) - unbind(device) + @staticmethod + def _fmt_bytes(n): + if n == resource.RLIM_INFINITY: + return "unlimited" + for unit in ("B", "kB", "MB", "GB", "TB"): + if n < 1024: + return f"{n} {unit}" + n //= 1024 + return f"{n} PB" - log.info(f"Binding({device.bdf}) to '{driver_name}'") + def pp(self): + print("system:") + print(" drivers:") + for driver_name, props in self.drivers.items(): + print(f" - {driver_name}: {props}") + print(" limits:") + for name, val in self.limits.items(): + print(f" {name}: {self._fmt_bytes(val)}") - sysfs = Path("/sys") / "bus" / "pci" - sysfs_write(sysfs / "devices" / device.bdf / "driver_override", driver_name) +def print_props(args, device: Device): + """Pretty-print the properties of a device""" - max_attempts = 10 - for attempt in range(1, max_attempts + 1): - try: - sysfs_write(sysfs / "drivers" / driver_name / "bind", device.bdf) - break - except OSError as exc: - if attempt == max_attempts or exc.errno != errno.EBUSY: - log.error(f"Could not bind despite {max_attempts} retries.") - raise - delay = attempt * 1 - log.info(f"Retrying in in {delay} second(s)") - time.sleep(delay) - - # Enable BUS-mastering (tell it that it can initiate DMA) - if driver_name == "uio_pci_generic": - log.info(f"Running setpci to enable bus-mastering; driver_name({driver_name})") - run(f"setpci -s {device.bdf} COMMAND=0x06") - else: - log.info(f"Not running setpci; driver_name({driver_name})") + print("props:") + for key, val in asdict(device).items(): + if isinstance(val, int) or isinstance(val, list): + print(f" {key}: {val}") + else: + print(f" {key}: '{val}'") def parse_args(): @@ -325,7 +386,7 @@ def parse_args(): parser.add_argument("--unbind", action="store_true", help="Unbind if bound.") def parse_bind(value): - if value in System.DRIVERS: + if value in KNOWN_DRIVERS: return value return Path(value) @@ -363,14 +424,16 @@ def main(): log.error("Binding/unbinding PCIe devices requires root. Re-run with sudo.") sys.exit(errno.EPERM) + backend = get_backend() + system = System() - system.probe_drivers() - system.probe_limits() + system.drivers = backend.probe_drivers() + system.probe_limits(backend.memlock_remediation_hint()) if args.list: system.pp() - devices = list(device_scan(args.classcode, args.device)) + devices = list(backend.scan_devices(args.classcode, args.device)) try: for cur, device in enumerate(devices, 1): @@ -383,13 +446,13 @@ def main(): if device.is_used: log.info(f"Skipping unbind({device.driver}); device is in use.") else: - unbind(device) + backend.unbind(device) if args.bind: if device.is_used: log.info(f"Skipping bind({args.bind}); device is in use.") else: - bind(device, args.bind) + backend.bind(device, args.bind) except PermissionError as exc: log.error(str(exc)) log.error("Binding/unbinding PCIe devices requires root. Re-run with sudo.") diff --git a/tests/test_linux.py b/tests/test_linux.py index 3a9e46d..cee3a51 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -2,9 +2,11 @@ # Copyright (c) Jaeyoon Choi """Unit tests for the Linux backend (runnable on any platform).""" +import sys from types import SimpleNamespace from devbind import devbind +from devbind.devbind import LinuxBackend, Device # Representative `lspci -Dvmmnk` output: an NVMe controller and a @@ -27,17 +29,65 @@ ) -def test_device_scan_named_device_bypasses_class_filter(monkeypatch): +def test_scan_devices_parses_and_filters(monkeypatch): monkeypatch.setattr( devbind, "run", lambda cmd: SimpleNamespace(stdout=LSPCI_DVMMNK, stderr="", returncode=0), ) # Avoid touching the real /sys and /dev during probing - for name in ("probe_handles", "probe_usage", "probe_driver", "probe_iommugroup"): - monkeypatch.setattr(devbind.Device, name, lambda self: None) + for name in ("_probe_handles", "_probe_usage", "_probe_driver", "_probe_iommugroup"): + monkeypatch.setattr(LinuxBackend, name, staticmethod(lambda device: None)) - devices = list(devbind.device_scan(0x0108, "0000:03:00.0")) + devices = list(LinuxBackend().scan_devices(0x0108)) + + # The ethernet device (class 0x0200) is filtered out + assert [d.bdf for d in devices] == ["0000:01:00.0"] + + nvme = devices[0] + assert nvme.vendor == "144d" + assert nvme.device == "a808" + assert nvme.classcode == "0108" + + +def test_bind_writes_override_then_bind_then_setpci(monkeypatch): + writes = [] + calls = [] + monkeypatch.setattr(devbind, "sysfs_write", lambda path, text: writes.append((str(path), text))) + monkeypatch.setattr( + devbind, + "run", + lambda cmd: calls.append(cmd) or SimpleNamespace(stdout="", stderr="", returncode=0), + ) + monkeypatch.setattr(LinuxBackend, "unbind", lambda self, device: None) + + device = Device( + bdf="0000:01:00.0", vendor="144d", device="a808", classcode="0108", driver="nvme" + ) + LinuxBackend().bind(device, "uio_pci_generic") + + assert writes == [ + ("/sys/bus/pci/devices/0000:01:00.0/driver_override", "uio_pci_generic"), + ("/sys/bus/pci/drivers/uio_pci_generic/bind", "0000:01:00.0"), + ] + assert "setpci -s 0000:01:00.0 COMMAND=0x06" in calls + + +def test_get_backend_selects_linux(monkeypatch): + monkeypatch.setattr(sys, "platform", "linux") + assert isinstance(devbind.get_backend(), LinuxBackend) + + +def test_scan_devices_named_device_bypasses_class_filter(monkeypatch): + monkeypatch.setattr( + devbind, + "run", + lambda cmd: SimpleNamespace(stdout=LSPCI_DVMMNK, stderr="", returncode=0), + ) + for name in ("_probe_handles", "_probe_usage", "_probe_driver", "_probe_iommugroup"): + monkeypatch.setattr(LinuxBackend, name, staticmethod(lambda device: None)) + + devices = list(LinuxBackend().scan_devices(0x0108, "0000:03:00.0")) # The ethernet device is found by BDF even though its class is 0x0200 assert [d.bdf for d in devices] == ["0000:03:00.0"] From 3b23b7d0766d10717e52dec37383204d093107c4 Mon Sep 17 00:00:00 2001 From: Jaeyoon Choi Date: Wed, 29 Jul 2026 16:19:31 +0900 Subject: [PATCH 4/5] fix: exit with one-line errors instead of tracebacks An unsupported platform, a --bind name the platform does not have, and a failing bind or unbind all ended in a Python traceback. The driver-name case gets worse once more platforms exist. A name owned by another platform would reach the device before failing. Catch the backend errors in main() and exit with a one-line error. Validate --bind against the active backend's driver_names() before touching the device. Signed-off-by: Jaeyoon Choi --- src/devbind/devbind.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/src/devbind/devbind.py b/src/devbind/devbind.py index c4f4db1..da52d2b 100644 --- a/src/devbind/devbind.py +++ b/src/devbind/devbind.py @@ -424,7 +424,18 @@ def main(): log.error("Binding/unbinding PCIe devices requires root. Re-run with sudo.") sys.exit(errno.EPERM) - backend = get_backend() + try: + backend = get_backend() + except NotImplementedError as exc: + log.error(str(exc)) + sys.exit(errno.ENOSYS) + + if isinstance(args.bind, str) and args.bind not in backend.driver_names(): + log.error( + f"driver '{args.bind}' is not supported on this platform; " + f"expected one of: {', '.join(sorted(backend.driver_names()))}" + ) + sys.exit(errno.EINVAL) system = System() system.drivers = backend.probe_drivers() @@ -457,6 +468,9 @@ def main(): log.error(str(exc)) log.error("Binding/unbinding PCIe devices requires root. Re-run with sudo.") sys.exit(errno.EPERM) + except OSError as exc: + log.error(str(exc)) + sys.exit(exc.errno or 1) if __name__ == "__main__": From 3e27d8f1bc313cd45563947a1bb7684aadd0c697 Mon Sep 17 00:00:00 2001 From: Jaeyoon Choi Date: Wed, 29 Jul 2026 18:06:41 +0900 Subject: [PATCH 5/5] feat: add a FreeBSD backend (pciconf/devctl/camcontrol/fstat) Implement the Backend interface with FreeBSD's native tools: - pciconf -l enumerates devices. Both output formats are parsed: the packed chip=/card= fields of 14.3 and earlier, and the split vendor=/device= fields of 15.x and 14.4. - devctl detach / devctl set driver implement unbind and bind. - camcontrol devlist maps an nvmeX controller to its CAM disk (ndaY), so the fstat in-use check also sees users of the disk device. - kldstat reports whether nic_uio is loaded. - scan_devices() honors the --device class-filter bypass, matching the Linux backend. - pciconf -w enables bus-mastering after binding to nic_uio. A failed write is reported instead of ignored. Helpers convert between the pciX:B:S:F selector and the domain:bus:device.function bdf form. nic_uio is added to KNOWN_DRIVERS, completion, and the docs. User-facing text drops its Linux assumptions: the --bind help says driver file instead of .ko file, and the memlock warning says DMA mapping instead of VFIO_IOMMU_MAP_DMA. Unit tests cover both pciconf formats, the fstat heuristic, kldstat probing, camcontrol parsing, and the devctl bind sequence. Signed-off-by: Jaeyoon Choi --- README.md | 39 ++++++-- pyproject.toml | 3 +- src/devbind/devbind.py | 203 ++++++++++++++++++++++++++++++++++++++--- tests/test_freebsd.py | 175 +++++++++++++++++++++++++++++++++++ 4 files changed, 397 insertions(+), 23 deletions(-) create mode 100644 tests/test_freebsd.py diff --git a/README.md b/README.md index ed4d602..2e172b3 100644 --- a/README.md +++ b/README.md @@ -7,12 +7,23 @@ [![Test](https://github.com/xnvme/devbind/actions/workflows/test.yml/badge.svg)](https://github.com/xnvme/devbind/actions/workflows/test.yml) `devbind` is a small CLI for binding and unbinding PCI devices to a -chosen kernel driver via sysfs. The typical use is moving a device -between its native driver (e.g. `nvme`) and a user space driver -framework (`vfio-pci`, `uio_pci_generic`) for DPDK/SPDK and xNVMe/uPCIe -workloads. `devbind --list` also reports the process `RLIMIT_MEMLOCK` -and warns when the soft limit is below the 64 MiB threshold those -frameworks inherit. +chosen kernel driver. The typical use is moving a device between its +native driver (e.g. `nvme`) and a user space driver framework for +DPDK/SPDK and xNVMe/uPCIe workloads. `devbind --list` also reports the +process `RLIMIT_MEMLOCK` and warns when the soft limit is below the +64 MiB threshold those frameworks inherit. + +Both **Linux** and **FreeBSD** are supported; the platform-specific +operations are handled by a backend selected at runtime: + +| | Linux | FreeBSD | +|---|---|---| +| enumerate / inspect | `lspci`, sysfs | `pciconf -l`, `camcontrol` | +| user space framework | `vfio-pci`, `uio_pci_generic` | `nic_uio` | +| unbind | sysfs `driver/unbind` | `devctl detach` | +| bind | sysfs `drivers//bind` | `devctl set driver` | + +(The Linux-only `iommugroup` is reported as `None` on FreeBSD.) ## Install @@ -33,7 +44,7 @@ curl -fsSL https://raw.githubusercontent.com/xnvme/devbind/main/src/devbind/devb devbind --print-completion bash > ~/.local/share/bash-completion/completions/devbind ``` -Open a new shell (or `source` the file) and tab-completion is live: `devbind --bind ` lists `nvme vfio-pci vfio-noiommu uio_pci_generic`. +Open a new shell (or `source` the file) and tab-completion is live: `devbind --bind ` lists `nvme vfio-pci vfio-noiommu uio_pci_generic nic_uio`. ## Usage @@ -43,7 +54,7 @@ usage: devbind [-h] [--version] [--classcode CLASSCODE] [--device DEVICE] [--list] [--unbind] [--bind BIND] [--verbose] [--print-completion SHELL] -Inspect and control PCI device-driver binding in Linux +Inspect and control PCI device-driver binding on Linux and FreeBSD options: -h, --help show this help message and exit @@ -55,8 +66,8 @@ options: association. --unbind Unbind if bound. --bind BIND Unbind if bound; then bind to the given driver-name - [nvme, vfio-pci, uio_pci_generic] or to a .ko driver - file (path) + [nvme, vfio-pci, uio_pci_generic, nic_uio] or to a + driver file (path) --verbose Enable verbose logging --print-completion SHELL Print shell completion script to stdout and exit @@ -71,6 +82,14 @@ sudo devbind --bind nvme --device 0000:01:00.0 # rebind to the native driv sudo devbind --unbind --device 0000:01:00.0 # unbind without rebinding ``` +On FreeBSD, bind to `nic_uio` instead of `vfio-pci`/`uio_pci_generic` +(`--device` always takes the `domain:bus:device.function` form `0000:01:00.0` on both platforms): + +``` +sudo devbind --bind nic_uio --device 0000:01:00.0 # hand device to DPDK/SPDK +sudo devbind --bind nvme --device 0000:01:00.0 # rebind to the native driver +``` + `devbind --list` sample output (stock WSL host, no NVMe devices visible): ``` diff --git a/pyproject.toml b/pyproject.toml index ed1b3fb..b6ba120 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "devbind" dynamic = ["version"] -description = "Inspect and control PCI device-driver binding in Linux" +description = "Inspect and control PCI device-driver binding on Linux and FreeBSD" readme = "README.md" license = "BSD-3-Clause" requires-python = ">=3.10" @@ -18,6 +18,7 @@ classifiers = [ "Intended Audience :: Developers", "Intended Audience :: System Administrators", "Operating System :: POSIX :: Linux", + "Operating System :: POSIX :: BSD :: FreeBSD", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", diff --git a/src/devbind/devbind.py b/src/devbind/devbind.py index da52d2b..14cdaa2 100644 --- a/src/devbind/devbind.py +++ b/src/devbind/devbind.py @@ -8,12 +8,15 @@ # a Backend selected at runtime via get_backend(): # # * Linux -- sysfs (/sys/bus/pci/...) + lspci / lsof / setpci +# * FreeBSD -- pciconf / devctl / camcontrol / fstat, with nic_uio as the +# userspace stub driver # # Kept as a single, stdlib-only file so it can be installed by copying this script. # import sys import os import abc +import shlex import subprocess import argparse import errno @@ -31,13 +34,13 @@ # Driver-names recognized across platforms; used for argument parsing and # completion. The active backend reports which are actually available. -KNOWN_DRIVERS = {"nvme", "vfio-pci", "vfio-noiommu", "uio_pci_generic"} +KNOWN_DRIVERS = {"nvme", "vfio-pci", "vfio-noiommu", "uio_pci_generic", "nic_uio"} BASH_COMPLETION = r"""# bash completion for devbind _devbind() { local cur="${COMP_WORDS[COMP_CWORD]}" local prev="${COMP_WORDS[COMP_CWORD-1]}" - local drivers="nvme vfio-pci vfio-noiommu uio_pci_generic" + local drivers="nvme vfio-pci vfio-noiommu uio_pci_generic nic_uio" local opts="--classcode --device --list --unbind --bind --verbose --help --print-completion" case "${prev}" in --bind) @@ -62,15 +65,15 @@ def run(cmd: str): @dataclass class Device: - """Encapsulation of a PCIe device""" + """Encapsulation of a PCIe device (platform-neutral)""" - bdf: str # PCI address of the device, e.g. "0000:02:00.0" + bdf: str # PCI address as domain:bus:device.function, e.g. "0000:02:00.0" vendor: str # Vendor ID (hex), e.g. "144d" for Samsung device: str # Device ID (hex), identifies the specific device model classcode: str # PCI class code (hex), e.g. "0108" for NVMe controller driver: Optional[str] = None # Name of the driver bound to the device, e.g. "nvme" - iommugroup: Optional[int] = None # IOMMU group number the device belongs to + iommugroup: Optional[int] = None # IOMMU group number (Linux-only; None elsewhere) is_used: bool = True # Whether or not the device is in use; assume it is handles: list = field(default_factory=list) @@ -293,18 +296,193 @@ def bind(self, device: Device, driver_name: str): log.info(f"Not running setpci; driver_name({driver_name})") +# --- FreeBSD backend --------------------------------------------------------- + +# Userspace stub driver used by DPDK/SPDK on FreeBSD (analogous to uio_pci_generic) +FREEBSD_USERSPACE_DRIVER = "nic_uio" + + +def selector_to_bdf(selector: str) -> str: + """Convert a FreeBSD selector 'pci0:1:0:0' to the bdf form '0000:01:00.0'""" + domain, bus, slot, func = (int(part) for part in selector[len("pci") :].split(":")) + return f"{domain:04x}:{bus:02x}:{slot:02x}.{func:x}" + + +def bdf_to_selector(bdf: str) -> str: + """Convert a bdf '0000:01:00.0' to a FreeBSD selector 'pci0:1:0:0'""" + dom_bus_slot, func = bdf.split(".") + domain, bus, slot = dom_bus_slot.split(":") + return f"pci{int(domain, 16)}:{int(bus, 16)}:{int(slot, 16)}:{int(func, 16)}" + + +class FreeBsdBackend(Backend): + DRIVERS = {"nvme", FREEBSD_USERSPACE_DRIVER} + + @staticmethod + def _module_loaded(name: str) -> bool: + return run(f"kldstat -q -n {name}").returncode == 0 + + def probe_drivers(self): + # nvme ships in GENERIC; nic_uio is an out-of-tree module that must be loaded + return { + "nvme": {"available": True}, + FREEBSD_USERSPACE_DRIVER: {"available": self._module_loaded(FREEBSD_USERSPACE_DRIVER)}, + } + + def memlock_remediation_hint(self): + return "Raise via the 'memorylocked' capability in /etc/login.conf or /boot/loader.conf" + + @staticmethod + def _cam_disks(instance: str) -> list: + """Map an nvmeX controller instance to its CAM disk names (e.g. ['nda0']) + + Parses ``camcontrol devlist -v``, whose output groups peripherals under + bus headers like ``scbus1 on nvme0 bus 0:``. Legacy nvd(4) disks do not + attach via CAM and are not mapped. Openers of /dev/nvd* are therefore + invisible to the in-use check. + """ + disks = [] + on_our_bus = False + for line in run("camcontrol devlist -v").stdout.splitlines(): + stripped = line.strip() + parts = stripped.split() + if stripped.startswith("scbus") and len(parts) >= 3 and parts[1] == "on": + on_our_bus = parts[2] == instance + continue + if not on_our_bus or "(" not in stripped: + continue + names = stripped.rsplit("(", 1)[1].rstrip(")").split(",") + disks.extend(name for name in names if name.startswith("nda")) + return disks + + @staticmethod + def _probe_handles(device: Device, instance: str): + """Map the pciconf instance (e.g. 'nvme0') to its /dev nodes""" + if not instance.startswith("nvme"): + return + for stem in [instance] + FreeBsdBackend._cam_disks(instance): + for path in Path("/dev").glob(f"{stem}*"): + device.handles.append(str(path)) + + @staticmethod + def _probe_usage(device: Device): + """Attempt to determine whether the device is in use via fstat""" + if not device.handles: + device.is_used = False + return + handles = " ".join(device.handles) + proc = run(f"fstat {handles}") + # fstat always prints a header line; any further rows mean an open handle + rows = [line for line in proc.stdout.splitlines() if line.strip()] + device.is_used = len(rows) > 1 + + def scan_devices(self, classcode: int, bdf: Optional[str] = None): + # Each `pciconf -l` line, FreeBSD 14.3 and earlier: + # nvme0@pci0:1:0:0:\tclass=0x010802 card=0x... chip=0xDDDDVVVV rev=0x.. hdr=0x.. + # 15.x and 14.4 replaced the packed chip=/card= fields: + # nvme0@pci0:1:0:0:\tclass=0x010802 rev=0x.. hdr=0x.. vendor=0xVVVV device=0xDDDD ... + proc = run("pciconf -l") + + for line in proc.stdout.splitlines(): + if "@pci" not in line: + continue + + parts = line.split() + instance, _, selector = parts[0].partition("@") + selector = selector.rstrip(":") + + fields = dict(tok.split("=", 1) for tok in parts[1:] if "=" in tok) + if "class" not in fields: + continue + + if "chip" in fields: + chip = int(fields["chip"], 16) + vendor = chip & 0xFFFF + device_id = (chip >> 16) & 0xFFFF + elif "vendor" in fields and "device" in fields: + vendor = int(fields["vendor"], 16) + device_id = int(fields["device"], 16) + else: + continue + + cls = int(fields["class"], 16) + if bdf: + if selector_to_bdf(selector) != bdf: + continue + elif (cls >> 8) != classcode: + continue + # 'none0' (or 'none1', ...) denotes a device without an attached driver + stem = instance.rstrip("0123456789") + driver = None if stem == "none" else stem + + device = Device( + bdf=selector_to_bdf(selector), + vendor=f"{vendor:04x}", + device=f"{device_id:04x}", + classcode=f"{cls >> 8:04x}", + driver=driver, + ) + self._probe_handles(device, instance) + self._probe_usage(device) + + yield device + + def unbind(self, device: Device): + if not device.driver: + log.info("Not bound; skipping unbind()") + return + + selector = bdf_to_selector(device.bdf) + log.info(f"Unbinding({device.bdf}) from '{device.driver}'") + + proc = run(f"devctl detach {selector}") + if proc.returncode != 0: + message = proc.stderr.strip() or f"devctl detach {selector} failed" + log.error(message) + raise OSError(message) + + def bind(self, device: Device, driver_name: str): + """Bind the driver named 'driver_name' with 'device'""" + + self.unbind(device) + + selector = bdf_to_selector(device.bdf) + log.info(f"Binding({device.bdf}) to '{driver_name}'") + + driver_arg = shlex.quote(str(driver_name)) + proc = run(f"devctl set driver {selector} {driver_arg}") + if proc.returncode != 0: + message = proc.stderr.strip() or f"devctl set driver {selector} {driver_arg} failed" + log.error(message) + raise OSError(message) + + # Enable BUS-mastering (memory space + bus master) for the userspace driver + if driver_name == FREEBSD_USERSPACE_DRIVER: + log.info(f"Running pciconf to enable bus-mastering; driver_name({driver_name})") + proc = run(f"pciconf -w -h {selector} 0x4 0x6") + if proc.returncode != 0: + log.error( + proc.stderr.strip() + or f"pciconf write failed; bus-mastering not enabled for {device.bdf}" + ) + else: + log.info(f"Not enabling bus-mastering; driver_name({driver_name})") + + def get_backend() -> Backend: """Return the Backend implementation for the running platform""" if sys.platform.startswith("linux"): return LinuxBackend() + if sys.platform.startswith("freebsd"): + return FreeBsdBackend() raise NotImplementedError(f"devbind has no backend for platform '{sys.platform}'") class System: - # DPDK/SPDK and xNVMe/uPCIe convention: VFIO_IOMMU_MAP_DMA pins user - # space pages against RLIMIT_MEMLOCK. Below 64 MiB the buffer-pool - # allocation fails outright. + # DPDK/SPDK and xNVMe/uPCIe convention: pinning user space pages for DMA + # (VFIO_IOMMU_MAP_DMA on Linux, nic_uio/contigmem on FreeBSD) counts against + # RLIMIT_MEMLOCK. Below 64 MiB the buffer-pool allocation fails outright. MEMLOCK_MIN_BYTES = 64 * 1024 * 1024 def __init__(self): @@ -312,7 +490,7 @@ def __init__(self): self.limits: dict = {} def probe_limits(self, remediation_hint: str = ""): - """Read process resource limits relevant to vfio-pci consumers""" + """Read process resource limits relevant to userspace-driver consumers""" soft, hard = resource.getrlimit(resource.RLIMIT_MEMLOCK) self.limits["memlock_soft"] = soft @@ -322,7 +500,7 @@ def probe_limits(self, remediation_hint: str = ""): log.warning( f"memlock soft limit ({self._fmt_bytes(soft)}) is below " f"{self._fmt_bytes(self.MEMLOCK_MIN_BYTES)}; " - "VFIO_IOMMU_MAP_DMA will fail for DPDK/SPDK and xNVMe/uPCIe. " + "DMA mapping will fail for DPDK/SPDK and xNVMe/uPCIe. " f"{remediation_hint}" ) @@ -359,7 +537,7 @@ def print_props(args, device: Device): def parse_args(): parser = argparse.ArgumentParser( - description="Inspect and control PCI device-driver binding in Linux" + description="Inspect and control PCI device-driver binding on Linux and FreeBSD" ) parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") @@ -393,7 +571,8 @@ def parse_bind(value): parser.add_argument( "--bind", type=parse_bind, - help="Unbind if bound; then bind to the given driver-name [nvme, vfio-pci, uio_pci_generic] or to a .ko driver file (path)", + help="Unbind if bound; then bind to the given driver-name " + "[nvme, vfio-pci, uio_pci_generic, nic_uio] or to a driver file (path)", ) parser.add_argument("--verbose", action="store_true", help="Enable verbose logging") diff --git a/tests/test_freebsd.py b/tests/test_freebsd.py new file mode 100644 index 0000000..cf6c55a --- /dev/null +++ b/tests/test_freebsd.py @@ -0,0 +1,175 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) Jaeyoon Choi +"""Unit tests for the FreeBSD backend (runnable on any platform).""" + +import sys +from types import SimpleNamespace + +import pytest + +from devbind import devbind +from devbind.devbind import FreeBsdBackend, Device, bdf_to_selector, selector_to_bdf + + +# Representative `pciconf -l` output: an NVMe controller (bound), an NVMe +# controller without a driver, and a non-matching ethernet device. The first +# two lines use the packed chip=/card= format of FreeBSD <= 14; the last NVMe +# line uses the split vendor=/device= format that FreeBSD 15 switched to. +PCICONF_L = ( + "nvme0@pci0:1:0:0:\tclass=0x010802 card=0xa801144d chip=0xa808144d rev=0x00 hdr=0x00\n" + "none0@pci0:2:0:0:\tclass=0x010802 card=0x0000 chip=0x540a1b96 rev=0x00 hdr=0x00\n" + "em0@pci0:3:0:0:\tclass=0x020000 card=0x00008086 chip=0x10d38086 rev=0x00 hdr=0x00\n" + "nvme1@pci0:4:0:0:\tclass=0x010802 rev=0x02 hdr=0x00 vendor=0x1b36 device=0x0010 " + "subvendor=0x1af4 subdevice=0x1100\n" +) + + +def _fake_run(outputs): + """Build a run() replacement dispatching by command prefix + + Commands matching no prefix fail (returncode 1), so a test cannot pass by + accident when the code under test never issues the expected command. + """ + + def runner(cmd): + for prefix, result in outputs.items(): + if cmd.startswith(prefix): + return SimpleNamespace(stdout=result, stderr="", returncode=0) + return SimpleNamespace(stdout="", stderr="", returncode=1) + + return runner + + +def test_selector_bdf_roundtrip(): + assert selector_to_bdf("pci0:1:0:0") == "0000:01:00.0" + assert bdf_to_selector("0000:01:00.0") == "pci0:1:0:0" + assert selector_to_bdf("pci1:255:31:7") == "0001:ff:1f.7" + assert bdf_to_selector(selector_to_bdf("pci0:130:5:3")) == "pci0:130:5:3" + + +def test_scan_devices_parses_and_filters(monkeypatch): + monkeypatch.setattr( + devbind, "run", _fake_run({"pciconf -l": PCICONF_L, "fstat": "USER CMD ...\n"}) + ) + # Avoid touching the real /dev during handle probing + monkeypatch.setattr(FreeBsdBackend, "_probe_handles", staticmethod(lambda device, inst: None)) + + devices = list(FreeBsdBackend().scan_devices(0x0108)) + + # The ethernet device (class 0x0200) is filtered out + assert [d.bdf for d in devices] == ["0000:01:00.0", "0000:02:00.0", "0000:04:00.0"] + + nvme = devices[0] + assert nvme.vendor == "144d" # low 16 bits of chip + assert nvme.device == "a808" # high 16 bits of chip + assert nvme.classcode == "0108" + assert nvme.driver == "nvme" + + assert devices[1].driver is None # 'none0' -> no driver attached + + fb15 = devices[2] # FreeBSD 15 vendor=/device= format + assert fb15.vendor == "1b36" + assert fb15.device == "0010" + assert fb15.driver == "nvme" + + +def test_probe_drivers_nic_uio_loaded(monkeypatch): + monkeypatch.setattr( + devbind, "run", _fake_run({"kldstat -q -n nic_uio": ""}) + ) # returncode 0 -> loaded + drivers = FreeBsdBackend().probe_drivers() + assert drivers["nvme"]["available"] is True + assert drivers["nic_uio"]["available"] is True + + +def test_probe_drivers_nic_uio_not_loaded(monkeypatch): + monkeypatch.setattr(devbind, "run", _fake_run({})) # kldstat fails -> not loaded + drivers = FreeBsdBackend().probe_drivers() + assert drivers["nvme"]["available"] is True + assert drivers["nic_uio"]["available"] is False + + +def test_probe_usage_via_fstat(monkeypatch): + device = Device(bdf="0000:01:00.0", vendor="144d", device="a808", classcode="0108") + device.handles = ["/dev/nvme0", "/dev/nda0"] + + header = "USER CMD PID FD PATH\n" + + monkeypatch.setattr(devbind, "run", _fake_run({"fstat": header})) + FreeBsdBackend._probe_usage(device) + assert device.is_used is False # header only -> no open handles + + monkeypatch.setattr( + devbind, "run", _fake_run({"fstat": header + "root dd 71 3 /dev/nda0\n"}) + ) + FreeBsdBackend._probe_usage(device) + assert device.is_used is True + + +CAMCONTROL_V = ( + "scbus0 on ahcich0 bus 0:\n" + " at scbus0 target 0 lun 0 (pass0,ada0)\n" + "<> at scbus0 target -1 lun ffffffff ()\n" + "scbus1 on nvme0 bus 0:\n" + " at scbus1 target 0 lun 1 (pass1,nda0)\n" + "<> at scbus1 target -1 lun ffffffff ()\n" + "scbus-1 on xpt0 bus 0:\n" + "<> at scbus-1 target -1 lun ffffffff (xpt0)\n" +) + + +def test_cam_disks_maps_controller_to_disks(monkeypatch): + monkeypatch.setattr(devbind, "run", _fake_run({"camcontrol devlist -v": CAMCONTROL_V})) + assert FreeBsdBackend._cam_disks("nvme0") == ["nda0"] + assert FreeBsdBackend._cam_disks("ahcich0") == [] # ada is not an NVMe disk + assert FreeBsdBackend._cam_disks("nvme1") == [] # no such bus + + +def test_bind_invokes_devctl_and_busmaster(monkeypatch): + calls = [] + + def recording_run(cmd): + calls.append(cmd) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(devbind, "run", recording_run) + + device = Device( + bdf="0000:01:00.0", vendor="144d", device="a808", classcode="0108", driver="nvme" + ) + FreeBsdBackend().bind(device, "nic_uio") + + assert "devctl detach pci0:1:0:0" in calls + assert "devctl set driver pci0:1:0:0 nic_uio" in calls + # pciconf -w takes the selector as a positional argument (no -s flag) + assert "pciconf -w -h pci0:1:0:0 0x4 0x6" in calls + + +def test_unbind_skips_when_unbound(monkeypatch): + calls = [] + monkeypatch.setattr(devbind, "run", lambda cmd: calls.append(cmd) or SimpleNamespace()) + device = Device( + bdf="0000:01:00.0", vendor="144d", device="a808", classcode="0108", driver=None + ) + FreeBsdBackend().unbind(device) + assert calls == [] + + +def test_get_backend_selection(monkeypatch): + monkeypatch.setattr(sys, "platform", "freebsd14") + assert isinstance(devbind.get_backend(), FreeBsdBackend) + + monkeypatch.setattr(sys, "platform", "win32") + with pytest.raises(NotImplementedError): + devbind.get_backend() + + +def test_scan_devices_named_device_bypasses_class_filter(monkeypatch): + monkeypatch.setattr(devbind, "run", _fake_run({"pciconf -l": PCICONF_L})) + monkeypatch.setattr(FreeBsdBackend, "_probe_handles", staticmethod(lambda device, inst: None)) + + devices = list(FreeBsdBackend().scan_devices(0x0108, "0000:03:00.0")) + + # The ethernet device is found by BDF even though its class is 0x0200 + assert [d.bdf for d in devices] == ["0000:03:00.0"] + assert devices[0].driver == "em"