Unsolicited write test

tests/efa_base.py

@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB)
-# Copyright 2020-2023 Amazon.com, Inc. or its affiliates. All rights reserved.
+# Copyright 2020-2025 Amazon.com, Inc. or its affiliates. All rights reserved.
 
 import unittest
 import random
@@ -8,7 +8,7 @@
 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.efa_enums as efa_enums
 import pyverbs.providers.efa.efadv as efa
 import pyverbs.device as d
 from pyverbs.libibverbs_enums import ibv_qp_init_attr_mask, ibv_qp_type, ibv_qp_create_send_ops_flags,\
@@ -49,10 +49,22 @@ class SRDResources(TrafficResources):
     SRD_QKEY = 0x11111111
     SRD_PKEY_INDEX = 0
     def __init__(self, dev_name, ib_port, gid_index, send_ops_flags,
-                 qp_count=1):
+                 qp_count=1, required_dev_cap=None, wc_flags=0, qp_flags=0):
         self.send_ops_flags = send_ops_flags
+        self.qp_flags = qp_flags
+        self.required_dev_cap = required_dev_cap
+        self.efa_wc_flags = wc_flags
         super().__init__(dev_name, ib_port, gid_index, qp_count=qp_count)
 
+    def create_context(self):
+        super().create_context()
+        if self.required_dev_cap:
+            with efa.EfaContext(name=self.ctx.name) as efa_ctx:
+                device_caps = efa_ctx.query_efa_device().device_caps
+                if (device_caps & self.required_dev_cap) != self.required_dev_cap:
+                    miss_caps = efa.dev_cap_to_str(self.required_dev_cap)
+                    raise unittest.SkipTest(f"Device caps doesn't support {miss_caps}")
+
     def create_qp_attr(self):
         attr = QPAttr(port_num=self.ib_port)
         attr.qkey = self.SRD_QKEY
@@ -76,7 +88,8 @@ def create_qps(self):
                                        rcq=self.cq, pd=self.pd, send_ops_flags=self.send_ops_flags,
                                        comp_mask=comp_mask)
         efa_init_attr_ex = efa.EfaQPInitAttr()
-        efa_init_attr_ex.driver_qp_type = efa_e.EFADV_QP_DRIVER_TYPE_SRD
+        efa_init_attr_ex.driver_qp_type = efa_enums.EFADV_QP_DRIVER_TYPE_SRD
+        efa_init_attr_ex.flags |= self.qp_flags
         try:
             for _ in range(self.qp_count):
                 qp = efa.SRDQPEx(self.ctx, qp_init_attr_ex, efa_init_attr_ex)
@@ -96,34 +109,15 @@ def create_mr(self):
             additional_access_flags = ibv_access_flags.IBV_ACCESS_REMOTE_WRITE
         self.mr = tests.utils.create_custom_mr(self, additional_access_flags)
 
-
-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=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
+        if self.efa_wc_flags == 0:
+            super().create_cq()
+        else:
+            cia = CqInitAttrEx(wc_flags=IBV_WC_STANDARD_FLAGS)
+            efa_cia = efa.EfaDVCQInitAttr(wc_flags=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_efa_srd.py

@@ -1,12 +1,14 @@
 # SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB)
-# Copyright 2020-2023 Amazon.com, Inc. or its affiliates. All rights reserved.
+# Copyright 2020-2025 Amazon.com, Inc. or its affiliates. All rights reserved.
 
 import unittest
 import errno
 
 from pyverbs.cq import CQ, CompChannel
-from pyverbs.pyverbs_error import PyverbsRDMAError
 from pyverbs.libibverbs_enums import ibv_qp_create_send_ops_flags, ibv_wr_opcode, ibv_qp_attr_mask
+from pyverbs.pyverbs_error import PyverbsRDMAError, PyverbsError
+import pyverbs.providers.efa.efa_enums as efa_enums
+import pyverbs.providers.efa.efadv as efa
 
 from tests.efa_base import EfaRDMATestCase
 from tests.efa_base import SRDResources
@@ -42,8 +44,8 @@ def setUp(self):
         self.server = None
         self.client = None
 
-    def create_players(self, send_ops_flags=0, qp_count=8):
-        super().create_players(SRDResources, send_ops_flags=send_ops_flags, qp_count=qp_count)
+    def create_players(self, send_ops_flags=0, qp_count=8, dev_cap=None, wc_flags=0, qp_flags=0):
+        super().create_players(SRDResources, send_ops_flags=send_ops_flags, qp_count=qp_count, required_dev_cap=dev_cap, wc_flags=wc_flags, qp_flags=qp_flags)
 
     def full_sq_bad_flow(self):
         """
@@ -62,6 +64,29 @@ def full_sq_bad_flow(self):
                 u.send(self.client, c_sg, send_op, new_send=True, qp_idx=qp_idx, ah=ah)
         self.assertEqual(ex.exception.error_code, errno.ENOMEM)
 
+    def unsolicited_rdma_write_traffic(self, client, server, iters, gid_idx, port):
+        """
+        Runs unsolicited rdma write traffic between two sides
+        """
+        ah_client = u.get_global_ah(client, gid_idx, port)
+        ah_server = u.get_global_ah(server, gid_idx, port)
+        poll = u.poll_cq_ex
+        imm_data = u.IMM_DATA
+        send_op=ibv_wr_opcode.IBV_WR_RDMA_WRITE_WITH_IMM
+        read_offset = 0
+        for _ in range(iters):
+            for qp_idx in range(server.qp_count):
+                _, c_send_object = u.get_send_elements(client, False, send_op)
+                u.send(client, c_send_object, send_op, True, qp_idx, ah_client, is_imm=True)
+                poll(client.cq)
+                poll(server.cq, data=imm_data)
+                # Validate that the CQE is marked as unsolicited
+                if not server.cq.is_unsolicited():
+                    raise PyverbsError("Completion was not marked as unsolicited")
+                msg_received_list = u.get_msg_received(server, read_offset)
+                for msg in msg_received_list:
+                    u.validate(msg, True, server.msg_size)
+
     def test_qp_ex_srd_send(self):
         self.create_players(ibv_qp_create_send_ops_flags.IBV_QP_EX_WITH_SEND)
         u.traffic(**self.traffic_args, new_send=True, send_op=ibv_wr_opcode.IBV_WR_SEND)
@@ -83,6 +108,13 @@ def test_qp_ex_srd_rdma_write_with_imm(self):
         self.create_players(ibv_qp_create_send_ops_flags.IBV_QP_EX_WITH_RDMA_WRITE_WITH_IMM)
         u.traffic(**self.traffic_args, new_send=True, send_op=ibv_wr_opcode.IBV_WR_RDMA_WRITE_WITH_IMM)
 
+    def test_qp_ex_srd_rdma_unsolicited_write_with_imm(self):
+        wc_flag = efa_enums.EFADV_WC_EX_WITH_IS_UNSOLICITED
+        dev_cap = efa_enums.EFADV_DEVICE_ATTR_CAPS_UNSOLICITED_WRITE_RECV
+        qp_flags = efa_enums.EFADV_QP_FLAGS_UNSOLICITED_WRITE_RECV
+        self.create_players(ibv_qp_create_send_ops_flags.IBV_QP_EX_WITH_RDMA_WRITE_WITH_IMM, dev_cap=dev_cap, wc_flags=wc_flag, qp_flags=qp_flags)
+        self.unsolicited_rdma_write_traffic(**self.traffic_args)
+
     def test_qp_ex_srd_old_send(self):
         self.create_players()
         u.traffic(**self.traffic_args, new_send=False)

tests/test_efadv.py

@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB)
-# Copyright 2020-2024 Amazon.com, Inc. or its affiliates. All rights reserved.
+# Copyright 2020-2025 Amazon.com, Inc. or its affiliates. All rights reserved.
 """
 Test module for efa direct-verbs.
 """
@@ -18,7 +18,7 @@
     ibv_wr_opcode, ibv_access_flags
 from pyverbs.pd import PD
 
-from tests.efa_base import EfaAPITestCase, EfaRDMATestCase, EfaCQRes
+from tests.efa_base import EfaAPITestCase, EfaRDMATestCase, SRDResources
 import tests.utils as u
 
 
@@ -138,8 +138,8 @@ def setUp(self):
         self.client = None
 
     def create_players(self, dev_cap, wc_flags, send_ops_flags, qp_count=8):
-        super().create_players(EfaCQRes, send_ops_flags=send_ops_flags, qp_count=qp_count,
-                               requested_dev_cap=dev_cap, wc_flags=wc_flags)
+        super().create_players(SRDResources, send_ops_flags=send_ops_flags, qp_count=qp_count,
+                               required_dev_cap=dev_cap, wc_flags=wc_flags)
         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):