| # SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB) |
| # Copyright 2026 Advanced Micro Devices, Inc. All rights reserved. |
| |
| import unittest |
| import datetime |
| import time |
| import os |
| import errno |
| |
| from pyverbs.libibverbs_enums import ibv_qp_type, ibv_wc_status, ibv_wr_opcode, ibv_create_cq_wc_flags, \ |
| IBV_WC_STANDARD_FLAGS |
| from tests.base import RCResources, RDMATestCase |
| from pyverbs.pyverbs_error import PyverbsRDMAError |
| from pyverbs.cq import CqInitAttrEx, CQEX |
| from pyverbs.qp import QPInitAttr |
| from pyverbs.cq import PollCqAttr |
| import tests.utils as u |
| |
| GIGA = 1000000000 |
| CLOCKFD = 3 |
| CLOCKID_TYPE_BITS = 3 |
| |
| 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 |
| self.phc_file = None |
| self.open_phc_dev(dev_name) |
| super().__init__(dev_name=dev_name, ib_port=ib_port, gid_index=gid_index) |
| |
| def close_files(self): |
| if self.phc_file is not None: |
| self.phc_file.close() |
| self.phc_file = None |
| |
| def open_phc_dev(self, dev_name): |
| try: |
| phc_name = os.listdir(f'/sys/class/infiniband/{dev_name}/device/ptp')[0] |
| self.phc_file = open(f'/dev/{phc_name}', 'rb') |
| self.phc_clkid = (~self.phc_file.fileno() << CLOCKID_TYPE_BITS) | CLOCKFD |
| except (FileNotFoundError, OSError, IndexError): |
| raise unittest.SkipTest('No PHC or failed to open') |
| |
| def read_phc(self): |
| return time.clock_gettime(self.phc_clkid) |
| |
| 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 |
| |
| def tearDown(self): |
| for player in [getattr(self, 'client', None), getattr(self, 'server', None)]: |
| if player is not None and hasattr(player, 'close_files'): |
| player.close_files() |
| super().tearDown() |
| |
| @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. |
| The timestamp is returned as a raw HW PHC tick value, so this test |
| only verifies that polling succeeds; value validation is done by the |
| real-time variant. |
| """ |
| 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() |
| |
| def test_timestamp_real_time_rc_traffic(self): |
| """ |
| Test real-time timestamp on RC traffic. |
| The timestamp is returned in nanoseconds and validated against the |
| live PHC read from /dev/ptpN. |
| """ |
| 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 / GIGA, self.client.read_phc()) |
| self.verify_ts(self.server.timestamp / GIGA, self.server.read_phc()) |
| |
| def verify_ts(self, timestamp, phc_now): |
| """ |
| Verify that the timestamp is in the past one second |
| """ |
| if self.config['verbosity']: |
| print(f'timestamp {timestamp} current {phc_now} difference {phc_now - timestamp}') |
| if timestamp > phc_now: |
| raise PyverbsRDMAError(f'Completion timestamp is in the future: {timestamp} > {phc_now}') |
| if timestamp < phc_now - 1: |
| raise PyverbsRDMAError(f'Completion timestamp is too far in the past: {timestamp} < {phc_now - 1}') |
| |
| @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(f'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) |
| 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) |
