ch4: add CS_YIELD in netmod dynamic_sendrecv

The dynamic_sendrecv is used in MPI_Intercomm_create. The mismatching
between threads are protected by the user provided tag, thus it is okay
to yield during the blocking progress. Without the yield,
MPI_Intercomm_create may block another thread's progress when the remote
processes are not present (blocked by other communications).

In the dynamic process accept/connect path, we force peer_comm's context
id to 0.  This is okay because the leader exchange is established with a
specific pair of addresses and there is no other communications yet
during leader_exchange.

Author: hzhou

src/mpid/ch4/ch4_api.txt

@@ -91,7 +91,7 @@ Non Native API:
   dynamic_recv : int
       NM : tag, buf-2, size, timeout
   dynamic_sendrecv : int
-      NM : remote_lpid, tag, send_buf, send_size, recv_buf, recv_size, timeout
+      NM : remote_lpid, peer_comm, tag, send_buf, send_size, recv_buf, recv_size, timeout
   mpi_comm_commit_pre_hook : int
       NM : comm
      SHM : comm
@@ -489,6 +489,7 @@ PARAM:
     origin_count: MPI_Aint
     origin_datatype: MPI_Datatype
     partner: MPIR_Request *
+    peer_comm: MPIR_Comm *
     port_name: const char *
     port_name-2: char *
     ptr: void *

src/mpid/ch4/netmod/ofi/ofi_spawn.c

@@ -17,7 +17,7 @@
 
 static int cancel_dynamic_request(MPIDI_OFI_dynamic_process_request_t * dynamic_req, bool is_send);
 static uint64_t get_dynamic_connection_match_bits(int tag);
-static uint64_t get_dynamic_match_bits(MPIR_Lpid lpid, int tag);
+static uint64_t get_dynamic_match_bits(MPIR_Lpid lpid, int context_id, int tag);
 
 int MPIDI_OFI_dynamic_send(MPIR_Lpid remote_lpid, int tag, const void *buf, int size, int timeout)
 {
@@ -111,7 +111,7 @@ int MPIDI_OFI_dynamic_recv(int tag, void *buf, int size, int timeout)
     goto fn_exit;
 }
 
-int MPIDI_OFI_dynamic_sendrecv(MPIR_Lpid remote_lpid, int tag,
+int MPIDI_OFI_dynamic_sendrecv(MPIR_Lpid remote_lpid, MPIR_Comm * peer_comm, int tag,
                                const void *send_buf, int send_size, void *recv_buf, int recv_size,
                                int timeout)
 {
@@ -132,7 +132,7 @@ int MPIDI_OFI_dynamic_sendrecv(MPIR_Lpid remote_lpid, int tag,
     send_req.event_id = MPIDI_OFI_EVENT_DYNPROC_DONE;
 
     if (send_size > 0) {
-        uint64_t match_bits = get_dynamic_match_bits(MPIR_Process.rank, tag);
+        uint64_t match_bits = get_dynamic_match_bits(MPIR_Process.rank, peer_comm->context_id, tag);
         MPIDI_OFI_CALL_RETRY(fi_tsend(MPIDI_OFI_global.ctx[ctx_idx].tx,
                                       send_buf, send_size, NULL,
                                       remote_addr, match_bits, (void *) &send_req.context),
@@ -147,7 +147,7 @@ int MPIDI_OFI_dynamic_sendrecv(MPIR_Lpid remote_lpid, int tag,
 
     if (recv_size > 0) {
         uint64_t mask_bits = 0;
-        uint64_t match_bits = get_dynamic_match_bits(remote_lpid, tag);
+        uint64_t match_bits = get_dynamic_match_bits(remote_lpid, peer_comm->recvcontext_id, tag);
         MPIDI_OFI_CALL_RETRY(fi_trecv(MPIDI_OFI_global.ctx[ctx_idx].rx,
                                       recv_buf, recv_size, NULL,
                                       remote_addr, match_bits, mask_bits, &recv_req.context),
@@ -182,6 +182,8 @@ int MPIDI_OFI_dynamic_sendrecv(MPIR_Lpid remote_lpid, int tag,
                 break;
             }
         }
+        MPID_THREAD_CS_YIELD(GLOBAL, MPIR_THREAD_GLOBAL_ALLFUNC_MUTEX);
+        MPID_THREAD_CS_YIELD(VCI, MPIDI_VCI_LOCK(vci));
     }
 
   fn_exit:
@@ -295,12 +297,13 @@ int MPIDI_OFI_insert_upid(MPIR_Lpid lpid, const char *upid, int upid_len)
 /* -- internal static routines */
 
 /* NOTE: used by MPIDI_OFI_dynamic_sendrecv, exact source match */
-static uint64_t get_dynamic_match_bits(MPIR_Lpid lpid, int tag)
+static uint64_t get_dynamic_match_bits(MPIR_Lpid lpid, int context_id, int tag)
 {
     /* normalize tag within (MPIDI_OFI_TAG_BITS - 1) bits, reserve 1 bit for dynamic connect/accept */
     tag &= (1 << (MPIDI_OFI_TAG_BITS - 1)) - 1;
 
-    uint64_t match_bits = MPIDI_OFI_DYNPROC_SEND | tag;
+    uint64_t match_bits;
+    match_bits = context_id;
 
     if (!MPIDI_OFI_ENABLE_DATA) {
         /* FI_DIRECTED_RECV is not enabled, we have to embed source in the match_bits */
@@ -314,9 +317,15 @@ static uint64_t get_dynamic_match_bits(MPIR_Lpid lpid, int tag)
         HASH_VALUE(upid, sz, upid_hash);
         upid_hash &= (1 << MPIDI_OFI_SOURCE_BITS) - 1;
 
-        match_bits |= (upid_hash << MPIDI_OFI_TAG_BITS);
+        match_bits <<= MPIDI_OFI_SOURCE_BITS;
+        match_bits |= upid_hash;
     }
 
+    match_bits <<= MPIDI_OFI_TAG_BITS;
+    match_bits |= tag;
+
+    match_bits |= MPIDI_OFI_DYNPROC_SEND;
+
     return match_bits;
 }
 

src/mpid/ch4/netmod/ucx/ucx_spawn.c

@@ -120,7 +120,12 @@ int MPIDI_UCX_dynamic_recv(int tag, void *buf, int size, int timeout)
     return mpi_errno;
 }
 
-int MPIDI_UCX_dynamic_sendrecv(MPIR_Lpid remote_lpid, int tag,
+static uint64_t get_dynamic_match_bits(int context_id, int tag)
+{
+    return MPIDI_UCX_DYNPROC_MASK | ((uint64_t) context_id << MPIDI_UCX_TAG_BITS) | tag;
+}
+
+int MPIDI_UCX_dynamic_sendrecv(MPIR_Lpid remote_lpid, MPIR_Comm * peer_comm, int tag,
                                const void *send_buf, int send_size, void *recv_buf, int recv_size,
                                int timeout)
 {
@@ -132,7 +137,6 @@ int MPIDI_UCX_dynamic_sendrecv(MPIR_Lpid remote_lpid, int tag,
     MPID_THREAD_ASSERT_IN_CS(VCI, MPIDI_VCI_LOCK(vci));
 #endif
 
-    uint64_t ucx_tag = MPIDI_UCX_DYNPROC_MASK + tag;
     uint64_t tag_mask = 0xffffffffffffffff;     /* for recv */
     MPIDI_av_entry_t *av = MPIDIU_lpid_to_av_slow(remote_lpid);
     ucp_ep_h ep = MPIDI_UCX_AV_TO_EP(av, vci, vci);
@@ -142,6 +146,8 @@ int MPIDI_UCX_dynamic_sendrecv(MPIR_Lpid remote_lpid, int tag,
     /* send */
     bool send_done = false;
     if (send_size > 0) {
+        uint64_t ucx_tag = get_dynamic_match_bits(peer_comm->context_id, tag);
+
         ucp_request_param_t send_param = {
             .op_attr_mask = UCP_OP_ATTR_FIELD_CALLBACK | UCP_OP_ATTR_FIELD_USER_DATA,
             .cb.send = dynamic_send_cb,
@@ -163,6 +169,8 @@ int MPIDI_UCX_dynamic_sendrecv(MPIR_Lpid remote_lpid, int tag,
     /* recv */
     bool recv_done = false;
     if (recv_size > 0) {
+        uint64_t ucx_tag = get_dynamic_match_bits(peer_comm->recvcontext_id, tag);
+
         ucp_request_param_t recv_param = {
             .op_attr_mask = UCP_OP_ATTR_FIELD_CALLBACK | UCP_OP_ATTR_FIELD_USER_DATA,
             .cb.recv = dynamic_recv_cb,
@@ -198,6 +206,8 @@ int MPIDI_UCX_dynamic_sendrecv(MPIR_Lpid remote_lpid, int tag,
                 break;
             }
         }
+        MPID_THREAD_CS_YIELD(GLOBAL, MPIR_THREAD_GLOBAL_ALLFUNC_MUTEX);
+        MPID_THREAD_CS_YIELD(VCI, MPIDI_VCI_LOCK(vci));
     }
 
   fn_exit:

src/mpid/ch4/src/ch4_comm.c

@@ -324,7 +324,8 @@ int MPID_Comm_set_hints(MPIR_Comm * comm_ptr, MPIR_Info * info_ptr)
  * 1. leader exchange data.
  * 2. leader broadcast over local_comm.
  */
-static int leader_exchange(MPIR_Comm * local_comm, MPIR_Lpid remote_lpid, int tag,
+static int leader_exchange(MPIR_Comm * local_comm, MPIR_Lpid remote_lpid,
+                           MPIR_Comm * peer_comm, int tag,
                            int context_id, int *remote_data_size_out, void **remote_data_out,
                            int timeout);
 static int prepare_local_lpids(MPIR_Comm * local_comm, MPIR_Lpid ** lpids_out,
@@ -357,7 +358,7 @@ int MPID_Intercomm_exchange(MPIR_Comm * local_comm, int local_leader,
     void *remote_data = NULL;
     if (is_local_leader) {
         MPIR_Lpid remote_lpid = MPIR_comm_rank_to_lpid(peer_comm, remote_leader);
-        mpi_errno = leader_exchange(local_comm, remote_lpid, tag, context_id,
+        mpi_errno = leader_exchange(local_comm, remote_lpid, peer_comm, tag, context_id,
                                     &remote_data_size, &remote_data, timeout);
     }
 
@@ -625,7 +626,8 @@ static int extract_remote_data(void *remote_data, int *remote_size_out,
 }
 
 /* exchange data between leaders */
-static int leader_exchange(MPIR_Comm * local_comm, MPIR_Lpid remote_lpid, int tag, int context_id,
+static int leader_exchange(MPIR_Comm * local_comm, MPIR_Lpid remote_lpid,
+                           MPIR_Comm * peer_comm, int tag, int context_id,
                            int *remote_data_size_out, void **remote_data_out, int timeout)
 {
     int mpi_errno = MPI_SUCCESS;
@@ -673,14 +675,16 @@ static int leader_exchange(MPIR_Comm * local_comm, MPIR_Lpid remote_lpid, int ta
     /* exchange */
     int remote_data_size;
     void *remote_data;
-    mpi_errno = MPIDI_NM_dynamic_sendrecv(remote_lpid, tag, &local_data_size, sizeof(int),
+    mpi_errno = MPIDI_NM_dynamic_sendrecv(remote_lpid, peer_comm, tag,
+                                          &local_data_size, sizeof(int),
                                           &remote_data_size, sizeof(int), timeout);
     MPIR_ERR_CHECK(mpi_errno);
 
     remote_data = MPL_malloc(remote_data_size, MPL_MEM_OTHER);
     MPIR_ERR_CHKANDJUMP(!remote_data, mpi_errno, MPI_ERR_OTHER, "**nomem");
 
-    mpi_errno = MPIDI_NM_dynamic_sendrecv(remote_lpid, tag, local_data, local_data_size,
+    mpi_errno = MPIDI_NM_dynamic_sendrecv(remote_lpid, peer_comm, tag,
+                                          local_data, local_data_size,
                                           remote_data, remote_data_size, timeout);
     MPIR_ERR_CHECK(mpi_errno);
 

src/mpid/ch4/src/ch4_spawn.c

@@ -373,6 +373,11 @@ static int dynamic_intercomm_create(const char *port_name, MPIR_Info * info, int
         peer_comm->local_size = 1;
         peer_comm->rank = 0;
         peer_comm->local_group = NULL;
+        /* We have not exchanged context_id yet, set them to 0. This is okay since
+         * the dynamic exchange is established between a pair of addresses (lpids) that
+         * no other communications can happen yet. */
+        peer_comm->context_id = 0;
+        peer_comm->recvcontext_id = 0;
 
         MPIR_Group_create_stride(1, 0, NULL, remote_lpid, 1, &peer_comm->remote_group);