| # SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB) |
| # Copyright (c) 2019 Mellanox Technologies, Inc. All rights reserved. See COPYING file |
| """ |
| Provide some useful helper function for pyverbs rdmacm' tests. |
| """ |
| import sys |
| from tests.utils import validate, poll_cq, get_send_elements, get_recv_wr |
| from tests.base_rdmacm import AsyncCMResources, SyncCMResources |
| from pyverbs.cmid import CMEvent, AddrInfo, JoinMCAttrEx |
| from pyverbs.pyverbs_error import PyverbsError, PyverbsRDMAError |
| import pyverbs.cm_enums as ce |
| from pyverbs.addr import AH |
| import pyverbs.enums as e |
| import abc |
| import errno |
| |
| GRH_SIZE = 40 |
| MULTICAST_QPN = 0xffffff |
| REJECT_MSG = 'connection rejected' |
| |
| |
| class CMConnection(abc.ABC): |
| """ |
| RDMA CM base abstract connection class. The class contains the rdmacm |
| resources and other methods to easily establish a connection and run |
| traffic using the rdmacm resources. Each type of connection or traffic |
| should inherit from this class and implement the necessary methods such as |
| connection establishment and traffic. |
| """ |
| def __init__(self, syncer=None, notifier=None): |
| """ |
| Initializes a connection object. |
| :param syncer: Barrier object to sync between all the test processes. |
| :param notifier: Queue object to pass objects between the connection |
| sides. |
| """ |
| self.syncer = syncer |
| self.notifier = notifier |
| self.cm_res = None |
| |
| def rdmacm_traffic(self, server=None, multicast=False): |
| """ |
| Run rdmacm traffic. This method runs the compatible traffic flow |
| depending on the CMResources. If self.with_ext_qp is set the traffic |
| will go through the external QP. |
| :param server: Run as server. |
| :param multicast: Run multicast traffic. |
| """ |
| server = server if server is not None else self.cm_res.passive |
| if self.cm_res.with_ext_qp: |
| if server: |
| self._ext_qp_server_traffic() |
| else: |
| self._ext_qp_client_traffic() |
| elif multicast: |
| if server: |
| self._cmid_server_multicast_traffic() |
| else: |
| self._cmid_client_multicast_traffic() |
| else: |
| if server: |
| self._cmid_server_traffic() |
| else: |
| self._cmid_client_traffic() |
| |
| def remote_traffic(self, passive, remote_op='write'): |
| """ |
| Run rdmacm remote traffic. This method runs RDMA remote traffic from |
| the active to the passive. |
| :param passive: If True, run as server. |
| :param remote_op: 'write'/'read', The type of the RDMA remote operation. |
| """ |
| msg_size = self.cm_res.msg_size |
| if passive: |
| self.cm_res.mr.write((msg_size) * 's', msg_size) |
| mr_details = (self.cm_res.mr.rkey, self.cm_res.mr.buf) |
| self.notifier.put(mr_details) |
| self.syncer.wait() |
| self.syncer.wait() |
| if remote_op == 'write': |
| msg_received = self.cm_res.mr.read(msg_size, 0) |
| validate(msg_received, True, msg_size) |
| else: |
| self.cm_res.mr.write((msg_size) * 'c', msg_size) |
| self.syncer.wait() |
| rkey, remote_addr = self.notifier.get() |
| cmid = self.cm_res.cmid |
| post_func = cmid.post_write if remote_op == 'write' else \ |
| cmid.post_read |
| for _ in range(self.cm_res.num_msgs): |
| post_func(self.cm_res.mr, msg_size, remote_addr, rkey, |
| flags=e.IBV_SEND_SIGNALED) |
| cmid.get_send_comp() |
| self.syncer.wait() |
| if remote_op == 'read': |
| msg_received = self.cm_res.mr.read(msg_size, 0) |
| validate(msg_received, False, msg_size) |
| |
| def _ext_qp_server_traffic(self): |
| """ |
| RDMACM server side traffic function which sends and receives a message, |
| and then validates the received message. This traffic method uses the CM |
| external QP and CQ for send, recv and get_completion. |
| :return: None |
| """ |
| recv_wr = get_recv_wr(self.cm_res) |
| self.cm_res.qp.post_recv(recv_wr) |
| self.syncer.wait() |
| for _ in range(self.cm_res.num_msgs): |
| poll_cq(self.cm_res.cq) |
| self.cm_res.qp.post_recv(recv_wr) |
| msg_received = self.cm_res.mr.read(self.cm_res.msg_size, 0) |
| validate(msg_received, self.cm_res.passive, self.cm_res.msg_size) |
| send_wr = get_send_elements(self.cm_res, self.cm_res.passive)[0] |
| self.cm_res.qp.post_send(send_wr) |
| poll_cq(self.cm_res.cq) |
| |
| def _ext_qp_client_traffic(self): |
| """ |
| RDMACM client side traffic function which sends and receives a message, |
| and then validates the received message. This traffic method uses the CM |
| external QP and CQ for send, recv and get_completion. |
| :return: None |
| """ |
| recv_wr = get_recv_wr(self.cm_res) |
| self.syncer.wait() |
| for _ in range(self.cm_res.num_msgs): |
| send_wr = get_send_elements(self.cm_res, self.cm_res.passive)[0] |
| self.cm_res.qp.post_send(send_wr) |
| poll_cq(self.cm_res.cq) |
| self.cm_res.qp.post_recv(recv_wr) |
| poll_cq(self.cm_res.cq) |
| msg_received = self.cm_res.mr.read(self.cm_res.msg_size, 0) |
| validate(msg_received, self.cm_res.passive, self.cm_res.msg_size) |
| |
| def _cmid_server_traffic(self): |
| """ |
| RDMACM server side traffic function which sends and receives a message, |
| and then validates the received message. This traffic method uses the |
| RDMACM API for send, recv and get_completion. |
| :return: None |
| """ |
| grh_offset = GRH_SIZE if self.cm_res.qp_type == e.IBV_QPT_UD else 0 |
| send_msg = (self.cm_res.msg_size + grh_offset) * 's' |
| cmid = self.cm_res.child_id |
| for _ in range(self.cm_res.num_msgs): |
| cmid.post_recv(self.cm_res.mr) |
| self.syncer.wait() |
| self.syncer.wait() |
| wc = cmid.get_recv_comp() |
| msg_received = self.cm_res.mr.read(self.cm_res.msg_size, grh_offset) |
| validate(msg_received, True, self.cm_res.msg_size) |
| if self.cm_res.port_space == ce.RDMA_PS_TCP: |
| self.cm_res.mr.write(send_msg, self.cm_res.msg_size) |
| cmid.post_send(self.cm_res.mr) |
| else: |
| ah = AH(cmid.pd, wc=wc, port_num=self.cm_res.ib_port, grh=self.cm_res.mr.buf) |
| rqpn = self.cm_res.remote_qpn |
| self.cm_res.mr.write(send_msg, self.cm_res.msg_size + GRH_SIZE) |
| cmid.post_ud_send(self.cm_res.mr, ah, rqpn=rqpn, |
| length=self.cm_res.msg_size) |
| cmid.get_send_comp() |
| self.syncer.wait() |
| |
| def _cmid_client_traffic(self): |
| """ |
| RDMACM client side traffic function which sends and receives a message, |
| and then validates the received message. This traffic method uses the |
| RDMACM API for send, recv and get_completion. |
| :return: None |
| """ |
| grh_offset = GRH_SIZE if self.cm_res.qp_type == e.IBV_QPT_UD else 0 |
| send_msg = (self.cm_res.msg_size + grh_offset) * 'c' |
| cmid = self.cm_res.cmid |
| for _ in range(self.cm_res.num_msgs): |
| self.cm_res.mr.write(send_msg, self.cm_res.msg_size + grh_offset) |
| self.syncer.wait() |
| if self.cm_res.port_space == ce.RDMA_PS_TCP: |
| cmid.post_send(self.cm_res.mr) |
| else: |
| ah = AH(cmid.pd, attr=self.cm_res.ud_params.ah_attr) |
| cmid.post_ud_send(self.cm_res.mr, ah, rqpn=self.cm_res.ud_params.qp_num, |
| length=self.cm_res.msg_size) |
| cmid.get_send_comp() |
| cmid.post_recv(self.cm_res.mr) |
| self.syncer.wait() |
| self.syncer.wait() |
| cmid.get_recv_comp() |
| msg_received = self.cm_res.mr.read(self.cm_res.msg_size, grh_offset) |
| validate(msg_received, False, self.cm_res.msg_size) |
| |
| def _cmid_server_multicast_traffic(self): |
| """ |
| RDMACM server side multicast traffic function which receives a message, |
| and then validates its data. |
| """ |
| for _ in range(self.cm_res.num_msgs): |
| self.cm_res.cmid.post_recv(self.cm_res.mr) |
| self.syncer.wait() |
| self.syncer.wait() |
| self.cm_res.cmid.get_recv_comp() |
| msg_received = self.cm_res.mr.read(self.cm_res.msg_size, GRH_SIZE) |
| validate(msg_received, True, self.cm_res.msg_size) |
| |
| def _cmid_client_multicast_traffic(self): |
| """ |
| RDMACM client side multicast traffic function which sends a message to |
| the multicast group. |
| """ |
| send_msg = (self.cm_res.msg_size + GRH_SIZE) * 'c' |
| for _ in range(self.cm_res.num_msgs): |
| self.cm_res.mr.write(send_msg, self.cm_res.msg_size + GRH_SIZE) |
| self.syncer.wait() |
| ah = AH(self.cm_res.cmid.pd, attr=self.cm_res.ud_params.ah_attr) |
| self.cm_res.cmid.post_ud_send(self.cm_res.mr, ah, rqpn=MULTICAST_QPN, |
| length=self.cm_res.msg_size) |
| self.cm_res.cmid.get_send_comp() |
| self.syncer.wait() |
| |
| def event_handler(self, expected_event=None): |
| """ |
| Handle and execute corresponding API for RDMACM events of asynchronous |
| communication. |
| :param expected_event: The user expected event. |
| :return: None |
| """ |
| cm_event = CMEvent(self.cm_res.cmid.event_channel) |
| if cm_event.event_type == ce.RDMA_CM_EVENT_CONNECT_REQUEST: |
| self.cm_res.create_child_id(cm_event) |
| elif cm_event.event_type in [ce.RDMA_CM_EVENT_ESTABLISHED, |
| ce.RDMA_CM_EVENT_MULTICAST_JOIN]: |
| self.cm_res.set_ud_params(cm_event) |
| if expected_event and expected_event != cm_event.event_type: |
| raise PyverbsError('Expected this event: {}, got this event: {}'. |
| format(expected_event, cm_event.event_str())) |
| if expected_event == ce.RDMA_CM_EVENT_REJECTED: |
| assert cm_event.private_data[:len(REJECT_MSG)].decode() == REJECT_MSG, \ |
| f'CM event data ({cm_event.private_data}) is different than the expected ({REJECT_MSG})' |
| cm_event.ack_cm_event() |
| |
| @abc.abstractmethod |
| def establish_connection(self): |
| pass |
| |
| @abc.abstractmethod |
| def disconnect(self): |
| pass |
| |
| |
| class CMAsyncConnection(CMConnection): |
| """ |
| Implement RDMACM connection management for asynchronous CMIDs. It includes |
| connection establishment, disconnection and other methods such as traffic. |
| """ |
| def __init__(self, ip_addr, syncer=None, notifier=None, passive=False, |
| num_conns=1, qp_timeout=-1, reject_conn=False, **kwargs): |
| """ |
| Init the CMConnection and then init the AsyncCMResources. |
| :param ip_addr: IP address to use. |
| :param syncer: Barrier object to sync between all the test processes. |
| :param notifier: Queue object to pass objects between the connection |
| sides. |
| :param passive: Indicate if it's a passive side. |
| :param num_conns: Number of connections. |
| :param qp_timeout: Value of the QP timeout. |
| :param reject_conn: True if the server will reject the connection. |
| :param kwargs: Arguments used to initialize the CM resources. For more |
| info please check CMResources. |
| """ |
| super(CMAsyncConnection, self).__init__(syncer=syncer, notifier=notifier) |
| self.num_conns = num_conns |
| self.create_cm_res(ip_addr, passive=passive, **kwargs) |
| self.qp_timeout = qp_timeout |
| self.reject_conn = reject_conn |
| |
| def create_cm_res(self, ip_addr, passive, **kwargs): |
| self.cm_res = AsyncCMResources(addr=ip_addr, passive=passive, **kwargs) |
| if passive: |
| self.cm_res.create_cmid() |
| else: |
| for i in range(self.num_conns): |
| self.cm_res.create_cmid(i) |
| |
| def join_to_multicast(self, mc_addr=None, src_addr=None, extended=False): |
| """ |
| Join the CMID to multicast group. |
| :param mc_addr: The multicast IP address. |
| :param src_addr: The CMIDs source address. |
| :param extended: Use the join_multicast_ex API. |
| """ |
| self.cm_res.cmid.bind_addr(self.cm_res.ai) |
| resolve_addr_info = AddrInfo(src=src_addr, dst=mc_addr) |
| self.cm_res.cmid.resolve_addr(resolve_addr_info) |
| self.event_handler(expected_event=ce.RDMA_CM_EVENT_ADDR_RESOLVED) |
| self.cm_res.create_qp() |
| mc_addr_info = AddrInfo(src=mc_addr) |
| if not extended: |
| self.cm_res.cmid.join_multicast(addr=mc_addr_info) |
| else: |
| flags = ce.RDMA_MC_JOIN_FLAG_FULLMEMBER |
| comp_mask = ce.RDMA_CM_JOIN_MC_ATTR_ADDRESS | \ |
| ce.RDMA_CM_JOIN_MC_ATTR_JOIN_FLAGS |
| mcattr = JoinMCAttrEx(addr=mc_addr_info, comp_mask=comp_mask, |
| join_flags=flags) |
| self.cm_res.cmid.join_multicast(mc_attr=mcattr) |
| self.event_handler(expected_event=ce.RDMA_CM_EVENT_MULTICAST_JOIN) |
| self.cm_res.create_mr() |
| |
| def leave_multicast(self, mc_addr=None): |
| """ |
| Leave multicast group. |
| :param mc_addr: The multicast IP address. |
| """ |
| mc_addr_info = AddrInfo(src=mc_addr) |
| self.cm_res.cmid.leave_multicast(mc_addr_info) |
| |
| def establish_connection(self): |
| """ |
| Establish RDMACM connection between two Async CMIDs. |
| """ |
| if self.cm_res.passive: |
| self.cm_res.cmid.bind_addr(self.cm_res.ai) |
| self.cm_res.cmid.listen() |
| for conn_idx in range(self.num_conns): |
| if self.cm_res.passive: |
| self.syncer.wait() |
| self.event_handler(expected_event=ce.RDMA_CM_EVENT_CONNECT_REQUEST) |
| self.cm_res.create_qp(conn_idx=conn_idx) |
| if self.qp_timeout >= 0: |
| self.set_qp_timeout(self.cm_res.child_ids[conn_idx], self.qp_timeout) |
| if self.cm_res.with_ext_qp: |
| self.set_cmids_qp_ece(self.cm_res.passive) |
| self.cm_res.modify_ext_qp_to_rts(conn_idx=conn_idx) |
| self.set_cmid_ece(self.cm_res.passive) |
| child_id = self.cm_res.child_ids[conn_idx] |
| if self.reject_conn: |
| child_id.reject(REJECT_MSG.encode()) |
| return |
| child_id.accept(self.cm_res.create_conn_param(conn_idx=conn_idx)) |
| if self.qp_timeout >= 0: |
| attr, _ = child_id.query_qp(e.IBV_QP_TIMEOUT) |
| assert self.qp_timeout == attr.timeout |
| if self.cm_res.port_space == ce.RDMA_PS_TCP: |
| self.event_handler(expected_event=ce.RDMA_CM_EVENT_ESTABLISHED) |
| else: |
| cmid = self.cm_res.cmids[conn_idx] |
| cmid.resolve_addr(self.cm_res.ai) |
| self.event_handler(expected_event=ce.RDMA_CM_EVENT_ADDR_RESOLVED) |
| self.syncer.wait() |
| cmid.resolve_route() |
| self.event_handler(expected_event=ce.RDMA_CM_EVENT_ROUTE_RESOLVED) |
| self.cm_res.create_qp(conn_idx=conn_idx) |
| if self.qp_timeout >= 0: |
| self.set_qp_timeout(self.cm_res.cmid, self.qp_timeout) |
| if self.cm_res.with_ext_qp: |
| self.set_cmid_ece(self.cm_res.passive) |
| cmid.connect(self.cm_res.create_conn_param(conn_idx=conn_idx)) |
| if self.cm_res.with_ext_qp: |
| self.event_handler(expected_event=\ |
| ce.RDMA_CM_EVENT_CONNECT_RESPONSE) |
| self.set_cmids_qp_ece(self.cm_res.passive) |
| self.cm_res.modify_ext_qp_to_rts(conn_idx=conn_idx) |
| cmid.establish() |
| else: |
| if self.reject_conn: |
| self.event_handler(expected_event=ce.RDMA_CM_EVENT_REJECTED) |
| return |
| self.event_handler(expected_event=ce.RDMA_CM_EVENT_ESTABLISHED) |
| if self.qp_timeout >= 0: |
| attr, _ = self.cm_res.cmid.query_qp(e.IBV_QP_TIMEOUT) |
| assert self.qp_timeout == attr.timeout |
| self.cm_res.create_mr() |
| self.sync_qp_numbers() |
| |
| def set_qp_timeout(self, cm_id, ack_timeout): |
| cm_id.set_option(ce.RDMA_OPTION_ID, ce.RDMA_OPTION_ID_ACK_TIMEOUT, ack_timeout, 1) |
| |
| def sync_qp_numbers(self): |
| """ |
| Sync the QP numbers of the connections sides. |
| """ |
| if self.cm_res.passive: |
| self.syncer.wait() |
| self.notifier.put(self.cm_res.my_qp_number()) |
| self.syncer.wait() |
| self.cm_res.remote_qpn = self.notifier.get() |
| else: |
| self.syncer.wait() |
| self.cm_res.remote_qpn = self.notifier.get() |
| self.notifier.put(self.cm_res.my_qp_number()) |
| self.syncer.wait() |
| |
| def disconnect(self): |
| """ |
| Disconnect the connection. |
| """ |
| if self.cm_res.port_space == ce.RDMA_PS_TCP: |
| if self.cm_res.passive: |
| for child_id in self.cm_res.child_ids.values(): |
| child_id.disconnect() |
| else: |
| self.event_handler(expected_event=ce.RDMA_CM_EVENT_DISCONNECTED) |
| for cmid in self.cm_res.cmids.values(): |
| cmid.disconnect() |
| |
| def set_cmid_ece(self, passive): |
| """ |
| Set the local CMIDs ECE. The ECE is taken from the CMIDs QP ECE. |
| :param passive: Indicates if this CMID is participate as passive in |
| this connection. |
| """ |
| cmid = self.cm_res.child_id if passive else self.cm_res.cmid |
| try: |
| ece = self.cm_res.qp.query_ece() |
| cmid.set_local_ece(ece) |
| except PyverbsRDMAError as ex: |
| if ex.error_code != errno.EOPNOTSUPP: |
| raise ex |
| |
| def set_cmids_qp_ece(self, passive): |
| """ |
| Set the CMIDs QP ECE. |
| :param passive: Indicates if this CMID is participate as passive in |
| this connection. |
| """ |
| cmid = self.cm_res.child_id if passive else self.cm_res.cmid |
| try: |
| ece = cmid.get_remote_ece() |
| self.cm_res.qp.set_ece(ece) |
| except PyverbsRDMAError as ex: |
| if ex.error_code != errno.EOPNOTSUPP: |
| raise ex |
| |
| class CMSyncConnection(CMConnection): |
| """ |
| Implement RDMACM connection management for synchronous CMIDs. It includes |
| connection establishment, disconnection and other methods such as traffic. |
| """ |
| def __init__(self, ip_addr, syncer=None, notifier=None, passive=False, **kwargs): |
| """ |
| Init the CMConnection and then init the SyncCMResources. |
| :param ip_addr: IP address to use. |
| :param syncer: Barrier object to sync between all the test processes. |
| :param notifier: Queue object to pass objects between the connection |
| sides. |
| :param passive: Indicate if it's a passive side. |
| :param kwargs: Arguments used to initialize the CM resources. For more |
| info please check CMResources. |
| """ |
| super(CMSyncConnection, self).__init__(syncer=syncer, notifier=notifier) |
| self.create_cm_res(ip_addr, passive=passive, **kwargs) |
| |
| def create_cm_res(self, ip_addr, passive, **kwargs): |
| self.cm_res = SyncCMResources(addr=ip_addr, passive=passive, **kwargs) |
| self.cm_res.create_cmid() |
| |
| def establish_connection(self): |
| """ |
| Establish RDMACM connection between two Sync CMIDs. |
| """ |
| if self.cm_res.passive: |
| self.cm_res.cmid.listen() |
| self.syncer.wait() |
| self.cm_res.create_child_id() |
| self.cm_res.child_id.accept() |
| self.cm_res.create_mr() |
| else: |
| self.syncer.wait() |
| self.cm_res.cmid.connect() |
| self.cm_res.create_mr() |
| |
| def disconnect(self): |
| """ |
| Disconnect the connection. |
| """ |
| if self.cm_res.port_space == ce.RDMA_PS_TCP: |
| if self.cm_res.passive: |
| self.cm_res.child_id.disconnect() |
| else: |
| self.cm_res.cmid.disconnect() |
