Skip to contentSkip to conversation

Rewrite devbind

PATCH Superseded Anatoly Burakov 3 participants 22 messages Dec 2, 2024 – Jan 14

lore
Share & tools
lore
#0 lore raw

cover Rewrite devbind

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort.

The script has become slightly bigger - 1000 lines instead of 800, however I would argue that since most of that increase is infrastructure, comments, and sacrificing code golf for code readability (such as expanding one-liners into multiple lines), the trade-off between being able to read and reason about what happens in the script is worth the added line count.

[1] https://patches.dpdk.org/project/dpdk/patch/c2bf00195c2d43833a831a9cc9346b4606d6ea2e.1723810613.git.anatoly.burakov@intel.com/

Anatoly Burakov (1):
usertools/devbind: update coding style

 usertools/dpdk-devbind.py | 1736 +++++++++++++++++++++----------------
 1 file changed, 968 insertions(+), 768 deletions(-)
Signature · 1 line
2.43.5
#1 lore raw

PATCH 1/1 usertools/devbind: update coding style

Devbind is one of the oldest tools in DPDK, and is written in a way that uses a lot of string matching, no type safety, lots of global variables, and has a few inconsistencies in the way it handles data (such as differences between lspci calls and parsing in different circumstances).

This patch is a nigh complete rewrite of devbind, with full 100% feature and command-line compatibility with the old version, albeit with a few differences in formatting and error messages. All file handling code has also been replaced with context managers.

What's different from old code:
- Full PEP-484 compliance
- Formatted with Ruff
- Much better structured code
- Clean and consistent control flow
- More comments
- Better error handling
- Fewer lspci calls
- Unified lspci parsing
- Using /sys/bus/pci/drivers as a source of truth about kernel modules
- Check for iproute2 package
- Deprecate --status-dev in favor of optional --status argument

---
 usertools/dpdk-devbind.py | 1736 +++++++++++++++++++++----------------
 1 file changed, 968 insertions(+), 768 deletions(-)
usertools/dpdk-devbind.py +968 -768
index f2a2a9a12f..fe4b60a541 100755
@@ -1,705 +1,898 @@
 #!/usr/bin/env python3
 # SPDX-License-Identifier: BSD-3-Clause
-# Copyright(c) 2010-2014 Intel Corporation
-#
+# Copyright(c) 2010-2024 Intel Corporation
 
-import sys
+import argparse
+import glob
+import grp
 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 this system has NUMA support
-def is_numa():
-    return os.path.exists('/sys/devices/system/node')
-
-
-# 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):
-            continue
-
-        # 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
-
-        # 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 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 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"]
+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,
+]
+
+
+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 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
+    return value
+
+
+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_lines(args: T.List[str]) -> T.List[str]:
+    """Run a subprocess, collect its output, and return it as a list of lines."""
     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()
+        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.splitlines()
 
 
-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
+def read_routed_interfaces() -> T.List[str]:
+    """Find interfaces with active routes."""
+    route_info = [
+        token.strip()
+        # read 'ip -o route' output
+        for line in read_output_lines(["ip", "-o", "route"])
+        # skip routes that are not of interest to us
+        if not line.startswith("169.254")
+        # split into individual tokens
+        for token in line.split()
+    ]
+    routed_ifs: T.List[str] = []
+    # find devices with active routes
+    for node, iface in zip(route_info, route_info[1:]):
+        if node == "dev":
+            routed_ifs.append(iface)
+    return routed_ifs
 
-    # 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
+def sysfs_read_pci_drivers() -> T.List[str]:
+    """Gather all PCI modules loaded on the system."""
+    return os.listdir("/sys/bus/pci/drivers")
 
-    # 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 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 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 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 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 _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."""
+        lspci_output = read_output_lines(["lspci", "-vmmnnks", self.slot])
+        for line in lspci_output:
+            if not line:
+                continue
+            self._dev_dict.update(parse_lspci_line(line))
+        self._update_interfaces()
+
+    def get_str(self, key: str) -> str:
+        """Get string value as it appears in the device dictionary."""
+        return self._get_str(key)
+
+    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_lines(["lspci", "-Dvmmnnk"])
+    cur_device: T.Dict[str, str] = {}
+    for line in lspci_output:
+        if not line:
+            dev = Device(cur_device)
+            yield dev
+            cur_device = {}
+        else:
+            cur_device.update(parse_lspci_line(line))
+
+
+class Devbind(object):
+    """Class to cover various devbind-related operations and data."""
+
+    def __init__(self) -> None:
+        self.uio_modules: T.List[str]
+        # all detected PCI devices, keyed by PCI D:B:D.F
+        self.pci_devices: T.Dict[str, Device]
+        # matched categories of devices
+        self.network_devices: T.List[Device] = []
+        self.baseband_devices: T.List[Device] = []
+        self.crypto_devices: T.List[Device] = []
+        self.dma_devices: T.List[Device] = []
+        self.eventdev_devices: T.List[Device] = []
+        self.mempool_devices: T.List[Device] = []
+        self.compress_devices: T.List[Device] = []
+        self.regex_devices: T.List[Device] = []
+        self.ml_devices: T.List[Device] = []
+        self.misc_devices: T.List[Device] = []
+        # whether to use driver_override
+        self.use_driver_override: bool
+        # gather all loaded kernel modules
+        self.loaded_pci_modules = sysfs_read_pci_drivers()
+        # find which ones are UIO modules
+        self._scan_uio_modules()
+        # gather all routed interfaces
+        self._routed_interfaces = read_routed_interfaces()
+        # initialize data
+        self._scan_pci_devices()
+
+    def _categorize_pci_device(self, dev: Device) -> None:
+        pattern_to_list = [
+            (NETWORK_DEVICES, self.network_devices),
+            (BASEDBAND_DEVICES, self.baseband_devices),
+            (CRYPTO_DEVICES, self.crypto_devices),
+            (DMA_DEVICES, self.dma_devices),
+            (EVENTDEV_DEVICES, self.eventdev_devices),
+            (MEMPOOL_DEVICES, self.mempool_devices),
+            (COMPRESS_DEVICES, self.compress_devices),
+            (REGEX_DEVICES, self.regex_devices),
+            (ML_DEVICES, self.ml_devices),
+            (MISC_DEVICES, self.misc_devices),
+        ]
+        for patterns, devices in pattern_to_list:
+            for pattern in patterns:
+                if dev.match(pattern):
+                    devices.append(dev)
+                    break
+        # special case: if this was a network device, find if any interfaces are active. non-network
+        # interfaces will not have any interfaces at all, so it's safe to check all devices
+        if any(iface in self._routed_interfaces for iface in dev.interfaces):
+            dev.active_interface = True
+
+    def _scan_uio_modules(self) -> None:
+        loaded = set(self.loaded_pci_modules)
+        supported = set(DPDK_KERNEL_MODULES)
+        self.uio_modules = 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 _scan_pci_devices(self) -> None:
+        """Find all devices on the system."""
+        self.pci_devices = {}
+
+        for dev in read_devices_from_lspci():
+            # categorize device
+            self._categorize_pci_device(dev)
+
+            # fixup module and driver fields
+            self.add_uio_modules(dev)
+
+            # save the device in common list
+            self.pci_devices[dev.slot] = dev
+
+            # do we use driver_override?
+            self.use_driver_override = os.path.exists(
+                sysfs_device_get_path(dev.slot, "driver_override")
+            )
+
+    def resolve_device(self, devstr: str) -> str:
+        """Try to resolve a device into a PCI D:B:D:F."""
+        if not devstr:
+            raise ValueError(
+                f"Unknown device '{devstr}'. Please specify device in 'bus:slot.func' format."
+            )
+        # 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.network_devices:
+            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 unbind(self, dbdf: str, force: bool) -> None:
+        """Unbind one device from its current driver."""
+        dev = self.pci_devices[dbdf]
+
+        # are we allowed to unbind 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
+
+        # is this device bound to anything?
+        cur_drv = dev.driver
+        if not cur_drv:
+            print(
+                f"Notice: {dev} is not currently managed by any driver, skipping",
+                file=sys.stderr,
+            )
+            return
+
+        # we're OK, unbind
+        print(f"Unbinding {dev} from {cur_drv}...")
+
+        try:
+            unbind_path = sysfs_driver_get_path(cur_drv, "unbind")
+            with open(unbind_path, "w", encoding="utf-8") as f:
+                f.write(dbdf)
+        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, dbdf: str, driver: str, force: bool) -> None:
+        """Bind one device to the specified driver."""
+        dev = self.pci_devices[dbdf]
+        rollback_driver = ""
+        override_path = sysfs_device_get_path(dev.slot, "driver_override")
+        new_id_path = sysfs_driver_get_path(driver, "new_id")
+        bind_path = sysfs_driver_get_path(driver, "bind")
+
+        # are we allowed to bind 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
+
+        # check if device is bound to anything
+        rollback_driver = dev.driver
+        if driver == rollback_driver:
+            print(
+                f"Notice: {dev} is already bound to driver {driver}, skipping",
+                file=sys.stderr,
+            )
+            return
+        self.unbind(dbdf, force)
+
+        print(f"Binding {dev} to {driver}...")
+
+        # 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.
+
+        # are we binding to UIO module?
+        if driver in self.uio_modules:
+            try:
+                if self.use_driver_override:
+                    with open(override_path, "w", encoding="utf-8") as f:
+                        f.write(driver)
+                else:
+                    # For kernels < 3.15 use new_id to add PCI id's to the driver
+                    with open(new_id_path, "w", encoding="utf-8") as f:
+                        ven_id = int(dev.get_str("Vendor"), 16)
+                        dev_id = int(dev.get_str("Device"), 16)
+                        f.write(f"{ven_id:04x} {dev_id:04x}")
+            except OSError as e:
+                raise DevbindError(f"Bind failed for {dev}: {e}") from e
+
+        try:
+            with open(bind_path, "a", encoding="utf-8") as f:
+                f.write(dbdf)
+            dev.update()
+            self.add_uio_modules(dev)
+        except OSError as e:
+            # for some reason, closing fd after adding a new PCI ID to new_id results in IOError on
+            # some kernel versions. however, if the device was successfully bound, we don't care
+            # for any errors and can safely proceed.
+
+            # update device from lspci and fixup modules
+            dev.update()
+            self.add_uio_modules(dev)
+
+            # check if the correct driver is bound
+            cur_drv = dev.driver
+
+            if cur_drv == driver:
+                # we're OK, ignore error and don't rollback
+                rollback_driver = ""
+            elif cur_drv:
+                # bound to wrong driver
+                print(
+                    f"Error: bind failed for {dev}: bound to unexpected driver {cur_drv}",
+                    file=sys.stderr,
+                )
+            else:
+                # bind failed
+                print(f"Error: bind failed for {dev}: {e}", file=sys.stderr)
+            if rollback_driver:
+                self.bind(dbdf, rollback_driver, force)
+
+        # 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
+        if self.use_driver_override:
+            try:
+                with open(override_path, "w", encoding="utf-8") as f:
+                    f.write("\00")
+            except OSError as e:
+                raise DevbindError(f"Bind failed for {dev}: {e}") from e
+
+
+class DevbindCtx:
+    """POD class to keep commmand-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
+
+    # do we want to bind or unbind?
+    if not ctx.driver:
+        # unbind devices
+        for dbdf in ctx.devices:
+            devbind.unbind(dbdf, 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 dbdf in ctx.devices:
+        devbind.bind(dbdf, ctx.driver, ctx.force)
+        # 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(dbdf, ctx.vfio_uid, ctx.vfio_gid)
 
-    # 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():
+    # 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 devbind.use_driver_override:
+        for dbdf, dev in devbind.pci_devices.items():
             # skip devices that were already bound or that we know should be bound
-            if "Driver_str" in devices[d] or d in dev_list:
+            if dev.driver or dbdf in ctx.devices:
                 continue
-
-            # update information about this device
-            devices[d] = dict(devices[d].items()
-                              + get_pci_device_details(d, True).items())
-
+            dev.update()
+            devbind.add_uio_modules(dev)
             # check if updated information indicates that the device was bound
-            if "Driver_str" in devices[d]:
-                unbind_one(d, force)
+            if dev.driver:
+                devbind.unbind(dbdf, ctx.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))
-            else:
-                strings.append("%s '%s'" % (dev["Slot"], dev["Device_str"]))
-    # sort before printing, so that the entries appear in PCI order
+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] = []
+
+    # generate device strings
+    for dev in section_devs:
+        # construct strings
+        devstr = f'{dev.get_str("Device_str")} {dev.get_str("Device")}'
+        strs = [
+            dev.slot,
+            f"'{devstr}'",
+            fmt_key_val("drv", dev.get_str("Driver")),
+            fmt_key_val("unused", dev.get_str("Module")),
+            fmt_key_val("if", dev.get_str("Interface")),
+            fmt_key_val("numa_node", dev.get_str("NUMANode")),
+            fmt_key_val("", dev.get_str("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 = []
-
-    # split our list of network devices into the three categories above
-    for d in devices.keys():
-        if device_type_match(devices[d], devices_type):
-            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_name: 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_name} devices found."
         print(msg)
-        print("".join('=' * len(msg)))
+        print("=" * len(msg))
+        print()
         return
 
-    print_numa = is_numa()
-
-    # 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
-        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
+    devbind = ctx.devbind
+
+    # 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 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_name} devices using DPDK-compatible driver", dpdk)
+    if kernel:
+        print_status_section(f"{group_name} devices using kernel driver", kernel)
+    if unbound:
+        print_status_section(f"Other {group_name} devices", unbound)
+
+
+def print_status(ctx: DevbindCtx) -> None:
+    """Print status of all devices."""
+    status_groups = [
+        ("Network", "net", ctx.devbind.network_devices),
+        ("Baseband", "baseband", ctx.devbind.baseband_devices),
+        ("Crypto", "crypto", ctx.devbind.crypto_devices),
+        ("DMA", "dma", ctx.devbind.dma_devices),
+        ("Eventdev", "event", ctx.devbind.eventdev_devices),
+        ("Mempool", "mempool", ctx.devbind.mempool_devices),
+        ("Compress", "compress", ctx.devbind.compress_devices),
+        ("Misc (rawdev)", "misc", ctx.devbind.misc_devices),
+        ("Regex", "regex", ctx.devbind.regex_devices),
+        ("ML", "ml", ctx.devbind.ml_devices),
+    ]
+    for group_name, group_type, group_devs in status_groups:
+        if ctx.status_group in ["all", group_type]:
+            print_status_group(ctx, group_name, 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:
@@ -709,7 +902,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
@@ -719,137 +912,144 @@ 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
-""")
+""",
+    )
 
     parser.add_argument(
-        '-s',
-        '--status',
-        action='store_true',
-        help="Print the current status of all known devices.")
+        "-s",
+        "--status",
+        # backwards compatibility
+        "--status-dev",
+        # None if flag was not specified
+        default=None,
+        # "all" if flag was specified without arguments
+        const="all",
+        # otherwise, match against the choice table
+        nargs="?",
+        choices=[
+            "baseband",
+            "compress",
+            "crypto",
+            "dma",
+            "event",
+            "mempool",
+            "misc",
+            "net",
+            "regex",
+            "ml",
+            "all",
+        ],
+        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(
-        '--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\")")
+        "--noiommu-mode",
+        action="store_true",
+        help="If IOMMU is not available, enable no IOMMU mode for VFIO drivers",
+    )
     parser.add_argument(
-        '--noiommu-mode',
-        action='store_true',
-        help="If IOMMU is not available, enable no IOMMU mode for VFIO drivers")
+        "--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(
-        '--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.
-""")
+        "-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(
-        'devices',
-        metavar='DEVICE',
-        nargs='*',
-        help="""
+        "-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()
 
-    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}")
Signature · 1 line
2.43.5
#2 lore raw

Re: [PATCH v1 1/1] usertools/devbind: update coding style

usertools/dpdk-devbind.py line 796

Quoted 5 lines · On Mon, 2 Dec 2024 15:09:34 +0000 Anatoly Burakov <[email protected]>…
usertools/dpdk-devbind.py +5 -0
+    # 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 devbind.use_driver_override:

Why is tool still supporting out of date and no longer supported kernel?

usertools/dpdk-devbind.py def parse_args(): · line 941

Quoted 13 lines
usertools/dpdk-devbind.py +13 -0
+        choices=[
+            "baseband",
+            "compress",
+            "crypto",
+            "dma",
+            "event",
+            "mempool",
+            "misc",
+            "net",
+            "regex",
+            "ml",
+            "all",
+        ],

Would prefer that all the types are in table/list and the help just references that list. The next time a type is added, only one place needs to change.

Also, I would not trust the output format of ip route not to change. If the utility has to parse output of ip command, use json (-j) instead.

This whole section of code is quite fragile:

Quoted 12 lines
    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])
#3 lore raw

Re: [PATCH v1 1/1] usertools/devbind: update coding style

Quoted 10 lines · Stephen Hemminger

usertools/dpdk-devbind.py line 796

Quoted 5 lines · On Mon, 2 Dec 2024 15:09:34 +0000 Anatoly Burakov <[email protected]>…
usertools/dpdk-devbind.py +5 -0
+    # 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 devbind.use_driver_override:

Why is tool still supporting out of date and no longer supported kernel?

The aim was 100% compatibility with the old script, but I agree these parts can be taken out as this kernel is no longer supported. This will definitely make the binding code simpler.

Quoted 17 lines
Quoted 6 lines
usertools/dpdk-devbind.py +3 -0
+        choices=[
+            "baseband",
+            "compress",

usertools/dpdk-devbind.py def parse_args(): · line 941

Quoted 10 lines
usertools/dpdk-devbind.py +10 -0
+            "crypto",
+            "dma",
+            "event",
+            "mempool",
+            "misc",
+            "net",
+            "regex",
+            "ml",
+            "all",
+        ],
Quoted 3 lines

Would prefer that all the types are in table/list and the help just references that list. The next time a type is added, only one place needs to change.

It's a bit difficult to have *everything* as one list, as there are multiple places where we use this:

1) initial declarations at the top of the file (which I treat as "ground truth" for what sort of devices devbind aims to recognize)
2) categorization rules (which are inside Devbind class)
3) command line arguments
4) printouts

I suppose I can merge 3 and 4, but I don't see a neat way to specify 1) and 2) in a way that we can reuse elsewhere. I'll think on this though, thanks for the suggestion.

Quoted 18 lines

Also, I would not trust the output format of ip route not to change. If the utility has to parse output of ip command, use json (-j) instead.

This whole section of code is quite fragile:

Quoted 12 lines
     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])

The quoted code is from old devbind code, but I agree that relying on -o output is not ideal, and using -j will be better. I'll fix it in v2.

Thanks for your feedback!

Signature · 2 lines
Thanks,
Anatoly
#4 lore raw
In reply to Anatoly Burakov
Quoted 11 lines · Anatoly Burakov

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort.

One suggestion here which might help reviewing. Since it is essentially a rewrite, is it worth making this a two-patch set, where:

Patch 1: introduces a new script called e.g. dpdk-devbind-new.py, which
contains just the rewrite without any of old code. This then can be reviewed in isolation
Patch 2: moves dpdk-devbind-new.py to overwrite dpdk-devbind.py

WDYT?

Regards,
/Bruce

#5 lore raw
Quoted 20 lines · Bruce Richardson
Quoted 11 lines · Anatoly Burakov

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort.

One suggestion here which might help reviewing. Since it is essentially a rewrite, is it worth making this a two-patch set, where:

Patch 1: introduces a new script called e.g. dpdk-devbind-new.py, which
contains just the rewrite without any of old code. This then can be reviewed in isolation
Patch 2: moves dpdk-devbind-new.py to overwrite dpdk-devbind.py

Quoted 4 lines

WDYT?

Regards,
/Bruce

I think it's a good idea, provided it gets squashed on apply.

Signature · 2 lines
Thanks,
Anatoly
#6 lore raw
In reply to Anatoly Burakov
Quoted 17 lines · On Mon, 2 Dec 2024 15:09:33 +0000 Anatoly Burakov <[email protected]>…

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort.

The script has become slightly bigger - 1000 lines instead of 800, however I would argue that since most of that increase is infrastructure, comments, and sacrificing code golf for code readability (such as expanding one-liners into multiple lines), the trade-off between being able to read and reason about what happens in the script is worth the added line count.

Quoted 8 lines

[1] https://patches.dpdk.org/project/dpdk/patch/c2bf00195c2d43833a831a9cc9346b4606d6ea2e.1723810613.git.anatoly.burakov@intel.com/

Anatoly Burakov (1):
usertools/devbind: update coding style

 usertools/dpdk-devbind.py | 1736 +++++++++++++++++++++----------------
 1 file changed, 968 insertions(+), 768 deletions(-)

Looks good, but needs to be rebased to current 24.11

#7 lore raw

v2 cover Rewrite devbind

In reply to Anatoly Burakov

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort. Instead, as has been suggested [2], the patchset now consists of creating a new file, followed by a removal of old file and rename of the new file. It is expected that this will be squashed on apply.

The script has become slightly bigger - 1000 lines instead of 800, however I would argue that since most of that increase is infrastructure, comments, and sacrificing code golf for code readability (such as expanding one-liners into multiple lines), the trade-off between being able to read and reason about what happens in the script is worth the added line count.

[1] https://patches.dpdk.org/project/dpdk/patch/c2bf00195c2d43833a831a9cc9346b4606d6ea2e.1723810613.git.anatoly.burakov@intel.com/
[2] https://patches.dpdk.org/project/dpdk/cover/[email protected]/

Anatoly Burakov (2):
usertools/devbind: update coding style
usertools/devbind: replace devbind

 usertools/dpdk-devbind.py | 1678 ++++++++++++++++++++-----------------
 1 file changed, 911 insertions(+), 767 deletions(-)
Signature · 1 line
2.43.5
#8 lore raw

PATCH v2 1/2 usertools/devbind: update coding style

Devbind is one of the oldest tools in DPDK, and is written in a way that uses a lot of string matching, no type safety, lots of global variables, and has a few inconsistencies in the way it handles data (such as differences between lspci calls and parsing in different circumstances).

This patch is a nigh complete rewrite of devbind, with full 100% feature and command-line compatibility with the old version (except for dropping older kernel support), albeit with a few differences in formatting and error messages. All file handling code has also been replaced with context managers.

What's different from old code:
- Full PEP-484 compliance
- Formatted with Ruff
- Much better structured code
- Clean and consistent control flow
- More comments
- Better error handling
- Fewer lspci calls
- Unified lspci parsing
- Using /sys/bus/pci/drivers as a source of truth about kernel modules
- Check for iproute2 package
- Use JSON parsing for iproute2 output
- Deprecate --status-dev in favor of optional --status argument
- Deprecate kernel <3.15 support and only use driver_override

---

Notes:
v1 -> v2:
- Use dictionary syntax to get raw string values from devices
- Fixed rollback not working correctly due to stale device state
- Fixed attempts to bind to empty driver on rollback
- Simplified bind/rollback and removed recursion
- Unified command-line and device type handling
- Dropped support for kernels <3.15
- Use JSON parsing for ip route output
- Used a new filename to aid in review, rename in next patch

v1 -> v2:
- Fixed rollback not working correctly
- Fixed attempts to bind to empty driver
- Unified command-line and device type handling
- Dropped support for kernels <3.15
- Use JSON parsing for ip route output
- Used a new filename to aid in review, rename in next patch

v1 -> v2:
- Use dictionary syntax to get raw string values from devices
- Fixed rollback not working correctly
- Fixed attempts to bind to empty driver on rollback
- Unified command-line and device type handling
- Dropped support for kernels <3.15
- Use JSON parsing for ip route output
- Used a new filename to aid in review, rename in next patch

v1 -> v2:
- Fixed rollback not working correctly
- Fixed attempts to bind to empty driver
- Unified command-line and device type handling
- Dropped support for kernels <3.15
- Use JSON parsing for ip route output
- Used a new filename to aid in review, rename in next patch

 usertools/dpdk-devbind-new.py | 996 ++++++++++++++++++++++++++++++++++
 1 file changed, 996 insertions(+)
 create mode 100755 usertools/dpdk-devbind-new.py
usertools/dpdk-devbind-new.py new file +996 -0
new file mode 100755
index 0000000000..9f2ee27cf3
@@ -0,0 +1,996 @@
+#!/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:
+        if "dev" in route:
+            routed_ifs.append(route["dev"])
+    return 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 _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 __getitem__(self, key: str) -> str:
+        """Get value as it appears in the device dictionary."""
+        return self._get_str(key)
+
+    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:
+        self.uio_modules: T.List[str]
+        # 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
+        }
+        # gather all loaded kernel modules
+        self.loaded_pci_modules = sysfs_read_pci_drivers()
+        # find which ones are UIO modules
+        self._scan_uio_modules()
+        # gather all routed interfaces
+        self._routed_interfaces = read_routed_interfaces()
+        # initialize data
+        self._scan_pci_devices()
+
+    def _categorize_pci_device(self, dev: Device) -> None:
+        for devtype, dt_tup in DEVICE_TYPES.items():
+            _, patterns = dt_tup
+            for pattern in patterns:
+                if dev.match(pattern):
+                    self.devices_by_type[devtype].append(dev)
+                    break
+        # 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
+        if any(iface in self._routed_interfaces for iface in dev.interfaces):
+            dev.active_interface = True
+
+    def _scan_uio_modules(self) -> None:
+        loaded = set(self.loaded_pci_modules)
+        supported = set(DPDK_KERNEL_MODULES)
+        self.uio_modules = 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 _scan_pci_devices(self) -> None:
+        """Find all devices on the system."""
+        self.pci_devices = {}
+
+        for dev in read_devices_from_lspci():
+            # categorize device
+            self._categorize_pci_device(dev)
+
+            # fixup module and driver fields
+            self.add_uio_modules(dev)
+
+            # save the device in common list
+            self.pci_devices[dev.slot] = dev
+
+    def resolve_device(self, devstr: str) -> str:
+        """Try to resolve a device into a PCI D:B:D:F."""
+        if not devstr:
+            raise ValueError(
+                f"Unknown device '{devstr}'. Please specify device in 'bus:slot.func' format."
+            )
+        # 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_name: 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_name} devices found."
+        print(msg)
+        print("=" * len(msg))
+        print()
+        return
+
+    devbind = ctx.devbind
+
+    # 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 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_name} devices using DPDK-compatible driver", dpdk)
+    if kernel:
+        print_status_section(f"{group_name} devices using kernel driver", kernel)
+    if unbound:
+        print_status_section(f"Other {group_name} 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}")
Signature · 1 line
2.43.5
#9 lore raw

Re: [PATCH v2 1/2] usertools/devbind: update coding style

usertools/dpdk-devbind-new.py line 302

Quoted 8 lines · On Tue, 3 Dec 2024 11:25:00 +0000 Anatoly Burakov <[email protected]>…
usertools/dpdk-devbind-new.py new file +8 -0
+
+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}'.")
+

Apparently the posix way to do this is to use command -v not "which"

command [-pVv] command [arg ...]
Run command with args suppressing the normal shell function lookup. Only builtin commands or commands found in the PATH are
executed. If the -p option is given, the search for command is
performed using a default value for PATH that is guaranteed to find all of the standard utilities. If either the -V or -v op‐
tion is supplied, a description of command is printed. The -v
option causes a single word indicating the command or filename used to invoke command to be displayed; the -V option produces a
more verbose description. If the -V or -v option is supplied,
the exit status is 0 if command was found, and 1 if not. If nei‐ ther option is supplied and an error occurred or command cannot be found, the exit status is 127. Otherwise, the exit status of the command builtin is the exit status of command.

#10 lore raw

Re: [PATCH v2 1/2] usertools/devbind: update coding style

Quoted 15 lines · Stephen Hemminger

usertools/dpdk-devbind-new.py line 302

Quoted 8 lines · On Tue, 3 Dec 2024 11:25:00 +0000 Anatoly Burakov <[email protected]>…
usertools/dpdk-devbind-new.py new file +8 -0
+
+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}'.")
+

Apparently the posix way to do this is to use command -v not "which"

Quoted 14 lines

command [-pVv] command [arg ...]
Run command with args suppressing the normal shell function lookup. Only builtin commands or commands found in the PATH are
executed. If the -p option is given, the search for command is
performed using a default value for PATH that is guaranteed to find all of the standard utilities. If either the -V or -v op‐
tion is supplied, a description of command is printed. The -v
option causes a single word indicating the command or filename used to invoke command to be displayed; the -V option produces a
more verbose description. If the -V or -v option is supplied,
the exit status is 0 if command was found, and 1 if not. If nei‐ ther option is supplied and an error occurred or command cannot be found, the exit status is 127. Otherwise, the exit status of the command builtin is the exit status of command.

Yes but that's for shell, it is not accessible from Python. Well, it wouldn't be unless you're suggesting calling into bash and running command -v?

Signature · 2 lines
Thanks,
Anatoly
Reviewed-by on v2 1/2
#11 lore raw

Re: [PATCH v2 1/2] usertools/devbind: update coding style

In reply to Anatoly Burakov
Quoted 26 lines · On Tue, 3 Dec 2024 11:25:00 +0000 Anatoly Burakov <[email protected]>…

Devbind is one of the oldest tools in DPDK, and is written in a way that uses a lot of string matching, no type safety, lots of global variables, and has a few inconsistencies in the way it handles data (such as differences between lspci calls and parsing in different circumstances).

This patch is a nigh complete rewrite of devbind, with full 100% feature and command-line compatibility with the old version (except for dropping older kernel support), albeit with a few differences in formatting and error messages. All file handling code has also been replaced with context managers.

What's different from old code:
- Full PEP-484 compliance
- Formatted with Ruff
- Much better structured code
- Clean and consistent control flow
- More comments
- Better error handling
- Fewer lspci calls
- Unified lspci parsing
- Using /sys/bus/pci/drivers as a source of truth about kernel modules
- Check for iproute2 package
- Use JSON parsing for iproute2 output
- Deprecate --status-dev in favor of optional --status argument
- Deprecate kernel <3.15 support and only use driver_override

Quoted 2 lines

---

Looks great, like it.

Only suggestion (which you can ignore) would be to make DevbindCtx an object with methods bind_devices and print_status, that might simplify.

#12 lore raw

Re: [PATCH v2 1/2] usertools/devbind: update coding style

Quoted 29 lines · Stephen Hemminger
Quoted 26 lines · On Tue, 3 Dec 2024 11:25:00 +0000 Anatoly Burakov <[email protected]>…

Devbind is one of the oldest tools in DPDK, and is written in a way that uses a lot of string matching, no type safety, lots of global variables, and has a few inconsistencies in the way it handles data (such as differences between lspci calls and parsing in different circumstances).

This patch is a nigh complete rewrite of devbind, with full 100% feature and command-line compatibility with the old version (except for dropping older kernel support), albeit with a few differences in formatting and error messages. All file handling code has also been replaced with context managers.

What's different from old code:
- Full PEP-484 compliance
- Formatted with Ruff
- Much better structured code
- Clean and consistent control flow
- More comments
- Better error handling
- Fewer lspci calls
- Unified lspci parsing
- Using /sys/bus/pci/drivers as a source of truth about kernel modules
- Check for iproute2 package
- Use JSON parsing for iproute2 output
- Deprecate --status-dev in favor of optional --status argument
- Deprecate kernel <3.15 support and only use driver_override

Quoted 7 lines
Quoted 2 lines

---

Looks great, like it.

Only suggestion (which you can ignore) would be to make DevbindCtx an object with methods bind_devices and print_status, that might simplify.

The intention was that DevbindCtx is for processing command-line configuration and for keeping reference to Devbind which does actual work. I feel like the only thing it will simplify is instead of passing ctx around we'll be passing self. I will look into it though, maybe there are some opportunities that I'm missing.

Quoted 2 lines

Thanks!

Signature · 2 lines
Thanks,
Anatoly
#13 lore raw

PATCH v2 2/2 usertools/devbind: replace devbind

In reply to Anatoly Burakov

---

Notes:
v2:
- Added this patch to aid in review
- I believe it's better to squash it on apply

 usertools/dpdk-devbind-new.py |  996 -------------------
 usertools/dpdk-devbind.py     | 1678 ++++++++++++++++++---------------
 2 files changed, 911 insertions(+), 1763 deletions(-)
 delete mode 100755 usertools/dpdk-devbind-new.py
usertools/dpdk-devbind-new.py deleted +0 -996
deleted file mode 100755
index 9f2ee27cf3..0000000000
@@ -1,996 +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:
-        if "dev" in route:
-            routed_ifs.append(route["dev"])
-    return 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 _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 __getitem__(self, key: str) -> str:
-        """Get value as it appears in the device dictionary."""
-        return self._get_str(key)
-
-    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:
-        self.uio_modules: T.List[str]
-        # 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
-        }
-        # gather all loaded kernel modules
-        self.loaded_pci_modules = sysfs_read_pci_drivers()
-        # find which ones are UIO modules
-        self._scan_uio_modules()
-        # gather all routed interfaces
-        self._routed_interfaces = read_routed_interfaces()
-        # initialize data
-        self._scan_pci_devices()
-
-    def _categorize_pci_device(self, dev: Device) -> None:
-        for devtype, dt_tup in DEVICE_TYPES.items():
-            _, patterns = dt_tup
-            for pattern in patterns:
-                if dev.match(pattern):
-                    self.devices_by_type[devtype].append(dev)
-                    break
-        # 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
-        if any(iface in self._routed_interfaces for iface in dev.interfaces):
-            dev.active_interface = True
-
-    def _scan_uio_modules(self) -> None:
-        loaded = set(self.loaded_pci_modules)
-        supported = set(DPDK_KERNEL_MODULES)
-        self.uio_modules = 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 _scan_pci_devices(self) -> None:
-        """Find all devices on the system."""
-        self.pci_devices = {}
-
-        for dev in read_devices_from_lspci():
-            # categorize device
-            self._categorize_pci_device(dev)
-
-            # fixup module and driver fields
-            self.add_uio_modules(dev)
-
-            # save the device in common list
-            self.pci_devices[dev.slot] = dev
-
-    def resolve_device(self, devstr: str) -> str:
-        """Try to resolve a device into a PCI D:B:D:F."""
-        if not devstr:
-            raise ValueError(
-                f"Unknown device '{devstr}'. Please specify device in 'bus:slot.func' format."
-            )
-        # 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_name: 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_name} devices found."
-        print(msg)
-        print("=" * len(msg))
-        print()
-        return
-
-    devbind = ctx.devbind
-
-    # 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 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_name} devices using DPDK-compatible driver", dpdk)
-    if kernel:
-        print_status_section(f"{group_name} devices using kernel driver", kernel)
-    if unbound:
-        print_status_section(f"Other {group_name} 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}")
usertools/dpdk-devbind.py +911 -767

1,757 lines not on this page · raw message

Signature · 1 line
2.43.5
#14 lore raw

v3 cover Rewrite devbind

In reply to Anatoly Burakov

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort. Instead, as has been suggested [2], the patchset now consists of creating a new file, followed by a removal of old file and rename of the new file. It is expected that this will be squashed on apply.

The script has become slightly bigger - 1000 lines instead of 800, however I would argue that since most of that increase is infrastructure, comments, and sacrificing code golf for code readability (such as expanding one-liners into multiple lines), the trade-off between being able to read and reason about what happens in the script is worth the added line count.

[1] https://patches.dpdk.org/project/dpdk/patch/c2bf00195c2d43833a831a9cc9346b4606d6ea2e.1723810613.git.anatoly.burakov@intel.com/
[2] https://patches.dpdk.org/project/dpdk/cover/[email protected]/

Anatoly Burakov (1):
usertools/devbind: replace devbind

 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
Signature · 1 line
2.43.5
#15 lore raw

PATCH v3 1/1 usertools/devbind: replace devbind

---

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
usertools/dpdk-devbind-new.py deleted +0 -995
deleted file mode 100755
index 1f2d8cb118..0000000000
@@ -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}")
usertools/dpdk-devbind.py +913 -770

1,759 lines not on this page · raw message

Signature · 1 line
2.43.5
#16 lore raw
In reply to Anatoly Burakov
Quoted 22 lines · Anatoly Burakov

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort. Instead, as has been suggested [2], the patchset now consists of creating a new file, followed by a removal of old file and rename of the new file. It is expected that this will be squashed on apply.

The script has become slightly bigger - 1000 lines instead of 800, however I would argue that since most of that increase is infrastructure, comments, and sacrificing code golf for code readability (such as expanding one-liners into multiple lines), the trade-off between being able to read and reason about what happens in the script is worth the added line count.

[1] https://patches.dpdk.org/project/dpdk/patch/c2bf00195c2d43833a831a9cc9346b4606d6ea2e.1723810613.git.anatoly.burakov@intel.com/
[2] https://patches.dpdk.org/project/dpdk/cover/[email protected]/

Quoted 8 lines

Anatoly Burakov (1):
usertools/devbind: replace devbind

  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

Please disregard this, scripting mishap.

Signature · 2 lines
Thanks,
Anatoly
#17 lore raw

v3 cover Rewrite devbind

In reply to Anatoly Burakov

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort. Instead, as has been suggested [2], the patchset now consists of creating a new file, followed by a removal of old file and rename of the new file. It is expected that this will be squashed on apply.

The script has become slightly bigger - 1000 lines instead of 800, however I would argue that since most of that increase is infrastructure, comments, and sacrificing code golf for code readability (such as expanding one-liners into multiple lines), the trade-off between being able to read and reason about what happens in the script is worth the added line count.

[1] https://patches.dpdk.org/project/dpdk/patch/c2bf00195c2d43833a831a9cc9346b4606d6ea2e.1723810613.git.anatoly.burakov@intel.com/
[2] https://patches.dpdk.org/project/dpdk/cover/[email protected]/

Anatoly Burakov (2):
usertools/devbind: update coding style
usertools/devbind: replace devbind

 doc/guides/tools/devbind.rst |   11 +
 usertools/dpdk-devbind.py    | 1683 ++++++++++++++++++----------------
 2 files changed, 924 insertions(+), 770 deletions(-)
Signature · 1 line
2.43.5
#18 lore raw

PATCH v3 1/2 usertools/devbind: update coding style

Devbind is one of the oldest tools in DPDK, and is written in a way that uses a lot of string matching, no type safety, lots of global variables, and has a few inconsistencies in the way it handles data (such as differences between lspci calls and parsing in different circumstances).

This patch is a nigh complete rewrite of devbind, with full 100% feature and command-line compatibility with the old version (except for dropping older kernel support), albeit with a few differences in formatting and error messages. All file handling code has also been replaced with context managers.

What's different from old code:
- Full PEP-484 compliance
- Formatted with Ruff
- Much better structured code
- Clean and consistent control flow
- More comments
- Better error handling
- Fewer lspci calls
- Unified lspci parsing
- Using /sys/bus/pci/drivers as a source of truth about kernel modules
- Check for iproute2 package
- Use JSON parsing for iproute2 output
- Deprecate --status-dev in favor of optional --status argument
- Deprecate kernel <3.15 support and only use driver_override

---

Notes:
v2 - v3:
- Skip uninteresting routes reported by ip route
- Deduplicate active interface list
- Added missing documentation for no-IOMMU and UID/GID feature
- Slightly simplify device scanning

v1 -> v2:
- Use dictionary syntax to get raw string values from devices
- Fixed rollback not working correctly due to stale device state
- Fixed attempts to bind to empty driver on rollback
- Simplified bind/rollback and removed recursion
- Unified command-line and device type handling
- Dropped support for kernels <3.15
- Use JSON parsing for ip route output
- Used a new filename to aid in review, rename in next patch

 doc/guides/tools/devbind.rst  |  11 +
 usertools/dpdk-devbind-new.py | 995 ++++++++++++++++++++++++++++++++++
 2 files changed, 1006 insertions(+)
 create mode 100755 usertools/dpdk-devbind-new.py
doc/guides/tools/devbind.rst +11 -0
index df4f3505ac..ef9ed44fea 100644
@@ -56,6 +56,17 @@ OPTIONS
         WARNING: This can lead to loss of network connection and should be used
         with caution.
 
+* ``--noiommu-mode``
+
+        When using vfio-pci driver on a system with no IOMMU, this flag should be used to
+        enable unsafe no-IOMMU mode for vfio-pci driver.
+
+* ``--uid uid, --gid gid``
+
+        By default, devices which are bound to VFIO will be owned by ``root``.
+        Use this flag to change ownership to the specified user and group, so that
+        devices bound to VFIO would be usable by unprivileged users.
+
 
 .. warning::
 
usertools/dpdk-devbind-new.py new file +995 -0

995 lines not on this page · raw message

Signature · 1 line
2.43.5
#19 lore raw

PATCH v3 2/2 usertools/devbind: replace devbind

In reply to Anatoly Burakov

---

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
usertools/dpdk-devbind-new.py deleted +0 -995
deleted file mode 100755
index 1f2d8cb118..0000000000
@@ -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}")
usertools/dpdk-devbind.py +913 -770

1,759 lines not on this page · raw message

Signature · 1 line
2.43.5
#20 lore raw
In reply to Anatoly Burakov
Quoted 22 lines · Anatoly Burakov

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort. Instead, as has been suggested [2], the patchset now consists of creating a new file, followed by a removal of old file and rename of the new file. It is expected that this will be squashed on apply.

The script has become slightly bigger - 1000 lines instead of 800, however I would argue that since most of that increase is infrastructure, comments, and sacrificing code golf for code readability (such as expanding one-liners into multiple lines), the trade-off between being able to read and reason about what happens in the script is worth the added line count.

[1] https://patches.dpdk.org/project/dpdk/patch/c2bf00195c2d43833a831a9cc9346b4606d6ea2e.1723810613.git.anatoly.burakov@intel.com/
[2] https://patches.dpdk.org/project/dpdk/cover/[email protected]/

Quoted 8 lines

Anatoly Burakov (2):
usertools/devbind: update coding style
usertools/devbind: replace devbind

  doc/guides/tools/devbind.rst |   11 +
  usertools/dpdk-devbind.py    | 1683 ++++++++++++++++++----------------
  2 files changed, 924 insertions(+), 770 deletions(-)

Ping.

Signature · 2 lines
Thanks,
Anatoly
#21 lore raw
In reply to Anatoly Burakov
Quoted 22 lines · On Wed, 4 Dec 2024 09:47:36 +0000 Anatoly Burakov <[email protected]>…

It has been suggested [1] that a major cleanup/rewrite of devbind would be beneficial in terms of long term maintainability of the code. I was in a coding mood over the weekend, and so I've went ahead and rewritten devbind.

Note that this is one giant patch, rather than a series of patches adjusting existing code. Making it a patch series is possible, however the internal code architecture diverges quite significantly from the original devbind script due to its copious usage of string operations/pattern matching and global variables, so it is unclear whether subdividing this patch would be worth the effort. Instead, as has been suggested [2], the patchset now consists of creating a new file, followed by a removal of old file and rename of the new file. It is expected that this will be squashed on apply.

The script has become slightly bigger - 1000 lines instead of 800, however I would argue that since most of that increase is infrastructure, comments, and sacrificing code golf for code readability (such as expanding one-liners into multiple lines), the trade-off between being able to read and reason about what happens in the script is worth the added line count.

[1] https://patches.dpdk.org/project/dpdk/patch/c2bf00195c2d43833a831a9cc9346b4606d6ea2e.1723810613.git.anatoly.burakov@intel.com/
[2] https://patches.dpdk.org/project/dpdk/cover/[email protected]/

Quoted 8 lines

Anatoly Burakov (2):
usertools/devbind: update coding style
usertools/devbind: replace devbind

 doc/guides/tools/devbind.rst |   11 +
 usertools/dpdk-devbind.py    | 1683 ++++++++++++++++++----------------
 2 files changed, 924 insertions(+), 770 deletions(-)

This patch seemed to get no reviews. So sent it to AI; especially since my python-fu is rudimentary. It identified some small things, could you fix those and resubmit.

# DPDK Patch Review: usertools/devbind rewrite

**Series**: [PATCH v3 1/2] usertools/devbind: update coding style **Author**: Anatoly Burakov <[email protected]> **Reviewer**: Stephen Hemminger <[email protected]>

---

## Overview

This is a 2-patch series providing a complete rewrite of dpdk-devbind.py:

1. **Patch 1/2**: Introduces the new rewritten script as dpdk-devbind-new.py
2. **Patch 2/2**: Replaces the original script with the rewritten version

The rewrite brings significant improvements including PEP-484 type hints, modern Python idioms with context managers, structured classes, JSON parsing for ip route output, and deprecation of kernel <3.15 support.

---

## Errors (Must Fix)

### 1. Tag Formatting Error (Patch 1)
**Line 96:**

```
Reviewed-By: Stephen Hemminger <[email protected]>
```
Should be:
```
Reviewed-by: Stephen Hemminger <[email protected]>
```

The By should be lowercase. DPDK tag format is Reviewed-by:.

### 2. Typo in Variable Name
**Line 325:**

```python
BASEDBAND_DEVICES = [CLASS_ACCELERATION]
```
Should be:
```python
BASEBAND_DEVICES = [CLASS_ACCELERATION]
```

This typo is referenced at line 369 as well.

### 3. Typo in Error Message
**Line 820:**

```python
raise DevbindError(f"CLeanup failed for {dev}: {e}") from e
```
Should be:
```python
raise DevbindError(f"Cleanup failed for {dev}: {e}") from e
```

---

## Warnings (Should Fix)

### 4. Docstring Inconsistency
**Line 452-453:**

```python
def read_output(args: T.List[str]) -> str:
    """Run a subprocess, collect its output, and return it as a list of lines."""
```

The docstring says "list of lines" but the function returns a str. Either update the docstring to say "return it as a string" or change the return type.

### 5. --noiommu-mode Flag Not Checked
**Lines 876-877:**

```python
if use_vfio and not sysfs_iommu_enabled():
    sysfs_enable_unsafe_noiommu()
```
The `ctx.noiommu` flag is parsed (line 1100) but the code unconditionally enables no-IOMMU mode when IOMMU is disabled, regardless of whether the user specified `--noiommu-mode`. The original devbind required the flag to be explicitly set. Consider:
```python
if use_vfio and not sysfs_iommu_enabled():
    if not ctx.noiommu:
        raise DevbindError(
            "IOMMU not available. Use --noiommu-mode to enable unsafe no-IOMMU mode."
        )
    sysfs_enable_unsafe_noiommu()
```

### 6. No Unit Tests Added
A rewrite of this magnitude would benefit from unit tests in app/test or at minimum a standalone test script. Consider adding tests for:
- Device pattern matching
- lspci line parsing
- Argument parsing edge cases

### 7. Patch 2/2 Missing Commit Message Body
**Line 1213:**
The second patch has only Signed-off-by: with no commit message body explaining what it does. Even simple patches should have a brief description.

---

## Info (Suggestions)

### 8. File Mode for Bind Operation
**Line 807:**

```python
with open(bind_path, "a", encoding="utf-8") as f:
    f.write(dev.slot)
```

Using append mode ("a") for writing to sysfs bind file is unusual. While it likely works, write mode ("w") is more conventional and matches what the kernel expects. Consider changing to "w".

### 9. Null Byte for driver_override Cleanup
**Line 818:**

```python
f.write("\00")
```
Writing a null byte works, but an empty string or newline might be more portable/conventional:
```python
f.write("")
```

### 10. Generator Consumed Multiple Times Risk
**Line 862:**

```python
devices = (devbind.pci_devices[dbdf] for dbdf in ctx.devices)
```
This creates a generator that can only be iterated once. If the code path were to iterate it multiple times, this would silently fail. Consider using a list comprehension instead:
```python
devices = [devbind.pci_devices[dbdf] for dbdf in ctx.devices]
```

### 11. Consider Using shutil.which() Instead of subprocess **Lines 446-449:**

```python
if subprocess.call(
    ["which", program], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
):
```
Python 3.3+ provides `shutil.which()` which is more Pythonic:
```python
import shutil
if shutil.which(program) is None:
```

### 12. Documentation Formatting
**Lines 129-138 in doc/guides/tools/devbind.rst:**
The added documentation has a double blank line after line 128. RST typically uses single blank lines.

---

## Positive Observations

The rewrite demonstrates excellent software engineering:

- **Type Safety**: Comprehensive PEP-484 type annotations throughout
- **Modern Python**: Context managers, dataclasses pattern, proper exception hierarchy
- **Code Organization**: Clean separation with Device, Devbind, and DevbindCtx classes
- **Error Handling**: Consistent DevbindError exception with clear messages
- **Resource Management**: All file operations use context managers
- **JSON Parsing**: Using ip -j route for structured output is much better than text parsing
- **Reduced External Calls**: Fewer lspci invocations, unified parsing
- **Backward Compatibility**: Command-line interface remains compatible

---

## Summary

| Severity | Count |
|----------|-------|
| Error    | 3     |
| Warning  | 4     |
| Info     | 5     |

**Recommendation**: Fix the three errors (tag case, two typos) and address the --noiommu-mode logic, then this is ready for merge. The rewrite is a significant improvement over the original code.

---

## Checklist

- [x] SPDX license identifier present
- [x] Copyright line present
- [x] Signed-off-by present
- [ ] Reviewed-by tag correctly formatted
- [x] Subject line ≤60 characters
- [x] Subject lowercase (except acronyms)
- [x] Imperative mood
- [x] No trailing period
- [x] Lines ≤100 characters (Python allows this)
- [x] Documentation updated
- [ ] Tests added/updated