UCP: Route failover replay through protocol selection

Use owned UCT operation descriptors and internal replay requests so failover resend can reuse UCP protocol selection while remaining scoped to failover recovery.

Author: JeynmannZ

src/ucp/core/ucp_ep_failover.c

@@ -534,7 +534,7 @@ static ucs_status_t ucp_ep_failover_replay_lane(ucp_ep_failover_lane_ctx_t *lane
             return UCS_OK;
         } else if (status != UCS_OK) {
             ucs_debug("ep %p: lane %u failed to replay extracted op %d: %s",
-                      lane->ep, lane->lane, (int)op->operation,
+                      lane->ep, lane->lane, (int)op->info.operation,
                       ucs_status_string(status));
             lane->discard_status = status;
             ucp_ep_failover_replay_purge(lane);

src/ucp/core/ucp_request.h

@@ -24,6 +24,7 @@
 #include <ucp/wireup/wireup.h>
 #include <ucp/core/ucp_am.h>
 #include <ucp/core/ucp_worker.h>
+#include <uct/api/v2/uct_v2.h>
 
 
 #define ucp_trace_req(_sreq, _message, ...) \
@@ -403,6 +404,11 @@ struct ucp_request {
                     ucs_ptr_map_key_t remote_req_id;
                 } get_reply;
 
+                struct {
+                    const uct_ep_op_info_t *op_info;
+                    ucp_lane_index_t       failed_lane;
+                } failover;
+
                 struct {
                     /* Remote request ID received from a peer */
                     ucs_ptr_map_key_t  remote_req_id;

src/ucp/core/ucp_types.h

@@ -157,7 +157,14 @@ typedef enum {
     UCP_OP_ID_RNDV_RECV_DROP,
     UCP_OP_ID_RNDV_LAST,
 
-    UCP_OP_ID_LAST = UCP_OP_ID_RNDV_LAST
+    /* Internal failover replay operations */
+    UCP_OP_ID_FAILOVER_FIRST    = UCP_OP_ID_RNDV_LAST,
+    UCP_OP_ID_FAILOVER_AM_BCOPY = UCP_OP_ID_FAILOVER_FIRST,
+    UCP_OP_ID_FAILOVER_PUT_SHORT,
+    UCP_OP_ID_FAILOVER_PUT_BCOPY,
+    UCP_OP_ID_FAILOVER_LAST,
+
+    UCP_OP_ID_LAST = UCP_OP_ID_FAILOVER_LAST
 } ucp_operation_id_t;
 
 

src/ucp/proto/proto.c

@@ -71,6 +71,7 @@
     _macro(ucp_am_rndv_proto) \
     _macro(ucp_stream_multi_bcopy_proto) \
     _macro(ucp_stream_multi_zcopy_proto) \
+    _macro(ucp_failover_replay_proto) \
     UCP_PROTO_AMO_FOR_EACH(_macro, post) \
     UCP_PROTO_AMO_FOR_EACH(_macro, fetch) \
     UCP_PROTO_AMO_FOR_EACH(_macro, cswap)
@@ -87,20 +88,23 @@ const ucp_proto_t *ucp_protocols[] = {
 };
 
 const char *ucp_operation_names[] = {
-    [UCP_OP_ID_TAG_SEND]       = "tag_send",
-    [UCP_OP_ID_TAG_SEND_SYNC]  = "tag_send_sync",
-    [UCP_OP_ID_AM_SEND]        = "am_send",
-    [UCP_OP_ID_AM_SEND_REPLY]  = "am_send_reply",
-    [UCP_OP_ID_STREAM_SEND]    = "stream",
-    [UCP_OP_ID_PUT]            = "put",
-    [UCP_OP_ID_GET]            = "get",
-    [UCP_OP_ID_AMO_POST]       = "amo_post",
-    [UCP_OP_ID_AMO_FETCH]      = "amo_fetch",
-    [UCP_OP_ID_AMO_CSWAP]      = "amo_cswap",
-    [UCP_OP_ID_RNDV_SEND]      = "rndv_send",
-    [UCP_OP_ID_RNDV_RECV]      = "rndv_recv",
-    [UCP_OP_ID_RNDV_RECV_DROP] = "rndv_recv_drop",
-    [UCP_OP_ID_LAST]           = NULL
+    [UCP_OP_ID_TAG_SEND]           = "tag_send",
+    [UCP_OP_ID_TAG_SEND_SYNC]      = "tag_send_sync",
+    [UCP_OP_ID_AM_SEND]            = "am_send",
+    [UCP_OP_ID_AM_SEND_REPLY]      = "am_send_reply",
+    [UCP_OP_ID_STREAM_SEND]        = "stream",
+    [UCP_OP_ID_PUT]                = "put",
+    [UCP_OP_ID_GET]                = "get",
+    [UCP_OP_ID_AMO_POST]           = "amo_post",
+    [UCP_OP_ID_AMO_FETCH]          = "amo_fetch",
+    [UCP_OP_ID_AMO_CSWAP]          = "amo_cswap",
+    [UCP_OP_ID_RNDV_SEND]          = "rndv_send",
+    [UCP_OP_ID_RNDV_RECV]          = "rndv_recv",
+    [UCP_OP_ID_RNDV_RECV_DROP]     = "rndv_recv_drop",
+    [UCP_OP_ID_FAILOVER_AM_BCOPY]  = "failover_am_bcopy",
+    [UCP_OP_ID_FAILOVER_PUT_SHORT] = "failover_put_short",
+    [UCP_OP_ID_FAILOVER_PUT_BCOPY] = "failover_put_bcopy",
+    [UCP_OP_ID_LAST]               = NULL
 };
 
 const char *ucp_operation_descs[] = {
@@ -118,7 +122,10 @@ const char *ucp_operation_descs[] = {
     [UCP_OP_ID_RNDV_SEND]      = "rendezvous data send",
     [UCP_OP_ID_RNDV_RECV]      = "rendezvous data fetch",
     [UCP_OP_ID_RNDV_RECV_DROP] = "rendezvous data drop",
-    [UCP_OP_ID_LAST]           = NULL
+    [UCP_OP_ID_FAILOVER_AM_BCOPY]  = "failover replay active message",
+    [UCP_OP_ID_FAILOVER_PUT_SHORT] = "failover replay short put",
+    [UCP_OP_ID_FAILOVER_PUT_BCOPY] = "failover replay buffered put",
+    [UCP_OP_ID_LAST]               = NULL
 };
 
 unsigned ucp_protocols_count(void)

src/ucp/proto/proto.h

@@ -23,7 +23,7 @@
 
 
 /* Maximal number of protocols in total */
-#define UCP_PROTO_MAX_COUNT         65
+#define UCP_PROTO_MAX_COUNT 66
 
 
 /* Special value for non-existent protocol */