tests: Add read SGID test on EFA DV CQ

Run a simple traffic test using EfaCQRes class, in order to read SGID
from receive work completion.

Signed-off-by: Daniel Kranzdorf <[email protected]>

Author: dkkranz

tests/efa_base.py

@@ -1,11 +1,12 @@
 # SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB)
-# Copyright 2020 Amazon.com, Inc. or its affiliates. All rights reserved.
+# Copyright 2020-2022 Amazon.com, Inc. or its affiliates. All rights reserved.
 
 import unittest
 import random
 import errno
 
 from pyverbs.pyverbs_error import PyverbsRDMAError
+from pyverbs.cq import CqInitAttrEx
 from pyverbs.qp import QPAttr, QPCap, QPInitAttrEx
 import pyverbs.providers.efa.efa_enums as efa_e
 import pyverbs.providers.efa.efadv as efa
@@ -96,3 +97,35 @@ def create_mr(self):
             self.mr = tests.utils.create_custom_mr(self, e.IBV_ACCESS_REMOTE_READ)
         else:
             self.mr = tests.utils.create_custom_mr(self)
+
+
+class EfaCQRes(SRDResources):
+    def __init__(self, dev_name, ib_port, gid_index, send_ops_flags,
+                 qp_count=1, requested_dev_cap=None, wc_flags=None):
+        """
+        Initialize EFA DV CQ based on SRD resources.
+        :param requested_dev_cap: A necessary device cap. If it's not supported
+                                  by the device, the test will be skipped.
+        :param wc_flags: WC flags for EFA DV CQ.
+        """
+        self.requested_dev_cap = requested_dev_cap
+        self.efa_wc_flags = wc_flags
+        super().__init__(dev_name, ib_port, gid_index, send_ops_flags, qp_count=qp_count)
+
+    def create_context(self):
+        super().create_context()
+        if self.requested_dev_cap:
+            with efa.EfaContext(name=self.ctx.name) as efa_ctx:
+                if not efa_ctx.query_efa_device().device_caps & self.requested_dev_cap:
+                    miss_caps = efa.dev_cap_to_str(self.requested_dev_cap)
+                    raise unittest.SkipTest(f'Device caps doesn\'t support {miss_caps}')
+
+    def create_cq(self):
+        cia = CqInitAttrEx(wc_flags=e.IBV_WC_STANDARD_FLAGS)
+        efa_cia = efa.EfaDVCQInitAttr(self.efa_wc_flags)
+        try:
+            self.cq = efa.EfaCQ(self.ctx, cia, efa_cia)
+        except PyverbsRDMAError as ex:
+            if ex.error_code == errno.EOPNOTSUPP:
+                raise unittest.SkipTest('Create EFA DV CQ is not supported')
+            raise ex

tests/test_efadv.py

@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB)
-# Copyright 2020 Amazon.com, Inc. or its affiliates. All rights reserved.
+# Copyright 2020-2022 Amazon.com, Inc. or its affiliates. All rights reserved.
 """
 Test module for efa direct-verbs.
 """
@@ -17,7 +17,7 @@
 import pyverbs.enums as e
 from pyverbs.pd import PD
 
-from tests.efa_base import EfaAPITestCase
+from tests.efa_base import EfaAPITestCase, EfaRDMATestCase, EfaCQRes
 import tests.utils as u
 
 
@@ -125,3 +125,38 @@ def get_qp_init_attr_ex(cq, pd, attr):
     qia.recv_cq = cq
     qia.pd = pd
     return qia
+
+
+class EfaCqTest(EfaRDMATestCase):
+    def setUp(self):
+        super().setUp()
+        self.iters = 100
+        self.server = None
+        self.client = None
+
+    def create_players(self, dev_cap, wc_flags, send_ops_flags, qp_count=8):
+        try:
+            self.client = EfaCQRes(self.dev_name, self.ib_port, self.gid_index, send_ops_flags, qp_count,
+                                   dev_cap, wc_flags)
+            self.server = EfaCQRes(self.dev_name, self.ib_port, self.gid_index, send_ops_flags, qp_count,
+                                   dev_cap, wc_flags)
+        except PyverbsRDMAError as ex:
+            if ex.error_code == errno.EOPNOTSUPP:
+                raise unittest.SkipTest('Create EfaCQ Resources is not supported')
+            raise ex
+        self.client.pre_run(self.server.psns, self.server.qps_num)
+        self.server.pre_run(self.client.psns, self.client.qps_num)
+        self.server.remote_gid = self.client.ctx.query_gid(self.client.ib_port, self.client.gid_index)
+
+    def test_dv_cq_ex_with_sgid(self):
+        send_op = e.IBV_QP_EX_WITH_SEND
+        wc_flag = efa_e.EFADV_WC_EX_WITH_SGID
+        dev_cap = efa_e.EFADV_DEVICE_ATTR_CAPS_CQ_WITH_SGID
+        self.create_players(dev_cap, wc_flag, send_op, qp_count=1)
+        recv_wr = u.get_recv_wr(self.server)
+        self.server.qps[0].post_recv(recv_wr)
+        ah_client = u.get_global_ah(self.client, self.gid_index, self.ib_port)
+        _ , sg = u.get_send_elements(self.client, False)
+        u.send(self.client, sg, send_op, new_send=True, qp_idx=0, ah=ah_client)
+        u.poll_cq_ex(self.client.cq)
+        u.poll_cq_ex(self.server.cq, sgid=self.server.remote_gid)

tests/utils.py

@@ -20,6 +20,7 @@
 from pyverbs.addr import AHAttr, AH, GlobalRoute
 from tests.base import XRCResources, DCT_KEY
 from tests.efa_base import SRDResources
+from pyverbs.providers.efa.efadv import EfaCQ
 from pyverbs.wr import SGE, SendWR, RecvWR
 from pyverbs.qp import QPCap, QPInitAttr, QPInitAttrEx, QPAttr, QPEx, QP
 from tests.mlx5_base import Mlx5DcResources, Mlx5DcStreamsRes
@@ -572,13 +573,15 @@ def poll_cq(cq, count=1, data=None):
     return wcs
 
 
-def poll_cq_ex(cqex, count=1, data=None):
+def poll_cq_ex(cqex, count=1, data=None, sgid=None):
     """
     Poll <count> completions from the extended CQ.
     :param cq: CQEX to poll from
     :param count: How many completions to poll
     :param data: In case of a work request with immediate, the immediate data
                  to be compared after poll
+    :param sgid: In case of EFA receive completion, the sgid to be compared
+                 after poll
     :return: None
     """
     try:
@@ -595,6 +598,10 @@ def poll_cq_ex(cqex, count=1, data=None):
                                    format(s=cqex.status))
         if data:
             assert data == socket.ntohl(cqex.read_imm_data())
+
+        if isinstance(cqex, EfaCQ):
+            if sgid is not None and cqex.read_opcode() == e.IBV_WC_RECV:
+                assert sgid.gid == cqex.read_sgid().gid
         # Now poll the rest of the packets
         while count > 0 and (time.perf_counter() - start_poll_t < POLL_CQ_TIMEOUT):
             ret = cqex.poll_next()
@@ -608,6 +615,10 @@ def poll_cq_ex(cqex, count=1, data=None):
                                        format(s=cqex.status))
             if data:
                 assert data == socket.ntohl(cqex.read_imm_data())
+
+            if isinstance(cqex, EfaCQ):
+                if sgid is not None and cqex.read_opcode() == e.IBV_WC_RECV:
+                    assert sgid.gid == cqex.read_sgid().gid
             count -= 1
         if count > 0:
             raise PyverbsError(f'Got timeout on polling ({count} CQEs remaining)')