| import unittest | |
| import datetime | |
| import errno | |
| from tests.base import RCResources, RDMATestCase, PyverbsAPITestCase | |
| from pyverbs.providers.mlx5.mlx5dv import Mlx5Context | |
| from pyverbs.pyverbs_error import PyverbsRDMAError | |
| from pyverbs.cq import CqInitAttrEx, CQEX | |
| from tests.test_flow import FlowRes | |
| from pyverbs.qp import QPInitAttr | |
| from pyverbs.cq import PollCqAttr | |
| from pyverbs.libibverbs_enums import ibv_qp_type, ibv_wc_status, ibv_wr_opcode, ibv_create_cq_wc_flags, \ | |
| IBV_WC_STANDARD_FLAGS | |
| import tests.utils as u | |
| GIGA = 1000000000 | |
| def convert_ts_to_ns(ctx, device_ts): | |
| """ | |
| Convert device timestamp from HCA core clock units to corresponding | |
| nanosecond counts. | |
| :param ctx: The context that gets this timestamp. | |
| :param device_ts: The device timestamp to translate. | |
| :return: Timestamp in nanoseconds | |
| """ | |
| try: | |
| timestamp_in_ns = Mlx5Context.device_timestamp_to_ns(ctx, device_ts) | |
| except PyverbsRDMAError as ex: | |
| if ex.error_code == errno.EOPNOTSUPP: | |
| raise unittest.SkipTest('Converting timestamp to nanoseconds is not supported') | |
| raise ex | |
| return timestamp_in_ns | |
| def timestamp_res_cls(base_class): | |
| """ | |
| This is a factory function which creates a class that inherits base_class of | |
| any BaseResources type. | |
| :param base_class: The base resources class to inherit from. | |
| :return: TimeStampRes class. | |
| """ | |
| class TimeStampRes(base_class): | |
| def __init__(self, dev_name, ib_port, gid_index, qp_type, send_ts=None, | |
| recv_ts=None): | |
| self.qp_type = qp_type | |
| self.send_ts = send_ts | |
| self.recv_ts = recv_ts | |
| self.timestamp = None | |
| self.scq = None | |
| self.rcq = None | |
| super().__init__(dev_name=dev_name, ib_port=ib_port, gid_index=gid_index) | |
| def create_cq(self): | |
| self.scq = self._create_ex_cq(self.send_ts) | |
| self.rcq = self._create_ex_cq(self.recv_ts) | |
| def _create_ex_cq(self, timestamp=None): | |
| """ | |
| Create an Extended CQ. | |
| :param timestamp: If set, the timestamp type to use. | |
| """ | |
| wc_flags = IBV_WC_STANDARD_FLAGS | |
| if timestamp: | |
| wc_flags |= timestamp | |
| cia = CqInitAttrEx(cqe=self.num_msgs, wc_flags=wc_flags) | |
| try: | |
| cq = CQEX(self.ctx, cia) | |
| except PyverbsRDMAError as ex: | |
| if ex.error_code == errno.EOPNOTSUPP: | |
| raise unittest.SkipTest('Create Extended CQ is not supported') | |
| raise ex | |
| return cq | |
| def create_qp_init_attr(self): | |
| return QPInitAttr(qp_type=self.qp_type, scq=self.scq, | |
| rcq=self.rcq, srq=self.srq, cap=self.create_qp_cap()) | |
| return TimeStampRes | |
| class TimeStampTest(RDMATestCase): | |
| """ | |
| Test various types of timestamping formats. | |
| """ | |
| def setUp(self): | |
| super().setUp() | |
| self.send_ts = None | |
| self.recv_ts = None | |
| self.qp_type = None | |
| @property | |
| def resource_arg(self): | |
| return {'send_ts': self.send_ts, 'recv_ts': self.recv_ts, | |
| 'qp_type': self.qp_type} | |
| def test_timestamp_free_running_rc_traffic(self): | |
| """ | |
| Test free running timestamp on RC traffic. | |
| """ | |
| self.qp_type = ibv_qp_type.IBV_QPT_RC | |
| self.send_ts = self.recv_ts = ibv_create_cq_wc_flags.IBV_WC_EX_WITH_COMPLETION_TIMESTAMP | |
| self.create_players(timestamp_res_cls(RCResources), **self.resource_arg) | |
| self.ts_traffic() | |
| timestamp = convert_ts_to_ns(self.client.ctx, self.client.timestamp) | |
| self.verify_ts(timestamp) | |
| def test_timestamp_real_time_rc_traffic(self): | |
| """ | |
| Test real time timestamp on RC traffic. | |
| """ | |
| self.qp_type = ibv_qp_type.IBV_QPT_RC | |
| self.send_ts = self.recv_ts = ibv_create_cq_wc_flags.IBV_WC_EX_WITH_COMPLETION_TIMESTAMP_WALLCLOCK | |
| self.create_players(timestamp_res_cls(RCResources), **self.resource_arg) | |
| self.ts_traffic() | |
| self.verify_ts(self.client.timestamp) | |
| def test_timestamp_free_running_send_raw_traffic(self): | |
| """ | |
| Test timestamping on RAW traffic only on the send completions. | |
| """ | |
| self.qp_type = ibv_qp_type.IBV_QPT_RAW_PACKET | |
| self.send_ts = ibv_create_cq_wc_flags.IBV_WC_EX_WITH_COMPLETION_TIMESTAMP | |
| self.create_players(timestamp_res_cls(FlowRes), **self.resource_arg) | |
| self.flow = self.server.create_flow([self.server.create_eth_spec()]) | |
| self.ts_traffic() | |
| timestamp = convert_ts_to_ns(self.client.ctx, self.client.timestamp) | |
| self.verify_ts(timestamp) | |
| def test_timestamp_free_running_recv_raw_traffic(self): | |
| """ | |
| Test timestamping on RAW traffic only on the recv completions. | |
| """ | |
| self.qp_type = ibv_qp_type.IBV_QPT_RAW_PACKET | |
| self.recv_ts = ibv_create_cq_wc_flags.IBV_WC_EX_WITH_COMPLETION_TIMESTAMP | |
| self.create_players(timestamp_res_cls(FlowRes), **self.resource_arg) | |
| self.flow = self.server.create_flow([self.server.create_eth_spec()]) | |
| self.ts_traffic() | |
| timestamp = convert_ts_to_ns(self.server.ctx, self.server.timestamp) | |
| self.verify_ts(timestamp) | |
| def test_timestamp_real_time_raw_traffic(self): | |
| """ | |
| Test real time timestamp on RAW traffic. | |
| """ | |
| self.qp_type = ibv_qp_type.IBV_QPT_RAW_PACKET | |
| self.send_ts = self.recv_ts = ibv_create_cq_wc_flags.IBV_WC_EX_WITH_COMPLETION_TIMESTAMP_WALLCLOCK | |
| self.create_players(timestamp_res_cls(FlowRes), **self.resource_arg) | |
| self.flow = self.server.create_flow([self.server.create_eth_spec()]) | |
| self.ts_traffic() | |
| self.verify_ts(self.client.timestamp) | |
| @staticmethod | |
| def verify_ts(timestamp): | |
| """ | |
| Verify that the timestamp is a valid value of time. | |
| """ | |
| datetime.datetime.fromtimestamp(timestamp/GIGA) | |
| @staticmethod | |
| def poll_cq_ex_ts(cqex, ts_type=None): | |
| """ | |
| Poll completion from the extended CQ. | |
| :param cqex: CQEX to poll from | |
| :param ts_type: If set, read the CQE timestamp in this format | |
| :return: The CQE timestamp if it requested. | |
| """ | |
| polling_timeout = 10 | |
| start = datetime.datetime.now() | |
| ts = 0 | |
| poll_attr = PollCqAttr() | |
| ret = cqex.start_poll(poll_attr) | |
| while ret == 2 and (datetime.datetime.now() - start).seconds < polling_timeout: | |
| ret = cqex.start_poll(poll_attr) | |
| if ret == 2: | |
| raise PyverbsRDMAError('Failed to poll CQEX - Got timeout') | |
| if ret != 0: | |
| raise PyverbsRDMAError('Failed to poll CQEX') | |
| if cqex.status != ibv_wc_status.IBV_WC_SUCCESS: | |
| raise PyverbsRDMAError('Completion status is {cqex.status}') | |
| if ts_type == ibv_create_cq_wc_flags.IBV_WC_EX_WITH_COMPLETION_TIMESTAMP: | |
| ts = cqex.read_timestamp() | |
| if ts_type == ibv_create_cq_wc_flags.IBV_WC_EX_WITH_COMPLETION_TIMESTAMP_WALLCLOCK: | |
| ts = cqex.read_completion_wallclock_ns() | |
| cqex.end_poll() | |
| return ts | |
| def ts_traffic(self): | |
| """ | |
| Run RDMA traffic and read the completions timestamps. | |
| """ | |
| s_recv_wr = u.get_recv_wr(self.server) | |
| u.post_recv(self.server, s_recv_wr) | |
| if self.qp_type == ibv_qp_type.IBV_QPT_RAW_PACKET: | |
| c_send_wr, _, _ = u.get_send_elements_raw_qp(self.client) | |
| else: | |
| c_send_wr, _ = u.get_send_elements(self.client, False) | |
| u.send(self.client, c_send_wr, ibv_wr_opcode.IBV_WR_SEND, False, 0) | |
| self.client.timestamp = self.poll_cq_ex_ts(self.client.scq, ts_type=self.send_ts) | |
| self.server.timestamp = self.poll_cq_ex_ts(self.server.rcq, ts_type=self.recv_ts) | |
| class TimeAPITest(PyverbsAPITestCase): | |
| def test_query_rt_values(self): | |
| """ | |
| Test the ibv_query_rt_values_ex API. | |
| Query the device real-time values, convert them to ns and verify that | |
| the timestamp is a valid value of time.. | |
| """ | |
| try: | |
| _, hw_time = self.ctx.query_rt_values_ex() | |
| time_in_ns = convert_ts_to_ns(self.ctx, hw_time) | |
| datetime.datetime.fromtimestamp(time_in_ns/GIGA) | |
| except PyverbsRDMAError as ex: | |
| if ex.error_code == errno.EOPNOTSUPP: | |
| raise unittest.SkipTest('Query device real time is not supported') | |
| raise ex |