# 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 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 import pyverbs.cm_enums as ce import pyverbs.device as d import pyverbs.enums as e from pyverbs.pd import PD from pyverbs.cq import CQ from pyverbs.mr import MR PATH_MTU = e.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 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 == e.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 != e.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 == e.IBV_LINK_LAYER_ETHERNET and \ ctx.query_gid_type(port, idx) == \ e.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) == ce.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') # Check if the test processes raise exceptions. res_exception = False for side in proc_res: if 0 < proc_res[side] < 5: res_exception = True if res_exception: raise Exception('Exception in active/passive side occurred') # 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') 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, unittest.case.SkipTest): sys.exit(5) side = 'passive' if passive else 'active' print(f'Player {side} got: {exception}') sys.exit(2) class BaseResources(object): """ 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.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 - 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 - defined by each test. :return: None """ self.mr = MR(self.pd, self.msg_size, e.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=e.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 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 = MAX_DEST_RD_ATOMIC set_rnr_attributes(attr) attr.max_rd_atomic = 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, e.IBV_ACCESS_LOCAL_WRITE) def create_qp_init_attr(self): return QPInitAttr(qp_type=e.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=e.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=e.IBV_QPT_XRC_RECV, comp_mask=e.IBV_QP_INIT_ATTR_XRCD, xrcd=self.xrcd) qp_attr.qp_access_flags = e.IBV_ACCESS_LOCAL_WRITE | \ e.IBV_ACCESS_REMOTE_READ | \ e.IBV_ACCESS_REMOTE_WRITE | \ e.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=e.IBV_QPT_XRC_SEND, sq_sig_all=1, comp_mask=e.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( e.IBV_XRCD_INIT_ATTR_FD | e.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 = e.IBV_SRQT_XRC srq_attr.pd = self.pd srq_attr.xrcd = self.xrcd srq_attr.cq = self.cq srq_attr.comp_mask = e.IBV_SRQ_INIT_ATTR_TYPE | e.IBV_SRQ_INIT_ATTR_PD | \ e.IBV_SRQ_INIT_ATTR_CQ | e.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()