Merge pull request #647 from rhc54/topic/hangs

rhc54 · rhc54 · commit 5d38283c7f64 · 2015-06-18T10:25:07.000-07:00
Ensure we properly commit suicide if/when we lose connection to the daemon.
diff --git a/opal/mca/pmix/native/pmix_native.c b/opal/mca/pmix/native/pmix_native.c
@@ -102,8 +102,7 @@ const opal_pmix_base_module_t opal_pmix_native_module = {
 
 // local variables
 static int init_cntr = 0;
-opal_process_name_t native_pname = {0};
-
+static opal_process_name_t native_pname = {0};
 
 /* callback for wait completion */
 static void wait_cbfunc(opal_buffer_t *buf, void *cbdata)
@@ -213,7 +212,7 @@ static int native_fini(void)
 
     if (1 != init_cntr) {
         --init_cntr;
-       return OPAL_SUCCESS;
+        return OPAL_SUCCESS;
     }
     init_cntr = 0;
 
@@ -275,12 +274,20 @@ static bool native_initialized(void)
     return false;
 }
 
+static void timeout(int sd, short args, void *cbdata)
+{
+    pmix_cb_t *cb = (pmix_cb_t*)cbdata;
+    cb->active = false;
+}
+
 static int native_abort(int flag, const char msg[])
 {
     opal_buffer_t *bfr;
     pmix_cmd_t cmd = PMIX_ABORT_CMD;
     int rc;
     pmix_cb_t *cb;
+    opal_event_t ev;
+    struct timeval tv = {1, 0};
 
     opal_output_verbose(2, opal_pmix_base_framework.framework_output,
                         "%s pmix:native abort called",
@@ -291,39 +298,48 @@ static int native_abort(int flag, const char msg[])
         return OPAL_SUCCESS;
     }
 
-    /* create a buffer to hold the message */
-    bfr = OBJ_NEW(opal_buffer_t);
-    /* pack the cmd */
-    if (OPAL_SUCCESS != (rc = opal_dss.pack(bfr, &cmd, 1, PMIX_CMD_T))) {
-        OPAL_ERROR_LOG(rc);
-        OBJ_RELEASE(bfr);
-        return rc;
-    }
-    /* pack the status flag */
-    if (OPAL_SUCCESS != (rc = opal_dss.pack(bfr, &flag, 1, OPAL_INT))) {
-        OPAL_ERROR_LOG(rc);
-        OBJ_RELEASE(bfr);
-        return rc;
-    }
-    /* pack the string message - a NULL is okay */
-    if (OPAL_SUCCESS != (rc = opal_dss.pack(bfr, &msg, 1, OPAL_STRING))) {
-        OPAL_ERROR_LOG(rc);
-        OBJ_RELEASE(bfr);
-        return rc;
-    }
+    if (PMIX_USOCK_CONNECTED == mca_pmix_native_component.state) {
+        /* create a buffer to hold the message */
+        bfr = OBJ_NEW(opal_buffer_t);
+        /* pack the cmd */
+        if (OPAL_SUCCESS != (rc = opal_dss.pack(bfr, &cmd, 1, PMIX_CMD_T))) {
+            OPAL_ERROR_LOG(rc);
+            OBJ_RELEASE(bfr);
+            return rc;
+        }
+        /* pack the status flag */
+        if (OPAL_SUCCESS != (rc = opal_dss.pack(bfr, &flag, 1, OPAL_INT))) {
+            OPAL_ERROR_LOG(rc);
+            OBJ_RELEASE(bfr);
+            return rc;
+        }
+        /* pack the string message - a NULL is okay */
+        if (OPAL_SUCCESS != (rc = opal_dss.pack(bfr, &msg, 1, OPAL_STRING))) {
+            OPAL_ERROR_LOG(rc);
+            OBJ_RELEASE(bfr);
+            return rc;
+        }
 
-    /* create a callback object as we need to pass it to the
-     * recv routine so we know which callback to use when
-     * the return message is recvd */
-    cb = OBJ_NEW(pmix_cb_t);
-    cb->active = true;
+        /* create a callback object as we need to pass it to the
+         * recv routine so we know which callback to use when
+         * the return message is recvd */
+        cb = OBJ_NEW(pmix_cb_t);
+        cb->active = true;
 
-    /* push the message into our event base to send to the server */
-    PMIX_ACTIVATE_SEND_RECV(bfr, wait_cbfunc, cb);
+        /* push a timeout event to wake us up just in case this
+         * message cannot get thru - e.g., someone else may have
+         * detected the failure of the server and ordered an abort */
+        opal_event_evtimer_set(mca_pmix_native_component.evbase,
+                               &ev, timeout, cb);
+        opal_event_evtimer_add(&ev, &tv);
 
-    /* wait for the release */
-    PMIX_WAIT_FOR_COMPLETION(cb->active);
-    OBJ_RELEASE(cb);
+        /* push the message into our event base to send to the server */
+        PMIX_ACTIVATE_SEND_RECV(bfr, wait_cbfunc, cb);
+
+        /* wait for the release */
+        PMIX_WAIT_FOR_COMPLETION(cb->active);
+        OBJ_RELEASE(cb);
+    }
     return OPAL_SUCCESS;
 }
 
diff --git a/opal/mca/pmix/native/pmix_native.h b/opal/mca/pmix/native/pmix_native.h
@@ -41,6 +41,13 @@ typedef enum {
     PMIX_USOCK_ACCEPTING
 } pmix_usock_state_t;
 
+/* define a macro for abnormal termination */
+#define PMIX_NATIVE_ABNORMAL_TERM                               \
+    do {                                                        \
+        mca_pmix_native_component.state = PMIX_USOCK_FAILED;    \
+        opal_pmix_base_errhandler(OPAL_ERR_COMM_FAILURE);       \
+    } while(0);
+
 /* define a command type for communicating to the
  * pmix server */
 typedef uint8_t pmix_cmd_t;
@@ -202,12 +209,13 @@ OPAL_MODULE_DECLSPEC int usock_send_connect_ack(void);
         opal_event_active(&ms->ev, OPAL_EV_WRITE, 1);                   \
     } while(0);
 
-#define CLOSE_THE_SOCKET(socket)                                \
-    do {                                                        \
-        shutdown(socket, 2);                                    \
-        close(socket);                                          \
-        /* notify the error handler */                          \
-        opal_pmix_base_errhandler(OPAL_ERR_COMM_FAILURE);       \
+#define CLOSE_THE_SOCKET(socket)                \
+    do {                                        \
+        if (0 <= socket) {                      \
+            shutdown(socket, 2);                \
+            close(socket);                      \
+            socket = -1;                        \
+        }                                       \
     } while(0)
 
 
diff --git a/opal/mca/pmix/native/usock.c b/opal/mca/pmix/native/usock.c
@@ -35,6 +35,12 @@
 #ifdef HAVE_SYS_TYPES_H
 #include <sys/types.h>
 #endif
+#ifdef HAVE_SYS_STAT_H
+#include <sys/stat.h>
+#endif
+#ifdef HAVE_FCNTL_H
+#include <fcntl.h>
+#endif
 
 #include "opal_stdint.h"
 #include "opal/opal_socket_errno.h"
@@ -191,6 +197,7 @@ void pmix_usock_process_msg(int fd, short flags, void *cbdata)
     /* we get here if no matching recv was found - this is an error */
     opal_output(0, "%s UNEXPECTED MESSAGE",
                 OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
+    PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
     OBJ_RELEASE(msg);
 }
 
@@ -222,9 +229,10 @@ static void pmix_usock_try_connect(int fd, short args, void *cbdata)
                             "usock_peer_try_connect: attempting to connect to server on socket %d",
                             mca_pmix_native_component.sd);
         /* try to connect */
-        if (connect(mca_pmix_native_component.sd, &mca_pmix_native_component.address, addrlen) < 0) {
+        if (connect(mca_pmix_native_component.sd, (struct sockaddr*)&mca_pmix_native_component.address, addrlen) < 0) {
             if (opal_socket_errno == ETIMEDOUT) {
-                /* The server may be too busy to accept new connections */
+                /* The server may be too busy to accept new connections,
+                 * so cycle around and let it try again */
                 opal_output_verbose(2, opal_pmix_base_framework.framework_output,
                                     "timeout connecting to server");
                 CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
@@ -235,7 +243,7 @@ static void pmix_usock_try_connect(int fd, short args, void *cbdata)
                abort a connection that was ECONNREFUSED on the last
                attempt, without even trying to establish the
                connection.  Handle that case in a semi-rational
-               way by trying twice before giving up */
+               way by trying again before giving up */
             if (ECONNABORTED == opal_socket_errno) {
                 opal_output_verbose(2, opal_pmix_base_framework.framework_output,
                                     "connection to server aborted by OS - retrying");
@@ -255,6 +263,7 @@ static void pmix_usock_try_connect(int fd, short args, void *cbdata)
         if (0 <= mca_pmix_native_component.sd) {
             CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
         }
+        PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
         return;
     }
 
@@ -282,7 +291,7 @@ static void pmix_usock_try_connect(int fd, short args, void *cbdata)
     opal_event_set_priority(&mca_pmix_native_component.send_event, OPAL_EV_MSG_LO_PRI);
     mca_pmix_native_component.send_ev_active = false;
 
-     /* setup the socket as non-blocking */
+    /* setup the socket as non-blocking */
     if ((flags = fcntl(mca_pmix_native_component.sd, F_GETFL, 0)) < 0) {
         opal_output(0, "usock_peer_connect: fcntl(F_GETFL) failed: %s (%d)\n",
                     strerror(opal_socket_errno),
@@ -310,8 +319,8 @@ static void pmix_usock_try_connect(int fd, short args, void *cbdata)
                     "usock_send_connect_ack to server failed: %s (%d)",
                     OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
                     opal_strerror(rc), rc);
-        mca_pmix_native_component.state = PMIX_USOCK_FAILED;
         CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+        PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
         return;
     }
 }
diff --git a/opal/mca/pmix/native/usock_sendrecv.c b/opal/mca/pmix/native/usock_sendrecv.c
@@ -176,6 +176,8 @@ void pmix_usock_send_handler(int sd, short flags, void *cbdata)
                     mca_pmix_native_component.send_ev_active = false;
                     OBJ_RELEASE(msg);
                     mca_pmix_native_component.send_msg = NULL;
+                    CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+                    PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
                     return;
                 }
             }
@@ -208,6 +210,8 @@ void pmix_usock_send_handler(int sd, short flags, void *cbdata)
                     mca_pmix_native_component.send_ev_active = false;
                     OBJ_RELEASE(msg);
                     mca_pmix_native_component.send_msg = NULL;
+                    CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+                    PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
                     return;
                 }
             }
@@ -239,6 +243,8 @@ void pmix_usock_send_handler(int sd, short flags, void *cbdata)
             opal_event_del(&mca_pmix_native_component.send_event);
             mca_pmix_native_component.send_ev_active = false;
         }
+        CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+        PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
         break;
     }
 }
@@ -356,6 +362,8 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)
                                 OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
             opal_event_del(&mca_pmix_native_component.recv_event);
             mca_pmix_native_component.recv_ev_active = false;
+            CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+            PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
             return;
         }
         break;
@@ -372,6 +380,8 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)
             if (NULL == mca_pmix_native_component.recv_msg) {
                 opal_output(0, "%s usock_recv_handler: unable to allocate recv message\n",
                             OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
+                CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+                PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
                 return;
             }
             /* start by reading the header */
@@ -416,6 +426,7 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)
                                     "%s usock:recv:handler error reading bytes - closing connection",
                                     OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
                 CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+                PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
                 return;
             }
         }
@@ -447,6 +458,7 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)
                 opal_event_del(&mca_pmix_native_component.recv_event);
                 mca_pmix_native_component.recv_ev_active = false;
                 CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+                PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
                 return;
             }
         }
@@ -456,6 +468,7 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)
                     OPAL_NAME_PRINT(OPAL_PROC_MY_NAME), 
                     mca_pmix_native_component.state);
         CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+        PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
         break;
     }
 }
@@ -689,6 +702,7 @@ static void usock_complete_connect(void)
                     opal_socket_errno);
         mca_pmix_native_component.state = PMIX_USOCK_FAILED;
         CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+        PMIX_NATIVE_ABNORMAL_TERM;
         return;
     }
 
@@ -703,8 +717,8 @@ static void usock_complete_connect(void)
                             OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
                             strerror(so_error),
                             so_error);
-        mca_pmix_native_component.state = PMIX_USOCK_FAILED;
         CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+        PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
         return;
     } else if (so_error != 0) {
         /* No need to worry about the return code here - we return regardless
@@ -715,8 +729,8 @@ static void usock_complete_connect(void)
                             "connection to server failed with error %d",
                             OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),
                             so_error);
-        mca_pmix_native_component.state = PMIX_USOCK_FAILED;
         CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+        PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
         return;
     }
 
@@ -737,8 +751,8 @@ static void usock_complete_connect(void)
     } else {
         opal_output(0, "%s usock_complete_connect: unable to send connect ack to server",
                     OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));
-        mca_pmix_native_component.state = PMIX_USOCK_FAILED;
         CLOSE_THE_SOCKET(mca_pmix_native_component.sd);
+        PMIX_NATIVE_ABNORMAL_TERM;  // report the error upstream
     }
 }
 

Original file line number	Diff line number	Diff line change
`@@ -176,6 +176,8 @@ void pmix_usock_send_handler(int sd, short flags, void *cbdata)`
`176`	`176`	`mca_pmix_native_component.send_ev_active = false;`
`177`	`177`	`OBJ_RELEASE(msg);`
`178`	`178`	`mca_pmix_native_component.send_msg = NULL;`
	`179`	`+ CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`180`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`179`	`181`	`return;`
`180`	`182`	`}`
`181`	`183`	`}`
`@@ -208,6 +210,8 @@ void pmix_usock_send_handler(int sd, short flags, void *cbdata)`
`208`	`210`	`mca_pmix_native_component.send_ev_active = false;`
`209`	`211`	`OBJ_RELEASE(msg);`
`210`	`212`	`mca_pmix_native_component.send_msg = NULL;`
	`213`	`+ CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`214`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`211`	`215`	`return;`
`212`	`216`	`}`
`213`	`217`	`}`
`@@ -239,6 +243,8 @@ void pmix_usock_send_handler(int sd, short flags, void *cbdata)`
`239`	`243`	`opal_event_del(&mca_pmix_native_component.send_event);`
`240`	`244`	`mca_pmix_native_component.send_ev_active = false;`
`241`	`245`	`}`
	`246`	`+ CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`247`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`242`	`248`	`break;`
`243`	`249`	`}`
`244`	`250`	`}`
`@@ -356,6 +362,8 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)`
`356`	`362`	`OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));`
`357`	`363`	`opal_event_del(&mca_pmix_native_component.recv_event);`
`358`	`364`	`mca_pmix_native_component.recv_ev_active = false;`
	`365`	`+ CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`366`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`359`	`367`	`return;`
`360`	`368`	`}`
`361`	`369`	`break;`
`@@ -372,6 +380,8 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)`
`372`	`380`	`if (NULL == mca_pmix_native_component.recv_msg) {`
`373`	`381`	`opal_output(0, "%s usock_recv_handler: unable to allocate recv message\n",`
`374`	`382`	`OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));`
	`383`	`+ CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`384`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`375`	`385`	`return;`
`376`	`386`	`}`
`377`	`387`	`/* start by reading the header */`
`@@ -416,6 +426,7 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)`
`416`	`426`	`"%s usock:recv:handler error reading bytes - closing connection",`
`417`	`427`	`OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));`
`418`	`428`	`CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`429`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`419`	`430`	`return;`
`420`	`431`	`}`
`421`	`432`	`}`
`@@ -447,6 +458,7 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)`
`447`	`458`	`opal_event_del(&mca_pmix_native_component.recv_event);`
`448`	`459`	`mca_pmix_native_component.recv_ev_active = false;`
`449`	`460`	`CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`461`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`450`	`462`	`return;`
`451`	`463`	`}`
`452`	`464`	`}`
`@@ -456,6 +468,7 @@ void pmix_usock_recv_handler(int sd, short flags, void *cbdata)`
`456`	`468`	`OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),`
`457`	`469`	`mca_pmix_native_component.state);`
`458`	`470`	`CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`471`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`459`	`472`	`break;`
`460`	`473`	`}`
`461`	`474`	`}`
`@@ -689,6 +702,7 @@ static void usock_complete_connect(void)`
`689`	`702`	`opal_socket_errno);`
`690`	`703`	`mca_pmix_native_component.state = PMIX_USOCK_FAILED;`
`691`	`704`	`CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`705`	`+ PMIX_NATIVE_ABNORMAL_TERM;`
`692`	`706`	`return;`
`693`	`707`	`}`
`694`	`708`
`@@ -703,8 +717,8 @@ static void usock_complete_connect(void)`
`703`	`717`	`OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),`
`704`	`718`	`strerror(so_error),`
`705`	`719`	`so_error);`
`706`		`- mca_pmix_native_component.state = PMIX_USOCK_FAILED;`
`707`	`720`	`CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`721`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`708`	`722`	`return;`
`709`	`723`	`} else if (so_error != 0) {`
`710`	`724`	`/* No need to worry about the return code here - we return regardless`
`@@ -715,8 +729,8 @@ static void usock_complete_connect(void)`
`715`	`729`	`"connection to server failed with error %d",`
`716`	`730`	`OPAL_NAME_PRINT(OPAL_PROC_MY_NAME),`
`717`	`731`	`so_error);`
`718`		`- mca_pmix_native_component.state = PMIX_USOCK_FAILED;`
`719`	`732`	`CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`733`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`720`	`734`	`return;`
`721`	`735`	`}`
`722`	`736`
`@@ -737,8 +751,8 @@ static void usock_complete_connect(void)`
`737`	`751`	`} else {`
`738`	`752`	`opal_output(0, "%s usock_complete_connect: unable to send connect ack to server",`
`739`	`753`	`OPAL_NAME_PRINT(OPAL_PROC_MY_NAME));`
`740`		`- mca_pmix_native_component.state = PMIX_USOCK_FAILED;`
`741`	`754`	`CLOSE_THE_SOCKET(mca_pmix_native_component.sd);`
	`755`	`+ PMIX_NATIVE_ABNORMAL_TERM; // report the error upstream`
`742`	`756`	`}`
`743`	`757`	`}`
`744`	`758`