Allow user to invoke callbacks, better define callback behavior when nested faults occur

Author: Matthew-Whitlock

include/fenix.h

@@ -279,6 +279,11 @@ int Fenix_Callback_register(void (*recover)(MPI_Comm, int, void *),
  */
 int Fenix_Callback_pop();
 
+/**
+ * @brief Invoke all callbacks with information from the last recovered fault
+ */
+void Fenix_Callback_invoke_all();
+
 /**
  * @brief Check for any failed ranks
  *

include/fenix.hpp

@@ -87,23 +87,29 @@ constexpr ResumeMode JUMP   = FENIX_RESUME_JUMP;
 constexpr ResumeMode RETURN = FENIX_RESUME_RETURN;
 constexpr ResumeMode THROW  = FENIX_RESUME_THROW;
 
+enum CallbackExceptionMode {
+    RETHROW,
+    SQUASH
+};
+
 using UnhandledMode = Fenix_Unhandled_mode;
 constexpr UnhandledMode SILENT = FENIX_UNHANDLED_SILENT;
 constexpr UnhandledMode PRINT  = FENIX_UNHANDLED_PRINT;
 constexpr UnhandledMode ABORT  = FENIX_UNHANDLED_ABORT;
 
 namespace Args {
 struct FenixInitArgs {
-    int* role                    = nullptr;
-    MPI_Comm in_comm             = MPI_COMM_WORLD;
-    MPI_Comm* out_comm           = nullptr;
-    int* argc                    = nullptr;
-    char*** argv                 = nullptr;
-    int spares                   = 0;
-    int spawn                    = 0;
-    ResumeMode resume_mode       = THROW;
-    UnhandledMode unhandled_mode = ABORT;
-    int* err                     = nullptr;
+    int* role                                       = nullptr;
+    MPI_Comm in_comm                                = MPI_COMM_WORLD;
+    MPI_Comm* out_comm                              = nullptr;
+    int* argc                                       = nullptr;
+    char*** argv                                    = nullptr;
+    int spares                                      = 0;
+    int spawn                                       = 0;
+    ResumeMode resume_mode                          = THROW;
+    CallbackExceptionMode callback_exception_mode   = RETHROW;
+    UnhandledMode unhandled_mode                    = ABORT;
+    int* err                                        = nullptr;
 };
 }
 
@@ -127,6 +133,9 @@ int callback_register(std::function<void(MPI_Comm, int)> callback);
 //@!brief Overload of #Fenix_Callback_pop
 int callback_pop();
 
+//@!brief Overload of #Fenix_Callback_invoke_all
+void callback_invoke_all();
+
 /**
  * @brief Get the failed ranks from the most recent recovery
  * @return vector of failed ranks

include/fenix_ext.hpp

@@ -74,11 +74,13 @@ typedef struct {
     int spare_ranks;             // Spare ranks entered by user to repair failed ranks
 
     ResumeMode resume_mode = JUMP;
+    CallbackExceptionMode callback_exception_mode = RETHROW;
     UnhandledMode unhandled_mode = ABORT;
     int ignore_errs = false;       // Temporarily ignore all errors & recovery
     int spawn_policy;             // Indicate dynamic process spawning
     jmp_buf *recover_environment; // Calling environment to fill the jmp_buf structure
 
+    int mpi_fail_code = MPI_SUCCESS;
     int repair_result = FENIX_SUCCESS; // Internal variable to store the result of MPI comm repair
     int role = FENIX_ROLE_INITIAL_RANK;
 

include/fenix_process_recovery.hpp

@@ -98,7 +98,7 @@ int __fenix_callback_register(fenix_callback_func& recover);
 
 int __fenix_callback_pop();
 
-void __fenix_callback_invoke_all(int error);
+void __fenix_callback_invoke_all();
 
 int* __fenix_get_fail_ranks(int *, int, int);
 

src/fenix.cpp

@@ -88,6 +88,10 @@ int Fenix_Callback_pop() {
     return __fenix_callback_pop();
 }
 
+void Fenix_Callback_invoke_all() {
+    __fenix_callback_invoke_all();
+}
+
 int Fenix_Initialized(int *flag) {
     *flag = (fenix.fenix_init_flag) ? 1 : 0;
     return FENIX_SUCCESS;
@@ -298,6 +302,10 @@ int callback_pop() {
     return __fenix_callback_pop();
 }
 
+void callback_invoke_all() {
+    __fenix_callback_invoke_all();
+}
+
 std::vector<int> fail_list(){
     if(fenix.fail_world_size == 0) return {};
     return {fenix.fail_world, fenix.fail_world+fenix.fail_world_size};

src/fenix_callbacks.cpp

@@ -62,8 +62,10 @@
 #include "fenix_data_recovery.hpp"
 #include "fenix_opt.hpp"
 #include "fenix_util.hpp"
+#include "fenix_exception.hpp"
 #include <mpi.h>
 
+using namespace Fenix;
 
 int __fenix_callback_register(fenix_callback_func& recover)
 {
@@ -83,9 +85,28 @@ int __fenix_callback_pop(){
    return FENIX_SUCCESS;
 }
 
-void __fenix_callback_invoke_all(int error)
-{
-    for(auto it = fenix.callbacks.rbegin(); it != fenix.callbacks.rend(); it++){
-        (*it)(*fenix.user_world, error);
+void __fenix_callback_invoke_all(){
+    //If callbacks are invoked in a nested manner due to caught exceptions
+    //within a callback, we want to only finish the most recent call. All prior
+    //calls should exit as soon as control returns.
+    static int callbacks_depth = 0;
+    int m_callbacks_layer = callbacks_depth++;
+
+    try {
+        for(auto& cb : fenix.callbacks) {
+            if(callbacks_depth != m_callbacks_layer+1) break;
+            cb(*fenix.user_world, fenix.mpi_fail_code);
+        }
+    } catch (const CommException& e) {
+        switch(fenix.callback_exception_mode){
+            case(RETHROW):
+                if(m_callbacks_layer == 0) callbacks_depth = 0;
+                throw;
+            case(SQUASH):
+                break;
+        }
     }
+
+    //Reset the callback depth when leaving the outermost call
+    if(m_callbacks_layer == 0) callbacks_depth = 0;
 }

src/fenix_process_recovery.cpp

@@ -112,6 +112,7 @@ int fenix_preinit(const Args::FenixInitArgs& args, jmp_buf* jump_env){
     fenix.spawn_policy = args.spawn;
     fenix.recover_environment = jump_env;
     fenix.resume_mode = args.resume_mode;
+    fenix.callback_exception_mode = args.callback_exception_mode;
     fenix.unhandled_mode = args.unhandled_mode;
     fenix.ret_role = args.role ? args.role : &fenix.role;
     fenix.ret_error = args.err ? args.err : &fenix.repair_result;
@@ -663,9 +664,10 @@ void __fenix_postinit()
                   34095347, fenix.new_world, &fenix.check_failures_req);
     }
 
-    if (fenix.role == FENIX_ROLE_SURVIVOR_RANK) {
-        __fenix_callback_invoke_all(*fenix.ret_error);
+    if(fenix.role != FENIX_ROLE_INITIAL_RANK) {
+        __fenix_callback_invoke_all();
     }
+
     if (fenix.options.verbose == 9) {
         verbose_print("After barrier. current_rank: %d, role: %d\n", __fenix_get_current_rank(fenix.new_world),
                       fenix.role);
@@ -798,17 +800,17 @@ void __fenix_test_MPI(MPI_Comm *pcomm, int *pret, ...)
 {
     int ret_repair;
     int index;
-    int ret = *pret;
+    fenix.mpi_fail_code = *pret;
+
     if(!fenix.fenix_init_flag || __fenix_spare_rank() == 1 || fenix.ignore_errs) {
         return;
     }
 
-    switch (ret) {
+    switch (fenix.mpi_fail_code) {
         case MPI_ERR_PROC_FAILED_PENDING:
         case MPI_ERR_PROC_FAILED:
             MPIX_Comm_revoke(*fenix.world);
             MPIX_Comm_revoke(fenix.new_world);
-            
             if(fenix.user_world_exists) MPIX_Comm_revoke(*fenix.user_world);
 
             fenix.repair_result = __fenix_repair_ranks();
@@ -819,7 +821,7 @@ void __fenix_test_MPI(MPI_Comm *pcomm, int *pret, ...)
         default:
             int len;
             char errstr[MPI_MAX_ERROR_STRING];
-            MPI_Error_string(ret, errstr, &len);
+            MPI_Error_string(fenix.mpi_fail_code, errstr, &len);
             switch (fenix.unhandled_mode) {
                 case ABORT:
                     fprintf(stderr, "UNHANDLED ERR: %s\n", errstr);