ulfm: new functions comm_get_failed, comm_ack_failed C bindings

abouteiller · abouteiller · commit 8fe26b48743a · 2023-02-21T17:05:27.000-05:00
Add error management to ack_failed_internal

Use comm-&gt;num_acked to decide if any_source should be re-enabled

More compatibility with mixed use of failure_ack and ack_failed

Additional shortcut: bypass computing num_acked when no new faults

Recompute num_acked when the v1 interface got used inbetween

Signed-off-by: Aurelien Bouteiller &lt;bouteill@icl.utk.edu&gt;

Copyrights

The group needs to be initialized before the first goto.

Signed-off-by: George Bosilca &lt;bosilca@icl.utk.edu&gt;

Add #defines for the new API.

Signed-off-by: George Bosilca &lt;bosilca@icl.utk.edu&gt;
diff --git a/ompi/communicator/comm_init.c b/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;
diff --git a/ompi/communicator/communicator.h b/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()
diff --git a/ompi/communicator/ft/comm_ft.c b/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;
diff --git a/ompi/mpiext/ftmpi/c/Makefile.am b/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 = \
diff --git a/ompi/mpiext/ftmpi/c/comm_ack_failed.c b/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);
+}
+
diff --git a/ompi/mpiext/ftmpi/c/comm_get_failed.c b/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;
+}
+
diff --git a/ompi/mpiext/ftmpi/c/mpiext_ftmpi_c.h b/ompi/mpiext/ftmpi/c/mpiext_ftmpi_c.h
@@ -27,6 +27,8 @@ OMPI_DECLSPEC int MPIX_Comm_is_revoked(MPI_Comm comm, int *flag);
 OMPI_DECLSPEC int MPIX_Comm_shrink(MPI_Comm comm, MPI_Comm *newcomm);
 OMPI_DECLSPEC int MPIX_Comm_failure_ack(MPI_Comm comm);
 OMPI_DECLSPEC int MPIX_Comm_failure_get_acked(MPI_Comm comm, MPI_Group *failedgrp);
+OMPI_DECLSPEC int MPIX_Comm_get_failed(MPI_Comm comm, MPI_Group *failedgroup);
+OMPI_DECLSPEC int MPIX_Comm_ack_failed(MPI_Comm comm, int num_to_ack, int *num_acked);
 OMPI_DECLSPEC int MPIX_Comm_agree(MPI_Comm comm, int *flag);
 OMPI_DECLSPEC int MPIX_Comm_iagree(MPI_Comm comm, int *flag, MPI_Request *request);
 
@@ -35,9 +37,14 @@ OMPI_DECLSPEC int PMPIX_Comm_is_revoked(MPI_Comm comm, int *flag);
 OMPI_DECLSPEC int PMPIX_Comm_shrink(MPI_Comm comm, MPI_Comm *newcomm);
 OMPI_DECLSPEC int PMPIX_Comm_failure_ack(MPI_Comm comm);
 OMPI_DECLSPEC int PMPIX_Comm_failure_get_acked(MPI_Comm comm, MPI_Group *failedgrp);
+OMPI_DECLSPEC int PMPIX_Comm_get_failed(MPI_Comm comm, MPI_Group *failedgroup);
+OMPI_DECLSPEC int PMPIX_Comm_ack_failed(MPI_Comm comm, int num_to_ack, int *num_acked);
 OMPI_DECLSPEC int PMPIX_Comm_agree(MPI_Comm comm, int *flag);
 OMPI_DECLSPEC int PMPIX_Comm_iagree(MPI_Comm comm, int *flag, MPI_Request *request);
 
 #include <stdbool.h>
 OMPI_DECLSPEC int OMPI_Comm_failure_inject(MPI_Comm comm, bool notify);
+/* Provide defines to facilitate the detection of the new API */
+#define OMPI_HAVE_MPIX_COMM_GET_FAILED 1
+#define OMPI_HAVE_MPIX_COMM_ACK_FAILED 1
 
diff --git a/ompi/mpiext/ftmpi/c/profile/Makefile.am b/ompi/mpiext/ftmpi/c/profile/Makefile.am
@@ -1,5 +1,5 @@
 #
-# 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 (c) 2021 Cisco Systems, Inc.  All rights reserved.
@@ -21,6 +21,8 @@ nodist_libpmpiext_ftmpi_c_la_SOURCES = \
     pcomm_revoke.c \
     pcomm_is_revoked.c \
     pcomm_shrink.c \
+    pcomm_get_failed.c \
+    pcomm_ack_failed.c \
     pcomm_failure_ack.c \
     pcomm_failure_get_acked.c \
     pcomm_agree.c \
