UCP: tm offload rndv fixes p3

openucx · Jun 1, 2017 · 2ba0e9a · 2ba0e9a
1 parent bd9c7b8
commit 2ba0e9a
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 21 deletions.
diff --git a/src/ucp/tag/offload.c b/src/ucp/tag/offload.c
@@ -56,6 +56,7 @@ void ucp_tag_offload_rndv_cb(uct_tag_context_t *self, uct_tag_t stag,
                                                               ucp_worker_iface_t, queue);
     ucp_rndv_rts_hdr_t rts;
 
+    /* Emulate RTS without rkey (to be handled as AM-based RNDV). */
     rts.sreq      = sreq->super;
     rts.super.tag = stag;
     rts.flags     = 0;
@@ -68,11 +69,11 @@ void ucp_tag_offload_rndv_cb(uct_tag_context_t *self, uct_tag_t stag,
     ucp_rndv_matched(req->recv.worker, req, &rts);
 }
 
-ucs_status_t ucp_tag_offload_unexp_rndv(void *arg, unsigned flags,
-                                        uint64_t stag, const void *hdr,
-                                        unsigned hdr_length,
-                                        uint64_t remote_addr, size_t length,
-                                        const void *rkey_buf)
+UCS_PROFILE_FUNC(ucs_status_t, ucp_tag_offload_unexp_rndv,
+                 (arg, flags, stag, hdr, hdr_length, remote_addr, length, rkey_buf),
+                 void *arg, unsigned flags, uint64_t stag, const void *hdr,
+                 unsigned hdr_length, uint64_t remote_addr, size_t length,
+                 const void *rkey_buf)
 {
     ucp_worker_t *worker          = arg;
     ucp_request_hdr_t *rndv_hdr   = (ucp_request_hdr_t*)hdr;
@@ -115,17 +116,19 @@ void ucp_tag_offload_cancel(ucp_context_t *ctx, ucp_request_t *req, int force)
     ucp_worker_iface_t *ucp_iface;
     ucs_status_t status;
 
-    if (req->flags & UCP_REQUEST_FLAG_OFFLOADED) {
-        ucp_iface = ucs_queue_head_elem_non_empty(&ctx->tm.offload_ifaces,
-                                                   ucp_worker_iface_t, queue);
-        ucp_request_memory_dereg(ctx, ucp_iface->rsc_index, req->recv.datatype,
-                                 &req->recv.state);
-        status = uct_iface_tag_recv_cancel(ucp_iface->iface, &req->recv.uct_ctx,
-                                           force);
-        if (status != UCS_OK) {
-            ucs_error("Failed to cancel recv in the transport: %s",
-                    ucs_status_string(status));
-        }
+    if (!(req->flags & UCP_REQUEST_FLAG_OFFLOADED)) {
+        return;
+    }
+
+    ucp_iface = ucs_queue_head_elem_non_empty(&ctx->tm.offload_ifaces,
+                                              ucp_worker_iface_t, queue);
+    ucp_request_memory_dereg(ctx, ucp_iface->rsc_index, req->recv.datatype,
+                             &req->recv.state);
+    status = uct_iface_tag_recv_cancel(ucp_iface->iface, &req->recv.uct_ctx,
+                                       force);
+    if (status != UCS_OK) {
+        ucs_error("Failed to cancel recv in the transport: %s",
+                  ucs_status_string(status));
     }
 }
 

diff --git a/src/ucp/tag/rndv.c b/src/ucp/tag/rndv.c
@@ -431,9 +431,8 @@ UCS_PROFILE_FUNC_VOID(ucp_rndv_matched, (worker, rreq, rndv_rts_hdr),
     UCS_ASYNC_UNBLOCK(&worker->async);
 }
 
-UCS_PROFILE_FUNC(ucs_status_t, ucp_rndv_process_rts,
-                 (arg, data, length, tl_flags),
-                 void *arg, void *data, size_t length, unsigned tl_flags)
+ucs_status_t ucp_rndv_process_rts(void *arg, void *data, size_t length,
+                                  unsigned tl_flags)
 {
     const unsigned recv_flags = UCP_RECV_DESC_FLAG_FIRST |
                                 UCP_RECV_DESC_FLAG_LAST  |
@@ -466,8 +465,9 @@ UCS_PROFILE_FUNC(ucs_status_t, ucp_rndv_process_rts,
     return status;
 }
 
-ucs_status_t ucp_rndv_rts_handler(void *arg, void *data, size_t length,
-                                  unsigned tl_flags)
+UCS_PROFILE_FUNC(ucs_status_t, ucp_rndv_rts_handler,
+                 (arg, data, length, tl_flags),
+                 void *arg, void *data, size_t length, unsigned tl_flags)
 {
     return ucp_rndv_process_rts(arg, data, length, tl_flags);
 }