Fixes for SIGPIPE-related failures

src/cobo/cobo_comm.c

@@ -241,3 +241,61 @@ int write_msg(int fd, ldcs_message_t *msg)
 
    return 0;
 }
+
+/* ll_send_nosignal is a variation on ll_write which uses
+ * send(..., MSG_NOSIGNAL) to avoid SIGPIPE when writing to
+ * a broken socket.
+ */
+int ll_send_nosignal(int fd, void *buf, size_t count)
+{
+   int error;
+   ssize_t result;
+   size_t pos = 0;
+   unsigned char *cbuf = (unsigned char *) buf;
+
+   debug_printf3("Have %lu bytes at %p to write to network\n", count, buf);
+
+   while (pos < count) {
+      result = send(fd, cbuf + pos, count - pos, MSG_NOSIGNAL);
+      debug_printf3("Wrote %d bytes at %p to network: 0x%x 0x%x 0x%x 0x%x 0x%x 0x%x 0x%x 0x%x...\n", (int) result, cbuf + pos,
+                    result > 0 ? ((int) cbuf[pos+0]) : 0,
+                    result > 1 ? ((int) cbuf[pos+1]) : 0,
+                    result > 2 ? ((int) cbuf[pos+2]) : 0,
+                    result > 3 ? ((int) cbuf[pos+3]) : 0,
+                    result > 4 ? ((int) cbuf[pos+4]) : 0,
+                    result > 5 ? ((int) cbuf[pos+5]) : 0,
+                    result > 6 ? ((int) cbuf[pos+6]) : 0,
+                    result > 7 ? ((int) cbuf[pos+7]) : 0);
+      if (result == -1 && (errno == EINTR || errno == EAGAIN))
+         continue;
+      if (result <= 0) {
+         error = errno;
+         err_printf("Error writing to cobo FD %d: %s\n", fd, strerror(error));
+         return -1;
+      }
+      pos += result;
+   }
+   debug_printf3("Sent %lu bytes to fd %d\n", count, fd);
+   return 0;
+}
+
+/* write_msg_nosignal is a variation on write_msg
+ * which uses ll_send_nosignal instead of ll_send
+ * to avoid SIGPIPE when writing to a broken socket.
+ */
+int write_msg_nosignal(int fd, ldcs_message_t *msg)
+{
+   int result = ll_send_nosignal(fd, msg, sizeof(*msg));
+   if (result == -1) {
+      return -1;
+   }
+
+   if (msg->header.len && msg->data) {
+      result = ll_send_nosignal(fd, msg->data, msg->header.len);
+      if (result == -1) {
+         return -1;
+      }
+   }
+
+   return 0;
+}

src/cobo/cobo_comm.h

@@ -24,7 +24,9 @@ Place, Suite 330, Boston, MA 02111-1307 USA
 int ldcs_cobo_read_fd(int fd, void* buf, int size);
 int ldcs_cobo_write_fd(int fd, void* buf, int size);
 int ll_write(int fd, void *buf, size_t count);
+int ll_send_nosignal(int fd, void *buf, size_t count);
 int ll_read(int fd, void *buf, size_t count);
 int write_msg(int fd, ldcs_message_t *msg);
+int write_msg_nosignal(int fd, ldcs_message_t *msg);
 
 #endif /* _COBO_COMM_H */

src/fe/comlib/cobo_fe_comm.c

@@ -178,7 +178,9 @@ int ldcs_audit_server_fe_md_close ( void *data  ) {
    out_msg.data = NULL;
 
    cobo_server_get_root_socket(&root_fd);
-   write_msg(root_fd, &out_msg);
+   /* We use write_msg_nosignal to avoid SIGPIPE if the socket is broken.
+    * We're exiting here, so ignore if the server already exited. */
+   write_msg_nosignal(root_fd, &out_msg);
    return cobo_server_close();
 }
 

testsuite/run_driver_template

@@ -10,6 +10,40 @@ export SPINDLE_TEST=1
 export SPINDLE=SPINDLE_EXEC
 export PATH=$PATH:.
 
+SESSION_ACTIVE=false
+if [ "x$SESSION_ID" != "x" ] || [ "x$SPANK_SPINDLE_USE_SESSION" != "x" ] ; then
+    SESSION_ACTIVE=true
+fi
+
+# If we're not running in a session, wait, on every node in the job, for
+# the log daemon to exit before proceeding to the actual test; after a
+# delay, forcibly kill it.  This ensures that each test connects to a
+# fresh daemon.
+if [ $SESSION_ACTIVE == false ] && [ "x$1" != "x--end-session" ] ; then
+    # This is the command run on each node.
+    LOGD_WAIT_CMD='
+SECONDS=0
+TMP="${TMPDIR:-${TEMPDIR:-/tmp}}"
+LOCK="$TMP/spindle_log_lock"
+TIMEOUT="${SPINDLE_LOGD_SHUTDOWN_TIMEOUT:-15}"
+while PID=$(cat "$LOCK" 2>/dev/null) && kill -0 "$PID" 2>/dev/null ; do
+    if [ $SECONDS -ge $TIMEOUT ]; then
+        echo "WARNING: $(hostname): spindle_logd (pid $PID) still running after $TIMEOUT sec; killing it" >&2
+        kill -9 "$PID" 2>/dev/null
+        break
+    fi
+    sleep 0.1
+done
+rm -f "$TMP/spindle_log" "$TMP/spindle_test" "$TMP/spindle_log_lock" "$TMP/spindle_log_reset"
+'
+    bash -c "$LOGD_WAIT_CMD"
+    if [ "x$TEST_RM" == "xslurm" -o "x$TEST_RM" == "xslurm-plugin" ] && [ "x$SLURM_NNODES" != "x" ] ; then
+        srun --overlap -N $SLURM_NNODES -n $SLURM_NNODES bash -c "$LOGD_WAIT_CMD"
+    elif [ "x$TEST_RM" == "xflux" ] ; then
+        flux exec -r all bash -c "$LOGD_WAIT_CMD"
+    fi
+fi
+
 if [ $1 == --start-session ] ; then
     # With SPANK plugin, sessions are started by argument to salloc/sbatch instead
     if [ "x$TEST_RM" == "xslurm-plugin" ] ; then