/
githubmirror
/
rdma-core
Обзор
Документация
Войти
/
githubmirror
/
rdma-core
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
tests/rdmacm_utils.py
525 строк
23 KB
Mazal Or Yakuel
tests: Add resolve and query addr_info CM test
11 янв 2026, 11:56
11 янв 2026, 11:56
3f5b710
Код
Авторство
О чём код?
# 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 from pyverbs.librdmacm_enums import rdma_port_space, rdma_cm_event_type, rdma_cm_mc_join_flags, \ rdma_cm_join_mc_attr_mask, RDMA_OPTION_ID, RDMA_OPTION_ID_ACK_TIMEOUT, RAI_PASSIVE from pyverbs.addr import AH from pyverbs.libibverbs_enums import ibv_send_flags, ibv_qp_type, ibv_qp_attr_mask 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=ibv_send_flags.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 == ibv_qp_type.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 == rdma_port_space.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 == ibv_qp_type.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 == rdma_port_space.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 == rdma_cm_event_type.RDMA_CM_EVENT_CONNECT_REQUEST: self.cm_res.create_child_id(cm_event) elif cm_event.event_type in [rdma_cm_event_type.RDMA_CM_EVENT_ESTABLISHED, rdma_cm_event_type.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 == rdma_cm_event_type.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 def get_and_update_addrinfo(self, passive, ip_addr): """ Get and set address information for RDMA connection. :param passive: If True, get address info for passive side :param ip_addr: IP address to use for resolution """ if passive: self.cm_res.ai.get_addrinfo(src=ip_addr, src_service=self.cm_res.port, flags=RAI_PASSIVE, port_space=self.cm_res.port_space) else: self.cm_res.ai.get_addrinfo(src=ip_addr, dst=ip_addr, dst_service=self.cm_res.port, port_space=self.cm_res.port_space) 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(passive=passive, **kwargs) if passive: self.cm_res.create_cmid() else: for i in range(self.num_conns): self.cm_res.create_cmid(i) if kwargs.get('use_resolve_addrinfo'): self.resolve_and_update_addrinfo(passive, ip_addr) else: self.get_and_update_addrinfo(passive, ip_addr) 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=rdma_cm_event_type.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 = rdma_cm_mc_join_flags.RDMA_MC_JOIN_FLAG_FULLMEMBER comp_mask = rdma_cm_join_mc_attr_mask.RDMA_CM_JOIN_MC_ATTR_ADDRESS | \ rdma_cm_join_mc_attr_mask.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=rdma_cm_event_type.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=rdma_cm_event_type.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(ibv_qp_attr_mask.IBV_QP_TIMEOUT) assert self.qp_timeout == attr.timeout if self.cm_res.port_space == rdma_port_space.RDMA_PS_TCP: self.event_handler(expected_event=rdma_cm_event_type.RDMA_CM_EVENT_ESTABLISHED) else: cmid = self.cm_res.cmids[conn_idx] cmid.resolve_addr(self.cm_res.ai) self.event_handler(expected_event=rdma_cm_event_type.RDMA_CM_EVENT_ADDR_RESOLVED) self.syncer.wait() cmid.resolve_route() self.event_handler(expected_event=rdma_cm_event_type.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=\ rdma_cm_event_type.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=rdma_cm_event_type.RDMA_CM_EVENT_REJECTED) return self.event_handler(expected_event=rdma_cm_event_type.RDMA_CM_EVENT_ESTABLISHED) if self.qp_timeout >= 0: attr, _ = self.cm_res.cmid.query_qp(ibv_qp_attr_mask.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(RDMA_OPTION_ID, 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 == rdma_port_space.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=rdma_cm_event_type.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 def resolve_and_update_addrinfo(self, passive, ip_addr): """ Resolve and query and update the address information for the CMID. :param passive: If True, resolve as passive side :param ip_addr: IP address to resolve """ if passive: self.cm_res.cmid.resolve_addrinfo( node=ip_addr, service=self.cm_res.port, port_space=self.cm_res.port_space, flags=RAI_PASSIVE) else: self.cm_res.cmid.resolve_addrinfo( node=ip_addr, service=self.cm_res.port, port_space=self.cm_res.port_space) self.event_handler(expected_event=rdma_cm_event_type.RDMA_CM_EVENT_ADDRINFO_RESOLVED) self.cm_res.ai = self.cm_res.cmid.query_addrinfo() 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(passive=passive, **kwargs) if self.cm_res.passive: self.cm_res.ai = AddrInfo(src=ip_addr, src_service=self.cm_res.port, port_space=self.cm_res.port_space, flags=RAI_PASSIVE) else: self.cm_res.ai = AddrInfo(src=ip_addr, dst=ip_addr, dst_service=self.cm_res.port, port_space=self.cm_res.port_space) 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 == rdma_port_space.RDMA_PS_TCP: if self.cm_res.passive: self.cm_res.child_id.disconnect() else: self.cm_res.cmid.disconnect()