sessions: fixes to accomodate ULFM

- fix arguments to ompi_comm_set in comm_ft.c
- use accessor function to acquire local cid from a comm
- add back in PMIx event handler to handle PMIX_ERR_PROC_ABORTED
  default event handler approach doesn't work
- only use the excid algorithm for communicators being
  created from MPI_Comm_create_from_group to avoid issues with
  trying to use ULFM method on communicators not using a global cid

Improving interoperability of ULFM with communicators derived from
sessions can be addressed by later PRs.

Related to hpc#67

Note that at the time this commit was made the ulfm api tests
5, 10,11, 12 fail, but this is a subset of the failures seen with
master.  For master failures are observd with 5,9,10,11,12.

Signed-off-by: Howard Pritchard <[email protected]>

Author: hppritcha

ompi/communicator/comm_cid.c

@@ -421,10 +421,8 @@ int ompi_comm_nextcid_nb (ompi_communicator_t *newcomm, ompi_communicator_t *com
 {
     ompi_comm_cid_context_t *context;
     ompi_comm_request_t *request;
-    if (mca_pml_base_supports_extended_cid() && OMPI_COMM_CID_INTER != mode &&
-        OMPI_COMM_CID_INTRA_BRIDGE != mode && OMPI_COMM_CID_INTRA_PMIX != mode &&
-        OMPI_COMM_CID_INTRA_FT != mode && OMPI_COMM_CID_INTER_FT != mode &&
-         OMPI_COMM_CID_INTRA_PMIX_FT != mode) {
+
+    if (mca_pml_base_supports_extended_cid() && NULL == comm) {
         return ompi_comm_nextcid_ext_nb (newcomm, comm, bridgecomm, arg0, arg1, send_first, mode, req);
     }
 

ompi/communicator/ft/comm_ft.c

@@ -169,8 +169,8 @@ int ompi_comm_shrink_internal(ompi_communicator_t* comm, ompi_communicator_t** n
      */
     /* --------------------------------------------------------- */
     OPAL_OUTPUT_VERBOSE((5, ompi_ftmpi_output_handle,
-                         "%s ompi: comm_shrink: Determine ranking for new communicator",
-                         OMPI_NAME_PRINT(OMPI_PROC_MY_NAME) ));
+                         "%s ompi: comm_shrink: Determine ranking for new communicator intra %d",
+                         OMPI_NAME_PRINT(OMPI_PROC_MY_NAME), OMPI_COMM_IS_INTRA(comm)));
     start = PMPI_Wtime();
 
     /* Create 'alive' groups */
@@ -198,9 +198,9 @@ int ompi_comm_shrink_internal(ompi_communicator_t* comm, ompi_communicator_t** n
                          NULL,                     /* remote_ranks */
                          comm->c_keyhash,          /* attrs */
                          comm->error_handler,      /* error handler */
-                         NULL,                     /* topo component */
                          alive_group,              /* local group */
-                         alive_rgroup              /* remote group */
+                         alive_rgroup,             /* remote group */
+                         0                         /* flags */
                        );
     if( OMPI_SUCCESS != ret ) {
         exit_status = ret;

ompi/communicator/ft/comm_ft_detector.c

@@ -382,7 +382,7 @@ static int fd_heartbeat_request(comm_detector_t* detector) {
         detector->hb_observing = rank;
 
         ompi_comm_heartbeat_req_t* msg = calloc(sizeof(*msg)+regsize, 1);
-        msg->super.cid = comm->c_index;
+        msg->super.cid = ompi_comm_get_local_cid(comm);
         msg->super.epoch = comm->c_epoch;
         msg->super.type = comm_heartbeat_request_cb_type;
         msg->from = comm->c_my_rank;
@@ -676,7 +676,7 @@ static int fd_heartbeat_send(comm_detector_t* detector) {
 
     /* send the heartbeat with eager send */
     ompi_comm_heartbeat_message_t msg;
-    msg.super.cid = comm->c_index;
+    msg.super.cid = ompi_comm_get_local_cid(comm);
     msg.super.epoch = comm->c_epoch;
     msg.super.type = comm_heartbeat_recv_cb_type;
     msg.from = detector->hb_rdma_rank; /* comm->c_my_rank; except during finalize when it is equal to detector->hb_observer */

ompi/communicator/ft/comm_ft_propagator.c

@@ -74,7 +74,7 @@ int ompi_comm_failure_propagate(ompi_communicator_t* comm, ompi_proc_t* proc, in
 
     ompi_comm_failure_propagator_message_t msg;
     /* Broadcast the 'failure_propagator' signal to all other processes. */
-    msg.rbcast_msg.cid   = comm->c_index;
+    msg.rbcast_msg.cid   = ompi_comm_get_local_cid(comm);
     msg.rbcast_msg.epoch = comm->c_epoch;
     msg.rbcast_msg.type  = comm_failure_propagator_cb_type;
     msg.proc_name        = proc->super.proc_name;

ompi/communicator/ft/comm_ft_reliable_bcast.c

@@ -182,7 +182,7 @@ static void ompi_comm_rbcast_bml_recv_cb(
                              OMPI_NAME_PRINT(OMPI_PROC_MY_NAME), __func__, msg->cid, msg->epoch));
         return;
     }
-    if(OPAL_UNLIKELY( msg->cid != comm->c_index )) {
+    if(OPAL_UNLIKELY( msg->cid != ompi_comm_get_local_cid(comm))) {
         OPAL_OUTPUT_VERBOSE((2, ompi_ftmpi_output_handle,
                              "%s %s: Info: received a late rbcast message with CID %3d:%d during an MPI_COMM_DUP that is trying to reuse that CID (thus increasing the epoch) - ignoring, nothing to do",
                              OMPI_NAME_PRINT(OMPI_PROC_MY_NAME), __func__, msg->cid, msg->epoch));

ompi/communicator/ft/comm_ft_revoke.c

@@ -58,7 +58,7 @@ int ompi_comm_revoke_internal(ompi_communicator_t* comm)
     if( ompi_comm_revoke_local(comm, NULL) ) {
         /* Broadcast the 'revoke' signal to all other processes. */
         ompi_comm_rbcast_message_t msg;
-        msg.cid   = comm->c_index;
+        msg.cid   = ompi_comm_get_local_cid(comm);
         msg.epoch = comm->c_epoch;
         msg.type  = comm_revoke_cb_type;
         ret = ompi_comm_rbcast(comm, &msg, sizeof(msg));

ompi/instance/instance.c

@@ -376,12 +376,28 @@ static int ompi_mpi_instance_init_common (void)
     ompi_rte_initialized = true;
 
     /* Register the default errhandler callback  */
+    /* give it a name so we can distinguish it */
+    PMIX_INFO_LOAD(&info[0], PMIX_EVENT_HDLR_NAME, "MPI-Default", PMIX_STRING);
+    OPAL_PMIX_CONSTRUCT_LOCK(&mylock);
+    PMIx_Register_event_handler(NULL, 0, info, 1, ompi_errhandler_callback, evhandler_reg_callbk, (void*)&mylock);
+    OPAL_PMIX_WAIT_THREAD(&mylock);
+    rc = mylock.status;
+    OPAL_PMIX_DESTRUCT_LOCK(&mylock);
+    PMIX_INFO_DESTRUCT(&info[0]);
+    if (PMIX_SUCCESS != rc) {
+        ret = opal_pmix_convert_status(rc);
+        return ret;
+    }
+
+#if OPAL_ENABLE_FT_MPI
+    /* Register the ULFM errhandler callback  */
     /* we want to go first */
     PMIX_INFO_LOAD(&info[0], PMIX_EVENT_HDLR_PREPEND, NULL, PMIX_BOOL);
     /* give it a name so we can distinguish it */
-    PMIX_INFO_LOAD(&info[1], PMIX_EVENT_HDLR_NAME, "MPI-Default", PMIX_STRING);
+    PMIX_INFO_LOAD(&info[1], PMIX_EVENT_HDLR_NAME, "ULFM-Event-handler", PMIX_STRING);
     OPAL_PMIX_CONSTRUCT_LOCK(&mylock);
-    PMIx_Register_event_handler(NULL, 0, info, 2, ompi_errhandler_callback, evhandler_reg_callbk, (void*)&mylock);
+    pmix_status_t codes[1] = { PMIX_ERR_PROC_ABORTED };
+    PMIx_Register_event_handler(codes, 1, info, 2, ompi_errhandler_callback, evhandler_reg_callbk, (void*)&mylock);
     OPAL_PMIX_WAIT_THREAD(&mylock);
     rc = mylock.status;
     OPAL_PMIX_DESTRUCT_LOCK(&mylock);
@@ -391,6 +407,7 @@ static int ompi_mpi_instance_init_common (void)
         ret = opal_pmix_convert_status(rc);
         return ret;
     }
+#endif
 
     /* initialize info */
     if (OMPI_SUCCESS != (ret = ompi_mpiinfo_init_mpi3())) {