Added some fork tests, fixed some bugs it found

common.h

@@ -557,6 +557,12 @@ int read_blocked(int fd, void *buf, size_t count);
  */
 ssize_t write_loop(int fd, const char *buff, size_t count);
 
+/**
+   Loop a read request while failiure is non-critical. Return -1 and set errno
+   in case of critical error.
+ */
+ssize_t read_loop(int fd, void *buff, size_t count);
+
 
 /**
    Issue a debug message with printf-style string formating and
@@ -707,4 +713,9 @@ void assert_is_not_forked_child(const char *who);
 #define ASSERT_IS_NOT_FORKED_CHILD_TRAMPOLINE(x) assert_is_not_forked_child(x)
 #define ASSERT_IS_NOT_FORKED_CHILD() ASSERT_IS_NOT_FORKED_CHILD_TRAMPOLINE(__FUNCTION__)
 
+extern "C" {
+__attribute__((noinline)) void debug_thread_error(void);
+}
+
+
 #endif

fish_tests.cpp

@@ -51,7 +51,9 @@
 #include "path.h"
 #include "history.h"
 #include "highlight.h"
-
+#include "iothread.h"
+#include "postfork.h"
+#include "signal.h"
 /**
    The number of tests to run
  */
@@ -324,9 +326,58 @@ static void test_tok()
 			}
 		}
 	}
-	
+}
 
-		
+static int test_fork_helper(void *unused) {
+    size_t i;
+    for (i=0; i < 100000; i++) {
+        delete [] (new char[4 * 1024 * 1024]);
+    }
+    return 0;
+}
+
+static void test_fork(void) {
+    return; // Test is disabled until I can force it to fail
+    say(L"Testing fork");
+    size_t i, max = 100;
+    for (i=0; i < 100; i++) {
+        printf("%lu / %lu\n", i+1, max);
+        /* Do something horrible to try to trigger an error */
+#define THREAD_COUNT 8
+#define FORK_COUNT 200
+#define FORK_LOOP_COUNT 16
+        signal_block();
+        for (size_t i=0; i < THREAD_COUNT; i++) {
+            iothread_perform<void>(test_fork_helper, NULL, NULL);
+        }
+        for (size_t q = 0; q < FORK_LOOP_COUNT; q++) {
+            size_t pids[FORK_COUNT];
+            for (size_t i=0; i < FORK_COUNT; i++) {
+                pid_t pid = execute_fork(false);
+                if (pid > 0) {
+                    /* Parent */
+                    pids[i] = pid;
+                } else if (pid == 0) {
+                    /* Child */
+                    new char[4 * 1024 * 1024];
+                    exit_without_destructors(0);
+                } else {
+                    perror("fork");
+                }        
+            }
+            for (size_t i=0; i < FORK_COUNT; i++) {
+                int status = 0;
+                if (pids[i] != waitpid(pids[i], &status, 0)) {
+                    perror("waitpid");
+                    assert(0);
+                }
+                assert(WIFEXITED(status) && 0 == WEXITSTATUS(status));
+            }
+        }
+        iothread_drain_all();
+        signal_unblock();
+    }
+#undef FORK_COUNT
 }
 
 /**
@@ -719,11 +770,12 @@ int main( int argc, char **argv )
 	builtin_init();
 	reader_init();
 	env_init();
-
+    
     test_format();
 	test_escape();
 	test_convert();
 	test_tok();
+    test_fork();
 	test_parser();
 	test_lru();
 	test_expand();

iothread.cpp

@@ -9,10 +9,10 @@
 #include <stdlib.h>
 #include <unistd.h>
 #include <signal.h>
+#include <fcntl.h>
 #include <queue>
 
-
-#define VOMIT_ON_FAILURE(a) do { if (0 != (a)) { int err = errno; fprintf(stderr, "%s failed on line %d in file %s: %d (%s)\n", #a, __LINE__, __FILE__, err, strerror(err)); abort(); }} while (0)
+#define VOMIT_ON_FAILURE(a) do { if (0 != (a)) { int err = errno; fprintf(stderr, "%s failed on line %d in file %s: %d (%s). Break on debug_thread_error to debug.\n", #a, __LINE__, __FILE__, err, strerror(err)); debug_thread_error(); abort(); }} while (0)
 
 #ifdef _POSIX_THREAD_THREADS_MAX
   #if _POSIX_THREAD_THREADS_MAX < 64
@@ -67,6 +67,10 @@ static void iothread_init(void) {
 		VOMIT_ON_FAILURE(pipe(pipes));
 		s_read_pipe = pipes[0];
 		s_write_pipe = pipes[1];
+        
+        // 0 means success to VOMIT_ON_FAILURE. Arrange to pass 0 if fcntl returns anything other than -1.
+        VOMIT_ON_FAILURE(-1 == fcntl(s_read_pipe, F_SETFD, FD_CLOEXEC));
+        VOMIT_ON_FAILURE(-1 == fcntl(s_write_pipe, F_SETFD, FD_CLOEXEC));
 		
 		/* Tell each thread its index */
 		for (ThreadIndex_t i=0; i < IO_MAX_THREADS; i++) {
@@ -109,7 +113,7 @@ static void *iothread_worker(void *threadPtr) {
 	}
 	
 	/* Write our index to wake up the main thread */
-	VOMIT_ON_FAILURE(! write(s_write_pipe, &thread->idx, sizeof thread->idx));
+	VOMIT_ON_FAILURE(! write_loop(s_write_pipe, (const char *)&thread->idx, sizeof thread->idx));
 	
 	/* We're done */
 	return req;
@@ -168,7 +172,7 @@ int iothread_port(void) {
 void iothread_service_completion(void) {
     ASSERT_IS_MAIN_THREAD();
 	ThreadIndex_t threadIdx = (ThreadIndex_t)-1;
-	VOMIT_ON_FAILURE(! read(iothread_port(), &threadIdx, sizeof threadIdx));
+	VOMIT_ON_FAILURE(1 != read_loop(iothread_port(), &threadIdx, sizeof threadIdx));
 	assert(threadIdx < IO_MAX_THREADS);
 	
 	struct WorkerThread_t *thread = &threads[threadIdx];
@@ -197,7 +201,7 @@ void iothread_service_completion(void) {
 
 void iothread_drain_all(void) {
     ASSERT_IS_MAIN_THREAD();
-    ASSERT_IS_NOT_FORKED_CHILD();    
+    ASSERT_IS_NOT_FORKED_CHILD();
     while (s_active_thread_count > 0) {
         iothread_service_completion();
     }

iothread.h

@@ -25,7 +25,7 @@ int iothread_port(void);
 /** Services one iothread competion callback. */
 void iothread_service_completion(void);
 
-/** Cancels all outstanding requests and waits for all iothreads to terminate. */
+/** Waits for all iothreads to terminate. */
 void iothread_drain_all(void);
 
 /** Helper template */