osc/rdma: do not use local leader when use btl to update peer state

On each node, there is a local leader, which setup shared
memory, and each process that map its state to leader's
shared memory.

Peer's states are update through a local leader, when
peer is on a different node, or peer is on same node
but CPU atomics cannot be used.

This is incorrect for btl like TCP because TCP requires
all communication to the peer through the same orderred
channel.

This patch make the change that peer's state is updated using the
same channel data is transferred.

It introduced a new function gather_peer_state_and_handle(),
which uses allgather to gather peer's state and memory registration
handle (if memory registration is used).

Then it sets peer's state and handle using gathered information.

Signed-off-by: Wei Zhang <[email protected]>

Author: wzamazon

ompi/mca/osc/rdma/osc_rdma.h

@@ -255,6 +255,11 @@ struct ompi_osc_rdma_module_t {
     /** lock for peer hash table/array */
     opal_mutex_t peer_lock;
 
+    /** array of peers state point */
+    uintptr_t *peer_state_array;
+
+    /** array of peers state's memory reigstration handle */
+    char *peer_state_handle_array;
 
     /** BTL(s) in use. Currently this is only used to support RDMA emulation over
      * non-RDMA BTLs. The typical usage is btl/sm + btl/tcp. In the future this

ompi/mca/osc/rdma/osc_rdma_component.c

@@ -446,6 +446,60 @@ static int ompi_osc_rdma_initialize_region (ompi_osc_rdma_module_t *module, void
     return OMPI_SUCCESS;
 }
 
+/**
+ * @brief gather information of module state and module state handle inside a shared comm
+ *
+ * @param	module[in]	ompi osc rdma module
+ * @param	peer_state_array
+ */
+static int gather_peer_state_and_handle(ompi_osc_rdma_module_t *module)
+{
+    int ret, handle_size, comm_size;
+
+    comm_size = ompi_comm_size (module->comm);
+
+    module->peer_state_array = calloc(comm_size, sizeof(uintptr_t));
+    if (NULL == module->peer_state_array) {
+        OSC_RDMA_VERBOSE(MCA_BASE_VERBOSE_ERROR, "failed to allocate memory for module state array!");
+        return OMPI_ERR_OUT_OF_RESOURCE;
+    }
+
+    ret = module->comm->c_coll->coll_allgather(&module->state, sizeof(uintptr_t), MPI_BYTE,
+                                               module->peer_state_array, sizeof(uintptr_t), MPI_BYTE,
+                                               module->comm, module->comm->c_coll->coll_allgather_module);
+    if (OMPI_SUCCESS != ret) {
+        OSC_RDMA_VERBOSE(MCA_BASE_VERBOSE_ERROR, "module state allgather failed with ompi error code %d", ret);
+        return ret;
+    }
+
+    if (module->use_memory_registration) {
+        handle_size = module->selected_btls[0]->btl_registration_handle_size;
+        module->peer_state_handle_array = calloc(comm_size, handle_size);
+        if (NULL == module->peer_state_handle_array) {
+            OSC_RDMA_VERBOSE(MCA_BASE_VERBOSE_ERROR, "failed to allocate memory for module state handle array!");
+            return OMPI_ERR_OUT_OF_RESOURCE;
+        }
+
+        ret = ompi_osc_rdma_register (module, MCA_BTL_ENDPOINT_ANY, module->state, module->state_size,
+                                      MCA_BTL_REG_FLAG_ACCESS_ANY, &module->state_handle);
+        if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
+            OSC_RDMA_VERBOSE(MCA_BASE_VERBOSE_ERROR, "module state memory registration failed with ompi error code %d", ret);
+            return ret;
+        }
+
+        ret = module->comm->c_coll->coll_allgather(module->state_handle, handle_size, MPI_BYTE,
+                                                   module->peer_state_handle_array, handle_size, MPI_BYTE,
+                                                   module->comm, module->comm->c_coll->coll_allgather_module);
+        if (OMPI_SUCCESS != ret) {
+            OSC_RDMA_VERBOSE(MCA_BASE_VERBOSE_ERROR, "module state allgather failed with ompi error code %d", ret);
+            return ret;
+        }
+    }
+
+    return 0;
+}
+
+
 static int allocate_state_single (ompi_osc_rdma_module_t *module, void **base, size_t size)
 {
     size_t total_size, local_rank_array_size, leader_peer_data_size;
@@ -491,20 +545,19 @@ static int allocate_state_single (ompi_osc_rdma_module_t *module, void **base, s
         *base = (void *) ((intptr_t) module->node_comm_info + leader_peer_data_size);
     }
 
-    /* just go ahead and register the whole segment */
-    ret = ompi_osc_rdma_register (module, MCA_BTL_ENDPOINT_ANY, module->rank_array, total_size,
-                                  MCA_BTL_REG_FLAG_ACCESS_ANY, &module->state_handle);
-    if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
-        return ret;
-    }
-
     if (MPI_WIN_FLAVOR_DYNAMIC != module->flavor) {
         ret = ompi_osc_rdma_initialize_region (module, base, size);
         if (OMPI_SUCCESS != ret) {
             return ret;
         }
     }
 
+    ret = gather_peer_state_and_handle(module);
+    if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
+        OSC_RDMA_VERBOSE(MCA_BASE_VERBOSE_ERROR, "failed to create shared memory segment");
+        return ret;
+    }
+
     ret = ompi_osc_rdma_new_peer (module, my_rank, &my_peer);
     if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
         return ret;
@@ -711,16 +764,6 @@ static int allocate_state_shared (ompi_osc_rdma_module_t *module, void **base, s
         if (0 == local_rank) {
             /* unlink the shared memory backing file */
             opal_shmem_unlink (&module->seg_ds);
-            /* just go ahead and register the whole segment */
-            ret = ompi_osc_rdma_register (module, MCA_BTL_ENDPOINT_ANY, module->segment_base, total_size,
-                                          MCA_BTL_REG_FLAG_ACCESS_ANY, &module->state_handle);
-            if (OPAL_LIKELY(OMPI_SUCCESS == ret)) {
-                state_region->base = (intptr_t) module->segment_base;
-                if (module->state_handle) {
-                    memcpy (state_region->btl_handle_data, module->state_handle,
-                            module->selected_btls[0]->btl_registration_handle_size);
-                }
-            }
         }
 
         /* synchronization to make sure memory is registered */
@@ -749,6 +792,11 @@ static int allocate_state_shared (ompi_osc_rdma_module_t *module, void **base, s
             break;
         }
 
+        ret = gather_peer_state_and_handle(module);
+        if (OPAL_UNLIKELY(OMPI_SUCCESS !=ret)) {
+            break;
+        }
+
         offset = data_base;
         ompi_osc_rdma_peer_t *local_leader;
         for (int i = 0 ; i < local_size ; ++i) {
@@ -777,18 +825,15 @@ static int allocate_state_shared (ompi_osc_rdma_module_t *module, void **base, s
                 peer->state = (osc_rdma_counter_t) peer_state;
                 peer->state_endpoint = NULL;
             } else {
-                /* use my endpoint handle to modify the peer's state */
                 if (module->use_memory_registration) {
-                    peer->state_handle = (mca_btl_base_registration_handle_t *) state_region->btl_handle_data;
-                }
-                peer->state = (osc_rdma_counter_t) ((uintptr_t) state_region->base + state_base + module->state_size * i);
-                if (i==0) {
-                    peer->state_endpoint = peer->data_endpoint;
-                    peer->state_btl_index = peer->data_btl_index;
-                } else {
-                    peer->state_endpoint = local_leader->state_endpoint;
-                    peer->state_btl_index = local_leader->state_btl_index;
+                    assert(module->peer_state_handle_array);
+                    peer->state_handle = (mca_btl_base_registration_handle_t *)(module->peer_state_handle_array + peer_rank * module->selected_btls[0]->btl_registration_handle_size);
                 }
+
+                assert(NULL != module->peer_state_array);
+                peer->state = (osc_rdma_counter_t)module->peer_state_array[peer_rank];
+                peer->state_endpoint = peer->data_endpoint;
+                peer->state_btl_index = peer->data_btl_index;
             }
 
             if (my_rank == peer_rank) {

ompi/mca/osc/rdma/osc_rdma_peer.c

@@ -138,50 +138,16 @@ static int ompi_osc_rdma_peer_setup (ompi_osc_rdma_module_t *module, ompi_osc_rd
         registration_handle_size = module->selected_btls[0]->btl_registration_handle_size;
     }
 
-    /* each node is responsible for holding a part of the rank -> node/local rank mapping array. this code
-     * calculates the node and offset the mapping can be found. once the mapping has been read the state
-     * part of the peer structure can be initialized. */
-    node_id = peer->rank / RANK_ARRAY_COUNT(module);
-    array_peer_data = (ompi_osc_rdma_region_t *) ((intptr_t) module->node_comm_info + node_id * module->region_size);
-
-    /* the node leader rank is stored in the length field */
-    node_rank = NODE_ID_TO_RANK(module, array_peer_data, node_id);
-    array_index = peer->rank % RANK_ARRAY_COUNT(module);
-
-    array_pointer = array_peer_data->base + array_index * sizeof (rank_data);
-
-    /* lookup the btl endpoint needed to retrieve the mapping */
-    ret = ompi_osc_rdma_peer_btl_endpoint (module, node_rank, &array_btl_index, &array_endpoint);
-    if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
-        return OMPI_ERR_UNREACH;
-    }
-
-    OSC_RDMA_VERBOSE(MCA_BASE_VERBOSE_DEBUG, "reading region data for %d from rank: %d, index: %d, pointer: 0x%" PRIx64
-                     ", size: %lu", peer->rank, node_rank, array_index, array_pointer, sizeof (rank_data));
-
-    ret = ompi_osc_get_data_blocking (module, array_btl_index, array_endpoint, array_pointer,
-                                      (mca_btl_base_registration_handle_t *) array_peer_data->btl_handle_data,
-                                      &rank_data, sizeof (rank_data));
-    if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
-        return ret;
-    }
-
-    /* initialize the state part of the peer object. NTH: for now the state data is for every node is stored on
-     * every node. this gives a good balance of code complexity and memory usage at this time. we take advantage
-     * of this by re-using the endpoint and pointer stored in the node_comm_info array. */
-    node_peer_data = (ompi_osc_rdma_region_t *) ((intptr_t) module->node_comm_info + rank_data.node_id * module->region_size);
-
-    peer->state = node_peer_data->base + module->state_offset + module->state_size * rank_data.rank;
+    assert(NULL != module->peer_stat_array);
+    peer->state = module->peer_state_array[peer->rank];
 
     if (registration_handle_size) {
-        peer->state_handle = (mca_btl_base_registration_handle_t *) node_peer_data->btl_handle_data;
+	assert(NULL != module->peer_state_array);
+        peer->state_handle = (mca_btl_base_registration_handle_t *)(module->peer_state_handle_array + peer->rank * registration_handle_size);
     }
 
-    ret = ompi_osc_rdma_peer_btl_endpoint (module, NODE_ID_TO_RANK(module, node_peer_data, rank_data.node_id),
-                                           &peer->state_btl_index, &peer->state_endpoint);
-    if (OPAL_UNLIKELY(OMPI_SUCCESS != ret)) {
-        return OPAL_ERR_UNREACH;
-    }
+    peer->state_btl_index = peer->data_btl_index;
+    peer->state_endpoint = peer->data_endpoint;
 
     /* nothing more to do for dynamic memory windows */
     if (MPI_WIN_FLAVOR_DYNAMIC == module->flavor) {