Initial fault tolerance commit

Author: Matthew-Whitlock

src/mca/errmgr/dvm/errmgr_dvm.c

@@ -278,7 +278,7 @@ static void proc_errors(int fd, short args, void *cbdata)
                 /* remove from dependent routes, if it is one */
                 prte_rml_route_lost(proc->rank);
                 /* if all my routes and local children are gone, then terminate ourselves */
-                if (0 == pmix_list_get_size(&prte_rml_base.children)) {
+                if (0 == prte_rml_base.n_children) {
                     for (i = 0; i < prte_local_children->size; i++) {
                         proct = (prte_proc_t *) pmix_pointer_array_get_item(prte_local_children, i);
                         if (NULL != proct &&
@@ -301,7 +301,7 @@ static void proc_errors(int fd, short args, void *cbdata)
                     PMIX_OUTPUT_VERBOSE((5, prte_errmgr_base_framework.framework_output,
                                          "%s Comm failure: %d routes remain alive",
                                          PRTE_NAME_PRINT(PRTE_PROC_MY_NAME),
-                                         (int) pmix_list_get_size(&prte_rml_base.children)));
+                                         prte_rml_base.n_children));
                 }
                 goto cleanup;
             }
@@ -316,6 +316,10 @@ static void proc_errors(int fd, short args, void *cbdata)
                                    PRTE_NAME_PRINT(PRTE_PROC_MY_NAME), prte_process_info.nodename,
                                    PRTE_NAME_PRINT(proc), pptr->node->name);
                 }
+
+                if(PRTE_SUCCESS == prte_rml_route_lost(proc->rank))
+                    goto cleanup;
+
                 /* mark the daemon job as failed */
                 jdata->state = PRTE_JOB_STATE_COMM_FAILED;
                 /* point to the lowest rank to cause the problem */
@@ -364,7 +368,7 @@ static void proc_errors(int fd, short args, void *cbdata)
         }
         /* if all my routes and children are gone, then terminate
            ourselves nicely (i.e., this is a normal termination) */
-        if (0 == pmix_list_get_size(&prte_rml_base.children)) {
+        if (0 == prte_rml_base.n_children) {
             PMIX_OUTPUT_VERBOSE((2, prte_errmgr_base_framework.framework_output,
                                  "%s errmgr:default:dvm all routes gone - exiting",
                                  PRTE_NAME_PRINT(PRTE_PROC_MY_NAME)));

src/mca/errmgr/prted/errmgr_prted.c

@@ -435,7 +435,7 @@ static void proc_errors(int fd, short args, void *cbdata)
             }
             /* if all my routes and children are gone, then terminate
                ourselves nicely (i.e., this is a normal termination) */
-            if (0 == pmix_list_get_size(&prte_rml_base.children)) {
+            if (0 == prte_rml_base.n_children) {
                 PMIX_OUTPUT_VERBOSE((2, prte_errmgr_base_framework.framework_output,
                                      "%s errmgr:default:prted all routes gone - exiting",
                                      PRTE_NAME_PRINT(PRTE_PROC_MY_NAME)));
@@ -444,7 +444,7 @@ static void proc_errors(int fd, short args, void *cbdata)
                 PMIX_OUTPUT_VERBOSE((2, prte_errmgr_base_framework.framework_output,
                                      "%s errmgr:default:prted not exiting, num_routes() == %d",
                                      PRTE_NAME_PRINT(PRTE_PROC_MY_NAME),
-                                     (int) pmix_list_get_size(&prte_rml_base.children)));
+                                     prte_rml_base.n_children));
             }
         }
         /* if not, then we can continue */
@@ -627,7 +627,7 @@ static void proc_errors(int fd, short args, void *cbdata)
             }
             /* if all my routes and children are gone, then terminate
                ourselves nicely (i.e., this is a normal termination) */
-            if (0 == pmix_list_get_size(&prte_rml_base.children)) {
+            if (0 == prte_rml_base.n_children) {
                 PMIX_OUTPUT_VERBOSE((2, prte_errmgr_base_framework.framework_output,
                                      "%s errmgr:default:prted all routes gone - exiting",
                                      PRTE_NAME_PRINT(PRTE_PROC_MY_NAME)));

src/mca/filem/filem.h

@@ -209,6 +209,12 @@ typedef int (*prte_filem_base_module_init_fn_t)(void);
  */
 typedef int (*prte_filem_base_module_finalize_fn_t)(void);
 
+/**
+ * Recovers filem operations from given faults or else activates relevant job
+ * failure state.
+ */
+typedef void (*prte_filem_base_fault_handler_fn_t)(const prte_rml_recovery_status_t *status);
+
 /**
  * Put a file or directory on the remote machine
  *
@@ -350,6 +356,9 @@ struct prte_filem_base_module_1_0_0_t {
     /** Finalization Function */
     prte_filem_base_module_finalize_fn_t filem_finalize;
 
+    /** Respond to daemon failures */
+    prte_filem_base_fault_handler_fn_t fault_handler;
+
     /** Put a file on the remote machine */
     prte_filem_base_put_fn_t put;
     prte_filem_base_put_nb_fn_t put_nb;

src/mca/filem/raw/filem_raw_module.c

@@ -63,12 +63,14 @@
 
 static int raw_init(void);
 static int raw_finalize(void);
+static void raw_fault_handler(const prte_rml_recovery_status_t *status);
 static int raw_preposition_files(prte_job_t *jdata, prte_filem_completion_cbfunc_t cbfunc,
                                  void *cbdata);
 static int raw_link_local_files(prte_job_t *jdata, prte_app_context_t *app);
 
 prte_filem_base_module_t prte_filem_raw_module = {.filem_init = raw_init,
                                                   .filem_finalize = raw_finalize,
+                                                  .fault_handler = raw_fault_handler,
                                                   /* we don't use any of the following */
                                                   .put = prte_filem_base_none_put,
                                                   .put_nb = prte_filem_base_none_put_nb,
@@ -135,6 +137,20 @@ static int raw_finalize(void)
     return PRTE_SUCCESS;
 }
 
+static void raw_fault_handler(const prte_rml_recovery_status_t *status){
+    PRTE_HIDE_UNUSED_PARAMS(status);
+    /* TODO: Make this actually resilient. Seems pretty trivial, since xcast is
+     * already resilient */
+    if (0 < pmix_list_get_size(&incoming_files) ||
+        0 < pmix_list_get_size(&outbound_files)) {
+        PMIX_OUTPUT_VERBOSE((0, prte_filem_base_framework.framework_output,
+                             "%s filem:raw daemon failed during active file"
+                             " transfer operation(s)",
+                             PRTE_NAME_PRINT(PRTE_PROC_MY_NAME)));
+        PRTE_ACTIVATE_JOB_STATE(NULL, PRTE_JOB_STATE_COMM_FAILED);
+    }
+}
+
 static void xfer_complete(int status, prte_filem_raw_xfer_t *xfer)
 {
     prte_filem_raw_outbound_t *outbound = xfer->outbound;

src/mca/grpcomm/direct/grpcomm_direct.c

@@ -40,11 +40,13 @@
 /* Static API's */
 static int init(void);
 static void finalize(void);
+static void fault_handler(const prte_rml_recovery_status_t* status);
 
 /* Module def */
 prte_grpcomm_base_module_t prte_grpcomm_direct_module = {
     .init = init,
     .finalize = finalize,
+    .fault_handler = fault_handler,
     .xcast = prte_grpcomm_direct_xcast,
     .fence = prte_grpcomm_direct_fence,
     .group = prte_grpcomm_direct_group
@@ -56,12 +58,16 @@ prte_grpcomm_base_module_t prte_grpcomm_direct_module = {
 static int init(void)
 {
     /* setup the trackers */
+    PMIX_CONSTRUCT(&prte_mca_grpcomm_direct_component.xcast_ops,
+                   prte_grpcomm_xcast_t);
     PMIX_CONSTRUCT(&prte_mca_grpcomm_direct_component.fence_ops, pmix_list_t);
     PMIX_CONSTRUCT(&prte_mca_grpcomm_direct_component.group_ops, pmix_list_t);
 
-    /* xcast receive */
+    /* xcast receives */
     PRTE_RML_RECV(PRTE_NAME_WILDCARD, PRTE_RML_TAG_XCAST,
                   PRTE_RML_PERSISTENT, prte_grpcomm_direct_xcast_recv, NULL);
+    PRTE_RML_RECV(PRTE_NAME_WILDCARD, PRTE_RML_TAG_XCAST_ACK,
+                  PRTE_RML_PERSISTENT, prte_grpcomm_direct_xcast_ack, NULL);
 
     /* fence receives */
     PRTE_RML_RECV(PRTE_NAME_WILDCARD, PRTE_RML_TAG_FENCE,
@@ -86,14 +92,22 @@ static int init(void)
  */
 static void finalize(void)
 {
-
+    PMIX_DESTRUCT(&prte_mca_grpcomm_direct_component.xcast_ops);
     PMIX_LIST_DESTRUCT(&prte_mca_grpcomm_direct_component.fence_ops);
     PMIX_LIST_DESTRUCT(&prte_mca_grpcomm_direct_component.group_ops);
 
     PRTE_RML_CANCEL(PRTE_NAME_WILDCARD, PRTE_RML_TAG_XCAST);
+    PRTE_RML_CANCEL(PRTE_NAME_WILDCARD, PRTE_RML_TAG_XCAST_ACK);
     PRTE_RML_CANCEL(PRTE_NAME_WILDCARD, PRTE_RML_TAG_FENCE);
     PRTE_RML_CANCEL(PRTE_NAME_WILDCARD, PRTE_RML_TAG_FENCE_RELEASE);
     PRTE_RML_CANCEL(PRTE_NAME_WILDCARD, PRTE_RML_TAG_GROUP);
     PRTE_RML_CANCEL(PRTE_NAME_WILDCARD, PRTE_RML_TAG_GROUP_RELEASE);
     return;
 }
+
+static void fault_handler(const prte_rml_recovery_status_t* status)
+{
+    prte_grpcomm_direct_xcast_fault_handler(status);
+    prte_grpcomm_direct_fence_fault_handler(status);
+    prte_grpcomm_direct_group_fault_handler(status);
+}

src/mca/grpcomm/direct/grpcomm_direct.h

@@ -21,16 +21,37 @@
 
 BEGIN_C_DECLS
 
+/* Tracks ongoing xcast operations to ensure all messages are delivered exactly
+ * once to all daemons even in the presence of daemon failures */
+typedef struct {
+    pmix_object_t super;
+    // list of ongoing operations, defined in grpcomm_direct_xcast.c
+    pmix_list_t ops;
+    // list of operations sent to HNP to be started, but not seen since
+    pmix_list_t pending_ops;
+    // global op id of the last known completed (in our subtree) operation
+    size_t op_id_completed;
+    // global op id of what was completed (in our subtree) when we were last
+    // promoted (meaning our subtree grew, so we can't assume completion in the
+    // new subtree)
+    size_t op_id_completed_at_promotion;
+    // local op id of the last op generated here
+    size_t op_id_local;
+    // used by HNP to assign global op id
+    size_t op_id_global;
+} prte_grpcomm_xcast_t;
+PRTE_MODULE_EXPORT PMIX_CLASS_DECLARATION(prte_grpcomm_xcast_t);
+
 /*
  * Grpcomm interfaces
  */
-
 typedef struct {
-	prte_grpcomm_base_component_t super;
-	// track ongoing fence operations - list of prte_grpcomm_fence_t
-	pmix_list_t fence_ops;
-	// track ongoiong group operations - list of prte_grpcomm_group_t
-	pmix_list_t group_ops;
+    prte_grpcomm_base_component_t super;
+    prte_grpcomm_xcast_t xcast_ops;
+    // track ongoing fence operations - list of prte_grpcomm_fence_t
+    pmix_list_t fence_ops;
+    // track ongoiong group operations - list of prte_grpcomm_group_t
+    pmix_list_t group_ops;
 } prte_grpcomm_direct_component_t;
 
 PRTE_MODULE_EXPORT extern prte_grpcomm_direct_component_t prte_mca_grpcomm_direct_component;
@@ -65,7 +86,6 @@ typedef struct {
 } prte_grpcomm_direct_group_signature_t;
 PRTE_MODULE_EXPORT PMIX_CLASS_DECLARATION(prte_grpcomm_direct_group_signature_t);
 
-
 /* Internal component object for tracking ongoing
  * allgather operations */
 typedef struct {
@@ -169,6 +189,14 @@ void prte_grpcomm_direct_xcast_recv(int status, pmix_proc_t *sender,
                                     pmix_data_buffer_t *buffer,
                                     prte_rml_tag_t tg, void *cbdata);
 
+PRTE_MODULE_EXPORT extern
+void prte_grpcomm_direct_xcast_ack(int status, pmix_proc_t *sender,
+                                   pmix_data_buffer_t *buffer,
+                                   prte_rml_tag_t tg, void *cbdata);
+
+PRTE_MODULE_EXPORT extern
+void prte_grpcomm_direct_xcast_fault_handler(const prte_rml_recovery_status_t* status);
+
 /* fence functions */
 PRTE_MODULE_EXPORT extern
 int prte_grpcomm_direct_fence(const pmix_proc_t procs[], size_t nprocs,
@@ -185,6 +213,8 @@ void prte_grpcomm_direct_fence_release(int status, pmix_proc_t *sender,
                                 	   pmix_data_buffer_t *buffer,
                                 	   prte_rml_tag_t tag, void *cbdata);
 
+PRTE_MODULE_EXPORT extern
+void prte_grpcomm_direct_fence_fault_handler(const prte_rml_recovery_status_t* status);
 
 /* group functions */
 PRTE_MODULE_EXPORT extern
@@ -193,6 +223,9 @@ int prte_grpcomm_direct_group(pmix_group_operation_t op, char *grpid,
                               const pmix_info_t directives[], size_t ndirs,
                               pmix_info_cbfunc_t cbfunc, void *cbdata);
 
+PRTE_MODULE_EXPORT extern
+void prte_grpcomm_direct_group_fault_handler(const prte_rml_recovery_status_t* status);
+
 #if PMIX_NUMERIC_VERSION >= 0x00060000
 
 PRTE_MODULE_EXPORT extern

src/mca/grpcomm/direct/grpcomm_direct_component.c

@@ -162,7 +162,6 @@ PMIX_CLASS_INSTANCE(prte_grpcomm_group_t,
                     pmix_list_item_t,
                     gccon, gcdes);
 
-
 static void mdcon(prte_pmix_fence_caddy_t *p)
 {
     p->sig = NULL;

src/mca/grpcomm/direct/grpcomm_direct_fence.c

@@ -81,6 +81,20 @@ int prte_grpcomm_direct_fence(const pmix_proc_t procs[], size_t nprocs,
     return PRTE_SUCCESS;
 }
 
+void prte_grpcomm_direct_fence_fault_handler(const prte_rml_recovery_status_t* status)
+{
+    PRTE_HIDE_UNUSED_PARAMS(status);
+    /* TODO: make this actually resilient
+     * For now, we'll just kill the job if any ops are active */
+    if(0 < pmix_list_get_size(&prte_mca_grpcomm_direct_component.fence_ops)){
+        PMIX_OUTPUT_VERBOSE((0, prte_grpcomm_base_framework.framework_output,
+                             "%s grpcomm:direct:fence daemon failed during"
+                             " active fence operation(s)",
+                             PRTE_NAME_PRINT(PRTE_PROC_MY_NAME)));
+        PRTE_ACTIVATE_JOB_STATE(NULL, PRTE_JOB_STATE_COMM_FAILED);
+    }
+}
+
 static void fence(int sd, short args, void *cbdata)
 {
     prte_pmix_fence_caddy_t *cd = (prte_pmix_fence_caddy_t *) cbdata;

src/mca/grpcomm/direct/grpcomm_direct_group.c

@@ -95,6 +95,21 @@ int prte_grpcomm_direct_group(pmix_group_operation_t op, char *grpid,
     return PRTE_SUCCESS;
 }
 
+void prte_grpcomm_direct_group_fault_handler(const prte_rml_recovery_status_t* status)
+{
+    PRTE_HIDE_UNUSED_PARAMS(status);
+    /* TODO: make this actually resilient
+     * For now, we'll just kill the job if any ops are active */
+    if(0 < pmix_list_get_size(&prte_mca_grpcomm_direct_component.group_ops)){
+        PMIX_OUTPUT_VERBOSE((0, prte_grpcomm_base_framework.framework_output,
+                             "%s grpcomm:direct:group daemon failed during"
+                             " active group operation(s)",
+                             PRTE_NAME_PRINT(PRTE_PROC_MY_NAME)));
+        PRTE_ACTIVATE_JOB_STATE(NULL, PRTE_JOB_STATE_COMM_FAILED);
+    }
+}
+
+
 static void group(int sd, short args, void *cbdata)
 {
     prte_pmix_grp_caddy_t *cd = (prte_pmix_grp_caddy_t*)cbdata;