Skip to content

RDMA

RDMA (Remote Direct Memory Access) device and link management — covers InfiniBand and RoCE adapters. Used for NVMe-oF RDMA transport and RDMA-capable storage network testing.

sts.rdma

RDMA device discovery and configuration via sysfs (InfiniBand, RoCE, iWARP).

NetDev pydantic-model

Bases: _SysfsModel

Network interface associated with an RDMA device.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Network interface associated with an RDMA device.",
  "properties": {
    "path": {
      "format": "path",
      "title": "Path",
      "type": "string"
    },
    "sysfs_attrs": {
      "additionalProperties": {
        "type": "string"
      },
      "title": "Sysfs Attrs",
      "type": "object"
    },
    "dev_port": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dev Port"
    }
  },
  "required": [
    "path"
  ],
  "title": "NetDev",
  "type": "object"
}

Fields:

  • path (Path)
  • sysfs_attrs (dict[str, str])
  • dev_port (str | None)
Source code in sts_libs/src/sts/rdma.py 
126
127
128
129
130
131
132
133
134
class NetDev(_SysfsModel):
    """Network interface associated with an RDMA device."""

    dev_port: str | None = None  # Associated port number

    def discover(self) -> Self:
        """Read network device attributes from sysfs."""
        self.read_sysfs()
        return self

discover()

Read network device attributes from sysfs.

Source code in sts_libs/src/sts/rdma.py 
131
132
133
134
def discover(self) -> Self:
    """Read network device attributes from sysfs."""
    self.read_sysfs()
    return self

Port pydantic-model

Bases: _SysfsModel

RDMA port with link rate, state, and physical state parsed from sysfs.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "RDMA port with link rate, state, and physical state parsed from sysfs.",
  "properties": {
    "path": {
      "format": "path",
      "title": "Path",
      "type": "string"
    },
    "sysfs_attrs": {
      "additionalProperties": {
        "type": "string"
      },
      "title": "Sysfs Attrs",
      "type": "object"
    },
    "rate": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Rate"
    },
    "state": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "State"
    },
    "phys_state": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Phys State"
    },
    "rate_speed": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Rate Speed"
    },
    "rate_unit": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Rate Unit"
    },
    "rate_info": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Rate Info"
    },
    "state_num": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "State Num"
    },
    "state_str": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "State Str"
    },
    "phys_state_num": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Phys State Num"
    },
    "phys_state_str": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Phys State Str"
    }
  },
  "required": [
    "path"
  ],
  "title": "Port",
  "type": "object"
}

Fields:

  • path (Path)
  • sysfs_attrs (dict[str, str])
  • rate (str | None)
  • state (str | None)
  • phys_state (str | None)
  • rate_speed (str | None)
  • rate_unit (str | None)
  • rate_info (str | None)
  • state_num (str | None)
  • state_str (str | None)
  • phys_state_num (str | None)
  • phys_state_str (str | None)
Source code in sts_libs/src/sts/rdma.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class Port(_SysfsModel):
    """RDMA port with link rate, state, and physical state parsed from sysfs."""

    rate: str | None = None  # Link rate (e.g., '100 Gb/sec (4X EDR)')
    state: str | None = None  # Port state (e.g., '4: ACTIVE')
    phys_state: str | None = None  # Physical state (e.g., '5: LinkUp')
    rate_speed: str | None = None  # Numeric rate (e.g., '100')
    rate_unit: str | None = None  # Rate unit (e.g., 'Gb/sec')
    rate_info: str | None = None  # Link info (e.g., '4X EDR')
    state_num: str | None = None  # State number (e.g., '4')
    state_str: str | None = None  # State string (e.g., 'ACTIVE')
    phys_state_num: str | None = None  # Physical state number (e.g., '5')
    phys_state_str: str | None = None  # Physical state string (e.g., 'LinkUp')

    @computed_field  # type: ignore[prop-decorator]
    @property
    def name(self) -> str:
        """Port name, derived from the sysfs path."""
        return self.path.name

    def discover(self) -> Self:
        """Read and parse port attributes from sysfs."""
        self.read_sysfs()

        # Parse rate into components
        if self.rate:
            # Example: '100 Gb/sec (4X EDR)'
            rate_split = self.rate.split()
            self.rate_speed = rate_split[0]  # Numeric rate
            self.rate_unit = rate_split[1]  # Rate unit
            # Keep original format for link width and type
            self.rate_info = f'{rate_split[2].strip("(")} {rate_split[3].strip(")")}'

        # Parse port state
        if self.state:
            self.state_num = self.state.split(':')[0]
            self.state_str = self.state.split(': ')[1]

        # Parse physical state
        if self.phys_state:
            self.phys_state_num = self.phys_state.split(':')[0]
            self.phys_state_str = self.phys_state.split(': ')[1]

        return self

name property

Port name, derived from the sysfs path.

discover()

Read and parse port attributes from sysfs.

Source code in sts_libs/src/sts/rdma.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def discover(self) -> Self:
    """Read and parse port attributes from sysfs."""
    self.read_sysfs()

    # Parse rate into components
    if self.rate:
        # Example: '100 Gb/sec (4X EDR)'
        rate_split = self.rate.split()
        self.rate_speed = rate_split[0]  # Numeric rate
        self.rate_unit = rate_split[1]  # Rate unit
        # Keep original format for link width and type
        self.rate_info = f'{rate_split[2].strip("(")} {rate_split[3].strip(")")}'

    # Parse port state
    if self.state:
        self.state_num = self.state.split(':')[0]
        self.state_str = self.state.split(': ')[1]

    # Parse physical state
    if self.phys_state:
        self.phys_state_num = self.phys_state.split(':')[0]
        self.phys_state_str = self.phys_state.split(': ')[1]

    return self

Power pydantic-model

Bases: _SysfsModel

RDMA power management attributes from sysfs.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "RDMA power management attributes from sysfs.",
  "properties": {
    "path": {
      "format": "path",
      "title": "Path",
      "type": "string"
    },
    "sysfs_attrs": {
      "additionalProperties": {
        "type": "string"
      },
      "title": "Sysfs Attrs",
      "type": "object"
    }
  },
  "required": [
    "path"
  ],
  "title": "Power",
  "type": "object"
}

Fields:

  • path (Path)
  • sysfs_attrs (dict[str, str])
Source code in sts_libs/src/sts/rdma.py 
115
116
117
118
119
120
121
122
123
class Power(_SysfsModel):
    """RDMA power management attributes from sysfs."""

    def discover(self) -> Self:
        """Read power attributes from sysfs."""
        power_path = self.path / 'power'
        if power_path.is_dir():
            self.read_sysfs(power_path)
        return self

discover()

Read power attributes from sysfs.

Source code in sts_libs/src/sts/rdma.py 
118
119
120
121
122
123
def discover(self) -> Self:
    """Read power attributes from sysfs."""
    power_path = self.path / 'power'
    if power_path.is_dir():
        self.read_sysfs(power_path)
    return self

RdmaDevice pydantic-model

Bases: Device

RDMA device with ports, network interfaces, SR-IOV, and power management.

Attributes:

Name Type Description
ibdev str

Device ID (e.g., 'mlx5_0')
Show JSON schema: 
{
  "additionalProperties": false,
  "description": "RDMA device with ports, network interfaces, SR-IOV, and power management.\n\nAttributes:\n    ibdev: Device ID (e.g., 'mlx5_0')",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "ibdev": {
      "title": "Ibdev",
      "type": "string"
    },
    "ports_path": {
      "anyOf": [
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Ports Path"
    },
    "device_path": {
      "anyOf": [
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Device Path"
    },
    "net_path": {
      "anyOf": [
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Net Path"
    },
    "ports": {
      "items": {
        "format": "path",
        "type": "string"
      },
      "title": "Ports",
      "type": "array"
    },
    "port_numbers": {
      "items": {
        "type": "string"
      },
      "title": "Port Numbers",
      "type": "array"
    },
    "is_sriov_capable": {
      "default": false,
      "title": "Is Sriov Capable",
      "type": "boolean"
    },
    "sysfs_attrs": {
      "additionalProperties": {
        "type": "string"
      },
      "title": "Sysfs Attrs",
      "type": "object"
    }
  },
  "required": [
    "ibdev"
  ],
  "title": "RdmaDevice",
  "type": "object"
}

Fields:

  • path (PathOrStr | None)
  • name (str | None)
  • ibdev (str)
  • ports_path (Path | None)
  • device_path (Path | None)
  • net_path (Path | None)
  • ports (list[Path])
  • port_numbers (list[str])
  • is_sriov_capable (bool)
  • sysfs_attrs (dict[str, str])

Validators:

  • _derive_fields
Source code in sts_libs/src/sts/rdma.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
class RdmaDevice(Device):
    """RDMA device with ports, network interfaces, SR-IOV, and power management.

    Attributes:
        ibdev: Device ID (e.g., 'mlx5_0')
    """

    # Class-level paths
    RDMA_PATH: ClassVar[Path] = Path(RDMA_SYSFS_BASE)

    # Instance variables
    ibdev: str  # Device identifier
    ports_path: Path | None = None  # Path to ports directory
    device_path: Path | None = None  # Path to PCI device
    net_path: Path | None = None  # Path to network interfaces
    ports: list[Path] = Field(default_factory=list)  # Available ports
    port_numbers: list[str] = Field(default_factory=list)  # Port numbers
    is_sriov_capable: bool = False  # SR-IOV support
    sysfs_attrs: dict[str, str] = Field(default_factory=dict, repr=False)

    @model_validator(mode='after')
    def _derive_fields(self) -> Self:
        """Derive path and name from ibdev.

        No filesystem I/O happens here. Call discover() to read sysfs.
        """
        self.path = Path(f'{RDMA_SYSFS_BASE}{self.ibdev}')
        self.name = self.ibdev
        return self

    def discover(self) -> Self:
        """Load device state from sysfs (paths, ports, SR-IOV capability)."""
        assert isinstance(self.path, Path)

        _read_sysfs(self, self.path, self.sysfs_attrs)

        # Initialize component paths
        self.ports_path = self.path / 'ports'
        self.device_path = (self.path / 'device').resolve()
        self.net_path = self.device_path / 'net'

        # Discover ports
        self.ports = [port for port in self.ports_path.iterdir() if port.is_dir()]
        self.port_numbers = [port.name for port in self.ports]

        # Check SR-IOV support
        self.is_sriov_capable = (self.device_path / 'sriov_numvfs').is_file()
        return self

    def get_netdevs(self) -> list[NetDev]:
        """Get all associated network interfaces."""
        assert isinstance(self.net_path, Path)
        return [NetDev(path=eth).discover() for eth in self.net_path.iterdir() if eth.is_dir()]

    def get_netdev(self, port_id: str) -> NetDev | None:
        """Get network device by port ID.

        Args:
            port_id: Port number (1-based, converted internally to 0-based dev_port)
        """
        for dev in self.get_netdevs():
            if dev.dev_port == str(int(port_id) - 1):  # Convert to 0-based index
                return dev
        return None

    def get_ports(self) -> list[Port] | None:
        """Get all ports."""
        return [Port(path=port).discover() for port in self.ports] if self.ports else None

    def get_port(self, port: str) -> Port | None:
        """Get port by number."""
        assert isinstance(self.ports_path, Path)
        path = self.ports_path / port
        return Port(path=path).discover() if path.is_dir() else None

    def get_power(self) -> Power:
        """Get power management interface."""
        assert isinstance(self.path, Path)
        return Power(path=self.path).discover()

    def get_sriov(self) -> Sriov | None:
        """Get SR-IOV configuration interface, or None if device lacks SR-IOV."""
        assert isinstance(self.device_path, Path)
        return Sriov(path=self.device_path).discover() if self.is_sriov_capable else None

discover()

Load device state from sysfs (paths, ports, SR-IOV capability).

Source code in sts_libs/src/sts/rdma.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def discover(self) -> Self:
    """Load device state from sysfs (paths, ports, SR-IOV capability)."""
    assert isinstance(self.path, Path)

    _read_sysfs(self, self.path, self.sysfs_attrs)

    # Initialize component paths
    self.ports_path = self.path / 'ports'
    self.device_path = (self.path / 'device').resolve()
    self.net_path = self.device_path / 'net'

    # Discover ports
    self.ports = [port for port in self.ports_path.iterdir() if port.is_dir()]
    self.port_numbers = [port.name for port in self.ports]

    # Check SR-IOV support
    self.is_sriov_capable = (self.device_path / 'sriov_numvfs').is_file()
    return self

get_netdev(port_id)

Get network device by port ID.

Parameters:

Name Type Description Default
port_id str

Port number (1-based, converted internally to 0-based dev_port)

 required
Source code in sts_libs/src/sts/rdma.py 
224
225
226
227
228
229
230
231
232
233
def get_netdev(self, port_id: str) -> NetDev | None:
    """Get network device by port ID.

    Args:
        port_id: Port number (1-based, converted internally to 0-based dev_port)
    """
    for dev in self.get_netdevs():
        if dev.dev_port == str(int(port_id) - 1):  # Convert to 0-based index
            return dev
    return None

get_netdevs()

Get all associated network interfaces.

Source code in sts_libs/src/sts/rdma.py 
219
220
221
222
def get_netdevs(self) -> list[NetDev]:
    """Get all associated network interfaces."""
    assert isinstance(self.net_path, Path)
    return [NetDev(path=eth).discover() for eth in self.net_path.iterdir() if eth.is_dir()]

get_port(port)

Get port by number.

Source code in sts_libs/src/sts/rdma.py 
239
240
241
242
243
def get_port(self, port: str) -> Port | None:
    """Get port by number."""
    assert isinstance(self.ports_path, Path)
    path = self.ports_path / port
    return Port(path=path).discover() if path.is_dir() else None

get_ports()

Get all ports.

Source code in sts_libs/src/sts/rdma.py 
235
236
237
def get_ports(self) -> list[Port] | None:
    """Get all ports."""
    return [Port(path=port).discover() for port in self.ports] if self.ports else None

get_power()

Get power management interface.

Source code in sts_libs/src/sts/rdma.py 
245
246
247
248
def get_power(self) -> Power:
    """Get power management interface."""
    assert isinstance(self.path, Path)
    return Power(path=self.path).discover()

get_sriov()

Get SR-IOV configuration interface, or None if device lacks SR-IOV.

Source code in sts_libs/src/sts/rdma.py 
250
251
252
253
def get_sriov(self) -> Sriov | None:
    """Get SR-IOV configuration interface, or None if device lacks SR-IOV."""
    assert isinstance(self.device_path, Path)
    return Sriov(path=self.device_path).discover() if self.is_sriov_capable else None

Sriov pydantic-model

Bases: _SysfsModel

SR-IOV (Single Root I/O Virtualization) configuration via sysfs.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "SR-IOV (Single Root I/O Virtualization) configuration via sysfs.",
  "properties": {
    "path": {
      "format": "path",
      "title": "Path",
      "type": "string"
    },
    "sysfs_attrs": {
      "additionalProperties": {
        "type": "string"
      },
      "title": "Sysfs Attrs",
      "type": "object"
    },
    "sriov_numvfs": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Sriov Numvfs"
    },
    "sriov_totalvfs": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Sriov Totalvfs"
    }
  },
  "required": [
    "path"
  ],
  "title": "Sriov",
  "type": "object"
}

Fields:

  • path (Path)
  • sysfs_attrs (dict[str, str])
  • sriov_numvfs (str | None)
  • sriov_totalvfs (str | None)
Source code in sts_libs/src/sts/rdma.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class Sriov(_SysfsModel):
    """SR-IOV (Single Root I/O Virtualization) configuration via sysfs."""

    sriov_numvfs: str | None = None  # Current number of VFs
    sriov_totalvfs: str | None = None  # Maximum VFs supported

    @computed_field  # type: ignore[prop-decorator]
    @property
    def sriov_numvfs_path(self) -> Path:
        """Path to the numvfs control file, derived from path."""
        return self.path / 'sriov_numvfs'

    def discover(self) -> Self:
        """Read SR-IOV attributes from sysfs."""
        if self.path.is_dir():
            self.read_sysfs()
        return self

    def set_numvfs(self, num: str = '1') -> None:
        """Set number of Virtual Functions.

        Resets existing VFs before allocating the new count.
        """
        if self.sriov_numvfs and num != self.sriov_numvfs:
            self.sriov_numvfs_path.write_text('0')  # Reset VFs
            self.sriov_numvfs_path.write_text(num)  # Allocate new VFs
            self.sriov_numvfs = self.read_numvfs()  # Update state

    def read_numvfs(self) -> str | None:
        """Read current number of Virtual Functions."""
        return self.sriov_numvfs_path.read_text().strip() if self.sriov_numvfs_path.is_file() else None

sriov_numvfs_path property

Path to the numvfs control file, derived from path.

discover()

Read SR-IOV attributes from sysfs.

Source code in sts_libs/src/sts/rdma.py 
149
150
151
152
153
def discover(self) -> Self:
    """Read SR-IOV attributes from sysfs."""
    if self.path.is_dir():
        self.read_sysfs()
    return self

read_numvfs()

Read current number of Virtual Functions.

Source code in sts_libs/src/sts/rdma.py 
165
166
167
def read_numvfs(self) -> str | None:
    """Read current number of Virtual Functions."""
    return self.sriov_numvfs_path.read_text().strip() if self.sriov_numvfs_path.is_file() else None

set_numvfs(num='1')

Set number of Virtual Functions.

Resets existing VFs before allocating the new count.

Source code in sts_libs/src/sts/rdma.py 
155
156
157
158
159
160
161
162
163
def set_numvfs(self, num: str = '1') -> None:
    """Set number of Virtual Functions.

    Resets existing VFs before allocating the new count.
    """
    if self.sriov_numvfs and num != self.sriov_numvfs:
        self.sriov_numvfs_path.write_text('0')  # Reset VFs
        self.sriov_numvfs_path.write_text(num)  # Allocate new VFs
        self.sriov_numvfs = self.read_numvfs()  # Update state

exists_device(ibdev)

Check whether specific RDMA device exists.

Parameters:

Name Type Description Default
ibdev str

RDMA device ID (e.g., 'mlx5_0')

 required
Source code in sts_libs/src/sts/rdma.py 
25
26
27
28
29
30
31
def exists_device(ibdev: str) -> bool:
    """Check whether specific RDMA device exists.

    Args:
        ibdev: RDMA device ID (e.g., 'mlx5_0')
    """
    return Path(f'{RDMA_SYSFS_BASE}{ibdev}').is_dir()

exists_rdma()

Check whether system has RDMA devices.

Source code in sts_libs/src/sts/rdma.py 
20
21
22
def exists_rdma() -> bool:
    """Check whether system has RDMA devices."""
    return Path(RDMA_SYSFS_BASE).is_dir()