utils/vfio: Add APIs for VFIO container and device management

dhsrivas · dhsrivas · commit d3cfbf3c423c · 2026-01-24T16:35:15.000+05:30
This patch introduces a set of VFIO (Virtual Function I/O) helper APIs
to simplify container, IOMMU group, and device management. The new
functions provide a clean abstraction over common VFIO operations:

* get_vfio_container_fd() – open and validate a VFIO container
* check_vfio_container() – verify API version and IOMMU extension support
* get_iommu_group_fd() – obtain an IOMMU group file descriptor
* attach_group_to_container()/detach_group_from_container() – manage group
  association with a VFIO container
* get_device_fd() – retrieve a VFIO device file descriptor
* vfio_device_supports_irq() – check if a device supports the requested
  number of MSI-X interrupts

Signed-off-by: Dheeraj Kumar Srivastava &lt;dheerajkumar.srivastava@amd.com&gt;
diff --git a/avocado/utils/vfio.py b/avocado/utils/vfio.py
@@ -0,0 +1,193 @@
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+#
+# See LICENSE for more details.
+#
+# Copyright: 2025 Advanced Micro Devices, Inc.
+# Author: Dheeraj Kumar Srivastava <dheerajkumar.srivastava@amd.com> # pylint: disable=C0401
+
+# pylint: disable=C0402
+
+"""APIs for virtual function I/O management."""
+
+
+import ctypes
+import os
+import struct
+from fcntl import ioctl
+
+from avocado.utils import pci
+
+
+def get_vfio_container_fd():
+    """get vfio container file descriptor
+
+    :return: file descriptor of the vfio container.
+    :rtype: int
+    :raises ValueError: if the vfio container cannot be opened.
+    """
+    try:
+        return os.open("/dev/vfio/vfio", os.O_RDWR | os.O_CLOEXEC)
+    except OSError as e:
+        raise ValueError(f"Failed to open VFIO container: {e}") from e
+
+
+def check_vfio_container(
+    container_fd,
+    vfio_get_api_version,
+    vfio_api_version,
+    vfio_check_extension,
+    vfio_type_iommu,
+):
+    """Validate a vfio container by verifying the api version and ensuring that the
+    required iommu extension is supported. If either validation fails, an exception
+    is raised with an appropriate message
+
+    :param container_fd: vfio container file descriptor
+    :type container_fd: int
+    :param vfio_get_api_version: ioctl to retrieve the vfio container api version
+    :type vfio_get_api_version: int
+    :param vfio_api_version: expected vfio api version
+    :type vfio_api_version: int
+    :param vfio_check_extension:  ioctl to check iommu extension support
+    :type vfio_check_extension: int
+    :param vfio_type_iommu: expected vfio iommu type
+    :type vfio_type_iommu: int
+    :raises ValueError: if the vfio api version is invalid or if the required
+                        iommu extension is not supported.
+    :return: True if vfio container fd check passes.
+    :rtype: bool
+    """
+    if ioctl(container_fd, vfio_get_api_version) != vfio_api_version:
+        raise ValueError("Failed to get right API version")
+
+    if not ioctl(container_fd, vfio_check_extension, vfio_type_iommu):
+        raise ValueError("does not support type 1 iommu")
+
+    return True
+
+
+def get_iommu_group_fd(device, vfio_group_get_status, vfio_group_flags_viable):
+    """get iommu group fd for the pci device
+
+    :param device: full pci address including domain (0000:03:00.0)
+    :type device: str
+    :param vfio_group_get_status: ioctl to get iommu group status
+    :type vfio_group_get_status: int
+    :param vfio_group_flags_viable: ioctl to check if iommu group is viable
+    :type vfio_group_flags_viable: int
+    :raises ValueError: if the vfio group device cannot be opened or the group
+                        is not viable.
+    :return: file descriptor for the iommu group.
+    :rtype: int
+    """
+    vfio_group = f"/dev/vfio/{pci.get_iommu_group(device)}"
+    try:
+        group_fd = os.open(vfio_group, os.O_RDWR)
+    except OSError as e:
+        raise ValueError(f"Failed to open {vfio_group}: {e}") from e
+
+    argsz = struct.calcsize("II")
+    group_status_request = struct.pack("II", argsz, 2)
+    group_status_response = ioctl(group_fd, vfio_group_get_status, group_status_request)
+    group_status = struct.unpack("II", group_status_response)
+
+    if not group_status[1] & vfio_group_flags_viable:
+        raise ValueError("Group not viable, are all devices attached to vfio?")
+
+    return group_fd
+
+
+def attach_group_to_container(group_fd, container_fd, vfio_group_set_container):
+    """attach the iommu group of pci device to the vfio container.
+
+    :param group_fd: iommu group file descriptor
+    :type group_fd: int
+    :param container_fd: vfio container file descriptor
+    :type container_fd: int
+    :param vfio_group_set_container: vfio ioctl to add iommu group to the container fd
+    :type vfio_group_set_container: int
+    :raises ValueError: if attaching the group to the container fails.
+    """
+
+    try:
+        ioctl(group_fd, vfio_group_set_container, ctypes.c_void_p(container_fd))
+    except OSError as e:
+        raise ValueError(
+            f"failed to attach pci device's iommu group to the vfio container: {e}"
+        ) from e
+
+
+def detach_group_from_container(group_fd, container_fd, vfio_group_unset_container):
+    """detach the iommu group of pci device from vfio container
+
+    :param group_fd: iommu group file descriptor
+    :type group_fd: int
+    :param container_fd: vfio container file descriptor
+    :type container_fd: int
+    :param vfio_group_unset_container: vfio ioctl to detach iommu group from the
+                                       container fd
+    :type vfio_group_unset_container: int
+    :raises ValueError: if detaching the group to the container fails.
+    """
+
+    try:
+        ioctl(group_fd, vfio_group_unset_container, ctypes.c_void_p(container_fd))
+    except OSError as e:
+        raise ValueError(
+            f"failed to detach pci device's iommu group from vfio container: {e}"
+        ) from e
+
+
+def get_device_fd(device, group_fd, vfio_group_get_device_fd):
+    """Get device file descriptor
+
+    :param device: full pci address including domain (0000:03:00.0)
+    :type device: str
+    :param group_fd: iommu group file descriptor
+    :type group_fd: int
+    :param vfio_group_get_device_fd: ioctl to get device fd
+    :type vfio_group_get_device_fd: int
+    :raises ValueError: if not able to get device descriptor
+    :return: device descriptor
+    :rtype: int
+    """
+    buf = ctypes.create_string_buffer(device.encode("utf-8") + b"\x00")
+    try:
+        device_fd = ioctl(group_fd, vfio_group_get_device_fd, buf)
+    except OSError as e:
+        raise ValueError("failed to get vfio device fd") from e
+
+    return device_fd
+
+
+def vfio_device_supports_irq(
+    device_fd, vfio_pci_msix_irq_index, vfio_device_get_irq_info, count
+):
+    """Check if device supports at least count number of interrupts
+
+    :param device_fd: device file descriptor
+    :type device_fd: int
+    :param vfio_pci_msix_irq_index: vfio ioctl to get irq index for msix
+    :type vfio_pci_msix_irq_index: int
+    :param vfio_device_get_irq_info: vfio ioctl to get vfio device irq information
+    :type vfio_device_get_irq_info: int
+    :param count: number of irqs the device should support
+    :type count: int
+    :return: true if supported, false otherwise
+    :rtype: bool
+    """
+    argsz = struct.calcsize("IIII")
+    index = vfio_pci_msix_irq_index
+    irq_info_request = struct.pack("IIII", argsz, 1, index, 1)
+    irq_info_response = ioctl(device_fd, vfio_device_get_irq_info, irq_info_request)
+    nirq = (struct.unpack("IIII", irq_info_response))[3]
+    if nirq < count:
+        return False
+    return True
diff --git a/selftests/check.py b/selftests/check.py
@@ -27,7 +27,7 @@
     "job-api-check-tmp-directory-exists": 1,
     "nrunner-interface": 90,
     "nrunner-requirement": 28,
-    "unit": 934,
+    "unit": 948,
     "jobs": 11,
     "functional-parallel": 353,
     "functional-serial": 7,
diff --git a/selftests/unit/utils/vfio.py b/selftests/unit/utils/vfio.py
@@ -0,0 +1,175 @@
+import os
+import struct
+import unittest
+from unittest import mock
+
+from avocado import Test
+from avocado.utils import vfio
+
+
+class VfioUtilsTests(Test):
+    @mock.patch("os.open")
+    def test_get_vfio_container_fd_fail(self, mock_open):
+        mock_open.side_effect = OSError("No such file or directory")
+        with self.assertRaises(ValueError) as e:
+            vfio.get_vfio_container_fd()
+        self.assertIn("Failed to open VFIO container", str(e.exception))
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_invalid_api_version(self, mock_ioctl):
+        mock_ioctl.side_effect = [1, True]
+        with self.assertRaises(ValueError) as e:
+            vfio.check_vfio_container(
+                container_fd=3,
+                vfio_get_api_version=15204,
+                vfio_api_version=0,
+                vfio_check_extension=15205,
+                vfio_type_iommu=1,
+            )
+        self.assertIn("Failed to get right API version", str(e.exception))
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_missing_iommu_extension(self, mock_ioctl):
+        mock_ioctl.side_effect = [0, False]
+        with self.assertRaises(ValueError) as e:
+            vfio.check_vfio_container(
+                container_fd=3,
+                vfio_get_api_version=15204,
+                vfio_api_version=0,
+                vfio_check_extension=15205,
+                vfio_type_iommu=1,
+            )
+        self.assertIn("does not support type 1 iommu", str(e.exception))
+
+    @mock.patch("avocado.utils.vfio.os.open")
+    @mock.patch("avocado.utils.vfio.pci.get_iommu_group")
+    def test_get_group_fd_fail_1(self, mock_get_group, mock_open):
+        mock_open.side_effect = OSError("No such file or directory")
+        mock_get_group.return_value = "42"
+        with self.assertRaises(ValueError) as e:
+            vfio.get_iommu_group_fd(
+                device="0000:03:00.0",
+                vfio_group_get_status=100,
+                vfio_group_flags_viable=0x1,
+            )
+        self.assertIn("Failed to open /dev/vfio/42", str(e.exception))
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    @mock.patch("avocado.utils.vfio.os.open")
+    @mock.patch("avocado.utils.vfio.pci.get_iommu_group")
+    def test_get_group_fd_fail_2(self, mock_get_group, mock_open, mock_ioctl):
+        mock_open.return_value = 3
+        mock_get_group.return_value = "42"
+        argsz = struct.calcsize("II")
+        # Pack request, ioctl returns same but with flags = 0
+        mock_ioctl.return_value = struct.pack("II", argsz, 0)
+
+        with self.assertRaises(ValueError) as ctx:
+            vfio.get_iommu_group_fd(
+                device="0000:03:00.0",
+                vfio_group_get_status=100,
+                vfio_group_flags_viable=0x1,
+            )
+        self.assertIn("Group not viable", str(ctx.exception))
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    @mock.patch("avocado.utils.vfio.os.open")
+    @mock.patch("avocado.utils.vfio.pci.get_iommu_group")
+    def test_get_group_fd_success(self, mock_get_group, mock_open, mock_ioctl):
+        mock_open.return_value = 3
+        mock_get_group.return_value = "42"
+        argsz = struct.calcsize("II")
+        # Pack request, ioctl returns same but with flags = 1
+        mock_ioctl.return_value = struct.pack("II", argsz, 0x1)
+
+        fd = vfio.get_iommu_group_fd(
+            device="0000:03:00.0",
+            vfio_group_get_status=100,
+            vfio_group_flags_viable=0x1,
+        )
+
+        self.assertEqual(fd, 3)
+        mock_open.assert_called_once_with("/dev/vfio/42", os.O_RDWR)
+        mock_ioctl.assert_called_once()
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_attach_group_to_container_success(self, mock_ioctl):
+        mock_ioctl.return_value = 0
+        # Should not raise
+        vfio.attach_group_to_container(
+            group_fd=10, container_fd=20, vfio_group_set_container=100
+        )
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_attach_group_to_container_failure(self, mock_ioctl):
+        mock_ioctl.return_value = 1
+        with self.assertRaises(ValueError) as ctx:
+            vfio.attach_group_to_container(
+                group_fd=10, container_fd=20, vfio_group_set_container=100
+            )
+        self.assertIn("failed to attached pci device", str(ctx.exception))
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_detach_group_from_container_success(self, mock_ioctl):
+        mock_ioctl.return_value = 0
+        vfio.detach_group_from_container(
+            group_fd=10, container_fd=20, vfio_group_unset_container=200
+        )
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_detach_group_from_container_failure(self, mock_ioctl):
+        mock_ioctl.return_value = 1
+        with self.assertRaises(ValueError) as ctx:
+            vfio.detach_group_from_container(
+                group_fd=10, container_fd=20, vfio_group_unset_container=200
+            )
+        self.assertIn("failed to detach pci device", str(ctx.exception))
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_get_device_fd_success(self, mock_ioctl):
+        mock_ioctl.return_value = 50
+        device_fd = vfio.get_device_fd(
+            "0000:03:00.0", group_fd=10, vfio_group_get_device_fd=200
+        )
+        self.assertEqual(device_fd, 50)
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_get_device_fd_failure(self, mock_ioctl):
+        mock_ioctl.side_effect = OSError("Device not found")
+        with self.assertRaises(ValueError) as e:
+            vfio.get_device_fd(
+                "0000:03:00.0", group_fd=10, vfio_group_get_device_fd=200
+            )
+        self.assertIn("failed to get vfio device fd", str(e.exception))
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_vfio_device_supports_irq_success(self, mock_ioctl):
+        argsz = struct.calcsize("IIII")
+        response = struct.pack("IIII", argsz, 0, 0, 8)
+        mock_ioctl.return_value = response
+
+        result = vfio.vfio_device_supports_irq(
+            device_fd=30,
+            vfio_pci_msix_irq_index=100,
+            vfio_device_get_irq_info=300,
+            count=4,
+        )
+        self.assertTrue(result)
+
+    @mock.patch("avocado.utils.vfio.ioctl")
+    def test_vfio_device_supports_irq_fail(self, mock_ioctl):
+        argsz = struct.calcsize("IIII")
+        response = struct.pack("IIII", argsz, 0, 0, 2)
+        mock_ioctl.return_value = response
+
+        result = vfio.vfio_device_supports_irq(
+            device_fd=30,
+            vfio_pci_msix_irq_index=100,
+            vfio_device_get_irq_info=300,
+            count=4,
+        )
+        self.assertFalse(result)
+
+
+if __name__ == "__main__":
+    unittest.main()
