UCP/TAG: Add debug info for eager messages

yosefe · yosefe · commit f400075145d4 · 2020-09-16T19:39:21.000+03:00
diff --git a/src/ucp/core/ucp_request.inl b/src/ucp/core/ucp_request.inl
@@ -113,6 +113,8 @@ ucp_request_complete_tag_recv(ucp_worker_h worker, ucp_request_t *req,
          entry->send_tag   = req->recv.tag.info.sender_tag;
          entry->status     = state;
          entry->recvd_size = req->recv.tag.info.length;
+         memcpy(entry->udata, req->recv.buffer,
+                ucs_min(UCP_TAG_MAX_DATA, req->recv.tag.info.length));
      }
 
     UCS_PROFILE_REQUEST_EVENT(req, "complete_recv", status);
diff --git a/src/ucp/proto/proto_am.c b/src/ucp/proto/proto_am.c
@@ -88,6 +88,7 @@ void ucp_proto_am_zcopy_req_complete(ucp_request_t *req, ucs_status_t status)
 {
     ucs_assert(req->send.state.uct_comp.count == 0);
     ucp_request_send_buffer_dereg(req); /* TODO register+lane change */
+    ucp_send_request_update_data(req, "sent_zcopy");
     ucp_request_complete_send(req, status);
 }
 
diff --git a/src/ucp/proto/proto_am.inl b/src/ucp/proto/proto_am.inl
@@ -490,4 +490,21 @@ ucp_proto_ssend_ack_request_alloc(ucp_worker_h worker, uintptr_t ep_ptr)
     return req;
 }
 
+static inline void
+ucp_send_request_update_data(ucp_request_t *req, const char *status)
+{
+    ucp_worker_h worker = req->send.ep->worker;
+    ucp_tag_rndv_debug_entry_t *entry;
+
+    if (ucs_unlikely(worker->tm.rndv_debug.queue_length == 0)) {
+        return;
+    }
+
+    entry         = ucp_worker_rndv_debug_entry(worker, req->send.rndv_req_id);
+    entry->status = status;
+    memcpy(entry->ndata, req->send.buffer,
+           ucs_min(UCP_TAG_MAX_DATA, req->send.length));
+}
+
+
 #endif
diff --git a/src/ucp/tag/eager_rcv.c b/src/ucp/tag/eager_rcv.c
@@ -71,6 +71,21 @@ ucp_eager_offload_handler(void *arg, void *data, size_t length,
     return status;
 }
 
+static void ucp_recv_request_update_data(ucp_request_t *req, const char *status,
+                                         void *data, size_t length)
+{
+    ucp_worker_h worker = req->recv.worker;
+    ucp_tag_rndv_debug_entry_t *entry;
+
+    if (ucs_unlikely(worker->tm.rndv_debug.queue_length == 0)) {
+        return;
+    }
+
+    entry         = ucp_worker_rndv_debug_entry(worker, req->recv.req_id);
+    entry->status = status;
+    memcpy(entry->ndata, data, ucs_min(UCP_TAG_MAX_DATA, length));
+}
+
 static UCS_F_ALWAYS_INLINE ucs_status_t
 ucp_eager_tagged_handler(void *arg, void *data, size_t length, unsigned am_flags,
                          uint16_t flags, uint16_t hdr_len, uint16_t priv_length)
@@ -98,6 +113,9 @@ ucp_eager_tagged_handler(void *arg, void *data, size_t length, unsigned am_flags
     if (req != NULL) {
         ucp_eager_expected_handler(worker, req, data, recv_len, recv_tag, flags);
 
+        ucp_recv_request_update_data(req, "eager_recv",
+                                     UCS_PTR_BYTE_OFFSET(data, hdr_len), recv_len);
+
         if (flags & UCP_RECV_DESC_FLAG_EAGER_SYNC) {
             ucp_tag_eager_sync_send_ack(worker, data, flags);
         }
diff --git a/src/ucp/tag/eager_snd.c b/src/ucp/tag/eager_snd.c
@@ -133,6 +133,7 @@ static ucs_status_t ucp_tag_eager_contig_short(uct_pending_req_t *self)
         return status;
     }
 
+    ucp_send_request_update_data(req, "sent_short");
     ucp_request_complete_send(req, UCS_OK);
     return UCS_OK;
 }
@@ -144,6 +145,7 @@ static ucs_status_t ucp_tag_eager_bcopy_single(uct_pending_req_t *self)
     if (status == UCS_OK) {
         ucp_request_t *req = ucs_container_of(self, ucp_request_t, send.uct);
         ucp_request_send_generic_dt_finish(req);
+        ucp_send_request_update_data(req, "sent_bcopy_s");
         ucp_request_complete_send(req, UCS_OK);
     }
     return status;
@@ -159,6 +161,7 @@ static ucs_status_t ucp_tag_eager_bcopy_multi(uct_pending_req_t *self)
     if (status == UCS_OK) {
         ucp_request_t *req = ucs_container_of(self, ucp_request_t, send.uct);
         ucp_request_send_generic_dt_finish(req);
+        ucp_send_request_update_data(req, "sent_bcopy_m");
         ucp_request_complete_send(req, UCS_OK);
     } else if (status == UCP_STATUS_PENDING_SWITCH) {
         status = UCS_OK;
diff --git a/src/ucp/tag/rndv.c b/src/ucp/tag/rndv.c
@@ -62,6 +62,7 @@ static void ucp_rndv_complete_send(ucp_request_t *sreq, ucs_status_t status)
                  sreq, sreq->send.rndv_req_id, worker);
     }
 
+    ucp_send_request_update_data(sreq, "rndv_done");
     ucp_request_complete_send(sreq, status);
 }
 
@@ -117,30 +118,13 @@ ucp_rndv_get_req_add_debug_entry(ucp_request_t *rndv_req,
     entry->send_req       = NULL;
 }
 
-/* add debug entry for rndv send flow */
-static void
-ucp_rndv_send_add_debug_entry(ucp_request_t *req)
-{
-    ucp_tag_rndv_debug_entry_t *entry;
-
-    entry                 = ucp_rndv_add_debug_entry_common(req);
-    entry->type           = "rndv_send";
-    entry->rts_seq        = 0;
-    entry->send_tag       = req->send.msg_proto.tag.tag;
-    entry->recv_tag       = 0;
-    entry->remote_address = 0;
-    entry->remote_reqptr  = 0;
-    entry->rndv_get_req   = NULL;
-    entry->recv_req       = NULL;
-    entry->send_req       = req;
-}
-
 /* to be used from debugger */
 void ucp_rndv_print_debug_data(ucp_worker_h worker, const char *filename,
                                ucp_tag_t send_tag)
 {
     ucp_tag_rndv_debug_entry_t *entry;
     size_t i, count;
+    uint64_t ndata, udata;
     FILE *file;
 
     if (filename == NULL) {
@@ -159,13 +143,18 @@ void ucp_rndv_print_debug_data(ucp_worker_h worker, const char *filename,
         if ((send_tag != 0) && (send_tag != entry->send_tag)) {
             continue;
         }
+        memcpy(&udata, entry->udata, sizeof(udata));
+        memcpy(&ndata, entry->ndata, sizeof(ndata));
         fprintf(file,
-                "%s id %lu rts_seq %lu stag 0x%lx rtag 0x%lx rva 0x%lx rmreq 0x%lx "
-                "lva %p sz %zu greq %p rreq %p sreq %p\n",
-                entry->type, entry->id, entry->rts_seq, entry->send_tag,
-                entry->recv_tag, entry->remote_address, entry->remote_reqptr,
-                entry->local_address, entry->size, entry->rndv_get_req,
-                entry->recv_req, entry->send_req);
+                "id %lu %s st '%s' rts_seq %lu pend %u stag 0x%lx rtag 0x%lx rva 0x%lx "
+                "rmreq %lu lva %p sz %zu rxsz %zu greq %p sreq %p rreq %p "
+                "udata 0x%"PRIx64" ndata 0x%"PRIx64"\n",
+                entry->id, entry->type, entry->status, entry->rts_seq,
+                entry->pending_count, entry->send_tag, entry->recv_tag,
+                entry->remote_address, entry->remote_reqptr,
+                entry->local_address, entry->size, entry->recvd_size,
+                entry->rndv_get_req, entry->send_req, entry->recv_req,
+                udata, ndata);
     }
 
     if (filename != NULL) {
@@ -378,11 +367,6 @@ ucs_status_t ucp_tag_send_start_rndv(ucp_request_t *sreq)
 
     sreq->flags |= UCP_REQUEST_FLAG_SEND_RNDV;
 
-    sreq->send.rndv_req_id = worker->rndv_req_id++;
-    if (ucs_unlikely(worker->tm.rndv_debug.queue_length > 0)) {
-        ucp_rndv_send_add_debug_entry(sreq);
-    }
-
     status = ucp_ep_resolve_dest_ep_ptr(ep, sreq->send.lane);
     if (status != UCS_OK) {
         return status;
diff --git a/src/ucp/tag/tag_match.h b/src/ucp/tag/tag_match.h
@@ -53,12 +53,13 @@ typedef union {
 
 KHASH_INIT(ucp_tag_frag_hash, uint64_t, ucp_tag_frag_match_t, 1,
            kh_int64_hash_func, kh_int64_hash_equal);
+#define UCP_TAG_MAX_DATA 8
 
 
 typedef struct ucp_tag_rndv_debug_entry {
+    uint64_t          id;
     const char        *type;
     const char        *status;
-    uint64_t          id;
     uint64_t          rts_seq;
     unsigned          pending_count;
     ucp_ep_h          ep;
@@ -72,6 +73,8 @@ typedef struct ucp_tag_rndv_debug_entry {
     ucp_request_t     *rndv_get_req;
     ucp_request_t     *send_req;
     ucp_request_t     *recv_req;
+    uint8_t           udata[UCP_TAG_MAX_DATA];
+    uint8_t           ndata[UCP_TAG_MAX_DATA]; /* data for the network */
 } ucp_tag_rndv_debug_entry_t;
 
 
diff --git a/src/ucp/tag/tag_recv.c b/src/ucp/tag/tag_recv.c
@@ -32,8 +32,10 @@ ucp_tag_recv_request_completed(ucp_worker_h worker, ucp_request_t *req,
         ucp_tag_rndv_debug_entry_t *entry =
                 ucp_worker_rndv_debug_entry(worker, req->recv.req_id);
         entry->send_tag   = info->sender_tag;
-        entry->status     = "recv_completed1";
+        entry->status     = "recv_completed_unexp";
         entry->recvd_size = info->length;
+        memcpy(entry->udata, req->recv.buffer,
+               ucs_min(UCP_TAG_MAX_DATA, info->length));
     }
 
     req->status = status;
diff --git a/src/ucp/tag/tag_send.c b/src/ucp/tag/tag_send.c
@@ -120,11 +120,34 @@ ucp_tag_send_req(ucp_request_t *req, size_t dt_count,
     return req + 1;
 }
 
+static void ucp_tag_send_add_debug_entry(ucp_request_t *req)
+{
+    ucp_tag_rndv_debug_entry_t *entry = ucp_worker_rndv_debug_entry(req->send.ep->worker,
+                                                                    req->send.rndv_req_id);
+    entry->id             = req->send.rndv_req_id;
+    entry->type           = "tag_send";
+    entry->ep             = req->send.ep;
+    entry->local_address  = req->send.buffer;
+    entry->size           = req->send.length;
+    entry->rts_seq        = 0;
+    entry->send_tag       = req->send.msg_proto.tag.tag;
+    entry->recv_tag       = 0;
+    entry->remote_address = 0;
+    entry->remote_reqptr  = 0;
+    entry->rndv_get_req   = NULL;
+    entry->recv_req       = NULL;
+    entry->send_req       = req;
+    memcpy(entry->udata, req->send.buffer,
+           ucs_min(UCP_TAG_MAX_DATA, req->send.length));
+}
+
 static UCS_F_ALWAYS_INLINE void
 ucp_tag_send_req_init(ucp_request_t* req, ucp_ep_h ep, const void* buffer,
                       uintptr_t datatype, size_t count, ucp_tag_t tag,
                       uint32_t flags)
 {
+    ucp_worker_h worker = ep->worker;
+
     req->flags                  = flags | UCP_REQUEST_FLAG_SEND_TAG;
     req->send.ep                = ep;
     req->send.buffer            = (void*)buffer;
@@ -139,7 +162,11 @@ ucp_tag_send_req_init(ucp_request_t* req, ucp_ep_h ep, const void* buffer,
                                                     req->send.length);
     req->send.lane         = ucp_ep_config(ep)->tag.lane;
     req->send.pending_lane = UCP_NULL_LANE;
-    req->send.rndv_req_id  = 0;
+    req->send.rndv_req_id  = worker->rndv_req_id++;
+
+    if (ucs_unlikely(worker->tm.rndv_debug.queue_length > 0)) {
+        ucp_tag_send_add_debug_entry(req);
+    }
 }
 
 //static UCS_F_ALWAYS_INLINE int
diff --git a/test/apps/iodemo/io_demo.cc b/test/apps/iodemo/io_demo.cc
@@ -189,6 +189,7 @@ class P2pDemoCommon : public UcxContext {
         for (size_t i = 0; i < _data_buffers.size(); ++i) {
             std::string &data_buffer = _data_buffers[i];
             data_buffer.resize(opts().max_data_size + ALIGNMENT);
+            std::fill(data_buffer.begin(), data_buffer.end(), 'A');
             uintptr_t ptr = (uintptr_t)&data_buffer[0];
             _padding = ((ptr + ALIGNMENT - 1) & ~(ALIGNMENT - 1)) - ptr;
         }

Original file line number	Diff line number	Diff line change
`@@ -113,6 +113,8 @@ ucp_request_complete_tag_recv(ucp_worker_h worker, ucp_request_t *req,`
`113`	`113`	`entry->send_tag = req->recv.tag.info.sender_tag;`
`114`	`114`	`entry->status = state;`
`115`	`115`	`entry->recvd_size = req->recv.tag.info.length;`
	`116`	`+ memcpy(entry->udata, req->recv.buffer,`
	`117`	`+ ucs_min(UCP_TAG_MAX_DATA, req->recv.tag.info.length));`
`116`	`118`	`}`
`117`	`119`
`118`	`120`	`UCS_PROFILE_REQUEST_EVENT(req, "complete_recv", status);`
Original file line number	Diff line number	Diff line change
`@@ -88,6 +88,7 @@ void ucp_proto_am_zcopy_req_complete(ucp_request_t *req, ucs_status_t status)`
`88`	`88`	`{`
`89`	`89`	`ucs_assert(req->send.state.uct_comp.count == 0);`
`90`	`90`	`ucp_request_send_buffer_dereg(req); /* TODO register+lane change */`
	`91`	`+ ucp_send_request_update_data(req, "sent_zcopy");`
`91`	`92`	`ucp_request_complete_send(req, status);`
`92`	`93`	`}`
`93`	`94`
Original file line number	Diff line number	Diff line change
`@@ -189,6 +189,7 @@ class P2pDemoCommon : public UcxContext {`
`189`	`189`	`for (size_t i = 0; i < _data_buffers.size(); ++i) {`
`190`	`190`	`std::string &data_buffer = _data_buffers[i];`
`191`	`191`	`data_buffer.resize(opts().max_data_size + ALIGNMENT);`
	`192`	`+ std::fill(data_buffer.begin(), data_buffer.end(), 'A');`
`192`	`193`	`uintptr_t ptr = (uintptr_t)&data_buffer[0];`
`193`	`194`	`_padding = ((ptr + ALIGNMENT - 1) & ~(ALIGNMENT - 1)) - ptr;`
`194`	`195`	`}`