Implementation of the current state of MPI Continuations proposal [WIP]

ompi/communicator/ft/comm_ft_revoke.c

@@ -18,6 +18,10 @@
 #include "ompi/communicator/communicator.h"
 #include "ompi/mca/pml/pml.h"
 
+#if OMPI_HAVE_MPI_EXT_CONTINUE
+#include "ompi/mpiext/continue/c/continuation.h"
+#endif /* OMPI_HAVE_MPI_EXT_CONTINUE */
+
 static int ompi_comm_revoke_local(ompi_communicator_t* comm,
                                   ompi_comm_rbcast_message_t* msg);
 
@@ -93,6 +97,14 @@ static int ompi_comm_revoke_local(ompi_communicator_t* comm, ompi_comm_rbcast_me
     MCA_PML_CALL(revoke_comm(comm, false));
     /* Signal the point-to-point stack to recheck requests */
     wait_sync_global_wakeup(MPI_ERR_REVOKED);
+
+#ifdef OMPI_HAVE_MPI_EXT_CONTINUE
+    /* Continuations:
+     * Release continuations and mark them as failed.
+     */
+    ompi_continue_global_wakeup(MPI_ERR_PROC_FAILED);
+#endif // OMPI_HAVE_MPI_EXT_CONTINUE
+
     return true;
 }
 

ompi/errhandler/errhandler.c

@@ -40,6 +40,10 @@
 #include "opal/mca/backtrace/backtrace.h"
 #include "ompi/runtime/mpiruntime.h"
 
+#if OMPI_HAVE_MPI_EXT_CONTINUE
+#include "ompi/mpiext/continue/c/continuation.h"
+#endif /* OMPI_HAVE_MPI_EXT_CONTINUE */
+
 /*
  * Table for Fortran <-> C errhandler handle conversion
  */
@@ -415,6 +419,13 @@ int ompi_errhandler_proc_failed_internal(ompi_proc_t* ompi_proc, int status, boo
      */
     wait_sync_global_wakeup(PMIX_ERR_PROC_ABORTED == status? MPI_ERR_PROC_ABORTED: MPI_ERR_PROC_FAILED);
 
+#ifdef OMPI_HAVE_MPI_EXT_CONTINUE
+    /* Continuations:
+     * Release continuations and mark them as failed.
+     */
+    ompi_continue_global_wakeup(MPI_ERR_PROC_FAILED);
+#endif // OMPI_HAVE_MPI_EXT_CONTINUE
+
     /* Collectives:
      * Propagate the error (this has been selected rather than the "roll
      * forward through errors in collectives" as this is less intrusive to the

ompi/errhandler/errhandler_invoke.c

@@ -34,6 +34,9 @@
 #include "ompi/errhandler/errhandler.h"
 #include "ompi/mpi/fortran/base/fint_2_int.h"
 
+#if OMPI_HAVE_MPI_EXT_CONTINUE
+#include "ompi/mpiext/continue/c/continuation.h"
+#endif /* OMPI_HAVE_MPI_EXT_CONTINUE */
 
 int ompi_errhandler_invoke(ompi_errhandler_t *errhandler, void *mpi_object,
                            int object_type, int err_code, const char *message)
@@ -174,6 +177,13 @@ int ompi_errhandler_request_invoke(int count,
     mpi_object = requests[i]->req_mpi_object;
     type = requests[i]->req_type;
 
+#if OMPI_HAVE_MPI_EXT_CONTINUE
+    if (OMPI_REQUEST_CONT == type) {
+        /* take the mpi object stored in the continuation request */
+        ompi_continue_get_error_info(requests[i], &mpi_object, &type);
+    }
+#endif // OMPI_HAVE_MPI_EXT_CONTINUE
+
     /* Since errors on requests cause them to not be freed (until we
        can examine them here), go through and free all requests with
        errors.  We only invoke the error on the *first* request

ompi/include/mpi.h.in

@@ -754,6 +754,8 @@ enum {
 #define MPI_ERR_SESSION               78
 #define MPI_ERR_VALUE_TOO_LARGE       79
 
+#define MPI_ERR_CONT                  78
+
 /* Per MPI-3 p349 47, MPI_ERR_LASTCODE must be >= the last predefined
    MPI_ERR_<foo> code. Set the last code to allow some room for adding
    error codes without breaking ABI. */

ompi/mca/pml/ob1/pml_ob1_recvfrag.c

@@ -369,7 +369,7 @@ int mca_pml_ob1_revoke_comm( struct ompi_communicator_t* ompi_comm, bool coll_on
         /* note this is not an ompi_proc, but a ob1_comm_proc, thus we don't
          * use ompi_proc_is_sentinel to verify if initialized. */
         if( NULL == proc ) continue;
-        /* remove the frag from the unexpected list, add to the nack list
+        /* remove the frag from the unexpected list, add to the nack list 
          * so that we can send the nack as needed to remote cancel the send
          * from outside the match lock.
          */
@@ -384,7 +384,7 @@ int mca_pml_ob1_revoke_comm( struct ompi_communicator_t* ompi_comm, bool coll_on
             }
         }
         /* same for the cantmatch queue/heap; this list is more complicated
-         * Keep it simple: we pop all of the complex list, put the bad items
+         * Keep it simple: we pop all of the complex list, put the bad items 
          * in the nack_list, and keep the good items in the keep_list;
          * then we reinsert the good items in the cantmatch heaplist */
         mca_pml_ob1_recv_frag_t* frag;

ompi/mca/pml/ucx/pml_ucx_request.h

@@ -157,6 +157,7 @@ int mca_pml_ucx_request_cancel_send(ompi_request_t *req, int flag);
 
 static inline void mca_pml_ucx_request_reset(ompi_request_t *req)
 {
+    OMPI_REQUEST_INIT(req, req->req_persistent);
     req->req_complete          = REQUEST_PENDING;
 }
 

ompi/mpi/c/start.c

@@ -78,8 +78,8 @@ int MPI_Start(MPI_Request *request)
     case OMPI_REQUEST_PML:
     case OMPI_REQUEST_COLL:
     case OMPI_REQUEST_PART:
-        if ( MPI_PARAM_CHECK && !((*request)->req_persistent &&
-                                  OMPI_REQUEST_INACTIVE == (*request)->req_state)) {
+    case OMPI_REQUEST_CONT:
+        if ( MPI_PARAM_CHECK && !(*request)->req_persistent) {
             return OMPI_ERRHANDLER_NOHANDLE_INVOKE(MPI_ERR_REQUEST, FUNC_NAME);
         }
 

ompi/mpi/c/startall.c

@@ -70,6 +70,7 @@ int MPI_Startall(int count, MPI_Request requests[])
                     (OMPI_REQUEST_PML  != requests[i]->req_type &&
                      OMPI_REQUEST_COLL != requests[i]->req_type &&
                      OMPI_REQUEST_PART != requests[i]->req_type &&
+                     OMPI_REQUEST_CONT != requests[i]->req_type &&
                      OMPI_REQUEST_NOOP != requests[i]->req_type)) {
                     rc = MPI_ERR_REQUEST;
                     break;

ompi/mpiext/continue/Makefile.am

@@ -0,0 +1,22 @@
+# -*- shell-script -*-
+#
+# Copyright (c) 2021      The University of Tennessee and The University
+#                         of Tennessee Research Foundation.  All rights
+#                         reserved.
+# $COPYRIGHT$
+#
+# Additional copyrights may follow
+#
+# $HEADER$
+#
+
+# This Makefile is not traversed during a normal "make all" in an OMPI
+# build.  It *is* traversed during "make dist", however.  So you can
+# put EXTRA_DIST targets in here.
+#
+# You can also use this as a convenience for building this MPI
+# extension (i.e., "make all" in this directory to invoke "make all"
+# in all the subdirectories).
+
+SUBDIRS = c
+

ompi/mpiext/continue/c/Makefile.am

@@ -0,0 +1,43 @@
+#
+# Copyright (c) 2021      The University of Tennessee and The University
+#                         of Tennessee Research Foundation.  All rights
+#                         reserved.
+# $COPYRIGHT$
+#
+# Additional copyrights may follow
+#
+# $HEADER$
+#
+
+# OMPI_BUILD_MPI_PROFILING is enabled when we want our generated MPI_* symbols
+# to be replaced by PMPI_*.
+# In this directory, we need it to be 0
+
+AM_CPPFLAGS = -DOMPI_BUILD_MPI_PROFILING=0 -DOMPI_COMPILING_FORTRAN_WRAPPERS=0
+
+include $(top_srcdir)/Makefile.ompi-rules
+
+noinst_LTLIBRARIES = libmpiext_continue_c.la
+
+# This is where the top-level header file (that is included in
+# <mpi-ext.h>) must be installed.
+ompidir = $(ompiincludedir)/mpiext
+
+# This is the header file that is installed.
+nodist_ompi_HEADERS = mpiext_continue_c.h
+
+libmpiext_continue_c_la_SOURCES = \
+    continuation.c \
+    continue.c \
+    continueall.c \
+    continue_init.c \
+    mpiext_continue_module.c
+
+#libmpiext_continue_c_la_LDFLAGS = -module -avoid-version
+
+dist_ompidata_DATA = help-mpi-continue.txt
+
+ompi_HEADERS = $(headers)
+
+MAINTAINERCLEANFILES = $(nodist_libmpiext_continue_c_la_SOURCES)
+