Merge pull request #10507 from abouteiller/ulfm/get_failed

ULFM: new functions MPI_COMM_ACK/GET_FAILED (as per latest standard draft)

Author: abouteiller

ompi/communicator/comm_init.c

@@ -451,6 +451,7 @@ static void ompi_comm_construct(ompi_communicator_t* comm)
 #if OPAL_ENABLE_FT_MPI
     comm->any_source_enabled  = true;
     comm->any_source_offset   = 0;
+    comm->num_acked           = 0;
     comm->comm_revoked        = false;
     comm->coll_revoked        = false;
     comm->c_epoch             = 0;

ompi/communicator/communicator.h

@@ -329,6 +329,8 @@ struct ompi_communicator_t {
 #if OPAL_ENABLE_FT_MPI
     /** agreement caching info for topology and previous returned decisions */
     opal_object_t           *agreement_specific;
+    /** num_acked - OMPI_Comm_ack_failed */
+    int                      num_acked;
     /** MPI_ANY_SOURCE Failed Group Offset - OMPI_Comm_failure_get_acked */
     int                      any_source_offset;
     /** Are MPI_ANY_SOURCE operations enabled? - OMPI_Comm_failure_ack */
@@ -598,6 +600,17 @@ static inline bool ompi_comm_is_revoked(ompi_communicator_t* comm)
     return (comm->comm_revoked);
 }
 
+/*
+ * Obtain the group of locally known failed processes on comm
+ */
+OMPI_DECLSPEC int ompi_comm_get_failed_internal(ompi_communicator_t* comm, ompi_group_t** failed_group);
+
+/*
+ * Acknowledge failures and re-enable MPI_ANY_SOURCE
+ * Related to OMPI_Comm_failure_ack() and OMPI_Comm_failure_get_acked()
+ */
+OMPI_DECLSPEC int ompi_comm_ack_failed_internal(ompi_communicator_t* comm, int num_to_ack, int *num_acked);
+
 /*
  * Acknowledge failures and re-enable MPI_ANY_SOURCE
  * Related to OMPI_Comm_failure_ack() and OMPI_Comm_failure_get_acked()

ompi/communicator/ft/comm_ft.c

@@ -1,6 +1,6 @@
 /*
  * Copyright (c) 2010-2012 Oak Ridge National Labs.  All rights reserved.
- * Copyright (c) 2011-2020 The University of Tennessee and The University
+ * Copyright (c) 2011-2022 The University of Tennessee and The University
  *
  *                         of Tennessee Research Foundation.  All rights
  *                         reserved.
@@ -30,7 +30,7 @@ ompi_comm_rank_failure_callback_t *ompi_rank_failure_cbfunc = NULL;
  * The handling of known failed processes is based on a two level process. On one
  * side the MPI library itself must know the failed processes (in order to be able
  * to correctly handle complex operations such as shrink). On the other side, the
- * failed processes acknowledged by the users shuould not be altered during any of
+ * failed processes acknowledged by the users should not be altered during any of
  * the internal calls, as they must only be updated upon user request.
  * Thus, the global list (ompi_group_all_failed_procs) is the list of all known
  * failed processes (by the MPI library internals), and it is allegedly updated
@@ -39,6 +39,7 @@ ompi_comm_rank_failure_callback_t *ompi_rank_failure_cbfunc = NULL;
  * in the context of a communicator. Thus, using a single index to know the user-level
  * acknowledged failure is the simplest solution.
  */
+/* deprecated ulfm v1 API */
 int ompi_comm_failure_ack_internal(ompi_communicator_t* comm)
 {
     opal_mutex_lock(&ompi_group_afp_mutex);
@@ -49,11 +50,13 @@ int ompi_comm_failure_ack_internal(ompi_communicator_t* comm)
     /* use the AFP lock implicit memory barrier to propagate the update to
      * any_source_enabled at the same time.
      */
+    comm->num_acked = -1; /* compat with v2 API: force recompute next time ack_failed is called */
     opal_mutex_unlock(&ompi_group_afp_mutex);
 
     return OMPI_SUCCESS;
 }
 
+/* deprecated ulfm v1 API; used internally in MPI_COMM_AGREE as well */
 int ompi_comm_failure_get_acked_internal(ompi_communicator_t* comm, ompi_group_t **group )
 {
     int ret, exit_status = OMPI_SUCCESS;
@@ -114,6 +117,105 @@ int ompi_comm_failure_get_acked_internal(ompi_communicator_t* comm, ompi_group_t
     return exit_status;
 }
 
+/* New v2 interface get_failed/ack_failed.
+ * This interface uses a cached value comm->num_acked to track how many
+ * processes in the group of this comm have been acknowledged in prior calls.
+ * For compatibility with v1 interface (failure_get_acked), it still updates
+ * the comm->any_source_offset, and the v1 interface failure_ack may erase the
+ * cached value comm->num_acked with -1 to force recomputing this value in mixed
+ * use cases (that is calling failure_ack will force full recomputation of
+ * comm->num_acked during the next ack_failed call). */
+int ompi_comm_ack_failed_internal(ompi_communicator_t* comm, int num_to_ack, int *num_acked) {
+    int ret, exit_status = MPI_SUCCESS;
+    int nf = -1, na = -1;
+    ompi_group_t *c_group = OMPI_COMM_IS_INTER(comm)? comm->c_local_group: comm->c_remote_group;
+    ompi_group_t *failed_group = NULL;
+
+    opal_mutex_lock(&ompi_group_afp_mutex);
+
+    /* shortcut when reading only */
+    if(num_to_ack <= comm->num_acked)
+        goto return_num_acked;
+
+    /* shortcut when no new faults */
+    if(comm->any_source_offset == ompi_group_size(ompi_group_all_failed_procs)
+    && comm->num_acked >= 0 /* reset by a call to v1 API? */)
+        goto return_num_acked;
+
+    /* compute num_acked */
+    ret = ompi_group_intersection(ompi_group_all_failed_procs,
+                                  c_group, &failed_group);
+    if(OMPI_SUCCESS != ret) {
+        exit_status = ret;
+        goto cleanup;
+    }
+    nf = ompi_group_size(failed_group);
+    na = (num_to_ack < nf)? num_to_ack: nf; /* never ack more than requested */
+
+    if(comm->num_acked < 0) { /* reset by a call to the v1 API: recompute it */
+        if(0 == comm->any_source_offset) comm->num_acked = 0;
+        else {
+            int aso = comm->any_source_offset - 1;
+            ret = ompi_group_translate_ranks(ompi_group_all_failed_procs, 1, &aso,
+                                             failed_group, &comm->num_acked);
+            comm->num_acked++; /* make it a group size again */
+            if(OMPI_SUCCESS != ret) {
+                exit_status = ret;
+                goto cleanup;
+            }
+        }
+    }
+
+    if(comm->num_acked < na) { /* comm->num_acked needs to be updated during this call */
+        comm->num_acked = na;
+        if(nf == na) {
+            /* all faults on comm acknowledged, resume any source then */
+            comm->any_source_enabled = true;
+            comm->any_source_offset = ompi_group_size(ompi_group_all_failed_procs); // compat with v1 interface
+        }
+        else {
+            /* some faults not acknowledged, do not resume any source then, but
+             * still update any_source_offset */
+            assert(comm->num_acked > 0);
+            int cna = comm->num_acked - 1;
+            ret = ompi_group_translate_ranks(failed_group, 1, &cna,
+                                             ompi_group_all_failed_procs, &comm->any_source_offset); // compat with v1 interface
+            comm->any_source_offset++; /* make it a group size again */
+            if(OMPI_SUCCESS != ret) {
+                exit_status = ret;
+                goto cleanup;
+            }
+        }
+    }
+
+return_num_acked:
+    *num_acked = comm->num_acked;
+
+cleanup:
+    if(NULL != failed_group) OBJ_RELEASE(failed_group);
+    /* use the AFP lock implicit memory barrier to propagate the update to
+     * any_source_enabled, num_acked, etc. at the same time.
+     */
+    opal_mutex_unlock(&ompi_group_afp_mutex);
+
+    return exit_status;
+}
+
+int ompi_comm_get_failed_internal(ompi_communicator_t* comm, ompi_group_t **group)
+{
+    int ret, exit_status = OMPI_SUCCESS;
+    ompi_group_t *c_group = OMPI_COMM_IS_INTER(comm)? comm->c_local_group: comm->c_remote_group;
+    opal_mutex_lock(&ompi_group_afp_mutex);
+    ret = ompi_group_intersection(ompi_group_all_failed_procs,
+                                  c_group,
+                                  group);
+    opal_mutex_unlock(&ompi_group_afp_mutex);
+    if( OMPI_SUCCESS != ret ) {
+        exit_status = ret;
+    }
+    return exit_status;
+}
+
 int ompi_comm_shrink_internal(ompi_communicator_t* comm, ompi_communicator_t** newcomm)
 {
     int ret, exit_status = OMPI_SUCCESS;

ompi/mca/coll/han/coll_han_subcomms.c

@@ -52,7 +52,7 @@ int mca_coll_han_comm_create_new(struct ompi_communicator_t *comm,
     ompi_communicator_t **low_comm = &(han_module->sub_comm[INTRA_NODE]);
     ompi_communicator_t **up_comm = &(han_module->sub_comm[INTER_NODE]);
     mca_coll_han_collectives_fallback_t fallbacks;
-    int vrank, *vranks;
+    int rc = OMPI_SUCCESS, vrank, *vranks;
     opal_info_t comm_info;
 
     /* The sub communicators have already been created */
@@ -91,9 +91,12 @@ int mca_coll_han_comm_create_new(struct ompi_communicator_t *comm,
      * all participants.
      */
     int local_procs = ompi_group_count_local_peers(comm->c_local_group);
-    comm->c_coll->coll_allreduce(MPI_IN_PLACE, &local_procs, 1, MPI_INT,
-                                 MPI_MAX, comm,
-                                 comm->c_coll->coll_allreduce_module);
+    rc = comm->c_coll->coll_allreduce(MPI_IN_PLACE, &local_procs, 1, MPI_INT,
+                                      MPI_MAX, comm,
+                                      comm->c_coll->coll_allreduce_module);
+    if( OMPI_SUCCESS != rc ) {
+        goto return_with_error;
+    }
     if( local_procs == 1 ) {
         /* restore saved collectives */
         HAN_SUBCOM_LOAD_COLLECTIVE(fallbacks, comm, han_module, allgatherv);
@@ -118,8 +121,12 @@ int mca_coll_han_comm_create_new(struct ompi_communicator_t *comm,
      */
     opal_info_set(&comm_info, "ompi_comm_coll_preference", "han");
     opal_info_set(&comm_info, "ompi_comm_coll_han_topo_level", "INTRA_NODE");
-    ompi_comm_split_type(comm, MPI_COMM_TYPE_SHARED, 0,
-                         &comm_info, low_comm);
+    rc = ompi_comm_split_type(comm, MPI_COMM_TYPE_SHARED, 0,
+                              &comm_info, low_comm);
+    if( OMPI_SUCCESS != rc ) {
+        /* cannot create subcommunicators. Return the error upstream */
+        goto return_with_error;
+    }
 
     /*
      * Get my local rank and the local size
@@ -132,7 +139,11 @@ int mca_coll_han_comm_create_new(struct ompi_communicator_t *comm,
      * same intra-node rank id share such a sub-communicator
      */
     opal_info_set(&comm_info, "ompi_comm_coll_han_topo_level", "INTER_NODE");
-    ompi_comm_split_with_info(comm, low_rank, w_rank, &comm_info, up_comm, false);
+    rc = ompi_comm_split_with_info(comm, low_rank, w_rank, &comm_info, up_comm, false);
+    if( OMPI_SUCCESS != rc ) {
+        /* cannot create subcommunicators. Return the error upstream */
+        goto return_with_error;
+    }
 
     up_rank = ompi_comm_rank(*up_comm);
 
@@ -150,14 +161,13 @@ int mca_coll_han_comm_create_new(struct ompi_communicator_t *comm,
      * gather vrank from each process so every process will know other processes
      * vrank
      */
-    comm->c_coll->coll_allgather(&vrank,
-                                 1,
-                                 MPI_INT,
-                                 vranks,
-                                 1,
-                                 MPI_INT,
-                                 comm,
-                                 comm->c_coll->coll_allgather_module);
+    rc = comm->c_coll->coll_allgather(&vrank, 1, MPI_INT,
+                                 vranks, 1, MPI_INT,
+                                 comm, comm->c_coll->coll_allgather_module);
+    if( OMPI_SUCCESS != rc ) {
+        /* cannot create subcommunicators. Return the error upstream */
+        goto return_with_error;
+    }
 
     /*
      * Set the cached info
@@ -175,6 +185,17 @@ int mca_coll_han_comm_create_new(struct ompi_communicator_t *comm,
 
     OBJ_DESTRUCT(&comm_info);
     return OMPI_SUCCESS;
+
+return_with_error:
+    if( NULL != *low_comm ) {
+        ompi_comm_free(low_comm);
+        *low_comm = NULL;  /* don't leave the MPI_COMM_NULL set by ompi_comm_free */
+    }
+    if( NULL != *up_comm ) {
+        ompi_comm_free(up_comm);
+        *up_comm = NULL;  /* don't leave the MPI_COMM_NULL set by ompi_comm_free */
+    }
+    return rc;
 }
 
 /*

ompi/mpiext/ftmpi/c/Makefile.am

@@ -1,6 +1,6 @@
 #
 # Copyright (c) 2010-2012 Oak Ridge National Labs.  All rights reserved.
-# Copyright (c) 2016-2018 The University of Tennessee and The University
+# Copyright (c) 2016-2022 The University of Tennessee and The University
 #                         of Tennessee Research Foundation.  All rights
 #                         reserved.
 # $COPYRIGHT$
@@ -29,6 +29,8 @@ libmpiext_ftmpi_c_la_SOURCES = \
     comm_shrink.c \
     comm_failure_ack.c \
     comm_failure_get_acked.c \
+    comm_get_failed.c \
+    comm_ack_failed.c \
     comm_agree.c \
     comm_iagree.c
 libmpiext_ftmpi_c_la_LIBADD = \

ompi/mpiext/ftmpi/c/comm_ack_failed.c

@@ -0,0 +1,50 @@
+/*
+ * Copyright (c) 2022      The University of Tennessee and The University
+ *                         of Tennessee Research Foundation.  All rights
+ *                         reserved.
+ * $COPYRIGHT$
+ *
+ * Additional copyrights may follow
+ *
+ * $HEADER$
+ */
+#include "ompi_config.h"
+
+#include "ompi/mpi/c/bindings.h"
+#include "ompi/runtime/params.h"
+#include "ompi/communicator/communicator.h"
+#include "ompi/proc/proc.h"
+#include "ompi/errhandler/errhandler.h"
+
+#include "ompi/mpiext/ftmpi/c/mpiext_ftmpi_c.h"
+
+#if OMPI_BUILD_MPI_PROFILING
+#if OPAL_HAVE_WEAK_SYMBOLS
+#pragma weak MPIX_Comm_ack_failed = PMPIX_Comm_ack_failed
+#endif
+#define MPIX_Comm_ack_failed PMPIX_Comm_ack_failed
+#endif
+
+static const char FUNC_NAME[] = "MPIX_Comm_ack_failed";
+
+
+int MPIX_Comm_ack_failed(MPI_Comm comm, int num_to_ack, int* num_acked)
+{
+    int rc = MPI_SUCCESS;
+
+    /* Argument checking */
+    if (MPI_PARAM_CHECK) {
+        OMPI_ERR_INIT_FINALIZE(FUNC_NAME);
+        if (ompi_comm_invalid(comm)) {
+            return OMPI_ERRHANDLER_INVOKE(MPI_COMM_WORLD, MPI_ERR_COMM, FUNC_NAME);
+        }
+        if (num_to_ack > ompi_comm_size(comm)) {
+            rc = MPI_ERR_ARG;
+        }
+        OMPI_ERRHANDLER_CHECK(rc, comm, rc, FUNC_NAME);
+    }
+
+    rc = ompi_comm_ack_failed_internal( (ompi_communicator_t*)comm, num_to_ack, num_acked );
+    OMPI_ERRHANDLER_RETURN(rc, comm, rc, FUNC_NAME);
+}
+

ompi/mpiext/ftmpi/c/comm_get_failed.c

@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2022      The University of Tennessee and The University
+ *                         of Tennessee Research Foundation.  All rights
+ *                         reserved.
+ * $COPYRIGHT$
+ *
+ * Additional copyrights may follow
+ *
+ * $HEADER$
+ */
+#include "ompi_config.h"
+
+#include "ompi/mpi/c/bindings.h"
+#include "ompi/runtime/params.h"
+#include "ompi/communicator/communicator.h"
+#include "ompi/proc/proc.h"
+
+#include "ompi/mpiext/ftmpi/c/mpiext_ftmpi_c.h"
+
+#if OMPI_BUILD_MPI_PROFILING
+#if OPAL_HAVE_WEAK_SYMBOLS
+#pragma weak MPIX_Comm_get_failed = PMPIX_Comm_get_failed
+#endif
+#define MPIX_Comm_get_failed PMPIX_Comm_get_failed
+#endif
+
+static const char FUNC_NAME[] = "MPIX_Comm_get_failed";
+
+int MPIX_Comm_get_failed(MPI_Comm comm, MPI_Group *failedgrp)
+{
+    int rc = MPI_SUCCESS;
+
+    /* Argument checking */
+    if (MPI_PARAM_CHECK) {
+        OMPI_ERR_INIT_FINALIZE(FUNC_NAME);
+        if (ompi_comm_invalid(comm)) {
+            return OMPI_ERRHANDLER_INVOKE(MPI_COMM_WORLD, MPI_ERR_COMM, FUNC_NAME);
+        }
+        OMPI_ERRHANDLER_CHECK(rc, comm, rc, FUNC_NAME);
+    }
+
+    rc = ompi_comm_get_failed_internal( (ompi_communicator_t*)comm, (ompi_group_t**)failedgrp );
+    if( OMPI_SUCCESS != rc ) {
+        OMPI_ERRHANDLER_RETURN(rc, comm, rc, FUNC_NAME);
+    }
+
+    return MPI_SUCCESS;
+}
+