add facility to receive callback on disconnection from peer

This commit was SVN r7650.

orte/mca/oob/base/Makefile.am

@@ -29,6 +29,7 @@ libmca_oob_base_la_SOURCES = \
         $(headers) \
         oob_base_barrier.c \
         oob_base_close.c \
+        oob_base_except.c \
         oob_base_init.c \
         oob_base_open.c \
         oob_base_ping.c \

orte/mca/oob/base/base.h

@@ -409,9 +409,40 @@ OMPI_DECLSPEC int mca_oob_xcast(
     orte_buffer_t* buffer,
     orte_gpr_trigger_cb_fn_t cbfunc);
 
+/*
+ * Callback on exception condition.
+ */
+
+typedef enum {
+    MCA_OOB_PEER_UNREACH,
+    MCA_OOB_PEER_DISCONNECTED
+} mca_oob_base_exception_t;
+
+typedef int (*mca_oob_base_exception_fn_t)(const orte_process_name_t* peer, int exception);
+                                                                                                                 
+/**
+ *  Register a callback function on loss of a connection.
+ */
+
+OMPI_DECLSPEC int mca_oob_add_exception_handler(
+    mca_oob_base_exception_fn_t cbfunc);
+
+/**
+ * Remove a callback
+ */
+
+OMPI_DECLSPEC int mca_oob_del_exception_handler(
+    mca_oob_base_exception_fn_t cbfunc);
+
+/**
+ * Invoke exception handlers
+ */
+
+OMPI_DECLSPEC void mca_oob_call_exception_handlers(
+    orte_process_name_t* peer, int exception);
+
 #if defined(c_plusplus) || defined(__cplusplus)
 }
 #endif
-
 #endif
 

orte/mca/oob/base/oob_base_close.c

@@ -45,6 +45,7 @@ int mca_oob_base_close(void)
 
   OBJ_DESTRUCT(&mca_oob_base_modules);
   OBJ_DESTRUCT(&mca_oob_base_components);
+  OBJ_DESTRUCT(&mca_oob_base_exception_handlers);
 
   if( NULL != mca_oob_base_include )
     free(mca_oob_base_include);

orte/mca/oob/base/oob_base_except.c

@@ -0,0 +1,84 @@
+/*
+ * Copyright (c) 2004-2005 The Trustees of Indiana University.
+ *                         All rights reserved.
+ * Copyright (c) 2004-2005 The Trustees of the University of Tennessee.
+ *                         All rights reserved.
+ * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
+ *                         University of Stuttgart.  All rights reserved.
+ * Copyright (c) 2004-2005 The Regents of the University of California.
+ *                         All rights reserved.
+ * $COPYRIGHT$
+ *
+ * Additional copyrights may follow
+ *
+ * $HEADER$
+ */
+                                                                                                                 
+                                                                                                                 
+#include "ompi_config.h"
+#include "include/constants.h"
+#include "mca/mca.h"
+#include "mca/base/base.h"
+#include "mca/oob/oob.h"
+
+
+OBJ_CLASS_INSTANCE(
+    mca_oob_base_exception_handler_t,
+    opal_list_item_t,
+    NULL,
+    NULL
+);
+
+
+/**
+ *  Register a callback function on loss of a connection.
+ */
+                                                                                                                 
+OMPI_DECLSPEC int mca_oob_add_exception_handler(
+    mca_oob_base_exception_fn_t cbfunc)
+{
+    mca_oob_base_exception_handler_t *eh = OBJ_NEW(mca_oob_base_exception_handler_t);
+    eh->cbfunc = cbfunc;
+    opal_list_append(&mca_oob_base_exception_handlers, &eh->super);
+    return ORTE_SUCCESS;
+}
+                                                                                                                 
+/**
+ * Remove a callback
+ */
+                                                                                                                 
+OMPI_DECLSPEC int mca_oob_del_exception_handler(
+    mca_oob_base_exception_fn_t cbfunc)
+{
+    opal_list_item_t* item;
+    item = opal_list_get_first(&mca_oob_base_exception_handlers);
+    while(item != opal_list_get_end(&mca_oob_base_exception_handlers)) {
+        opal_list_item_t* next = opal_list_get_next(item);
+        mca_oob_base_exception_handler_t* eh = (mca_oob_base_exception_handler_t*)item;
+        if(eh->cbfunc == cbfunc) {
+            opal_list_remove_item(&mca_oob_base_exception_handlers, &eh->super);
+            OBJ_RELEASE(eh);
+        }
+        item = next;
+    }
+    return ORTE_SUCCESS;
+}
+
+/**
+ * Invoke exception handlers
+ */
+                                                                                                                 
+OMPI_DECLSPEC void mca_oob_call_exception_handlers(
+    orte_process_name_t* peer, int exception)
+{
+    opal_list_item_t* item;
+    item = opal_list_get_first(&mca_oob_base_exception_handlers);
+    while(item != opal_list_get_end(&mca_oob_base_exception_handlers)) {
+        opal_list_item_t* next = opal_list_get_next(item);
+        mca_oob_base_exception_handler_t* eh = (mca_oob_base_exception_handler_t*)item;
+        eh->cbfunc(peer,exception);
+        item = next;
+    }
+}
+                                                                                                                 
+

orte/mca/oob/base/oob_base_open.c

@@ -41,6 +41,7 @@ char* mca_oob_base_include = NULL;
 char* mca_oob_base_exclude = NULL;
 opal_list_t mca_oob_base_components;
 opal_list_t mca_oob_base_modules;
+opal_list_t mca_oob_base_exception_handlers;
 
 /**
  * Function for finding and opening either all MCA components, or the one
@@ -52,6 +53,7 @@ int mca_oob_base_open(void)
 
   OBJ_CONSTRUCT(&mca_oob_base_components, opal_list_t);
   OBJ_CONSTRUCT(&mca_oob_base_modules, opal_list_t);
+  OBJ_CONSTRUCT(&mca_oob_base_exception_handlers, opal_list_t);
 
   if (OMPI_SUCCESS != 
       mca_base_components_open("oob", mca_oob_base_output,

orte/mca/oob/oob.h

@@ -187,9 +187,9 @@ typedef int (*mca_oob_base_module_init_fn_t)(void);
  */
 typedef int (*mca_oob_base_module_fini_fn_t)(void);
 
-    /**
-     * xcast function for sending common messages to all processes
-     */
+/**
+ * xcast function for sending common messages to all processes
+ */
 typedef int (*mca_oob_base_module_xcast_fn_t)(orte_process_name_t* root, 
     orte_process_name_t* peers,
     size_t num_peers,
@@ -271,6 +271,24 @@ typedef struct mca_oob_base_info_t mca_oob_base_info_t;
  */
 OBJ_CLASS_DECLARATION(mca_oob_base_info_t);
 
+/**
+ * List of registrations of exception callbacks
+ */
+struct mca_oob_base_exception_handler_t {
+  opal_list_item_t super;
+  mca_oob_base_exception_fn_t cbfunc;
+};
+
+/**
+ * Convenience Typedef
+ */
+typedef struct mca_oob_base_exception_handler_t mca_oob_base_exception_handler_t;
+
+/**
+ * declare the association structure as a class
+ */
+OBJ_CLASS_DECLARATION(mca_oob_base_exception_handler_t);
+
 
 /*
  * Global functions for MCA overall collective open and close
@@ -289,6 +307,7 @@ OMPI_DECLSPEC extern char* mca_oob_base_include;
 OMPI_DECLSPEC extern char* mca_oob_base_exclude;
 OMPI_DECLSPEC extern opal_list_t mca_oob_base_components;
 OMPI_DECLSPEC extern opal_list_t mca_oob_base_modules;
+OMPI_DECLSPEC extern opal_list_t mca_oob_base_exception_handlers;
 
 #if defined(c_plusplus) || defined(__cplusplus)
 }

orte/mca/oob/tcp/oob_tcp_msg.c

@@ -286,6 +286,7 @@ static bool mca_oob_tcp_msg_recv(mca_oob_tcp_msg_t* msg, mca_oob_tcp_peer_t* pee
                     ORTE_NAME_ARGS(&(peer->peer_name)),
                     ompi_socket_errno);
                 mca_oob_tcp_peer_close(peer);
+                mca_oob_call_exception_handlers(&peer->peer_name, MCA_OOB_PEER_DISCONNECTED);
                 return false;
             }
         } else if (rc == 0)  {
@@ -296,6 +297,7 @@ static bool mca_oob_tcp_msg_recv(mca_oob_tcp_msg_t* msg, mca_oob_tcp_peer_t* pee
                    ompi_socket_errno);
             }
             mca_oob_tcp_peer_close(peer);
+            mca_oob_call_exception_handlers(&peer->peer_name, MCA_OOB_PEER_DISCONNECTED);
             return false;
         }
 

orte/mca/rml/oob/rml_oob_component.c

@@ -76,7 +76,9 @@ orte_rml_module_t orte_rml_oob_module = {
     (orte_rml_module_recv_buffer_nb_fn_t)mca_oob_recv_packed_nb,
     (orte_rml_module_recv_cancel_fn_t)mca_oob_recv_cancel,
     (orte_rml_module_barrier_fn_t)mca_oob_barrier,
-    (orte_rml_module_xcast_fn_t)mca_oob_xcast
+    (orte_rml_module_xcast_fn_t)mca_oob_xcast,
+    (orte_rml_module_exception_fn_t)mca_oob_add_exception_handler,
+    (orte_rml_module_exception_fn_t)mca_oob_del_exception_handler
 };
 
 

orte/mca/rml/rml.h

@@ -341,6 +341,26 @@ typedef int (*orte_rml_module_xcast_fn_t)(
     orte_gpr_trigger_cb_fn_t cbfunc,
     void *user_tag);
 
+/*
+ * Callback on exception condition.
+ */
+                                                                                                                 
+typedef enum {
+    ORTE_RML_PEER_UNREACH,
+    ORTE_RML_PEER_DISCONNECTED
+} orte_rml_exception_t;
+                                                                                                                 
+typedef void (*orte_rml_exception_callback_t)(
+    const orte_process_name_t* peer, 
+    orte_rml_exception_t exception);
+                                                                                                                 
+/**
+ *  Register a callback function on loss of a connection.
+ */
+                                                                                                                 
+typedef int (*orte_rml_module_exception_fn_t)(
+    orte_rml_exception_callback_t cbfunc);
+                                                                                                                 
 /*
  * Initialization/Cleanup
  */
@@ -377,6 +397,8 @@ struct orte_rml_module_t {
     orte_rml_module_recv_cancel_fn_t     recv_cancel;
     orte_rml_module_barrier_fn_t         barrier;
     orte_rml_module_xcast_fn_t           xcast;
+    orte_rml_module_exception_fn_t       add_exception_handler;
+    orte_rml_module_exception_fn_t       del_exception_handler;
 };
 typedef struct orte_rml_module_t orte_rml_module_t;