From: Anatoly Burakov <anatoly.burakov@intel.com>
To: dev@dpdk.org
Subject: [PATCH v3 2/2] usertools/devbind: replace devbind
Date: Wed, 4 Dec 2024 09:47:38 +0000 [thread overview]
Message-ID: <b8aa13c9ed8989e73e9352ea905f6956a863148d.1733305655.git.anatoly.burakov@intel.com> (raw)
In-Reply-To: <cover.1733305654.git.anatoly.burakov@intel.com>
Signed-off-by: Anatoly Burakov <anatoly.burakov@intel.com>
---
Notes:
v2:
- Added this patch to aid in review
- I believe it's better to squash it on apply
usertools/dpdk-devbind-new.py | 995 -------------------
usertools/dpdk-devbind.py | 1683 ++++++++++++++++++---------------
2 files changed, 913 insertions(+), 1765 deletions(-)
delete mode 100755 usertools/dpdk-devbind-new.py
diff --git a/usertools/dpdk-devbind-new.py b/usertools/dpdk-devbind-new.py
deleted file mode 100755
index 1f2d8cb118..0000000000
--- a/usertools/dpdk-devbind-new.py
+++ /dev/null
@@ -1,995 +0,0 @@
-#!/usr/bin/env python3
-# SPDX-License-Identifier: BSD-3-Clause
-# Copyright(c) 2010-2024 Intel Corporation
-#
-"""Script to bind PCI devices to DPDK-compatible userspace IO drivers."""
-
-import argparse
-import glob
-import grp
-import json
-import os
-import pwd
-import subprocess
-import sys
-import typing as T
-
-# the following list of modules is supported by DPDK
-DPDK_KERNEL_MODULES = {"igb_uio", "vfio-pci", "uio_pci_generic"}
-
-# pattern matching criteria for various devices and devices classes. keys are entries in lspci,
-# while values, if present are further matches for lspci criteria. values can be either strings or
-# list of strings, in which case any match is sufficient.
-StrOrList = T.Union[str, T.List[str]]
-DeviceMatchPattern = T.Dict[str, StrOrList]
-CLASS_NETWORK: DeviceMatchPattern = {
- "Class": "02",
-}
-CLASS_ACCELERATION: DeviceMatchPattern = {
- "Class": "12",
-}
-CLASS_IFPGA: DeviceMatchPattern = {
- "Class": "12",
- "Vendor": "8086",
- "Device": "0b30",
-}
-CLASS_ENCRYPTION: DeviceMatchPattern = {
- "Class": "10",
-}
-CLASS_INTEL_PROCESSOR: DeviceMatchPattern = {
- "Class": "0b",
- "Vendor": "8086",
-}
-DEVICE_CAVIUM_SSO: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": ["a04b", "a04d"],
-}
-DEVICE_CAVIUM_FPA: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": "a053",
-}
-DEVICE_CAVIUM_PKX: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": ["a0dd", "a049"],
-}
-DEVICE_CAVIUM_TIM: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": "a051",
-}
-DEVICE_CAVIUM_ZIP: DeviceMatchPattern = {
- "Class": "12",
- "Vendor": "177d",
- "Device": "a037",
-}
-DEVICE_AVP_VNIC: DeviceMatchPattern = {
- "Class": "05",
- "Vendor": "1af4",
- "Device": "1110",
-}
-DEVICE_CNXK_BPHY: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": "a089",
-}
-DEVICE_CNXK_BPHY_CGX: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": ["a059", "a060"],
-}
-DEVICE_CNXK_DMA: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": "a081",
-}
-DEVICE_CNXK_INL_DEV: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": ["a0f0", "a0f1"],
-}
-DEVICE_HISILICON_DMA: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "19e5",
- "Device": "a122",
-}
-DEVICE_ODM_DMA: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": "a08c",
-}
-DEVICE_INTEL_DLB: DeviceMatchPattern = {
- "Class": "0b",
- "Vendor": "8086",
- "Device": ["270b", "2710", "2714"],
-}
-DEVICE_INTEL_IOAT_BDW: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "8086",
- "Device": [
- "6f20",
- "6f21",
- "6f22",
- "6f23",
- "6f24",
- "6f25",
- "6f26",
- "6f27",
- "6f2e",
- "6f2f",
- ],
-}
-DEVICE_INTEL_IOAT_SKX: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "8086",
- "Device": "2021",
-}
-DEVICE_INTEL_IOAT_ICX: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "8086",
- "Device": "0b00",
-}
-DEVICE_INTEL_IDXD_SPR: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "8086",
- "Device": "0b25",
-}
-DEVICE_INTEL_NTB_SKX: DeviceMatchPattern = {
- "Class": "06",
- "Vendor": "8086",
- "Device": "201c",
-}
-DEVICE_INTEL_NTB_ICX: DeviceMatchPattern = {
- "Class": "06",
- "Vendor": "8086",
- "Device": "347e",
-}
-DEVICE_CNXK_SSO: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": ["a0f9", "a0fa"],
-}
-DEVICE_CNXK_NPA: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": ["a0fb", "a0fc"],
-}
-DEVICE_CN9K_REE: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": "a0f4",
-}
-DEVICE_VIRTIO_BLK: DeviceMatchPattern = {
- "Class": "01",
- "Vendor": "1af4",
- "Device": ["1001", "1042"],
-}
-DEVICE_CNXK_ML: DeviceMatchPattern = {
- "Class": "08",
- "Vendor": "177d",
- "Device": "a092",
-}
-
-# device types as recognized by devbind
-NETWORK_DEVICES = [CLASS_NETWORK, CLASS_IFPGA, DEVICE_CAVIUM_PKX, DEVICE_AVP_VNIC]
-BASEDBAND_DEVICES = [CLASS_ACCELERATION]
-CRYPTO_DEVICES = [CLASS_ENCRYPTION, CLASS_INTEL_PROCESSOR]
-DMA_DEVICES = [
- DEVICE_CNXK_DMA,
- DEVICE_HISILICON_DMA,
- DEVICE_INTEL_IDXD_SPR,
- DEVICE_INTEL_IOAT_BDW,
- DEVICE_INTEL_IOAT_ICX,
- DEVICE_INTEL_IOAT_SKX,
- DEVICE_ODM_DMA,
-]
-EVENTDEV_DEVICES = [
- DEVICE_CAVIUM_SSO,
- DEVICE_CAVIUM_TIM,
- DEVICE_INTEL_DLB,
- DEVICE_CNXK_SSO,
-]
-MEMPOOL_DEVICES = [DEVICE_CAVIUM_FPA, DEVICE_CNXK_NPA]
-COMPRESS_DEVICES = [DEVICE_CAVIUM_ZIP]
-REGEX_DEVICES = [DEVICE_CN9K_REE]
-ML_DEVICES = [DEVICE_CNXK_ML]
-MISC_DEVICES = [
- DEVICE_CNXK_BPHY,
- DEVICE_CNXK_BPHY_CGX,
- DEVICE_CNXK_INL_DEV,
- DEVICE_INTEL_NTB_SKX,
- DEVICE_INTEL_NTB_ICX,
- DEVICE_VIRTIO_BLK,
-]
-# which command line arguments/printouts correspond to which device types
-DEVICE_TYPE_ALL = "all"
-DEVICE_TYPE_NET = "net"
-DEVICE_TYPE_BASEBAND = "baseband"
-DEVICE_TYPE_CRYPTO = "crypto"
-DEVICE_TYPE_DMA = "dma"
-DEVICE_TYPE_EVENT = "event"
-DEVICE_TYPE_MEMPOOL = "mempool"
-DEVICE_TYPE_COMPRESS = "compress"
-DEVICE_TYPE_REGEX = "regex"
-DEVICE_TYPE_ML = "ml"
-DEVICE_TYPE_MISC = "misc"
-DEVICE_TYPES = {
- # device type: (printout name, device match pattern)
- DEVICE_TYPE_NET: ("Network", NETWORK_DEVICES),
- DEVICE_TYPE_BASEBAND: ("Baseband", BASEDBAND_DEVICES),
- DEVICE_TYPE_CRYPTO: ("Crypto", CRYPTO_DEVICES),
- DEVICE_TYPE_DMA: ("DMA", DMA_DEVICES),
- DEVICE_TYPE_EVENT: ("Eventdev", EVENTDEV_DEVICES),
- DEVICE_TYPE_MEMPOOL: ("Mempool", MEMPOOL_DEVICES),
- DEVICE_TYPE_COMPRESS: ("Compress", COMPRESS_DEVICES),
- DEVICE_TYPE_REGEX: ("Regex", REGEX_DEVICES),
- DEVICE_TYPE_ML: ("ML", ML_DEVICES),
- DEVICE_TYPE_MISC: ("Misc", MISC_DEVICES),
-}
-
-
-class DevbindError(Exception):
- """Generic error to be displayed by devbind."""
-
- def __init__(self, message: str):
- super().__init__(message)
- self.message = message
-
- def __str__(self) -> str:
- return self.message
-
-
-def category_key_match(key: str, value: str, pattern: StrOrList) -> bool:
- """Check if value matches the pattern according to key match rules."""
- # if pattern is a list of strings, recurse and check each item
- if isinstance(pattern, list):
- return any(
- category_key_match(key, value, pattern_item) for pattern_item in pattern
- )
- # pattern is a single string, use single string match rules
- if key == "Class":
- # special case for Class: it has to match from the beginning
- return value.startswith(pattern)
- # default case: exact match
- return value == pattern
-
-
-def parse_lspci_line(line: str) -> T.Dict[str, str]:
- """Parse lspci line and return a dictionary."""
- # the format can be either:
- # key: value
- # or
- # key: string representation [value]
- # we want to store both because we may want to display both
- res: T.Dict[str, str] = {}
- name, value = line.split("\t", 1)
- name = name.strip().rstrip(":")
- value = value.strip()
- # does this value have string representation?
- value_list = value.rsplit(" ", 1)
- if len(value_list) > 1:
- value_str, value = value_list
- # store string representation
- res[name + "_str"] = value_str
- # strip out brackets
- value = value.strip("[]")
- res[name] = value
-
- return res
-
-
-def resolve_pci_glob(dev: str) -> T.List[str]:
- """Returns a list of PCI devices matching a glob pattern."""
- pci_sysfs_path = "/sys/bus/pci/devices"
- for _glob in [dev, "0000:" + dev]:
- paths = [
- os.path.basename(path)
- for path in glob.glob(os.path.join(pci_sysfs_path, _glob))
- ]
- if paths:
- return paths
- return [dev]
-
-
-def check_installed(program: str, package: str) -> None:
- """Check if a program is installed."""
- if subprocess.call(
- ["which", program], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
- ):
- raise DevbindError(f"'{program}' not found - please install '{package}'.")
-
-
-def read_output(args: T.List[str]) -> str:
- """Run a subprocess, collect its output, and return it as a list of lines."""
- try:
- output = subprocess.check_output(args).decode("utf-8")
- except subprocess.CalledProcessError as e:
- raise DevbindError(f"Error running '{' '.join(args)}': {e}") from e
- return output
-
-
-def read_routed_interfaces() -> T.List[str]:
- """Find interfaces with active routes."""
- try:
- # use ip route's JSON output to get a list of active interfaces
- routes = json.loads(read_output(["ip", "-j", "route"]))
- except json.JSONDecodeError as e:
- raise DevbindError(f"Error parsing 'ip route' output: {e}") from e
- # find interfaces with active routes
- routed_ifs: T.List[str] = []
- for route in routes:
- # skip uninteresting routes
- if "169.254" in route["dst"]:
- continue
- if "dev" in route:
- routed_ifs.append(route["dev"])
- # dedupe list
- return list(set(routed_ifs))
-
-
-def sysfs_read_pci_drivers() -> T.List[str]:
- """Gather all PCI modules loaded on the system."""
- return os.listdir("/sys/bus/pci/drivers")
-
-
-def sysfs_device_get_path(dev: str, path: str) -> str:
- """Construct path in device sysfs directory."""
- return os.path.join("/sys/bus/pci/devices", dev, path)
-
-
-def sysfs_driver_get_path(driver: str, path: str) -> str:
- """Construct path in driver sysfs directory."""
- return os.path.join("/sys/bus/pci/drivers", driver, path)
-
-
-def sysfs_iommu_enabled() -> bool:
- """Check if IOMMU is enabled on the system."""
- return len(os.listdir("/sys/class/iommu")) > 0
-
-
-def sysfs_enable_unsafe_noiommu() -> None:
- """Enable unsafe no-IOMMU mode."""
- fname = "/sys/module/vfio/parameters/enable_unsafe_noiommu_mode"
- try:
- with open(fname, "r", encoding="utf-8") as f:
- val = f.read()
- if val in ["1", "Y", "y"]:
- # already enabled
- return
- except OSError as e:
- raise DevbindError(f"Cannot read unsafe no IOMMU mode status: {e}") from e
- try:
- with open(fname, "w", encoding="utf-8") as f:
- f.write("1")
- except OSError as e:
- raise DevbindError(f"Cannot write unsafe no IOMMU mode status: {e}") from e
- print(
- "Warning: IOMMU is not enabled, enabling unsafe no-IOMMU mode for VFIO drivers."
- )
-
-
-def sysfs_get_vfio_device(dev: str) -> str:
- """Get VFIO device file for a PCI device."""
- iommu_grp_base_path = sysfs_device_get_path(dev, "iommu_group")
- # extract group number from base path
- iommu_grp = os.path.basename(os.readlink(iommu_grp_base_path))
- # find VFIO device corresponding to this IOMMU group
- return os.path.join("/dev/vfio", iommu_grp)
-
-
-def device_vfio_set_ownership(dev: str, uid: int, gid: int) -> None:
- """Set device ownership."""
- try:
- os.chown(sysfs_get_vfio_device(dev), uid, gid)
- except OSError as e:
- raise DevbindError(f"Failed to set device ownership for {dev}: {e}") from e
-
-
-class Device:
- """Thin wrapper around a device dict read from lspci."""
-
- def __init__(self, dev_dict: T.Dict[str, str]):
- self._dev_dict = dev_dict
- self.slot = self._dev_dict["Slot"]
-
- # find kernel interfaces for this device
- self._update_interfaces()
-
- def __str__(self) -> str:
- return self.slot
-
- def __getitem__(self, key: str) -> str:
- """Get value as it appears in the device dictionary."""
- return self._get_str(key)
-
- def _set_value(self, key: str, value: StrOrList):
- """Generic setter for different fields."""
- if value:
- # value can be a list or a string
- if isinstance(value, list):
- self._dev_dict[key] = ",".join(value)
- else:
- self._dev_dict[key] = value
- elif key in self._dev_dict:
- # delete key if value is empty
- del self._dev_dict[key]
-
- def _get_str(self, key: str) -> str:
- """Generic getter for string fields."""
- if key in self._dev_dict:
- return self._dev_dict[key]
- return ""
-
- def _get_list(self, key: str) -> T.List[str]:
- """Generic getter for list fields."""
- if key in self._dev_dict:
- return [m.strip() for m in self._dev_dict[key].split(",")]
- return []
-
- def _update_interfaces(self):
- """Update interfaces for this device."""
- sysfs_path = sysfs_device_get_path(self.slot, "net")
- try:
- self.interfaces = os.listdir(sysfs_path)
- except OSError:
- pass
-
- def update(self):
- """Update device information from lspci."""
- self._dev_dict.clear()
- lspci_output = read_output(["lspci", "-Dvmmnnks", self.slot])
- for line in lspci_output.splitlines():
- if not line:
- continue
- self._dev_dict.update(parse_lspci_line(line))
- self._update_interfaces()
-
- def match(self, pattern: DeviceMatchPattern) -> bool:
- """Check if this device matches the pattern."""
- for key, match_pattern in pattern.items():
- if key not in self._dev_dict:
- return False
- value = self._dev_dict[key]
- if not category_key_match(key, value, match_pattern):
- return False
- return True
-
- @property
- def driver(self) -> str:
- """Get driver bound for this device."""
- return self._get_str("Driver")
-
- @driver.setter
- def driver(self, driver: str):
- """Set driver for this device."""
- self._set_value("Driver", driver)
-
- @property
- def modules(self) -> T.List[str]:
- """Get compatible modules for this device."""
- return self._get_list("Module")
-
- @modules.setter
- def modules(self, modules: T.List[str]):
- """Set compatible modules for this device."""
- self._set_value("Module", modules)
-
- @property
- def interfaces(self) -> T.List[str]:
- """Get interfaces for this device."""
- return self._get_list("Interface")
-
- @interfaces.setter
- def interfaces(self, interfaces: T.List[str]):
- """Set interfaces for this device."""
- self._set_value("Interface", interfaces)
-
- @property
- def active_interface(self) -> bool:
- """Return active interface information."""
- return bool(self._get_str("Active"))
-
- @active_interface.setter
- def active_interface(self, active: bool):
- """Set active interface information."""
- self._set_value("Active", "*Active*" if active else "")
-
-
-def read_devices_from_lspci() -> T.Iterable[Device]:
- """Read devices from lspci."""
- lspci_output = read_output(["lspci", "-Dvmmnnk"])
- cur_device: T.Dict[str, str] = {}
- for line in lspci_output.splitlines():
- if not line:
- dev = Device(cur_device)
- yield dev
- cur_device = {}
- else:
- cur_device.update(parse_lspci_line(line))
-
-
-class Devbind:
- """Class to cover various devbind-related operations and data."""
-
- def __init__(self) -> None:
- # gather all loaded kernel modules
- self.loaded_pci_modules = sysfs_read_pci_drivers()
- # find which ones are UIO modules
- self.uio_modules = self._find_loaded_uio_modules()
- # gather all routed interfaces
- self._routed_interfaces = read_routed_interfaces()
-
- # all detected PCI devices, keyed by PCI D:B:D.F
- self.pci_devices: T.Dict[str, Device] = {}
- # list of devices by type
- self.devices_by_type: T.Dict[str, T.List[Device]] = {
- devtype: [] for devtype in DEVICE_TYPES
- }
- # scan all PCI devices
- for dev in read_devices_from_lspci():
- # categorize device
- devtype = self._find_device_type(dev)
- if devtype:
- self.devices_by_type[devtype].append(dev)
-
- # fixup module and driver fields
- self._add_uio_modules(dev)
-
- # special case: find if any interfaces are active. non-network interfaces will not have
- # any interfaces at all, so it's safe to check all devices. we never update this
- # information, because once we start binding/unbinding, we have already acted on it.
- if any(iface in self._routed_interfaces for iface in dev.interfaces):
- dev.active_interface = True
-
- # save the device in common list
- self.pci_devices[dev.slot] = dev
-
- def _find_device_type(self, dev: Device) -> str:
- """Match a device against known device types."""
- for devtype, dt_tup in DEVICE_TYPES.items():
- _, patterns = dt_tup
- for pattern in patterns:
- if dev.match(pattern):
- return devtype
- return ""
-
- def _find_loaded_uio_modules(self) -> T.List[str]:
- loaded = set(self.loaded_pci_modules)
- supported = set(DPDK_KERNEL_MODULES)
- return list(loaded & supported)
-
- def _add_uio_modules(self, dev: Device) -> None:
- """Add loaded UIO modules to list of available modules."""
- # add UIO modules to list of supported modules
- modules = set(dev.modules + self.uio_modules)
-
- # make sure driver and module string do not have any duplicates
- if dev.driver in modules:
- modules.remove(dev.driver)
-
- # update list of compatible modules
- dev.modules = list(modules)
-
- def resolve_device(self, devstr: str) -> str:
- """Try to resolve a device into a PCI D:B:D:F."""
- # is this already a valid device?
- if devstr in self.pci_devices:
- return devstr
- # can we append domain to it?
- if "0000:" + devstr in self.pci_devices:
- return "0000:" + devstr
- # can we find a network interface name?
- for dev in self.devices_by_type[DEVICE_TYPE_NET]:
- if devstr in dev.interfaces:
- return dev.slot
- # we can't figure out what this is
- raise ValueError(
- f"Unknown device '{devstr}'. Please specify device in 'bus:slot.func' format."
- )
-
- def _can_modify(self, dev: Device, driver: str, force: bool) -> bool:
- """Check if we should attempt to modify this device."""
- # are we allowed to modify this device?
- if dev.active_interface and not force:
- print(
- f"Warning: routing table indicates that interface {dev} is active. "
- "Not modifying.",
- file=sys.stderr,
- )
- return False
-
- # does this device already use the driver we want to use?
- cur_driver = dev.driver
- if cur_driver == driver:
- # are we binding or unbinding?
- if driver:
- print(
- f"Notice: {dev} is already bound to driver {driver}, skipping bind",
- file=sys.stderr,
- )
- else:
- print(
- f"Notice: {dev} is not managed by any driver, skipping unbind",
- file=sys.stderr,
- )
- return False
-
- # all checks passed
- return True
-
- def unbind(self, dev: Device, force: bool) -> None:
- """Unbind one device from its current driver."""
- if not self._can_modify(dev, "", force):
- return
- cur_drv = dev.driver
- unbind_path = sysfs_driver_get_path(cur_drv, "unbind")
-
- print(f"Unbinding {dev} from {cur_drv}...")
-
- try:
- with open(unbind_path, "w", encoding="utf-8") as f:
- f.write(dev.slot)
- except OSError as e:
- raise DevbindError(f"Unbind failed for {dev}: {e}") from e
- # update device state
- dev.update()
- self._add_uio_modules(dev)
-
- def bind(self, dev: Device, driver: str, force: bool) -> None:
- """Bind one device to the specified driver."""
- if not self._can_modify(dev, driver, force):
- return
- override_path = sysfs_device_get_path(dev.slot, "driver_override")
- bind_path = sysfs_driver_get_path(driver, "bind")
-
- print(f"Binding {dev} to {driver}...")
-
- # are we binding to UIO module?
- if driver in self.uio_modules:
- try:
- with open(override_path, "w", encoding="utf-8") as f:
- f.write(driver)
- except OSError as e:
- raise DevbindError(f"Driver override failed for {dev}: {e}") from e
-
- # bind driver to device
- try:
- with open(bind_path, "a", encoding="utf-8") as f:
- f.write(dev.slot)
- except OSError as e:
- raise DevbindError(f"Bind failed for {dev}: {e}") from e
- # update device state
- dev.update()
- self._add_uio_modules(dev)
-
- # driver_override caches its value, so clean up by writing empty string
- try:
- with open(override_path, "w", encoding="utf-8") as f:
- f.write("\00")
- except OSError as e:
- raise DevbindError(f"CLeanup failed for {dev}: {e}") from e
-
-
-class DevbindCtx:
- """POD class to keep command-line arguments and context."""
-
- def __init__(self) -> None:
- self.status = False
- self.bind = False
- self.status_group: str
- self.driver: str
- self.devices: T.List[str]
- self.force: bool
- self.noiommu: bool
- self.vfio_uid: int
- self.vfio_gid: int
-
- self.devbind: Devbind
-
-
-def bind_devices(ctx: DevbindCtx) -> None:
- """Bind devices to the specified driver."""
- devbind = ctx.devbind
- use_vfio = ctx.driver == "vfio-pci"
-
- # a common user error is to forget to specify the driver the devices need to be bound to. check
- # if the driver is a valid device, and if it is, show a meaningful error.
- try:
- devbind.resolve_device(ctx.driver)
- # if we got here, the driver is a valid device, which is an error
- raise DevbindError(f"""\
-Driver '{ctx.driver}' does not look like a valid driver. Did you
-forget to specify the driver to bind the devices to?""")
- except ValueError:
- # driver generated error - it's not a valid device
- pass
-
- # validate all devices
- try:
- ctx.devices = [devbind.resolve_device(dev) for dev in ctx.devices]
- except ValueError as e:
- raise DevbindError(str(e)) from e
- devices = (devbind.pci_devices[dbdf] for dbdf in ctx.devices)
-
- # do we want to unbind?
- if not ctx.driver:
- # unbind devices
- for dev in devices:
- devbind.unbind(dev, ctx.force)
- return
-
- # validate driver
- if ctx.driver not in devbind.loaded_pci_modules:
- raise DevbindError(f"Driver '{ctx.driver}' is not loaded.")
-
- # check for IOMMU support
- if use_vfio and not sysfs_iommu_enabled():
- sysfs_enable_unsafe_noiommu()
-
- # bind all devices
- for dev in devices:
- rollback_driver = dev.driver
- # does this device have a driver already? if so, unbind
- if rollback_driver:
- devbind.unbind(dev, ctx.force)
-
- # device doesn't have any driver now, bind it
- try:
- devbind.bind(dev, ctx.driver, ctx.force)
- # bind succeeded, rollback no longer necessary
- rollback_driver = ""
- except DevbindError as e:
- # should we roll back?
- if rollback_driver:
- print(f"Warning: {e}")
- else:
- # pass the error up the stack
- raise
- if rollback_driver:
- devbind.bind(dev, rollback_driver, ctx.force)
- return
-
- # if we're binding to vfio-pci, set IOMMU user/group ownership if one was specified
- if use_vfio and (ctx.vfio_uid != -1 or ctx.vfio_gid != -1):
- device_vfio_set_ownership(dev.slot, ctx.vfio_uid, ctx.vfio_gid)
-
-
-def print_status_section(title: str, section_devs: T.List[Device]) -> None:
- """Prints subsection of device status (e.g. only kernel devices)."""
- # we will sort strings before printing
- strings: T.List[str] = []
-
- def _fmt_key_val(name: str, value: str) -> str:
- """Generate a devbind device printout string for a particular value."""
- # if there's a name provided, include it in the output
- if name and value:
- return f"{name}={value}"
- # otherwise just print the value, including empty ones
- return value
-
- # generate device strings
- for dev in section_devs:
- # construct strings
- devstr = f'{dev["Device_str"]} {dev["Device"]}'
- strs = [
- dev.slot,
- f"'{devstr}'",
- _fmt_key_val("drv", dev["Driver"]),
- _fmt_key_val("unused", dev["Module"]),
- _fmt_key_val("if", dev["Interface"]),
- _fmt_key_val("numa_node", dev["NUMANode"]),
- _fmt_key_val("", dev["Active"]),
- ]
- # filter out empty strings and join
- strings.append(" ".join(filter(None, strs)))
- strings.sort()
- print(f"{title}")
- print("=" * len(title))
- print("\n".join(strings))
- print()
-
-
-def print_status_group(
- ctx: DevbindCtx, group_title: str, group_devs: T.List[Device]
-) -> None:
- """Print status for a specific device group."""
- # do we have any devices at all?
- if not group_devs:
- msg = f"No {group_title} devices found."
- print(msg)
- print("=" * len(msg))
- print()
- return
-
- # split out all devices into three groups: kernel, non-kernel, and unbound
- kernel: T.List[Device] = []
- dpdk: T.List[Device] = []
- unbound: T.List[Device] = []
-
- for dev in group_devs:
- driver = dev.driver
- if driver in ctx.devbind.uio_modules:
- dpdk.append(dev)
- elif driver:
- kernel.append(dev)
- else:
- unbound.append(dev)
-
- # print out each group
- if dpdk:
- print_status_section(
- f"{group_title} devices using DPDK-compatible driver", dpdk
- )
- if kernel:
- print_status_section(f"{group_title} devices using kernel driver", kernel)
- if unbound:
- print_status_section(f"Other {group_title} devices", unbound)
-
-
-def print_status(ctx: DevbindCtx) -> None:
- """Print status of all devices."""
- # device_type to devbind type mapping
- for group_name, dt_t in DEVICE_TYPES.items():
- if ctx.status_group in [DEVICE_TYPE_ALL, group_name]:
- group_title, _ = dt_t
- group_devs = ctx.devbind.devices_by_type[group_name]
- print_status_group(ctx, group_title, group_devs)
-
-
-def parse_args() -> DevbindCtx:
- """Parse command-line arguments into devbind context."""
-
- parser = argparse.ArgumentParser(
- description="Utility to bind and unbind devices from Linux kernel",
- formatter_class=argparse.RawDescriptionHelpFormatter,
- epilog="""
-Examples:
----------
-
-To display current device status:
- %(prog)s --status
-
-To display current network device status:
- %(prog)s --status net
-
-To bind eth1 from the current driver and move to use vfio-pci
- %(prog)s --bind=vfio-pci eth1
-
-To unbind 0000:01:00.0 from using any driver
- %(prog)s -u 0000:01:00.0
-
-To bind 0000:02:00.0 and 0000:02:00.1 to the ixgbe kernel driver
- %(prog)s -b ixgbe 02:00.0 02:00.1
-""",
- )
-
- status_choices = [DEVICE_TYPE_ALL] + list(DEVICE_TYPES.keys())
-
- parser.add_argument(
- "-s",
- "--status",
- # backwards compatibility
- "--status-dev",
- # None if flag was not specified
- default=None,
- # "all" if flag was specified without arguments
- const=DEVICE_TYPE_ALL,
- # otherwise, match against the choice table
- nargs="?",
- choices=status_choices,
- help="Print the status of device group (default: all devices).",
- )
- bind_action = parser.add_mutually_exclusive_group()
- bind_action.add_argument(
- "-b",
- "--bind",
- metavar="DRIVER",
- help='Select the driver to use ("none" to unbind the device)',
- )
- bind_action.add_argument(
- "-u",
- "--unbind",
- action="store_true",
- help='Unbind a device (equivalent to "-b none")',
- )
- parser.add_argument(
- "--noiommu-mode",
- action="store_true",
- help="If IOMMU is not available, enable no IOMMU mode for VFIO drivers",
- )
- parser.add_argument(
- "--force",
- action="store_true",
- help="""\
-Override restriction on binding devices in use by Linux. WARNING: This can lead
-to loss of network connection and should be used with caution.
-""",
- )
- parser.add_argument(
- "-G",
- "--gid",
- type=lambda g: grp.getgrnam(g).gr_gid,
- default=-1,
- help="For VFIO, specify the group ID to set IOMMU group ownership",
- )
- parser.add_argument(
- "-U",
- "--uid",
- type=lambda u: pwd.getpwnam(u).pw_uid,
- default=-1,
- help="For VFIO, specify the user ID to set IOMMU group ownership",
- )
- parser.add_argument(
- "devices",
- metavar="DEVICE",
- nargs="*",
- help="""\
-Device specified as PCI "domain:bus:slot.func" syntax or "bus:slot.func" syntax.
-For devices bound to Linux kernel drivers, they may be referred to by interface name.
-""",
- )
-
- opt = parser.parse_args()
-
- ctx = DevbindCtx()
-
- if opt.status:
- ctx.status = True
- ctx.status_group = opt.status
- if opt.bind or opt.unbind:
- ctx.bind = True
- ctx.driver = "" if opt.unbind else opt.bind
- # support any capitalization for binding to "none"
- if ctx.driver.lower() == "none":
- ctx.driver = ""
- if not ctx.status and not ctx.bind:
- print("Error: No action specified.", file=sys.stderr)
- parser.print_usage()
- sys.exit(1)
-
- ctx.noiommu = opt.noiommu_mode
- ctx.force = opt.force
- ctx.devices = opt.devices
- ctx.vfio_uid = opt.uid
- ctx.vfio_gid = opt.gid
-
- # if status is displayed, devices shouldn't be passed
- if not ctx.bind and ctx.devices:
- print("Error: Devices should not be specified with --status action.")
- parser.print_usage()
- sys.exit(1)
- # if bind is used, devices should be passed
- elif ctx.bind and not ctx.devices:
- print("Error: No devices specified for --bind/--unbind action.")
- parser.print_usage()
- sys.exit(1)
- return ctx
-
-
-def _main():
- ctx = parse_args()
-
- # initialize devbind data
- ctx.devbind = Devbind()
-
- if ctx.bind:
- # resolve any PCI globs in devices
- ctx.devices = [d for dev in ctx.devices for d in resolve_pci_glob(dev)]
- bind_devices(ctx)
- print()
- if ctx.status:
- print_status(ctx)
-
-
-if __name__ == "__main__":
- try:
- # check if lspci and ip are installed before doing anything
- check_installed("lspci", "pciutils")
- check_installed("ip", "iproute2")
-
- # run the main function
- _main()
- except DevbindError as e:
- sys.exit(f"Error: {e}")
diff --git a/usertools/dpdk-devbind.py b/usertools/dpdk-devbind.py
index 283707fc16..1f2d8cb118 100755
--- a/usertools/dpdk-devbind.py
+++ b/usertools/dpdk-devbind.py
@@ -1,702 +1,848 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: BSD-3-Clause
-# Copyright(c) 2010-2014 Intel Corporation
+# Copyright(c) 2010-2024 Intel Corporation
#
+"""Script to bind PCI devices to DPDK-compatible userspace IO drivers."""
-import sys
+import argparse
+import glob
+import grp
+import json
import os
+import pwd
import subprocess
-import argparse
-import platform
-
-from glob import glob
-from os.path import exists, basename
-from os.path import join as path_join
-
-# The PCI base class for all devices
-network_class = {'Class': '02', 'Vendor': None, 'Device': None,
- 'SVendor': None, 'SDevice': None}
-acceleration_class = {'Class': '12', 'Vendor': None, 'Device': None,
- 'SVendor': None, 'SDevice': None}
-ifpga_class = {'Class': '12', 'Vendor': '8086', 'Device': '0b30',
- 'SVendor': None, 'SDevice': None}
-encryption_class = {'Class': '10', 'Vendor': None, 'Device': None,
- 'SVendor': None, 'SDevice': None}
-intel_processor_class = {'Class': '0b', 'Vendor': '8086', 'Device': None,
- 'SVendor': None, 'SDevice': None}
-cavium_sso = {'Class': '08', 'Vendor': '177d', 'Device': 'a04b,a04d',
- 'SVendor': None, 'SDevice': None}
-cavium_fpa = {'Class': '08', 'Vendor': '177d', 'Device': 'a053',
- 'SVendor': None, 'SDevice': None}
-cavium_pkx = {'Class': '08', 'Vendor': '177d', 'Device': 'a0dd,a049',
- 'SVendor': None, 'SDevice': None}
-cavium_tim = {'Class': '08', 'Vendor': '177d', 'Device': 'a051',
- 'SVendor': None, 'SDevice': None}
-cavium_zip = {'Class': '12', 'Vendor': '177d', 'Device': 'a037',
- 'SVendor': None, 'SDevice': None}
-avp_vnic = {'Class': '05', 'Vendor': '1af4', 'Device': '1110',
- 'SVendor': None, 'SDevice': None}
-
-cnxk_bphy = {'Class': '08', 'Vendor': '177d', 'Device': 'a089',
- 'SVendor': None, 'SDevice': None}
-cnxk_bphy_cgx = {'Class': '08', 'Vendor': '177d', 'Device': 'a059,a060',
- 'SVendor': None, 'SDevice': None}
-cnxk_dma = {'Class': '08', 'Vendor': '177d', 'Device': 'a081',
- 'SVendor': None, 'SDevice': None}
-cnxk_inl_dev = {'Class': '08', 'Vendor': '177d', 'Device': 'a0f0,a0f1',
- 'SVendor': None, 'SDevice': None}
-
-hisilicon_dma = {'Class': '08', 'Vendor': '19e5', 'Device': 'a122',
- 'SVendor': None, 'SDevice': None}
-odm_dma = {'Class': '08', 'Vendor': '177d', 'Device': 'a08c',
- 'SVendor': None, 'SDevice': None}
-
-intel_dlb = {'Class': '0b', 'Vendor': '8086', 'Device': '270b,2710,2714',
- 'SVendor': None, 'SDevice': None}
-intel_ioat_bdw = {'Class': '08', 'Vendor': '8086',
- 'Device': '6f20,6f21,6f22,6f23,6f24,6f25,6f26,6f27,6f2e,6f2f',
- 'SVendor': None, 'SDevice': None}
-intel_ioat_skx = {'Class': '08', 'Vendor': '8086', 'Device': '2021',
- 'SVendor': None, 'SDevice': None}
-intel_ioat_icx = {'Class': '08', 'Vendor': '8086', 'Device': '0b00',
- 'SVendor': None, 'SDevice': None}
-intel_idxd_spr = {'Class': '08', 'Vendor': '8086', 'Device': '0b25',
- 'SVendor': None, 'SDevice': None}
-intel_ntb_skx = {'Class': '06', 'Vendor': '8086', 'Device': '201c',
- 'SVendor': None, 'SDevice': None}
-intel_ntb_icx = {'Class': '06', 'Vendor': '8086', 'Device': '347e',
- 'SVendor': None, 'SDevice': None}
-
-cnxk_sso = {'Class': '08', 'Vendor': '177d', 'Device': 'a0f9,a0fa',
- 'SVendor': None, 'SDevice': None}
-cnxk_npa = {'Class': '08', 'Vendor': '177d', 'Device': 'a0fb,a0fc',
- 'SVendor': None, 'SDevice': None}
-cn9k_ree = {'Class': '08', 'Vendor': '177d', 'Device': 'a0f4',
- 'SVendor': None, 'SDevice': None}
-
-virtio_blk = {'Class': '01', 'Vendor': "1af4", 'Device': '1001,1042',
- 'SVendor': None, 'SDevice': None}
-
-cnxk_ml = {'Class': '08', 'Vendor': '177d', 'Device': 'a092',
- 'SVendor': None, 'SDevice': None}
-
-network_devices = [network_class, cavium_pkx, avp_vnic, ifpga_class]
-baseband_devices = [acceleration_class]
-crypto_devices = [encryption_class, intel_processor_class]
-dma_devices = [cnxk_dma, hisilicon_dma,
- intel_idxd_spr, intel_ioat_bdw, intel_ioat_icx, intel_ioat_skx,
- odm_dma]
-eventdev_devices = [cavium_sso, cavium_tim, intel_dlb, cnxk_sso]
-mempool_devices = [cavium_fpa, cnxk_npa]
-compress_devices = [cavium_zip]
-regex_devices = [cn9k_ree]
-ml_devices = [cnxk_ml]
-misc_devices = [cnxk_bphy, cnxk_bphy_cgx, cnxk_inl_dev,
- intel_ntb_skx, intel_ntb_icx,
- virtio_blk]
-
-# global dict ethernet devices present. Dictionary indexed by PCI address.
-# Each device within this is itself a dictionary of device properties
-devices = {}
-# list of supported DPDK drivers
-dpdk_drivers = ["igb_uio", "vfio-pci", "uio_pci_generic"]
-# list of currently loaded kernel modules
-loaded_modules = None
-
-# command-line arg flags
-b_flag = None
-status_flag = False
-force_flag = False
-noiommu_flag = False
-args = []
-
-
-# check if a specific kernel module is loaded
-def module_is_loaded(module):
- global loaded_modules
-
- if module == 'vfio_pci':
- module = 'vfio-pci'
-
- if loaded_modules:
- return module in loaded_modules
-
- # Get list of sysfs modules (both built-in and dynamically loaded)
- sysfs_path = '/sys/module/'
-
- # Get the list of directories in sysfs_path
- sysfs_mods = [m for m in os.listdir(sysfs_path)
- if os.path.isdir(os.path.join(sysfs_path, m))]
-
- # special case for vfio_pci (module is named vfio-pci,
- # but its .ko is named vfio_pci)
- sysfs_mods = [a if a != 'vfio_pci' else 'vfio-pci' for a in sysfs_mods]
-
- loaded_modules = sysfs_mods
-
- # add built-in modules as loaded
- release = platform.uname().release
- filename = os.path.join("/lib/modules/", release, "modules.builtin")
- if os.path.exists(filename):
- try:
- with open(filename) as f:
- loaded_modules += [os.path.splitext(os.path.basename(mod))[0] for mod in f]
- except IOError:
- print("Warning: cannot read list of built-in kernel modules")
-
- return module in loaded_modules
-
-
-def check_modules():
- '''Checks that igb_uio is loaded'''
- global dpdk_drivers
-
- # list of supported modules
- mods = [{"Name": driver, "Found": False} for driver in dpdk_drivers]
-
- # first check if module is loaded
- for mod in mods:
- if module_is_loaded(mod["Name"]):
- mod["Found"] = True
-
- # check if we have at least one loaded module
- if True not in [mod["Found"] for mod in mods] and b_flag is not None:
- print("Warning: no supported DPDK kernel modules are loaded", file=sys.stderr)
-
- # change DPDK driver list to only contain drivers that are loaded
- dpdk_drivers = [mod["Name"] for mod in mods if mod["Found"]]
-
-
-def has_driver(dev_id):
- '''return true if a device is assigned to a driver. False otherwise'''
- return "Driver_str" in devices[dev_id]
-
-
-def get_pci_device_details(dev_id, probe_lspci):
- '''This function gets additional details for a PCI device'''
- device = {}
-
- if probe_lspci:
- extra_info = subprocess.check_output(["lspci", "-vmmks", dev_id]).splitlines()
- # parse lspci details
- for line in extra_info:
- if not line:
- continue
- name, value = line.decode("utf8").split("\t", 1)
- name = name.strip(":") + "_str"
- device[name] = value
- # check for a unix interface name
- device["Interface"] = ""
- for base, dirs, _ in os.walk("/sys/bus/pci/devices/%s/" % dev_id):
- if "net" in dirs:
- device["Interface"] = \
- ",".join(os.listdir(os.path.join(base, "net")))
- break
- # check if a port is used for ssh connection
- device["Ssh_if"] = False
- device["Active"] = ""
-
- return device
-
-
-def clear_data():
- '''This function clears any old data'''
- global devices
- devices = {}
-
-
-def get_device_details(devices_type):
- '''This function populates the "devices" dictionary. The keys used are
- the pci addresses (domain:bus:slot.func). The values are themselves
- dictionaries - one for each NIC.'''
- global devices
- global dpdk_drivers
-
- # first loop through and read details for all devices
- # request machine readable format, with numeric IDs and String
- dev = {}
- dev_lines = subprocess.check_output(["lspci", "-Dvmmnnk"]).splitlines()
- for dev_line in dev_lines:
- if not dev_line:
- if device_type_match(dev, devices_type):
- # Replace "Driver" with "Driver_str" to have consistency of
- # of dictionary key names
- if "Driver" in dev.keys():
- dev["Driver_str"] = dev.pop("Driver")
- if "Module" in dev.keys():
- dev["Module_str"] = dev.pop("Module")
- # use dict to make copy of dev
- devices[dev["Slot"]] = dict(dev)
- # Clear previous device's data
- dev = {}
- else:
- name, value = dev_line.decode("utf8").split("\t", 1)
- value_list = value.rsplit(' ', 1)
- if value_list:
- # String stored in <name>_str
- dev[name.rstrip(":") + '_str'] = value_list[0]
- # Numeric IDs
- dev[name.rstrip(":")] = value_list[len(value_list) - 1] \
- .rstrip("]").lstrip("[")
-
- if devices_type == network_devices:
- # check what is the interface if any for an ssh connection if
- # any to this host, so we can mark it later.
- ssh_if = []
- route = subprocess.check_output(["ip", "-o", "route"])
- # filter out all lines for 169.254 routes
- route = "\n".join(filter(lambda ln: not ln.startswith("169.254"),
- route.decode().splitlines()))
- rt_info = route.split()
- for i in range(len(rt_info) - 1):
- if rt_info[i] == "dev":
- ssh_if.append(rt_info[i + 1])
-
- # based on the basic info, get extended text details
- for d in devices.keys():
- if not device_type_match(devices[d], devices_type):
+import sys
+import typing as T
+
+# the following list of modules is supported by DPDK
+DPDK_KERNEL_MODULES = {"igb_uio", "vfio-pci", "uio_pci_generic"}
+
+# pattern matching criteria for various devices and devices classes. keys are entries in lspci,
+# while values, if present are further matches for lspci criteria. values can be either strings or
+# list of strings, in which case any match is sufficient.
+StrOrList = T.Union[str, T.List[str]]
+DeviceMatchPattern = T.Dict[str, StrOrList]
+CLASS_NETWORK: DeviceMatchPattern = {
+ "Class": "02",
+}
+CLASS_ACCELERATION: DeviceMatchPattern = {
+ "Class": "12",
+}
+CLASS_IFPGA: DeviceMatchPattern = {
+ "Class": "12",
+ "Vendor": "8086",
+ "Device": "0b30",
+}
+CLASS_ENCRYPTION: DeviceMatchPattern = {
+ "Class": "10",
+}
+CLASS_INTEL_PROCESSOR: DeviceMatchPattern = {
+ "Class": "0b",
+ "Vendor": "8086",
+}
+DEVICE_CAVIUM_SSO: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": ["a04b", "a04d"],
+}
+DEVICE_CAVIUM_FPA: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": "a053",
+}
+DEVICE_CAVIUM_PKX: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": ["a0dd", "a049"],
+}
+DEVICE_CAVIUM_TIM: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": "a051",
+}
+DEVICE_CAVIUM_ZIP: DeviceMatchPattern = {
+ "Class": "12",
+ "Vendor": "177d",
+ "Device": "a037",
+}
+DEVICE_AVP_VNIC: DeviceMatchPattern = {
+ "Class": "05",
+ "Vendor": "1af4",
+ "Device": "1110",
+}
+DEVICE_CNXK_BPHY: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": "a089",
+}
+DEVICE_CNXK_BPHY_CGX: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": ["a059", "a060"],
+}
+DEVICE_CNXK_DMA: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": "a081",
+}
+DEVICE_CNXK_INL_DEV: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": ["a0f0", "a0f1"],
+}
+DEVICE_HISILICON_DMA: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "19e5",
+ "Device": "a122",
+}
+DEVICE_ODM_DMA: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": "a08c",
+}
+DEVICE_INTEL_DLB: DeviceMatchPattern = {
+ "Class": "0b",
+ "Vendor": "8086",
+ "Device": ["270b", "2710", "2714"],
+}
+DEVICE_INTEL_IOAT_BDW: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "8086",
+ "Device": [
+ "6f20",
+ "6f21",
+ "6f22",
+ "6f23",
+ "6f24",
+ "6f25",
+ "6f26",
+ "6f27",
+ "6f2e",
+ "6f2f",
+ ],
+}
+DEVICE_INTEL_IOAT_SKX: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "8086",
+ "Device": "2021",
+}
+DEVICE_INTEL_IOAT_ICX: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "8086",
+ "Device": "0b00",
+}
+DEVICE_INTEL_IDXD_SPR: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "8086",
+ "Device": "0b25",
+}
+DEVICE_INTEL_NTB_SKX: DeviceMatchPattern = {
+ "Class": "06",
+ "Vendor": "8086",
+ "Device": "201c",
+}
+DEVICE_INTEL_NTB_ICX: DeviceMatchPattern = {
+ "Class": "06",
+ "Vendor": "8086",
+ "Device": "347e",
+}
+DEVICE_CNXK_SSO: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": ["a0f9", "a0fa"],
+}
+DEVICE_CNXK_NPA: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": ["a0fb", "a0fc"],
+}
+DEVICE_CN9K_REE: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": "a0f4",
+}
+DEVICE_VIRTIO_BLK: DeviceMatchPattern = {
+ "Class": "01",
+ "Vendor": "1af4",
+ "Device": ["1001", "1042"],
+}
+DEVICE_CNXK_ML: DeviceMatchPattern = {
+ "Class": "08",
+ "Vendor": "177d",
+ "Device": "a092",
+}
+
+# device types as recognized by devbind
+NETWORK_DEVICES = [CLASS_NETWORK, CLASS_IFPGA, DEVICE_CAVIUM_PKX, DEVICE_AVP_VNIC]
+BASEDBAND_DEVICES = [CLASS_ACCELERATION]
+CRYPTO_DEVICES = [CLASS_ENCRYPTION, CLASS_INTEL_PROCESSOR]
+DMA_DEVICES = [
+ DEVICE_CNXK_DMA,
+ DEVICE_HISILICON_DMA,
+ DEVICE_INTEL_IDXD_SPR,
+ DEVICE_INTEL_IOAT_BDW,
+ DEVICE_INTEL_IOAT_ICX,
+ DEVICE_INTEL_IOAT_SKX,
+ DEVICE_ODM_DMA,
+]
+EVENTDEV_DEVICES = [
+ DEVICE_CAVIUM_SSO,
+ DEVICE_CAVIUM_TIM,
+ DEVICE_INTEL_DLB,
+ DEVICE_CNXK_SSO,
+]
+MEMPOOL_DEVICES = [DEVICE_CAVIUM_FPA, DEVICE_CNXK_NPA]
+COMPRESS_DEVICES = [DEVICE_CAVIUM_ZIP]
+REGEX_DEVICES = [DEVICE_CN9K_REE]
+ML_DEVICES = [DEVICE_CNXK_ML]
+MISC_DEVICES = [
+ DEVICE_CNXK_BPHY,
+ DEVICE_CNXK_BPHY_CGX,
+ DEVICE_CNXK_INL_DEV,
+ DEVICE_INTEL_NTB_SKX,
+ DEVICE_INTEL_NTB_ICX,
+ DEVICE_VIRTIO_BLK,
+]
+# which command line arguments/printouts correspond to which device types
+DEVICE_TYPE_ALL = "all"
+DEVICE_TYPE_NET = "net"
+DEVICE_TYPE_BASEBAND = "baseband"
+DEVICE_TYPE_CRYPTO = "crypto"
+DEVICE_TYPE_DMA = "dma"
+DEVICE_TYPE_EVENT = "event"
+DEVICE_TYPE_MEMPOOL = "mempool"
+DEVICE_TYPE_COMPRESS = "compress"
+DEVICE_TYPE_REGEX = "regex"
+DEVICE_TYPE_ML = "ml"
+DEVICE_TYPE_MISC = "misc"
+DEVICE_TYPES = {
+ # device type: (printout name, device match pattern)
+ DEVICE_TYPE_NET: ("Network", NETWORK_DEVICES),
+ DEVICE_TYPE_BASEBAND: ("Baseband", BASEDBAND_DEVICES),
+ DEVICE_TYPE_CRYPTO: ("Crypto", CRYPTO_DEVICES),
+ DEVICE_TYPE_DMA: ("DMA", DMA_DEVICES),
+ DEVICE_TYPE_EVENT: ("Eventdev", EVENTDEV_DEVICES),
+ DEVICE_TYPE_MEMPOOL: ("Mempool", MEMPOOL_DEVICES),
+ DEVICE_TYPE_COMPRESS: ("Compress", COMPRESS_DEVICES),
+ DEVICE_TYPE_REGEX: ("Regex", REGEX_DEVICES),
+ DEVICE_TYPE_ML: ("ML", ML_DEVICES),
+ DEVICE_TYPE_MISC: ("Misc", MISC_DEVICES),
+}
+
+
+class DevbindError(Exception):
+ """Generic error to be displayed by devbind."""
+
+ def __init__(self, message: str):
+ super().__init__(message)
+ self.message = message
+
+ def __str__(self) -> str:
+ return self.message
+
+
+def category_key_match(key: str, value: str, pattern: StrOrList) -> bool:
+ """Check if value matches the pattern according to key match rules."""
+ # if pattern is a list of strings, recurse and check each item
+ if isinstance(pattern, list):
+ return any(
+ category_key_match(key, value, pattern_item) for pattern_item in pattern
+ )
+ # pattern is a single string, use single string match rules
+ if key == "Class":
+ # special case for Class: it has to match from the beginning
+ return value.startswith(pattern)
+ # default case: exact match
+ return value == pattern
+
+
+def parse_lspci_line(line: str) -> T.Dict[str, str]:
+ """Parse lspci line and return a dictionary."""
+ # the format can be either:
+ # key: value
+ # or
+ # key: string representation [value]
+ # we want to store both because we may want to display both
+ res: T.Dict[str, str] = {}
+ name, value = line.split("\t", 1)
+ name = name.strip().rstrip(":")
+ value = value.strip()
+ # does this value have string representation?
+ value_list = value.rsplit(" ", 1)
+ if len(value_list) > 1:
+ value_str, value = value_list
+ # store string representation
+ res[name + "_str"] = value_str
+ # strip out brackets
+ value = value.strip("[]")
+ res[name] = value
+
+ return res
+
+
+def resolve_pci_glob(dev: str) -> T.List[str]:
+ """Returns a list of PCI devices matching a glob pattern."""
+ pci_sysfs_path = "/sys/bus/pci/devices"
+ for _glob in [dev, "0000:" + dev]:
+ paths = [
+ os.path.basename(path)
+ for path in glob.glob(os.path.join(pci_sysfs_path, _glob))
+ ]
+ if paths:
+ return paths
+ return [dev]
+
+
+def check_installed(program: str, package: str) -> None:
+ """Check if a program is installed."""
+ if subprocess.call(
+ ["which", program], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
+ ):
+ raise DevbindError(f"'{program}' not found - please install '{package}'.")
+
+
+def read_output(args: T.List[str]) -> str:
+ """Run a subprocess, collect its output, and return it as a list of lines."""
+ try:
+ output = subprocess.check_output(args).decode("utf-8")
+ except subprocess.CalledProcessError as e:
+ raise DevbindError(f"Error running '{' '.join(args)}': {e}") from e
+ return output
+
+
+def read_routed_interfaces() -> T.List[str]:
+ """Find interfaces with active routes."""
+ try:
+ # use ip route's JSON output to get a list of active interfaces
+ routes = json.loads(read_output(["ip", "-j", "route"]))
+ except json.JSONDecodeError as e:
+ raise DevbindError(f"Error parsing 'ip route' output: {e}") from e
+ # find interfaces with active routes
+ routed_ifs: T.List[str] = []
+ for route in routes:
+ # skip uninteresting routes
+ if "169.254" in route["dst"]:
continue
+ if "dev" in route:
+ routed_ifs.append(route["dev"])
+ # dedupe list
+ return list(set(routed_ifs))
- # get additional info and add it to existing data
- devices[d] = devices[d].copy()
- # No need to probe lspci
- devices[d].update(get_pci_device_details(d, False).items())
- if devices_type == network_devices:
- for _if in ssh_if:
- if _if in devices[d]["Interface"].split(","):
- devices[d]["Ssh_if"] = True
- devices[d]["Active"] = "*Active*"
- break
+def sysfs_read_pci_drivers() -> T.List[str]:
+ """Gather all PCI modules loaded on the system."""
+ return os.listdir("/sys/bus/pci/drivers")
- # add igb_uio to list of supporting modules if needed
- if "Module_str" in devices[d]:
- for driver in dpdk_drivers:
- if driver not in devices[d]["Module_str"]:
- devices[d]["Module_str"] = \
- devices[d]["Module_str"] + ",%s" % driver
- else:
- devices[d]["Module_str"] = ",".join(dpdk_drivers)
- # make sure the driver and module strings do not have any duplicates
- if has_driver(d):
- modules = devices[d]["Module_str"].split(",")
- if devices[d]["Driver_str"] in modules:
- modules.remove(devices[d]["Driver_str"])
- devices[d]["Module_str"] = ",".join(modules)
+def sysfs_device_get_path(dev: str, path: str) -> str:
+ """Construct path in device sysfs directory."""
+ return os.path.join("/sys/bus/pci/devices", dev, path)
-def device_type_match(dev, devices_type):
- for i in range(len(devices_type)):
- param_count = len(
- [x for x in devices_type[i].values() if x is not None])
- match_count = 0
- if dev["Class"][0:2] == devices_type[i]["Class"]:
- match_count = match_count + 1
- for key in devices_type[i].keys():
- if key != 'Class' and devices_type[i][key]:
- value_list = devices_type[i][key].split(',')
- for value in value_list:
- if value.strip(' ') == dev[key]:
- match_count = match_count + 1
- # count must be the number of non None parameters to match
- if match_count == param_count:
- return True
- return False
+def sysfs_driver_get_path(driver: str, path: str) -> str:
+ """Construct path in driver sysfs directory."""
+ return os.path.join("/sys/bus/pci/drivers", driver, path)
-def dev_id_from_dev_name(dev_name):
- '''Take a device "name" - a string passed in by user to identify a NIC
- device, and determine the device id - i.e. the domain:bus:slot.func - for
- it, which can then be used to index into the devices array'''
-
- # check if it's already a suitable index
- if dev_name in devices:
- return dev_name
- # check if it's an index just missing the domain part
- if "0000:" + dev_name in devices:
- return "0000:" + dev_name
-
- # check if it's an interface name, e.g. eth1
- for d in devices.keys():
- if dev_name in devices[d]["Interface"].split(","):
- return devices[d]["Slot"]
- # if nothing else matches - error
- raise ValueError("Unknown device: %s. "
- "Please specify device in \"bus:slot.func\" format" % dev_name)
-
-
-def unbind_one(dev_id, force):
- '''Unbind the device identified by "dev_id" from its current driver'''
- dev = devices[dev_id]
- if not has_driver(dev_id):
- print("Notice: %s %s %s is not currently managed by any driver" %
- (dev["Slot"], dev["Device_str"], dev["Interface"]), file=sys.stderr)
- return
-
- # prevent us disconnecting ourselves
- if dev["Ssh_if"] and not force:
- print("Warning: routing table indicates that interface %s is active. "
- "Skipping unbind" % dev_id, file=sys.stderr)
- return
-
- # write to /sys to unbind
- filename = "/sys/bus/pci/drivers/%s/unbind" % dev["Driver_str"]
- try:
- f = open(filename, "a")
- except OSError as err:
- sys.exit("Error: unbind failed for %s - Cannot open %s: %s" %
- (dev_id, filename, err))
- f.write(dev_id)
- f.close()
-
-
-def bind_one(dev_id, driver, force):
- '''Bind the device given by "dev_id" to the driver "driver". If the device
- is already bound to a different driver, it will be unbound first'''
- dev = devices[dev_id]
- saved_driver = None # used to rollback any unbind in case of failure
-
- # prevent disconnection of our ssh session
- if dev["Ssh_if"] and not force:
- print("Warning: routing table indicates that interface %s is active. "
- "Not modifying" % dev_id, file=sys.stderr)
- return
-
- # unbind any existing drivers we don't want
- if has_driver(dev_id):
- if dev["Driver_str"] == driver:
- print("Notice: %s already bound to driver %s, skipping" %
- (dev_id, driver), file=sys.stderr)
- return
- saved_driver = dev["Driver_str"]
- unbind_one(dev_id, force)
- dev["Driver_str"] = "" # clear driver string
-
- # For kernels >= 3.15 driver_override can be used to specify the driver
- # for a device rather than relying on the driver to provide a positive
- # match of the device. The existing process of looking up
- # the vendor and device ID, adding them to the driver new_id,
- # will erroneously bind other devices too which has the additional burden
- # of unbinding those devices
- if driver in dpdk_drivers:
- filename = "/sys/bus/pci/devices/%s/driver_override" % dev_id
- if exists(filename):
- try:
- f = open(filename, "w")
- except OSError as err:
- print("Error: bind failed for %s - Cannot open %s: %s"
- % (dev_id, filename, err), file=sys.stderr)
- return
- try:
- f.write("%s" % driver)
- f.close()
- except OSError as err:
- print("Error: bind failed for %s - Cannot write driver %s to "
- "PCI ID: %s" % (dev_id, driver, err), file=sys.stderr)
- return
- # For kernels < 3.15 use new_id to add PCI id's to the driver
- else:
- filename = "/sys/bus/pci/drivers/%s/new_id" % driver
- try:
- f = open(filename, "w")
- except OSError as err:
- print("Error: bind failed for %s - Cannot open %s: %s"
- % (dev_id, filename, err), file=sys.stderr)
- return
- try:
- # Convert Device and Vendor Id to int to write to new_id
- f.write("%04x %04x" % (int(dev["Vendor"], 16),
- int(dev["Device"], 16)))
- f.close()
- except OSError as err:
- print("Error: bind failed for %s - Cannot write new PCI ID to "
- "driver %s: %s" % (dev_id, driver, err), file=sys.stderr)
- return
-
- # do the bind by writing to /sys
- filename = "/sys/bus/pci/drivers/%s/bind" % driver
- try:
- f = open(filename, "a")
- except OSError as err:
- print("Error: bind failed for %s - Cannot open %s: %s"
- % (dev_id, filename, err), file=sys.stderr)
- if saved_driver is not None: # restore any previous driver
- bind_one(dev_id, saved_driver, force)
- return
- try:
- f.write(dev_id)
- f.close()
- except OSError as err:
- # for some reason, closing dev_id after adding a new PCI ID to new_id
- # results in IOError. however, if the device was successfully bound,
- # we don't care for any errors and can safely ignore IOError
- tmp = get_pci_device_details(dev_id, True)
- if "Driver_str" in tmp and tmp["Driver_str"] == driver:
- return
- print("Error: bind failed for %s - Cannot bind to driver %s: %s"
- % (dev_id, driver, err), file=sys.stderr)
- if saved_driver is not None: # restore any previous driver
- bind_one(dev_id, saved_driver, force)
- return
-
- # For kernels > 3.15 driver_override is used to bind a device to a driver.
- # Before unbinding it, overwrite driver_override with empty string so that
- # the device can be bound to any other driver
- filename = "/sys/bus/pci/devices/%s/driver_override" % dev_id
- if exists(filename):
- try:
- f = open(filename, "w")
- except OSError as err:
- sys.exit("Error: unbind failed for %s - Cannot open %s: %s"
- % (dev_id, filename, err))
- try:
- f.write("\00")
- f.close()
- except OSError as err:
- sys.exit("Error: unbind failed for %s - Cannot write %s: %s"
- % (dev_id, filename, err))
-
-
-def unbind_all(dev_list, force=False):
- """Unbind method, takes a list of device locations"""
-
- if dev_list[0] == "dpdk":
- for d in devices.keys():
- if "Driver_str" in devices[d]:
- if devices[d]["Driver_str"] in dpdk_drivers:
- unbind_one(devices[d]["Slot"], force)
- return
-
- try:
- dev_list = map(dev_id_from_dev_name, dev_list)
- except ValueError as ex:
- print(ex)
- sys.exit(1)
-
- for d in dev_list:
- unbind_one(d, force)
-
-
-def has_iommu():
- """Check if IOMMU is enabled on system"""
+def sysfs_iommu_enabled() -> bool:
+ """Check if IOMMU is enabled on the system."""
return len(os.listdir("/sys/class/iommu")) > 0
-def check_noiommu_mode():
- """Check and enable the noiommu mode for VFIO drivers"""
- global noiommu_flag
- filename = "/sys/module/vfio/parameters/enable_unsafe_noiommu_mode"
-
+def sysfs_enable_unsafe_noiommu() -> None:
+ """Enable unsafe no-IOMMU mode."""
+ fname = "/sys/module/vfio/parameters/enable_unsafe_noiommu_mode"
try:
- with open(filename, "r") as f:
- value = f.read(1)
- if value in ("1", "y" ,"Y"):
- return
- except OSError as err:
- sys.exit(f"Error: failed to check unsafe noiommu mode - Cannot open {filename}: {err}")
-
- if not noiommu_flag:
- sys.exit("Error: IOMMU support is disabled, use --noiommu-mode for binding in noiommu mode")
-
+ with open(fname, "r", encoding="utf-8") as f:
+ val = f.read()
+ if val in ["1", "Y", "y"]:
+ # already enabled
+ return
+ except OSError as e:
+ raise DevbindError(f"Cannot read unsafe no IOMMU mode status: {e}") from e
try:
- with open(filename, "w") as f:
+ with open(fname, "w", encoding="utf-8") as f:
f.write("1")
- except OSError as err:
- sys.exit(f"Error: failed to enable unsafe noiommu mode - Cannot open {filename}: {err}")
- print("Warning: enabling unsafe no IOMMU mode for VFIO drivers")
+ except OSError as e:
+ raise DevbindError(f"Cannot write unsafe no IOMMU mode status: {e}") from e
+ print(
+ "Warning: IOMMU is not enabled, enabling unsafe no-IOMMU mode for VFIO drivers."
+ )
-def bind_all(dev_list, driver, force=False):
- """Bind method, takes a list of device locations"""
- global devices
+def sysfs_get_vfio_device(dev: str) -> str:
+ """Get VFIO device file for a PCI device."""
+ iommu_grp_base_path = sysfs_device_get_path(dev, "iommu_group")
+ # extract group number from base path
+ iommu_grp = os.path.basename(os.readlink(iommu_grp_base_path))
+ # find VFIO device corresponding to this IOMMU group
+ return os.path.join("/dev/vfio", iommu_grp)
- # a common user error is to forget to specify the driver the devices need to
- # be bound to. check if the driver is a valid device, and if it is, show
- # a meaningful error.
+
+def device_vfio_set_ownership(dev: str, uid: int, gid: int) -> None:
+ """Set device ownership."""
+ try:
+ os.chown(sysfs_get_vfio_device(dev), uid, gid)
+ except OSError as e:
+ raise DevbindError(f"Failed to set device ownership for {dev}: {e}") from e
+
+
+class Device:
+ """Thin wrapper around a device dict read from lspci."""
+
+ def __init__(self, dev_dict: T.Dict[str, str]):
+ self._dev_dict = dev_dict
+ self.slot = self._dev_dict["Slot"]
+
+ # find kernel interfaces for this device
+ self._update_interfaces()
+
+ def __str__(self) -> str:
+ return self.slot
+
+ def __getitem__(self, key: str) -> str:
+ """Get value as it appears in the device dictionary."""
+ return self._get_str(key)
+
+ def _set_value(self, key: str, value: StrOrList):
+ """Generic setter for different fields."""
+ if value:
+ # value can be a list or a string
+ if isinstance(value, list):
+ self._dev_dict[key] = ",".join(value)
+ else:
+ self._dev_dict[key] = value
+ elif key in self._dev_dict:
+ # delete key if value is empty
+ del self._dev_dict[key]
+
+ def _get_str(self, key: str) -> str:
+ """Generic getter for string fields."""
+ if key in self._dev_dict:
+ return self._dev_dict[key]
+ return ""
+
+ def _get_list(self, key: str) -> T.List[str]:
+ """Generic getter for list fields."""
+ if key in self._dev_dict:
+ return [m.strip() for m in self._dev_dict[key].split(",")]
+ return []
+
+ def _update_interfaces(self):
+ """Update interfaces for this device."""
+ sysfs_path = sysfs_device_get_path(self.slot, "net")
+ try:
+ self.interfaces = os.listdir(sysfs_path)
+ except OSError:
+ pass
+
+ def update(self):
+ """Update device information from lspci."""
+ self._dev_dict.clear()
+ lspci_output = read_output(["lspci", "-Dvmmnnks", self.slot])
+ for line in lspci_output.splitlines():
+ if not line:
+ continue
+ self._dev_dict.update(parse_lspci_line(line))
+ self._update_interfaces()
+
+ def match(self, pattern: DeviceMatchPattern) -> bool:
+ """Check if this device matches the pattern."""
+ for key, match_pattern in pattern.items():
+ if key not in self._dev_dict:
+ return False
+ value = self._dev_dict[key]
+ if not category_key_match(key, value, match_pattern):
+ return False
+ return True
+
+ @property
+ def driver(self) -> str:
+ """Get driver bound for this device."""
+ return self._get_str("Driver")
+
+ @driver.setter
+ def driver(self, driver: str):
+ """Set driver for this device."""
+ self._set_value("Driver", driver)
+
+ @property
+ def modules(self) -> T.List[str]:
+ """Get compatible modules for this device."""
+ return self._get_list("Module")
+
+ @modules.setter
+ def modules(self, modules: T.List[str]):
+ """Set compatible modules for this device."""
+ self._set_value("Module", modules)
+
+ @property
+ def interfaces(self) -> T.List[str]:
+ """Get interfaces for this device."""
+ return self._get_list("Interface")
+
+ @interfaces.setter
+ def interfaces(self, interfaces: T.List[str]):
+ """Set interfaces for this device."""
+ self._set_value("Interface", interfaces)
+
+ @property
+ def active_interface(self) -> bool:
+ """Return active interface information."""
+ return bool(self._get_str("Active"))
+
+ @active_interface.setter
+ def active_interface(self, active: bool):
+ """Set active interface information."""
+ self._set_value("Active", "*Active*" if active else "")
+
+
+def read_devices_from_lspci() -> T.Iterable[Device]:
+ """Read devices from lspci."""
+ lspci_output = read_output(["lspci", "-Dvmmnnk"])
+ cur_device: T.Dict[str, str] = {}
+ for line in lspci_output.splitlines():
+ if not line:
+ dev = Device(cur_device)
+ yield dev
+ cur_device = {}
+ else:
+ cur_device.update(parse_lspci_line(line))
+
+
+class Devbind:
+ """Class to cover various devbind-related operations and data."""
+
+ def __init__(self) -> None:
+ # gather all loaded kernel modules
+ self.loaded_pci_modules = sysfs_read_pci_drivers()
+ # find which ones are UIO modules
+ self.uio_modules = self._find_loaded_uio_modules()
+ # gather all routed interfaces
+ self._routed_interfaces = read_routed_interfaces()
+
+ # all detected PCI devices, keyed by PCI D:B:D.F
+ self.pci_devices: T.Dict[str, Device] = {}
+ # list of devices by type
+ self.devices_by_type: T.Dict[str, T.List[Device]] = {
+ devtype: [] for devtype in DEVICE_TYPES
+ }
+ # scan all PCI devices
+ for dev in read_devices_from_lspci():
+ # categorize device
+ devtype = self._find_device_type(dev)
+ if devtype:
+ self.devices_by_type[devtype].append(dev)
+
+ # fixup module and driver fields
+ self._add_uio_modules(dev)
+
+ # special case: find if any interfaces are active. non-network interfaces will not have
+ # any interfaces at all, so it's safe to check all devices. we never update this
+ # information, because once we start binding/unbinding, we have already acted on it.
+ if any(iface in self._routed_interfaces for iface in dev.interfaces):
+ dev.active_interface = True
+
+ # save the device in common list
+ self.pci_devices[dev.slot] = dev
+
+ def _find_device_type(self, dev: Device) -> str:
+ """Match a device against known device types."""
+ for devtype, dt_tup in DEVICE_TYPES.items():
+ _, patterns = dt_tup
+ for pattern in patterns:
+ if dev.match(pattern):
+ return devtype
+ return ""
+
+ def _find_loaded_uio_modules(self) -> T.List[str]:
+ loaded = set(self.loaded_pci_modules)
+ supported = set(DPDK_KERNEL_MODULES)
+ return list(loaded & supported)
+
+ def _add_uio_modules(self, dev: Device) -> None:
+ """Add loaded UIO modules to list of available modules."""
+ # add UIO modules to list of supported modules
+ modules = set(dev.modules + self.uio_modules)
+
+ # make sure driver and module string do not have any duplicates
+ if dev.driver in modules:
+ modules.remove(dev.driver)
+
+ # update list of compatible modules
+ dev.modules = list(modules)
+
+ def resolve_device(self, devstr: str) -> str:
+ """Try to resolve a device into a PCI D:B:D:F."""
+ # is this already a valid device?
+ if devstr in self.pci_devices:
+ return devstr
+ # can we append domain to it?
+ if "0000:" + devstr in self.pci_devices:
+ return "0000:" + devstr
+ # can we find a network interface name?
+ for dev in self.devices_by_type[DEVICE_TYPE_NET]:
+ if devstr in dev.interfaces:
+ return dev.slot
+ # we can't figure out what this is
+ raise ValueError(
+ f"Unknown device '{devstr}'. Please specify device in 'bus:slot.func' format."
+ )
+
+ def _can_modify(self, dev: Device, driver: str, force: bool) -> bool:
+ """Check if we should attempt to modify this device."""
+ # are we allowed to modify this device?
+ if dev.active_interface and not force:
+ print(
+ f"Warning: routing table indicates that interface {dev} is active. "
+ "Not modifying.",
+ file=sys.stderr,
+ )
+ return False
+
+ # does this device already use the driver we want to use?
+ cur_driver = dev.driver
+ if cur_driver == driver:
+ # are we binding or unbinding?
+ if driver:
+ print(
+ f"Notice: {dev} is already bound to driver {driver}, skipping bind",
+ file=sys.stderr,
+ )
+ else:
+ print(
+ f"Notice: {dev} is not managed by any driver, skipping unbind",
+ file=sys.stderr,
+ )
+ return False
+
+ # all checks passed
+ return True
+
+ def unbind(self, dev: Device, force: bool) -> None:
+ """Unbind one device from its current driver."""
+ if not self._can_modify(dev, "", force):
+ return
+ cur_drv = dev.driver
+ unbind_path = sysfs_driver_get_path(cur_drv, "unbind")
+
+ print(f"Unbinding {dev} from {cur_drv}...")
+
+ try:
+ with open(unbind_path, "w", encoding="utf-8") as f:
+ f.write(dev.slot)
+ except OSError as e:
+ raise DevbindError(f"Unbind failed for {dev}: {e}") from e
+ # update device state
+ dev.update()
+ self._add_uio_modules(dev)
+
+ def bind(self, dev: Device, driver: str, force: bool) -> None:
+ """Bind one device to the specified driver."""
+ if not self._can_modify(dev, driver, force):
+ return
+ override_path = sysfs_device_get_path(dev.slot, "driver_override")
+ bind_path = sysfs_driver_get_path(driver, "bind")
+
+ print(f"Binding {dev} to {driver}...")
+
+ # are we binding to UIO module?
+ if driver in self.uio_modules:
+ try:
+ with open(override_path, "w", encoding="utf-8") as f:
+ f.write(driver)
+ except OSError as e:
+ raise DevbindError(f"Driver override failed for {dev}: {e}") from e
+
+ # bind driver to device
+ try:
+ with open(bind_path, "a", encoding="utf-8") as f:
+ f.write(dev.slot)
+ except OSError as e:
+ raise DevbindError(f"Bind failed for {dev}: {e}") from e
+ # update device state
+ dev.update()
+ self._add_uio_modules(dev)
+
+ # driver_override caches its value, so clean up by writing empty string
+ try:
+ with open(override_path, "w", encoding="utf-8") as f:
+ f.write("\00")
+ except OSError as e:
+ raise DevbindError(f"CLeanup failed for {dev}: {e}") from e
+
+
+class DevbindCtx:
+ """POD class to keep command-line arguments and context."""
+
+ def __init__(self) -> None:
+ self.status = False
+ self.bind = False
+ self.status_group: str
+ self.driver: str
+ self.devices: T.List[str]
+ self.force: bool
+ self.noiommu: bool
+ self.vfio_uid: int
+ self.vfio_gid: int
+
+ self.devbind: Devbind
+
+
+def bind_devices(ctx: DevbindCtx) -> None:
+ """Bind devices to the specified driver."""
+ devbind = ctx.devbind
+ use_vfio = ctx.driver == "vfio-pci"
+
+ # a common user error is to forget to specify the driver the devices need to be bound to. check
+ # if the driver is a valid device, and if it is, show a meaningful error.
try:
- dev_id_from_dev_name(driver)
- # if we've made it this far, this means that the "driver" was a valid
- # device string, so it's probably not a valid driver name.
- sys.exit("Error: Driver '%s' does not look like a valid driver. "
- "Did you forget to specify the driver to bind devices to?" % driver)
+ devbind.resolve_device(ctx.driver)
+ # if we got here, the driver is a valid device, which is an error
+ raise DevbindError(f"""\
+Driver '{ctx.driver}' does not look like a valid driver. Did you
+forget to specify the driver to bind the devices to?""")
except ValueError:
- # driver generated error - it's not a valid device ID, so all is well
+ # driver generated error - it's not a valid device
pass
- # check if we're attempting to bind to a driver that isn't loaded
- if not module_is_loaded(driver.replace('-', '_')):
- sys.exit("Error: Driver '%s' is not loaded." % driver)
-
+ # validate all devices
try:
- dev_list = map(dev_id_from_dev_name, dev_list)
- except ValueError as ex:
- sys.exit(ex)
+ ctx.devices = [devbind.resolve_device(dev) for dev in ctx.devices]
+ except ValueError as e:
+ raise DevbindError(str(e)) from e
+ devices = (devbind.pci_devices[dbdf] for dbdf in ctx.devices)
+
+ # do we want to unbind?
+ if not ctx.driver:
+ # unbind devices
+ for dev in devices:
+ devbind.unbind(dev, ctx.force)
+ return
+
+ # validate driver
+ if ctx.driver not in devbind.loaded_pci_modules:
+ raise DevbindError(f"Driver '{ctx.driver}' is not loaded.")
# check for IOMMU support
- if driver == "vfio-pci" and not has_iommu():
- check_noiommu_mode()
+ if use_vfio and not sysfs_iommu_enabled():
+ sysfs_enable_unsafe_noiommu()
- for d in dev_list:
- bind_one(d, driver, force)
+ # bind all devices
+ for dev in devices:
+ rollback_driver = dev.driver
+ # does this device have a driver already? if so, unbind
+ if rollback_driver:
+ devbind.unbind(dev, ctx.force)
- # For kernels < 3.15 when binding devices to a generic driver
- # (i.e. one that doesn't have a PCI ID table) using new_id, some devices
- # that are not bound to any other driver could be bound even if no one has
- # asked them to. hence, we check the list of drivers again, and see if
- # some of the previously-unbound devices were erroneously bound.
- if not exists("/sys/bus/pci/devices/%s/driver_override" % d):
- for d in devices.keys():
- # skip devices that were already bound or that we know should be bound
- if "Driver_str" in devices[d] or d in dev_list:
- continue
-
- # update information about this device
- devices[d] = dict(devices[d].items()
- + get_pci_device_details(d, True).items())
-
- # check if updated information indicates that the device was bound
- if "Driver_str" in devices[d]:
- unbind_one(d, force)
-
-
-def display_devices(title, dev_list, extra_params=None):
- '''Displays to the user the details of a list of devices given in
- "dev_list". The "extra_params" parameter, if given, should contain a string
- with %()s fields in it for replacement by the named fields in each
- device's dictionary.'''
- strings = [] # this holds the strings to print. We sort before printing
- print("\n%s" % title)
- print("=" * len(title))
- if not dev_list:
- strings.append("<none>")
- else:
- for dev in dev_list:
- if extra_params is not None:
- strings.append("%s '%s %s' %s" % (dev["Slot"],
- dev["Device_str"],
- dev["Device"],
- extra_params % dev))
+ # device doesn't have any driver now, bind it
+ try:
+ devbind.bind(dev, ctx.driver, ctx.force)
+ # bind succeeded, rollback no longer necessary
+ rollback_driver = ""
+ except DevbindError as e:
+ # should we roll back?
+ if rollback_driver:
+ print(f"Warning: {e}")
else:
- strings.append("%s '%s'" % (dev["Slot"], dev["Device_str"]))
- # sort before printing, so that the entries appear in PCI order
+ # pass the error up the stack
+ raise
+ if rollback_driver:
+ devbind.bind(dev, rollback_driver, ctx.force)
+ return
+
+ # if we're binding to vfio-pci, set IOMMU user/group ownership if one was specified
+ if use_vfio and (ctx.vfio_uid != -1 or ctx.vfio_gid != -1):
+ device_vfio_set_ownership(dev.slot, ctx.vfio_uid, ctx.vfio_gid)
+
+
+def print_status_section(title: str, section_devs: T.List[Device]) -> None:
+ """Prints subsection of device status (e.g. only kernel devices)."""
+ # we will sort strings before printing
+ strings: T.List[str] = []
+
+ def _fmt_key_val(name: str, value: str) -> str:
+ """Generate a devbind device printout string for a particular value."""
+ # if there's a name provided, include it in the output
+ if name and value:
+ return f"{name}={value}"
+ # otherwise just print the value, including empty ones
+ return value
+
+ # generate device strings
+ for dev in section_devs:
+ # construct strings
+ devstr = f'{dev["Device_str"]} {dev["Device"]}'
+ strs = [
+ dev.slot,
+ f"'{devstr}'",
+ _fmt_key_val("drv", dev["Driver"]),
+ _fmt_key_val("unused", dev["Module"]),
+ _fmt_key_val("if", dev["Interface"]),
+ _fmt_key_val("numa_node", dev["NUMANode"]),
+ _fmt_key_val("", dev["Active"]),
+ ]
+ # filter out empty strings and join
+ strings.append(" ".join(filter(None, strs)))
strings.sort()
- print("\n".join(strings)) # print one per line
-
-
-def show_device_status(devices_type, device_name, if_field=False):
- global dpdk_drivers
- kernel_drv = []
- dpdk_drv = []
- no_drv = []
-
- print_numa = True # by default, assume we can print NUMA information
-
- # split our list of network devices into the three categories above
- for d in devices.keys():
- if device_type_match(devices[d], devices_type):
- print_numa &= "NUMANode" in devices[d]
- if not has_driver(d):
- no_drv.append(devices[d])
- continue
- if devices[d]["Driver_str"] in dpdk_drivers:
- dpdk_drv.append(devices[d])
- else:
- kernel_drv.append(devices[d])
-
- n_devs = len(dpdk_drv) + len(kernel_drv) + len(no_drv)
-
- # don't bother displaying anything if there are no devices
- if n_devs == 0:
- msg = "No '%s' devices detected" % device_name
- print("")
+ print(f"{title}")
+ print("=" * len(title))
+ print("\n".join(strings))
+ print()
+
+
+def print_status_group(
+ ctx: DevbindCtx, group_title: str, group_devs: T.List[Device]
+) -> None:
+ """Print status for a specific device group."""
+ # do we have any devices at all?
+ if not group_devs:
+ msg = f"No {group_title} devices found."
print(msg)
- print("".join('=' * len(msg)))
+ print("=" * len(msg))
+ print()
return
- # print each category separately, so we can clearly see what's used by DPDK
- if dpdk_drv:
- extra_param = "drv=%(Driver_str)s unused=%(Module_str)s"
- if print_numa:
- extra_param = "numa_node=%(NUMANode)s " + extra_param
- display_devices("%s devices using DPDK-compatible driver" % device_name,
- dpdk_drv, extra_param)
- if kernel_drv:
- extra_param = "drv=%(Driver_str)s unused=%(Module_str)s"
- if if_field:
- extra_param = "if=%(Interface)s " + extra_param
- if print_numa:
- extra_param = "numa_node=%(NUMANode)s " + extra_param
- extra_param += " %(Active)s"
- display_devices("%s devices using kernel driver" % device_name,
- kernel_drv, extra_param)
- if no_drv:
- extra_param = "unused=%(Module_str)s"
- if print_numa:
- extra_param = "numa_node=%(NUMANode)s " + extra_param
- display_devices("Other %s devices" % device_name, no_drv, extra_param)
-
-
-def show_status():
- '''Function called when the script is passed the "--status" option.
- Displays to the user what devices are bound to the igb_uio driver, the
- kernel driver or to no driver'''
-
- if status_dev in ["net", "all"]:
- show_device_status(network_devices, "Network", if_field=True)
-
- if status_dev in ["baseband", "all"]:
- show_device_status(baseband_devices, "Baseband")
-
- if status_dev in ["crypto", "all"]:
- show_device_status(crypto_devices, "Crypto")
-
- if status_dev in ["dma", "all"]:
- show_device_status(dma_devices, "DMA")
-
- if status_dev in ["event", "all"]:
- show_device_status(eventdev_devices, "Eventdev")
-
- if status_dev in ["mempool", "all"]:
- show_device_status(mempool_devices, "Mempool")
-
- if status_dev in ["compress", "all"]:
- show_device_status(compress_devices, "Compress")
-
- if status_dev in ["misc", "all"]:
- show_device_status(misc_devices, "Misc (rawdev)")
-
- if status_dev in ["regex", "all"]:
- show_device_status(regex_devices, "Regex")
-
- if status_dev in ["ml", "all"]:
- show_device_status(ml_devices, "ML")
-
-
-def pci_glob(arg):
- '''Returns a list containing either:
- * List of PCI B:D:F matching arg, using shell wildcards e.g. 80:04.*
- * Only the passed arg if matching list is empty'''
- sysfs_path = "/sys/bus/pci/devices"
- for _glob in [arg, '0000:' + arg]:
- paths = [basename(path) for path in glob(path_join(sysfs_path, _glob))]
- if paths:
- return paths
- return [arg]
-
-
-def parse_args():
- '''Parses the command-line arguments given by the user and takes the
- appropriate action for each'''
- global b_flag
- global status_flag
- global status_dev
- global force_flag
- global noiommu_flag
- global args
+ # split out all devices into three groups: kernel, non-kernel, and unbound
+ kernel: T.List[Device] = []
+ dpdk: T.List[Device] = []
+ unbound: T.List[Device] = []
+
+ for dev in group_devs:
+ driver = dev.driver
+ if driver in ctx.devbind.uio_modules:
+ dpdk.append(dev)
+ elif driver:
+ kernel.append(dev)
+ else:
+ unbound.append(dev)
+
+ # print out each group
+ if dpdk:
+ print_status_section(
+ f"{group_title} devices using DPDK-compatible driver", dpdk
+ )
+ if kernel:
+ print_status_section(f"{group_title} devices using kernel driver", kernel)
+ if unbound:
+ print_status_section(f"Other {group_title} devices", unbound)
+
+
+def print_status(ctx: DevbindCtx) -> None:
+ """Print status of all devices."""
+ # device_type to devbind type mapping
+ for group_name, dt_t in DEVICE_TYPES.items():
+ if ctx.status_group in [DEVICE_TYPE_ALL, group_name]:
+ group_title, _ = dt_t
+ group_devs = ctx.devbind.devices_by_type[group_name]
+ print_status_group(ctx, group_title, group_devs)
+
+
+def parse_args() -> DevbindCtx:
+ """Parse command-line arguments into devbind context."""
parser = argparse.ArgumentParser(
- description='Utility to bind and unbind devices from Linux kernel',
+ description="Utility to bind and unbind devices from Linux kernel",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
@@ -706,7 +852,7 @@ def parse_args():
%(prog)s --status
To display current network device status:
- %(prog)s --status-dev net
+ %(prog)s --status net
To bind eth1 from the current driver and move to use vfio-pci
%(prog)s --bind=vfio-pci eth1
@@ -716,137 +862,134 @@ def parse_args():
To bind 0000:02:00.0 and 0000:02:00.1 to the ixgbe kernel driver
%(prog)s -b ixgbe 02:00.0 02:00.1
-""")
+""",
+ )
+ status_choices = [DEVICE_TYPE_ALL] + list(DEVICE_TYPES.keys())
+
+ parser.add_argument(
+ "-s",
+ "--status",
+ # backwards compatibility
+ "--status-dev",
+ # None if flag was not specified
+ default=None,
+ # "all" if flag was specified without arguments
+ const=DEVICE_TYPE_ALL,
+ # otherwise, match against the choice table
+ nargs="?",
+ choices=status_choices,
+ help="Print the status of device group (default: all devices).",
+ )
+ bind_action = parser.add_mutually_exclusive_group()
+ bind_action.add_argument(
+ "-b",
+ "--bind",
+ metavar="DRIVER",
+ help='Select the driver to use ("none" to unbind the device)',
+ )
+ bind_action.add_argument(
+ "-u",
+ "--unbind",
+ action="store_true",
+ help='Unbind a device (equivalent to "-b none")',
+ )
parser.add_argument(
- '-s',
- '--status',
- action='store_true',
- help="Print the current status of all known devices.")
+ "--noiommu-mode",
+ action="store_true",
+ help="If IOMMU is not available, enable no IOMMU mode for VFIO drivers",
+ )
parser.add_argument(
- '--status-dev',
- help="Print the status of given device group.",
- choices=['baseband', 'compress', 'crypto', 'dma', 'event',
- 'mempool', 'misc', 'net', 'regex', 'ml'])
- bind_group = parser.add_mutually_exclusive_group()
- bind_group.add_argument(
- '-b',
- '--bind',
- metavar='DRIVER',
- help="Select the driver to use or \"none\" to unbind the device")
- bind_group.add_argument(
- '-u',
- '--unbind',
- action='store_true',
- help="Unbind a device (equivalent to \"-b none\")")
+ "--force",
+ action="store_true",
+ help="""\
+Override restriction on binding devices in use by Linux. WARNING: This can lead
+to loss of network connection and should be used with caution.
+""",
+ )
parser.add_argument(
- '--noiommu-mode',
- action='store_true',
- help="If IOMMU is not available, enable no IOMMU mode for VFIO drivers")
+ "-G",
+ "--gid",
+ type=lambda g: grp.getgrnam(g).gr_gid,
+ default=-1,
+ help="For VFIO, specify the group ID to set IOMMU group ownership",
+ )
parser.add_argument(
- '--force',
- action='store_true',
- help="""
-Override restriction on binding devices in use by Linux"
-WARNING: This can lead to loss of network connection and should be used with caution.
-""")
+ "-U",
+ "--uid",
+ type=lambda u: pwd.getpwnam(u).pw_uid,
+ default=-1,
+ help="For VFIO, specify the user ID to set IOMMU group ownership",
+ )
parser.add_argument(
- 'devices',
- metavar='DEVICE',
- nargs='*',
- help="""
+ "devices",
+ metavar="DEVICE",
+ nargs="*",
+ help="""\
Device specified as PCI "domain:bus:slot.func" syntax or "bus:slot.func" syntax.
For devices bound to Linux kernel drivers, they may be referred to by interface name.
-""")
+""",
+ )
opt = parser.parse_args()
- if opt.status_dev:
- status_flag = True
- status_dev = opt.status_dev
+ ctx = DevbindCtx()
+
if opt.status:
- status_flag = True
- status_dev = "all"
- if opt.force:
- force_flag = True
- if opt.noiommu_mode:
- noiommu_flag = True
- if opt.bind:
- b_flag = opt.bind
- elif opt.unbind:
- b_flag = "none"
- args = opt.devices
-
- if not b_flag and not status_flag:
- print("Error: No action specified for devices. "
- "Please give a --bind, --ubind or --status option",
- file=sys.stderr)
+ ctx.status = True
+ ctx.status_group = opt.status
+ if opt.bind or opt.unbind:
+ ctx.bind = True
+ ctx.driver = "" if opt.unbind else opt.bind
+ # support any capitalization for binding to "none"
+ if ctx.driver.lower() == "none":
+ ctx.driver = ""
+ if not ctx.status and not ctx.bind:
+ print("Error: No action specified.", file=sys.stderr)
parser.print_usage()
sys.exit(1)
- if b_flag and not args:
- print("Error: No devices specified.", file=sys.stderr)
+ ctx.noiommu = opt.noiommu_mode
+ ctx.force = opt.force
+ ctx.devices = opt.devices
+ ctx.vfio_uid = opt.uid
+ ctx.vfio_gid = opt.gid
+
+ # if status is displayed, devices shouldn't be passed
+ if not ctx.bind and ctx.devices:
+ print("Error: Devices should not be specified with --status action.")
parser.print_usage()
sys.exit(1)
+ # if bind is used, devices should be passed
+ elif ctx.bind and not ctx.devices:
+ print("Error: No devices specified for --bind/--unbind action.")
+ parser.print_usage()
+ sys.exit(1)
+ return ctx
- # resolve any PCI globs in the args
- new_args = []
- for arg in args:
- new_args.extend(pci_glob(arg))
- args = new_args
-
-
-def do_arg_actions():
- '''do the actual action requested by the user'''
- global b_flag
- global status_flag
- global force_flag
- global args
- if b_flag in ["none", "None"]:
- unbind_all(args, force_flag)
- elif b_flag is not None:
- bind_all(args, b_flag, force_flag)
- if status_flag:
- if b_flag is not None:
- clear_data()
- # refresh if we have changed anything
- get_device_details(network_devices)
- get_device_details(baseband_devices)
- get_device_details(crypto_devices)
- get_device_details(dma_devices)
- get_device_details(eventdev_devices)
- get_device_details(mempool_devices)
- get_device_details(compress_devices)
- get_device_details(regex_devices)
- get_device_details(ml_devices)
- get_device_details(misc_devices)
- show_status()
+def _main():
+ ctx = parse_args()
+ # initialize devbind data
+ ctx.devbind = Devbind()
-def main():
- '''program main function'''
- # check if lspci is installed, suppress any output
- with open(os.devnull, 'w') as devnull:
- ret = subprocess.call(['which', 'lspci'],
- stdout=devnull, stderr=devnull)
- if ret != 0:
- sys.exit("'lspci' not found - please install 'pciutils'")
- parse_args()
- check_modules()
- clear_data()
- get_device_details(network_devices)
- get_device_details(baseband_devices)
- get_device_details(crypto_devices)
- get_device_details(dma_devices)
- get_device_details(eventdev_devices)
- get_device_details(mempool_devices)
- get_device_details(compress_devices)
- get_device_details(regex_devices)
- get_device_details(ml_devices)
- get_device_details(misc_devices)
- do_arg_actions()
+ if ctx.bind:
+ # resolve any PCI globs in devices
+ ctx.devices = [d for dev in ctx.devices for d in resolve_pci_glob(dev)]
+ bind_devices(ctx)
+ print()
+ if ctx.status:
+ print_status(ctx)
if __name__ == "__main__":
- main()
+ try:
+ # check if lspci and ip are installed before doing anything
+ check_installed("lspci", "pciutils")
+ check_installed("ip", "iproute2")
+
+ # run the main function
+ _main()
+ except DevbindError as e:
+ sys.exit(f"Error: {e}")
--
2.43.5
prev parent reply other threads:[~2024-12-04 9:47 UTC|newest]
Thread overview: 20+ messages / expand[flat|nested] mbox.gz Atom feed top
2024-12-02 15:09 [PATCH v1 0/1] Rewrite devbind Anatoly Burakov
2024-12-02 15:09 ` [PATCH v1 1/1] usertools/devbind: update coding style Anatoly Burakov
2024-12-02 17:01 ` Stephen Hemminger
2024-12-03 8:55 ` Burakov, Anatoly
2024-12-02 16:14 ` [PATCH v1 0/1] Rewrite devbind Bruce Richardson
2024-12-03 8:51 ` Burakov, Anatoly
2024-12-02 16:54 ` Stephen Hemminger
2024-12-03 11:24 ` [PATCH v2 0/2] " Anatoly Burakov
2024-12-03 11:25 ` [PATCH v2 1/2] usertools/devbind: update coding style Anatoly Burakov
2024-12-03 17:07 ` Stephen Hemminger
2024-12-04 8:59 ` Burakov, Anatoly
2024-12-03 22:16 ` Stephen Hemminger
2024-12-04 9:02 ` Burakov, Anatoly
2024-12-03 11:25 ` [PATCH v2 2/2] usertools/devbind: replace devbind Anatoly Burakov
2024-12-04 9:45 ` [PATCH v3 0/1] Rewrite devbind Anatoly Burakov
2024-12-04 9:45 ` [PATCH v3 1/1] usertools/devbind: replace devbind Anatoly Burakov
2024-12-04 9:48 ` [PATCH v3 0/1] Rewrite devbind Burakov, Anatoly
2024-12-04 9:47 ` [PATCH v3 0/2] " Anatoly Burakov
2024-12-04 9:47 ` [PATCH v3 1/2] usertools/devbind: update coding style Anatoly Burakov
2024-12-04 9:47 ` Anatoly Burakov [this message]
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
Avoid top-posting and favor interleaved quoting:
https://en.wikipedia.org/wiki/Posting_style#Interleaved_style
* Reply using the --to, --cc, and --in-reply-to
switches of git-send-email(1):
git send-email \
--in-reply-to=b8aa13c9ed8989e73e9352ea905f6956a863148d.1733305655.git.anatoly.burakov@intel.com \
--to=anatoly.burakov@intel.com \
--cc=dev@dpdk.org \
/path/to/YOUR_REPLY
https://kernel.org/pub/software/scm/git/docs/git-send-email.html
* If your mail client supports setting the In-Reply-To header
via mailto: links, try the mailto: link
Be sure your reply has a Subject: header at the top and a blank line
before the message body.
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for NNTP newsgroup(s).