Centralize an entrypoint to locally revoking a comm

Signed-off-by: Matthew Whitlock <[email protected]>

Author: Matthew-Whitlock

ompi/communicator/communicator.h

@@ -728,6 +728,11 @@ OMPI_DECLSPEC bool ompi_comm_is_proc_active(ompi_communicator_t *comm, int peer_
  */
 OMPI_DECLSPEC int ompi_comm_set_rank_failed(ompi_communicator_t *comm, int peer_id, bool remote);
 
+/*
+ * Locally revoke a communicator
+ */
+OMPI_DECLSPEC bool ompi_comm_revoke_local(ompi_communicator_t* comm, bool coll_only);
+
 /*
  * Returns true if point-to-point communications with the target process
  * are supported (this means if the process is a valid peer, if the

ompi/communicator/ft/comm_ft.c

@@ -805,7 +805,7 @@ int ompi_comm_set_rank_failed(ompi_communicator_t *comm, int peer_id, bool remot
     opal_atomic_wmb(); /* non-locked update needs a memory barrier to propagate */
 
     /* Disable collectives */
-    MCA_PML_CALL(revoke_comm(comm, true));
+    ompi_comm_revoke_local(comm, true);
 
     if( NULL != ompi_rank_failure_cbfunc ) {
         (*ompi_rank_failure_cbfunc)(comm, peer_id, remote);

ompi/communicator/ft/comm_ft_revoke.c

@@ -18,7 +18,7 @@
 #include "ompi/communicator/communicator.h"
 #include "ompi/mca/pml/pml.h"
 
-static int ompi_comm_revoke_local(ompi_communicator_t* comm,
+static int ompi_comm_revoke_msg_cb(ompi_communicator_t* comm,
                                   ompi_comm_rbcast_message_t* msg);
 
 static int comm_revoke_cb_type = -1;
@@ -27,7 +27,7 @@ int ompi_comm_revoke_init(void)
 {
     int ret;
 
-    ret = ompi_comm_rbcast_register_cb_type(ompi_comm_revoke_local);
+    ret = ompi_comm_rbcast_register_cb_type(ompi_comm_revoke_msg_cb);
     if( 0 <= ret ) {
         comm_revoke_cb_type = ret;
         return OMPI_SUCCESS;
@@ -66,20 +66,22 @@ int ompi_comm_revoke_internal(ompi_communicator_t* comm)
     return ret;
 }
 
-
-/* internal code to revoke the communicator structure. Can be called from the
- * API or from receiving a revoke message */
-static int ompi_comm_revoke_local(ompi_communicator_t* comm, ompi_comm_rbcast_message_t* msg)
+/*
+ * Internal code to locally revoke a comm and update all necessary state
+ */
+bool ompi_comm_revoke_local(ompi_communicator_t* comm, bool coll_only)
 {
-    if( comm->comm_revoked ) {
+    if( comm->comm_revoked || (coll_only && comm->coll_revoked) ) {
         OPAL_OUTPUT_VERBOSE((9, ompi_ftmpi_output_handle,
-                             "%s %s: comm %s:%d is already revoked, nothing to do",
-                             OMPI_NAME_PRINT(OMPI_PROC_MY_NAME), __func__, ompi_comm_print_cid(comm), comm->c_epoch));
+                             "%s %s: comm %s:%d is already %srevoked, nothing to do",
+                             OMPI_NAME_PRINT(OMPI_PROC_MY_NAME), __func__, ompi_comm_print_cid(comm), comm->c_epoch,
+                             coll_only ? "coll " : ""));
         return false;
     }
     OPAL_OUTPUT_VERBOSE((9, ompi_ftmpi_output_handle,
-                         "%s %s: comm %s:%d is marked revoked locally",
-                         OMPI_NAME_PRINT(OMPI_PROC_MY_NAME), __func__, ompi_comm_print_cid(comm), comm->c_epoch));
+                         "%s %s: comm %s:%d is marked %srevoked locally",
+                         OMPI_NAME_PRINT(OMPI_PROC_MY_NAME), __func__, ompi_comm_print_cid(comm), comm->c_epoch,
+                         coll_only ? "coll " : ""));
     /*
      * Locally revoke the communicator
      *
@@ -90,11 +92,17 @@ static int ompi_comm_revoke_local(ompi_communicator_t* comm, ompi_comm_rbcast_me
      */
     comm->any_source_enabled = false;
     /* purge the communicator unexpected fragments and matching logic */
-    MCA_PML_CALL(revoke_comm(comm, false));
+    MCA_PML_CALL(revoke_comm(comm, coll_only));
     /* revoke any subcomms created by coll */
     comm->c_coll->coll_revoke_local(comm);
     /* Signal the point-to-point stack to recheck requests */
     wait_sync_global_wakeup(MPI_ERR_REVOKED);
     return true;
 }
 
+/* internal code to revoke the communicator structure. Can be called from the
+ * API or from receiving a revoke message */
+static int ompi_comm_revoke_msg_cb(ompi_communicator_t* comm, ompi_comm_rbcast_message_t* msg)
+{
+    return ompi_comm_revoke_local(comm, false);
+}

ompi/mca/coll/han/coll_han_subcomms.c

@@ -421,18 +421,18 @@ int mca_coll_han_revoke_local(ompi_communicator_t *comm,
     mca_coll_han_module_t *han_module = (mca_coll_han_module_t*) module;
     for(int i = 0; i < NB_TOPO_LVL; i++){
         if(NULL == han_module->sub_comm[i]) continue;
-        ompi_comm_revoke_internal(han_module->sub_comm[i]);
+        ompi_comm_revoke_local(han_module->sub_comm[i], true);
     }
     if(han_module->cached_low_comms != NULL){
         for(int i = 0; i < COLL_HAN_LOW_MODULES; i++){
             if(NULL == han_module->cached_low_comms[i]) continue;
-            ompi_comm_revoke_internal(han_module->cached_low_comms[i]);
+            ompi_comm_revoke_local(han_module->cached_low_comms[i], true);
         }
     }
     if(han_module->cached_up_comms != NULL){
         for(int i = 0; i < COLL_HAN_LOW_MODULES; i++){
             if(NULL == han_module->cached_low_comms[i]) continue;
-            ompi_comm_revoke_internal(han_module->cached_low_comms[i]);
+            ompi_comm_revoke_local(han_module->cached_low_comms[i], true);
         }
     }
     return MPI_SUCCESS;

ompi/mca/pml/ob1/pml_ob1_recvfrag.c

@@ -346,7 +346,7 @@ int mca_pml_ob1_revoke_comm( struct ompi_communicator_t* ompi_comm, bool coll_on
 
     /* For intercomm, also work with the local_comm */
     if( OMPI_COMM_IS_INTER(ompi_comm) ) {
-        mca_pml_ob1_revoke_comm(ompi_comm->c_local_comm, coll_only);
+        ompi_comm_revoke_local(ompi_comm->c_local_comm, coll_only);
     }
 
     OBJ_CONSTRUCT(&nack_list, opal_list_t);