tests: Add rdmacm remote traffic

Add remote read and write traffic between two CMIDs using the internal
QPs.

Signed-off-by: Ido Kalir <idok@mellanox.com>
Signed-off-by: Edward Srouji <edwards@mellanox.com>

tests/base_rdmacm.py

@@ -37,6 +37,7 @@ def __init__(self, addr=None, passive=None, **kwargs):
         self.with_ext_qp = kwargs.get('with_ext_qp', False)
         self.port = kwargs.get('port') if kwargs.get('port') else '7471'
         self.port_space = kwargs.get('port_space', ce.RDMA_PS_TCP)
+        self.remote_operation = kwargs.get('remote_op')
         self.qp_type = qp_type_per_ps[self.port_space]
         self.qp_init_attr = QPInitAttr(qp_type=self.qp_type, cap=QPCap())
         self.connected = False
@@ -60,10 +61,10 @@ def __init__(self, addr=None, passive=None, **kwargs):
                                port_space=self.port_space)
 
     def create_mr(self):
-        if self.passive:
-            self.mr = self.child_id.reg_msgs(self.msg_size + GRH_SIZE)
-        else:
-            self.mr = self.cmid.reg_msgs(self.msg_size + GRH_SIZE)
+        cmid = self.child_id if self.passive else self.cmid
+        mr_remote_function = {None: cmid.reg_msgs, 'read': cmid.reg_read,
+                              'write': cmid.reg_write}
+        self.mr = mr_remote_function[self.remote_operation](self.msg_size + GRH_SIZE)
 
     def create_event_channel(self):
         self.channel = CMEventChannel()

tests/rdmacm_utils.py

@@ -55,6 +55,39 @@ def rdmacm_traffic(self, server=None, multicast=False):
             else:
                 self._cmid_client_traffic(multicast)
 
+    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=e.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,

tests/test_rdmacm.py

@@ -114,6 +114,33 @@ def rdmacm_multicast_traffic(self, connection_resources=None, passive=None,
                         .format(side=side, pid=os.getpid()) +
                         'Exception message: {ex}'.format(ex=str(ex)))
 
+    def rdmacm_remote_traffic(self, connection_resources=None, passive=None,
+                              remote_op='write', **kwargs):
+        """
+        Run RDMACM remote traffic between two CMIDs.
+        :param connection_resources: The connection resources to use.
+        :param passive: Indicate if this CMID is the passive side.
+        :param remote_op: The remote operation in the traffic.
+        :param kwargs: Arguments to be passed to the connection_resources.
+        :return: None
+        """
+        try:
+            player = connection_resources(ip_addr=self.ip_addr,
+                                          syncer=self.syncer,
+                                          notifier=self.notifier,
+                                          passive=passive,
+                                          remote_op=remote_op, **kwargs)
+            player.establish_connection()
+            player.remote_traffic(passive=passive, remote_op=remote_op)
+            player.disconnect()
+        except Exception as ex:
+            while not self.notifier.empty():
+                self.notifier.get()
+            side = 'passive' if passive else 'active'
+            msg = f'Caught exception in {side} side process: pid {os.getpid()}\n' \
+                  f'Exception message: {str(ex)}'
+            self.notifier.put(msg)
+
 
     def test_rdmacm_sync_traffic(self):
         self.two_nodes_rdmacm_traffic(CMSyncConnection, self.rdmacm_traffic)
@@ -138,3 +165,13 @@ def test_rdmacm_async_traffic_external_qp(self):
     def test_rdmacm_async_udp_traffic(self):
         self.two_nodes_rdmacm_traffic(CMAsyncConnection, self.rdmacm_traffic,
                                       port_space=ce.RDMA_PS_UDP)
+
+    def test_rdmacm_async_read(self):
+        self.two_nodes_rdmacm_traffic(CMAsyncConnection,
+                                      self.rdmacm_remote_traffic,
+                                      remote_op='read')
+
+    def test_rdmacm_async_write(self):
+        self.two_nodes_rdmacm_traffic(CMAsyncConnection,
+                                      self.rdmacm_remote_traffic,
+                                      remote_op='write')