tests/base.py - master - githubmirror/rdma-core

tests/base.py
921 строка35 KB
Mazal Or Yakuel
tests: Fix process exit code checking
11 янв 2026, 11:56
11 янв 2026, 11:5692e7555
# SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB)
# Copyright (c) 2019 Mellanox Technologies, Inc . All rights reserved. See COPYING file

import multiprocessing as mp
import subprocess
import functools
import unittest
import tempfile
import random
import errno
import stat
import json
import sys
import os

from pyverbs.qp import QPCap, QPInitAttrEx, QPInitAttr, QPAttr, QP
from pyverbs.srq import SRQ, SrqInitAttrEx, SrqInitAttr, SrqAttr
from pyverbs.pyverbs_error import PyverbsRDMAError, PyverbsError
from pyverbs.addr import AHAttr, GlobalRoute
from pyverbs.xrcd import XRCD, XRCDInitAttr
from pyverbs.device import Context
from args_parser import parser
from pyverbs.librdmacm_enums import rdma_port_space
import pyverbs.device as d
from pyverbs.libibverbs_enums import ibv_mtu, ibv_port_state, ibv_gid_type_sysfs, ibv_access_flags,\
                                     ibv_qp_type, ibv_qp_init_attr_mask, ibv_xrcd_init_attr_mask,\
                                     ibv_srq_type, ibv_srq_init_attr_mask, IBV_LINK_LAYER_ETHERNET
from pyverbs.pd import PD
from pyverbs.cq import CQ
from pyverbs.mr import MR


PATH_MTU = ibv_mtu.IBV_MTU_1024
MAX_DEST_RD_ATOMIC = 1
NUM_OF_PROCESSES = 2
MC_IP_PREFIX = '230'
MAX_RDMA_ATOMIC = 20
MAX_RD_ATOMIC = 1
MIN_RNR_TIMER =12
RETRY_CNT = 7
RNR_RETRY = 7
TIMEOUT = 14
# Devices that don't support RoCEv2 should be added here
MLNX_VENDOR_ID = 0x02c9
CX3_MLNX_PART_ID = 4099
CX3Pro_MLNX_PART_ID = 4103
DCT_KEY = 0xbadc0de
# Dictionary: vendor_id -> array of part_ids of devices that lack RoCEv2 support
ROCEV2_UNSUPPORTED_DEVS = {MLNX_VENDOR_ID: [CX3Pro_MLNX_PART_ID,
                                            CX3_MLNX_PART_ID]}


def has_roce_hw_bug(vendor_id, vendor_part_id):
    return vendor_part_id in ROCEV2_UNSUPPORTED_DEVS.get(vendor_id, [])


def set_rnr_attributes(qp_attr):
    """
    Set default QP RNR attributes.
    :param qp_attr: The QPAttr to set its attributes
    :return: None
    """
    qp_attr.min_rnr_timer = MIN_RNR_TIMER
    qp_attr.timeout = TIMEOUT
    qp_attr.retry_cnt = RETRY_CNT
    qp_attr.rnr_retry = RNR_RETRY


def is_gid_available(gid_index):
    if gid_index is None:
        raise unittest.SkipTest(f'No relevant GID found')


class PyverbsAPITestCase(unittest.TestCase):
    def __init__(self, methodName='runTest'):
        super().__init__(methodName)
        # Hold the command line arguments
        self.config = parser.get_config()
        self.dev_name = None
        self.ctx = None
        self.attr = None
        self.attr_ex = None
        self.gid_index = 0
        self.pre_environment = {}

    def setUp(self):
        """
        Opens the device and queries it.
        The results of the query and query_ex are stored in attr and attr_ex
        instance attributes respectively.
        If the user didn't pass a device name, the first device is chosen by
        default.
        """
        self.ib_port = self.config['port']
        self.dev_name = self.config['dev']
        if not self.dev_name:
            dev_list = d.get_device_list()
            if not dev_list:
                raise unittest.SkipTest('No IB devices found')
            self.dev_name = dev_list[0].name.decode()

        if self.config['gid']:
            self.gid_index = self.config['gid']

        self.create_context()
        self.attr = self.ctx.query_device()
        self.attr_ex = self.ctx.query_device_ex()

    def create_context(self):
        self.ctx = d.Context(name=self.dev_name)

    def set_env_variable(self, var, value):
        """
        Set environment variable. The current value for each variable is stored
        and is set back at the end of the test.
        :param var: The name of the environment variable
        :param value: The requested new value of this environment variable
        """
        if var not in self.pre_environment.keys():
            self.pre_environment[var] = os.environ.get(var)
        os.environ[var] = value

    def tearDown(self):
        for k, v in self.pre_environment.items():
            if v is None:
                os.environ.pop(k)
            else:
                os.environ[k] = v
        self.ctx.close()


class RDMATestCase(unittest.TestCase):
    ZERO_GID = '0000:0000:0000:0000'

    def __init__(self, methodName='runTest', dev_name=None, ib_port=None,
                 gid_index=None, pkey_index=None, gid_type=None):
        """
        Initialize a RDMA test unit based on unittest.TestCase.
        If no device was provided, it iterates over the existing devices, for
        each port of each device, it checks which GID indexes are valid (in RoCE,
        only IPv4 and IPv6 based GIDs are used). Each <dev, port, gid> is added
        to an array and one entry is selected.
        If a device was provided, the same process is done for all ports of this
        device (in case they're not provided), and so on.
        If gid_type is provided by the user, only GIDs of that type would be
        be chosen (valid only if gid_index was not provided).
        :param methodName: The base method to be used by the unittest
        :param dev_name: Device name to use
        :param ib_port: IB port of the device to use
        :param gid_index: GID index to use
        :param pkey_index: PKEY index to use
        :param gid_type: If provided, only GIDs of gid_type will be chosen
                         (ignored if gid_index is provided by the user)
        """
        super(RDMATestCase, self).__init__(methodName)
        # Hold the command line arguments
        self.config = parser.get_config()
        dev = self.config['dev']
        self.dev_name = dev_name if dev_name else dev
        self.ib_port = ib_port if ib_port else self.config['port']
        self.gid_index = gid_index if gid_index else self.config['gid']
        self.pkey_index = pkey_index
        self.gid_type = gid_type if gid_index is None else None
        self.ip_addr = None
        self.mac_addr = None
        self.pre_environment = {}
        self.server = None
        self.client = None
        self.iters = 10


    def is_eth_and_has_roce_hw_bug(self):
        """
        Check if the link layer is Ethernet and the device lacks RoCEv2 support
        with a known HW bug.
        return: True if the link layer is Ethernet and device is not supported
        """
        ctx = d.Context(name=self.dev_name)
        port_attrs = ctx.query_port(self.ib_port)
        dev_attrs = ctx.query_device()
        vendor_id = dev_attrs.vendor_id
        vendor_pid = dev_attrs.vendor_part_id
        return port_attrs.link_layer == IBV_LINK_LAYER_ETHERNET and \
            has_roce_hw_bug(vendor_id, vendor_pid)

    @staticmethod
    def get_net_name(dev, port=None):
        if port is not None:
            out = subprocess.check_output(['rdma', 'link',  'show', '-j'])
            loaded_json = json.loads(out.decode())
            for row in loaded_json:
                try:
                    if row['ifname'] == dev and row['port'] == port:
                        return row['netdev']
                except KeyError:
                    pass

        if not os.path.exists(f'/sys/class/infiniband/{dev}/device/net/'):
            return None

        out = subprocess.check_output(['ls', f'/sys/class/infiniband/{dev}/device/net/'])
        return out.decode().split('\n')[0]

    @staticmethod
    def get_ip_mac_address(ifname):
        out = subprocess.check_output(['ip', '-j', 'addr', 'show', ifname])
        loaded_json = json.loads(out.decode())
        interface = loaded_json[0]['addr_info'][0]['local']
        mac = loaded_json[0]['address']
        if 'fe80::' in interface:
            interface = interface + '%' + ifname
        return interface, mac

    def setUp(self):
        """
        Verify that the test case has dev_name, ib_port, gid_index and pkey index.
        If not provided by the user, the first valid combination will be used.
        """
        if self.pkey_index is None:
            # To avoid iterating the entire pkeys table, if a pkey index wasn't
            # provided, use index 0 which is always valid
            self.pkey_index = 0

        self.args = []
        if self.dev_name is not None:
            ctx = d.Context(name=self.dev_name)
            if self.ib_port is not None:
                if self.gid_index is not None:
                    self._get_ip_mac(self.dev_name, self.ib_port, self.gid_index)
                else:
                    # Add avaiable GIDs of the given dev_name + port
                    self._add_gids_per_port(ctx, self.dev_name, self.ib_port)
            else:
                # Add available GIDs for each port of the given dev_name
                self._add_gids_per_device(ctx, self.dev_name)
        else:
            # Iterate available devices, add available GIDs for each of
            # their ports
            lst = d.get_device_list()
            for dev in lst:
                dev_name = dev.name.decode()
                ctx = d.Context(name=dev_name)
                self._add_gids_per_device(ctx, dev_name)

        if not self.args:
            raise unittest.SkipTest('No supported port is up, can\'t run traffic')
        # Choose one combination and use it
        self._select_config()
        self.dev_info = {'dev_name': self.dev_name, 'ib_port': self.ib_port,
                         'gid_index': self.gid_index}

    def _add_gids_per_port(self, ctx, dev, port):
        # Don't add ports which are not active
        port_attrs = ctx.query_port(port)
        if port_attrs.state != ibv_port_state.IBV_PORT_ACTIVE:
            return
        if not port_attrs.gid_tbl_len:
            self._get_ip_mac(dev, port, None)
            return
        dev_attrs = ctx.query_device()
        vendor_id = dev_attrs.vendor_id
        vendor_pid = dev_attrs.vendor_part_id
        for idx in range(port_attrs.gid_tbl_len):
            gid = ctx.query_gid(port, idx)
            # Avoid adding ZERO GIDs
            if gid.gid[-19:] == self.ZERO_GID:
                continue
            # Avoid RoCEv2 GIDs on unsupported devices
            if port_attrs.link_layer == IBV_LINK_LAYER_ETHERNET and \
                    ctx.query_gid_type(port, idx) == \
                    ibv_gid_type_sysfs.IBV_GID_TYPE_SYSFS_ROCE_V2 and \
                    has_roce_hw_bug(vendor_id, vendor_pid):
                continue
            if self.gid_type is not None and ctx.query_gid_type(port, idx) != \
                    self.gid_type:
                continue
            self._get_ip_mac(dev, port, idx)

    def _add_gids_per_device(self, ctx, dev):
        self._add_gids_per_port(ctx, dev, self.ib_port)

    def _get_ip_mac(self, dev, port, idx):
        net_name = self.get_net_name(dev, port)
        if net_name is None:
            self.args.append([dev, port, idx, None, None])
            return
        try:
            ip_addr, mac_addr = self.get_ip_mac_address(net_name)
        except (KeyError, IndexError):
            self.args.append([dev, port, idx, None, None])
        else:
            self.args.append([dev, port, idx, ip_addr, mac_addr])

    def _select_config(self):
        args_with_inet_ip = []
        for arg in self.args:
            if arg[3]:
                args_with_inet_ip.append(arg)
        if args_with_inet_ip:
            args = args_with_inet_ip[0]
        else:
            args = self.args[0]
        self.dev_name = args[0]
        self.ib_port = args[1]
        self.gid_index = args[2]
        self.ip_addr = args[3]
        self.mac_addr = args[4]

    def set_env_variable(self, var, value):
        """
        Set environment variable. The current value for each variable is stored
        and is set back at the end of the test.
        :param var: The name of the environment variable
        :param value: The requested new value of this environment variable
        """
        if var not in self.pre_environment.keys():
            self.pre_environment[var] = os.environ.get(var)
        os.environ[var] = value

    def sync_remote_attr(self):
        """
        Sync the MR remote attributes between the server and the client.
        """
        self.server.rkey = self.client.mr.rkey
        self.server.raddr = self.client.mr.buf
        self.client.rkey = self.server.mr.rkey
        self.client.raddr = self.server.mr.buf

    def pre_run(self):
        """
        Configure Resources before running traffic.
        pre_run() must be implemented by the client and server.
        """
        self.client.pre_run(self.server.psns, self.server.qps_num)
        self.server.pre_run(self.client.psns, self.client.qps_num)

    def create_players(self, resource, sync_attrs=True, **resource_arg):
        """
        Init test resources.
        :param resource: The RDMA resources to use.
        :param sync_attrs: If True, sync remote attrs such as rkey and raddr
        :param resource_arg: Dict of args that specify the resource specific
                             attributes.
        """
        try:
            self.client = resource(**self.dev_info, **resource_arg)
            self.server = resource(**self.dev_info, **resource_arg)
        except PyverbsRDMAError as ex:
            if ex.error_code == errno.EOPNOTSUPP:
                raise unittest.SkipTest(f'Create player of {resource.__name__} is not supported')
            raise ex
        self.pre_run()
        if sync_attrs:
            self.sync_remote_attr()
        self.traffic_args = {'client': self.client, 'server': self.server,
                             'iters': self.iters, 'gid_idx': self.gid_index,
                             'port': self.ib_port}

    def tearDown(self):
        """
        Restore the previous environment variables values before ending the test.
        """
        for k, v in self.pre_environment.items():
            if v is None:
                os.environ.pop(k)
            else:
                os.environ[k] = v
        if self.server:
            self.server.ctx.close()
        if self.client:
            self.client.ctx.close()
        super().tearDown()


class RDMACMBaseTest(RDMATestCase):
    """
    Base RDMACM test class.
    This class does not include any test, but rather implements generic
    connection and traffic methods that are needed by RDMACM tests in general.
    Each RDMACM test should have a class that inherits this class and extends
    its functionalities if needed.
    """
    def setUp(self):
        super().setUp()
        if not self.ip_addr:
            raise unittest.SkipTest('Device {} doesn\'t have net interface'
                                    .format(self.dev_name))
        is_gid_available(self.gid_index)

    def two_nodes_rdmacm_traffic(self, connection_resources, test_flow, bad_flow=False,
                                 **resource_kwargs):
        """
        Init and manage the rdmacm test processes. The exit code of the
        test processes indicates if exception was thrown.
        {0: pass, 2: exception was thrown, 5: skip test}
        If needed, terminate those processes and raise an exception.
        :param connection_resources: The CMConnection resources to use.
        :param test_flow: The target RDMACM flow method to run.
        :param bad_flow: If true, traffic is expected to fail.
        :param resource_kwargs: Dict of args that specify the CMResources
                                specific attributes. Each test case can pass
                                here as key words the specific CMResources
                                attributes that are requested.
        :return: None
        """
        if resource_kwargs.get('port_space', None) == rdma_port_space.RDMA_PS_UDP and \
            self.is_eth_and_has_roce_hw_bug():
            raise unittest.SkipTest('Device {} doesn\'t support UDP with RoCEv2'
                                    .format(self.dev_name))
        ctx = mp.get_context('fork')
        self.syncer = ctx.Barrier(NUM_OF_PROCESSES, timeout=15)
        self.notifier = ctx.Queue()
        passive = ctx.Process(target=test_flow,
                              kwargs={'connection_resources': connection_resources,
                                      'passive':True, **resource_kwargs})
        active = ctx.Process(target=test_flow,
                              kwargs={'connection_resources': connection_resources,
                                      'passive':False, **resource_kwargs})
        passive.start()
        active.start()
        repeat_times=150 if not bad_flow else 3
        proc_res = {}
        for _ in range(repeat_times):
            for proc in [passive, active]:
                proc.join(0.1)
                # Write the exit code of the proc.
                if not proc.is_alive():
                    side = 'passive' if proc == passive else 'active'
                    if side not in proc_res.keys():
                        proc_res[side] = proc.exitcode
        # If the processes is still alive kill them and fail the test.
        proc_killed = False
        for proc in [passive, active]:
            if proc.is_alive():
                proc.terminate()
                proc_killed = True
        # Check if need to skip this test
        for side in proc_res.keys():
            if proc_res[side] == 5:
                raise unittest.SkipTest(f'SkipTest occurred on {side} side')
       # Raise exeption if the test proceses was terminate.
        if bad_flow and not proc_killed:
            raise Exception('Bad flow: traffic passed which is not expected')
        if not bad_flow and proc_killed:
            raise Exception('RDMA CM test procces is stuck, kill the test')

        # Check if the test processes raise exceptions.
        for side in proc_res:
            if proc_res[side] != 0:
                raise Exception(f'Exception in {side} side occurred')

    def rdmacm_traffic(self, connection_resources=None, passive=None, **kwargs):
        """
        Run RDMACM traffic between two CMIDs.
        :param connection_resources: The connection resources to use.
        :param passive: Indicate if this CMID is this the passive side.
        :param kwargs: Arguments to be passed to the connection_resources.
        :return: None
        """
        try:
            player = connection_resources(ip_addr=self.ip_addr,
                                          syncer=self.syncer,
                                          notifier=self.notifier,
                                          passive=passive, **kwargs)
            player.establish_connection()
            if kwargs.get('reject_conn'):
                return
            player.rdmacm_traffic()
            player.disconnect()
        except Exception as ex:
            self._rdmacm_exception_handler(passive, ex)

    def rdmacm_multicast_traffic(self, connection_resources=None, passive=None,
                                 extended=False, leave_test=False, **kwargs):
        """
        Run RDMACM multicast traffic between two CMIDs.
        :param connection_resources: The connection resources to use.
        :param passive: Indicate if this CMID is the passive side.
        :param extended: Use exteneded multicast join request. This request
                         allows CMID to join with specific join flags.
        :param leave_test: Perform traffic after leaving the multicast group to
                           ensure leave works.
        :param kwargs: Arguments to be passed to the connection_resources.
        :return: None
        """
        try:
            player = connection_resources(ip_addr=self.ip_addr, syncer=self.syncer,
                                          notifier=self.notifier, passive=False,
                                          **kwargs)
            mc_addr = MC_IP_PREFIX + self.ip_addr[self.ip_addr.find('.'):]
            player.join_to_multicast(src_addr=self.ip_addr, mc_addr=mc_addr,
                                     extended=extended)
            player.rdmacm_traffic(server=passive, multicast=True)
            player.leave_multicast(mc_addr=mc_addr)
            if leave_test:
                player.rdmacm_traffic(server=passive, multicast=True)
        except Exception as ex:
            self._rdmacm_exception_handler(passive, ex)

    def rdmacm_remote_traffic(self, connection_resources=None, passive=None,
                              remote_op='write', **kwargs):
        """
        Run RDMACM remote traffic between two CMIDs.
        :param connection_resources: The connection resources to use.
        :param passive: Indicate if this CMID is the passive side.
        :param remote_op: The remote operation in the traffic.
        :param kwargs: Arguments to be passed to the connection_resources.
        :return: None
        """
        try:
            player = connection_resources(ip_addr=self.ip_addr,
                                          syncer=self.syncer,
                                          notifier=self.notifier,
                                          passive=passive,
                                          remote_op=remote_op, **kwargs)
            player.establish_connection()
            player.remote_traffic(passive=passive, remote_op=remote_op)
            player.disconnect()
        except Exception as ex:
            self._rdmacm_exception_handler(passive, ex)

    @staticmethod
    def _rdmacm_exception_handler(passive, exception):
        if isinstance(exception, PyverbsRDMAError):
            if exception.error_code in [errno.EOPNOTSUPP, errno.EPROTONOSUPPORT]:
                sys.exit(5)
        if isinstance(exception, unittest.case.SkipTest):
            sys.exit(5)
        side = 'passive' if passive else 'active'
        print(f'Player {side} got: {exception}')
        sys.exit(2)


def catch_skiptest(func):
    """
    Decorator to catch unittest.SkipTest in __init__ resource functions.
    It gracefully closes the context and all of its underlying resources.
    """
    @functools.wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            func(self, *args, **kwargs)
        except unittest.SkipTest as e:
            if hasattr(self, 'ctx') and self.ctx:
               self.ctx.close()
            raise e
    return wrapper


class SkipTestMeta(type):
    """
    Metaclass to automatically wrap __init__ in catch_skiptest.
    It should only be used in resource classes, such as those inheriting from
    BaseResources.
    """
    def __new__(cls, name, bases, dct):
        if "__init__" in dct:
            dct["__init__"] = catch_skiptest(dct["__init__"])
        return super().__new__(cls, name, bases, dct)


class BaseResources(object, metaclass=SkipTestMeta):
    """
    BaseResources class is a base aggregator object which contains basic
    resources like Context and PD. It opens a context over the given device
    and port and allocates a PD.
    """
    def __init__(self, dev_name, ib_port, gid_index):
        """
        Initializes a BaseResources object.
        :param dev_name: Device name to be used (default: 'ibp0s8f0')
        :param ib_port: IB port of the device to use (default: 1)
        :param gid_index: Which GID index to use (default: 0)
        """
        self.dev_name = dev_name
        self.gid_index = gid_index
        self.ib_port = ib_port
        self.ctx = None
        self.create_context()
        self.create_pd()

    def create_context(self):
        self.ctx = Context(name=self.dev_name)

    def create_pd(self):
        self.pd = PD(self.ctx)

    def mem_write(self, data, size, offset=0):
        self.mr.write(data, size, offset)

    def mem_read(self, size=None, offset=0):
        size_ = self.msg_size if size is None else size
        return self.mr.read(size_, offset)


class TrafficResources(BaseResources):
    """
    Basic traffic class. It provides the basic RDMA resources and operations
    needed for traffic.
    """
    def __init__(self, dev_name, ib_port, gid_index, with_srq=False,
                 qp_count=1, msg_size=1024):
        """
        Initializes a TrafficResources object with the given values and creates
        basic RDMA resources.
        :param dev_name: Device name to be used
        :param ib_port: IB port of the device to use
        :param gid_index: Which GID index to use
        :param with_srq: If True, create SRQ and attach to QPs
        :param qp_count: Number of QPs to create
        :param msg_size: Size of resource msg. If None, use 1024 as default.
        """
        super(TrafficResources, self).__init__(dev_name=dev_name,
                                               ib_port=ib_port,
                                               gid_index=gid_index)
        self.msg_size = msg_size
        self.num_msgs = 1000
        self.port_attr = None
        self.mr = None
        self.use_mr_prefetch = None
        self.srq = None
        self.cq = None
        self.qps = []
        self.qps_num = []
        self.psns = []
        self.rqps_num = None
        self.rpsns = None
        self.with_srq = with_srq
        self.qp_count = qp_count
        self.init_resources()

    @property
    def qp(self):
        return self.qps[0]

    @property
    def mr_lkey(self):
        if self.mr:
            return self.mr.lkey

    def init_resources(self):
        """
        Initializes a CQ, MR and an RC QP.
        :return: None
        """
        self.port_attr = self.ctx.query_port(self.ib_port)
        self.create_cq()
        if self.with_srq:
            self.create_srq()
        self.create_mr()
        self.create_qps()

    def create_cq(self):
        """
        Initializes self.cq with a CQ of depth <num_msgs> - defined by each
        test.
        :return: None
        """
        self.cq = CQ(self.ctx, self.num_msgs, None, None, 0)

    def create_mr(self):
        """
        Initializes self.mr with an MR of length <msg_size> - defined by each
        test.
        :return: None
        """
        self.mr = MR(self.pd, self.msg_size, ibv_access_flags.IBV_ACCESS_LOCAL_WRITE)

    def create_qp_cap(self):
        return QPCap(max_recv_wr=self.num_msgs)

    def create_qp_init_attr(self):
        return QPInitAttr(qp_type=ibv_qp_type.IBV_QPT_RC, scq=self.cq, rcq=self.cq,
                          srq=self.srq, cap=self.create_qp_cap())

    def create_qp_attr(self):
        return QPAttr(port_num=self.ib_port)

    def create_qps(self):
        """
        Initializes self.qps with RC QPs.
        :return: None
        """
        qp_init_attr = self.create_qp_init_attr()
        qp_attr = self.create_qp_attr()
        for _ in range(self.qp_count):
            try:
                qp = QP(self.pd, qp_init_attr, qp_attr)
                self.qps.append(qp)
                self.qps_num.append(qp.qp_num)
                self.psns.append(random.getrandbits(24))
            except PyverbsRDMAError as ex:
                if ex.error_code == errno.EOPNOTSUPP:
                    raise unittest.SkipTest(f'Create QP type {qp_init_attr.qp_type} is not supported')
                raise ex

    def create_srq_attr(self):
        return SrqAttr(max_wr=self.num_msgs*self.qp_count)

    def create_srq_init_attr(self):
        return SrqInitAttr(self.create_srq_attr())

    def create_srq(self):
        srq_init_attr = self.create_srq_init_attr()
        try:
            self.srq = SRQ(self.pd, srq_init_attr)
        except PyverbsRDMAError as ex:
            if ex.error_code == errno.EOPNOTSUPP:
                raise unittest.SkipTest('Create SRQ is not supported')
            raise ex

    def pre_run(self, rpsns, rqps_num):
        """
        Configure resources before running traffic and modifies the QP to RTS
        if required.
        :param rpsns: Remote PSNs (packet serial numbers)
        :param rqps_num: Remote QPs Number
        """
        self.rpsns = rpsns
        self.rqps_num = rqps_num
        self.to_rts()

    def to_rts(self):
        """
        Modify the QP's states to RTS and initialize it to be ready for traffic.
        If not required, can be "passed" but must be implemented.
        """
        raise NotImplementedError()


class RoCETrafficResources(TrafficResources):
    def __init__(self, dev_name, ib_port, gid_index, **kwargs):
        is_gid_available(gid_index)
        super(RoCETrafficResources, self).__init__(dev_name, ib_port, gid_index, **kwargs)


class RCResources(RoCETrafficResources):
    def __init__(self, dev_name, ib_port, gid_index,
                 max_dest_rd_atomic=MAX_DEST_RD_ATOMIC, max_rd_atomic=MAX_RD_ATOMIC, **kwargs):
        """
        Initializes a TrafficResources object with the given values and creates
        basic RDMA resources.
        :param dev_name: Device name to be used
        :param ib_port: IB port of the device to use
        :param gid_index: Which GID index to use
        :param max_dest_rd_atomic: Number of responder incoming rd_atomic operations that can
                                   be handled
        :param max_rd_atomic: Number of outstanding destination rd_atomic operations
        """
        self.max_dest_rd_atomic = max_dest_rd_atomic
        self.max_rd_atomic = max_rd_atomic
        super().__init__(dev_name, ib_port, gid_index, **kwargs)

    def to_rts(self):
        """
        Set the QP attributes' values to arbitrary values (same values used in
        ibv_rc_pingpong).
        :return: None
        """
        attr = self.create_qp_attr()
        attr.path_mtu = PATH_MTU
        attr.max_dest_rd_atomic = self.max_dest_rd_atomic
        set_rnr_attributes(attr)
        attr.max_rd_atomic = self.max_rd_atomic
        gr = GlobalRoute(dgid=self.ctx.query_gid(self.ib_port, self.gid_index),
                         sgid_index=self.gid_index)
        ah_attr = AHAttr(port_num=self.ib_port, is_global=1, gr=gr,
                         dlid=self.port_attr.lid)
        attr.ah_attr = ah_attr
        for i in range(self.qp_count):
            attr.dest_qp_num = self.rqps_num[i]
            attr.rq_psn = self.psns[i]
            attr.sq_psn = self.rpsns[i]
            self.qps[i].to_rts(attr)


class UDResources(RoCETrafficResources):
    UD_QKEY = 0x11111111
    UD_PKEY_INDEX = 0
    GRH_SIZE = 40

    def create_mr(self):
        self.mr = MR(self.pd, self.msg_size + self.GRH_SIZE,
                     ibv_access_flags.IBV_ACCESS_LOCAL_WRITE)

    def create_qp_init_attr(self):
        return QPInitAttr(qp_type=ibv_qp_type.IBV_QPT_UD, scq=self.cq,
                          rcq=self.cq, srq=self.srq, cap=self.create_qp_cap())

    def create_qps(self):
        qp_init_attr = self.create_qp_init_attr()
        qp_attr = self.create_qp_attr()
        qp_attr.qkey = self.UD_QKEY
        qp_attr.pkey_index = self.UD_PKEY_INDEX
        for _ in range(self.qp_count):
            try:
                qp = QP(self.pd, qp_init_attr, qp_attr)
                self.qps.append(qp)
                self.qps_num.append(qp.qp_num)
                self.psns.append(random.getrandbits(24))
            except PyverbsRDMAError as ex:
                if ex.error_code == errno.EOPNOTSUPP:
                    raise unittest.SkipTest(f'Create QP type {qp_init_attr.qp_type} is not supported')
                raise ex

    def to_rts(self):
        pass


class RawResources(TrafficResources):
    def create_qp_init_attr(self):
        return QPInitAttr(qp_type=ibv_qp_type.IBV_QPT_RAW_PACKET, scq=self.cq,
                          rcq=self.cq, srq=self.srq, cap=self.create_qp_cap())

    def pre_run(self, rpsns=None, rqps_num=None):
        pass


class XRCResources(RoCETrafficResources):
    def __init__(self, dev_name, ib_port, gid_index, qp_count=2):
        self.temp_file = None
        self.xrcd_fd = -1
        self.xrcd = None
        self.sqp_lst = []
        self.rqp_lst = []
        super(XRCResources, self).__init__(dev_name, ib_port, gid_index,
                                           qp_count=qp_count)

    def close(self):
        os.close(self.xrcd_fd)
        self.temp_file.close()

    @property
    def qp(self):
        return self.sqp_lst[0]

    def create_qps(self):
        """
        Initializes self.qps with an XRC SEND/RECV QPs.
        :return: None
        """
        qp_attr = QPAttr(port_num=self.ib_port)
        qp_attr.pkey_index = 0

        for _ in range(self.qp_count):
            attr_ex = QPInitAttrEx(qp_type=ibv_qp_type.IBV_QPT_XRC_RECV,
                                   comp_mask=ibv_qp_init_attr_mask.IBV_QP_INIT_ATTR_XRCD,
                                   xrcd=self.xrcd)
            qp_attr.qp_access_flags = ibv_access_flags.IBV_ACCESS_LOCAL_WRITE | \
                                      ibv_access_flags.IBV_ACCESS_REMOTE_READ | \
                                      ibv_access_flags.IBV_ACCESS_REMOTE_WRITE | \
                                      ibv_access_flags.IBV_ACCESS_REMOTE_ATOMIC
            recv_qp = QP(self.ctx, attr_ex, qp_attr)
            self.rqp_lst.append(recv_qp)

            qp_caps = QPCap(max_send_wr=self.num_msgs, max_recv_sge=0,
                            max_recv_wr=0)
            attr_ex = QPInitAttrEx(qp_type=ibv_qp_type.IBV_QPT_XRC_SEND, sq_sig_all=1,
                                   comp_mask=ibv_qp_init_attr_mask.IBV_QP_INIT_ATTR_PD,
                                   pd=self.pd, scq=self.cq, cap=qp_caps)
            qp_attr.qp_access_flags = 0
            send_qp =QP(self.ctx, attr_ex, qp_attr)
            self.sqp_lst.append(send_qp)
            self.qps_num.append((recv_qp.qp_num, send_qp.qp_num))
            self.psns.append(random.getrandbits(24))

    def create_xrcd(self):
        """
        Initializes self.xrcd with an XRC Domain object.
        :return: None
        """
        self.temp_file = tempfile.NamedTemporaryFile()
        self.xrcd_fd = os.open(self.temp_file.name, os.O_RDONLY | os.O_CREAT,
                               stat.S_IRUSR | stat.S_IRGRP)
        init = XRCDInitAttr(
            ibv_xrcd_init_attr_mask.IBV_XRCD_INIT_ATTR_FD | ibv_xrcd_init_attr_mask.IBV_XRCD_INIT_ATTR_OFLAGS,
            os.O_CREAT, self.xrcd_fd)
        try:
            self.xrcd = XRCD(self.ctx, init)
        except PyverbsRDMAError as ex:
            if ex.error_code == errno.EOPNOTSUPP:
                raise unittest.SkipTest('Create XRCD is not supported')
            raise ex

    def create_srq(self):
        """
        Initializes self.srq with a Shared Receive QP object.
        :return: None
        """
        srq_attr = SrqInitAttrEx(max_wr=self.qp_count*self.num_msgs)
        srq_attr.srq_type = ibv_srq_type.IBV_SRQT_XRC
        srq_attr.pd = self.pd
        srq_attr.xrcd = self.xrcd
        srq_attr.cq = self.cq
        srq_attr.comp_mask = ibv_srq_init_attr_mask.IBV_SRQ_INIT_ATTR_TYPE | ibv_srq_init_attr_mask.IBV_SRQ_INIT_ATTR_PD | \
                             ibv_srq_init_attr_mask.IBV_SRQ_INIT_ATTR_CQ | ibv_srq_init_attr_mask.IBV_SRQ_INIT_ATTR_XRCD
        self.srq = SRQ(self.ctx, srq_attr)

    def to_rts(self):
        gid = self.ctx.query_gid(self.ib_port, self.gid_index)
        gr = GlobalRoute(dgid=gid, sgid_index=self.gid_index)
        ah_attr = AHAttr(port_num=self.ib_port, is_global=True,
                         gr=gr, dlid=self.port_attr.lid)
        qp_attr = QPAttr()
        qp_attr.max_rd_atomic = MAX_RD_ATOMIC
        qp_attr.max_dest_rd_atomic = MAX_DEST_RD_ATOMIC
        qp_attr.path_mtu = PATH_MTU
        set_rnr_attributes(qp_attr)
        qp_attr.ah_attr = ah_attr
        for i in range(self.qp_count):
            qp_attr.dest_qp_num = self.rqps_num[i][1]
            qp_attr.rq_psn = self.psns[i]
            qp_attr.sq_psn = self.rpsns[i]
            self.rqp_lst[i].to_rts(qp_attr)
            qp_attr.dest_qp_num = self.rqps_num[i][0]
            self.sqp_lst[i].to_rts(qp_attr)

    def init_resources(self):
        self.create_xrcd()
        super(XRCResources, self).init_resources()
        self.create_srq()