Make reads non-blocking if async is specified.

appveyor.yml

@@ -42,7 +42,7 @@ build_script:
   - if NOT "%VSVERSION%"=="Visual Studio 16 2019" if "%PLATFORM%"=="x64" cmake -G "%VSVERSION% Win64" ../test
   - if NOT "%VSVERSION%"=="Visual Studio 16 2019" if "%PLATFORM%"=="Win32" cmake -G "%VSVERSION%" ../test
   - if "%VSVERSION%"=="Visual Studio 16 2019" cmake -G "%VSVERSION%" -A "%PLATFORM%" ../test
-  - msbuild /m /p:Configuration="%CONFIGURATION%" /p:Platform="%PLATFORM%" process.sln
+  - msbuild /m /p:Configuration="%CONFIGURATION%" /p:Platform="%PLATFORM%" subprocess.sln
   - cd %CONFIGURATION%
   - subprocess_test.exe
   - subprocess_mt_test.exe

subprocess.h

@@ -230,6 +230,7 @@ subprocess_weak int subprocess_alive(struct subprocess_s *const process);
 #endif
 
 #if !defined(_WIN32)
+#include <fcntl.h>
 #include <signal.h>
 #include <spawn.h>
 #include <stdlib.h>
@@ -433,7 +434,7 @@ int subprocess_create_named_pipe_helper(void **rd, void **wr) {
   const unsigned long pipeAccessInbound = 0x00000001;
   const unsigned long fileFlagOverlapped = 0x40000000;
   const unsigned long pipeTypeByte = 0x00000000;
-  const unsigned long pipeWait = 0x00000000;
+  const unsigned long pipeNoWait = 0x00000001;
   const unsigned long genericWrite = 0x40000000;
   const unsigned long openExisting = 3;
   const unsigned long fileAttributeNormal = 0x00000080;
@@ -460,7 +461,7 @@ int subprocess_create_named_pipe_helper(void **rd, void **wr) {
 
   *rd =
       CreateNamedPipeA(name, pipeAccessInbound | fileFlagOverlapped,
-                       pipeTypeByte | pipeWait, 1, 4096, 4096, SUBPROCESS_NULL,
+                       pipeTypeByte | pipeNoWait, 1, 4096, 4096, SUBPROCESS_NULL,
                        SUBPROCESS_PTR_CAST(LPSECURITY_ATTRIBUTES, &saAttr));
 
   if (invalidHandleValue == *rd) {
@@ -770,6 +771,7 @@ int subprocess_create_ex(const char *const commandLine[], int options,
   int stdinfd[2];
   int stdoutfd[2];
   int stderrfd[2];
+  int fd, fd_flags;
   pid_t child;
   extern char **environ;
   char *const empty_environment[1] = {SUBPROCESS_NULL};
@@ -899,6 +901,13 @@ int subprocess_create_ex(const char *const commandLine[], int options,
   // Store the stdout read end
   out_process->stdout_file = fdopen(stdoutfd[0], "rb");
 
+  // Set non blocking if we are async
+  if (options & subprocess_option_enable_async) {
+    fd = fileno(out_process->stdout_file);
+    fd_flags = fcntl(fd, F_GETFL, 0);
+    fcntl(fd, F_SETFL, fd_flags | O_NONBLOCK);
+  }
+
   if (subprocess_option_combined_stdout_stderr ==
       (options & subprocess_option_combined_stdout_stderr)) {
     out_process->stderr_file = out_process->stdout_file;
@@ -907,6 +916,13 @@ int subprocess_create_ex(const char *const commandLine[], int options,
     close(stderrfd[1]);
     // Store the stderr read end
     out_process->stderr_file = fdopen(stderrfd[0], "rb");
+
+    // Set non blocking if we are async
+    if (options & subprocess_option_enable_async) {
+      fd = fileno(out_process->stderr_file);
+      fd_flags = fcntl(fd, F_GETFL, 0);
+      fcntl(fd, F_SETFL, fd_flags | O_NONBLOCK);
+    }
   }
 
   // Store the child's pid
@@ -1070,14 +1086,19 @@ unsigned subprocess_read_stdout(struct subprocess_s *const process,
 
     // Means we've got an async read!
     if (error == errorIoPending) {
+      const uintptr_t statusPending = 0x00000103;
+
+      const int wait = statusPending == overlapped.Internal;
+
       if (!GetOverlappedResult(handle,
                                SUBPROCESS_PTR_CAST(LPOVERLAPPED, &overlapped),
-                               &bytes_read, 1)) {
-        const unsigned long errorIoIncomplete = 996;
+                               &bytes_read, wait)) {
         const unsigned long errorHandleEOF = 38;
+        const unsigned long errorBrokenPipe = 109;
+        const unsigned long errorIoIncomplete = 996;
         error = GetLastError();
 
-        if ((error != errorIoIncomplete) && (error != errorHandleEOF)) {
+        if ((errorHandleEOF != error) && (errorBrokenPipe != error) && (errorIoIncomplete != error)) {
           return 0;
         }
       }

test/CMakeLists.txt

@@ -23,9 +23,11 @@
 #
 # For more information, please refer to <http://unlicense.org/>
 
-project(process)
+project(subprocess)
 cmake_minimum_required(VERSION 3.15)
 
+set(SUBPROCESS_USE_SANITIZER "" CACHE STRING "Set which Clang Sanitizer to use")
+
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/../)
 
 add_executable(process_return_zero process_return_zero.c)
@@ -47,7 +49,9 @@ add_executable(process_fail_stackoverflow process_fail_stackoverflow.c)
 add_executable(process_hung process_hung.c)
 add_executable(process_stdout_data process_stdout_data.c)
 add_executable(process_stdout_poll process_stdout_poll.c)
+add_executable(process_stdout_poll_wait_first process_stdout_poll_wait_first.c)
 add_executable(process_stderr_poll process_stderr_poll.c)
+add_executable(process_stderr_poll_wait_first process_stderr_poll_wait_first.c)
 add_executable(process_stdout_large process_stdout_large.c)
 add_executable(process_call_return_argc process_call_return_argc.c)
 
@@ -61,6 +65,11 @@ add_executable(subprocess_test
   post_windows.cpp
 )
 
+if(NOT "${SUBPROCESS_USE_SANITIZER}" STREQUAL "")
+  target_compile_options(subprocess_test PUBLIC -fno-omit-frame-pointer -fsanitize=${SUBPROCESS_USE_SANITIZER})
+  target_link_options(subprocess_test PUBLIC -fno-omit-frame-pointer -fsanitize=${SUBPROCESS_USE_SANITIZER})
+endif()
+
 set(MSVC_FLAGS "/Wall /WX /wd4514 /wd4710 /wd4711 /wd5045")
 
 if("${CMAKE_C_COMPILER_ID}" STREQUAL "GNU")
@@ -117,6 +126,11 @@ add_executable(subprocess_mt_test
   post_windows.cpp
 )
 
+if(NOT "${SUBPROCESS_USE_SANITIZER}" STREQUAL "")
+  target_compile_options(subprocess_mt_test PUBLIC -fno-omit-frame-pointer -fsanitize=${SUBPROCESS_USE_SANITIZER})
+  target_link_options(subprocess_mt_test PUBLIC -fno-omit-frame-pointer -fsanitize=${SUBPROCESS_USE_SANITIZER})
+endif()
+
 if("${CMAKE_CXX_COMPILER_ID}" STREQUAL "Clang")
   if(CMAKE_CXX_COMPILER_FRONTEND_VARIANT STREQUAL "MSVC")
     target_compile_options(subprocess_mt_test PUBLIC "/MT")

test/main.c

@@ -291,8 +291,8 @@ UTEST(create, subprocess_return_special_argv) {
 }
 
 UTEST(create, subprocess_return_lpcmdline) {
-  const char *const commandLine[] = {"./process_return_lpcmdline",
-                                     "noquotes", "should be quoted", 0};
+  const char *const commandLine[] = {"./process_return_lpcmdline", "noquotes",
+                                     "should be quoted", 0};
   struct subprocess_s process;
   int ret = -1;
   size_t cmp_index, index;
@@ -313,7 +313,8 @@ UTEST(create, subprocess_return_lpcmdline) {
 
   // comparing from the back skips exe name
   cmp_index = strlen(compare) - 1;
-  for (index = strlen(temp) - 1; index != 0 && cmp_index != 0; index--,cmp_index--) {
+  for (index = strlen(temp) - 1; index != 0 && cmp_index != 0;
+       index--, cmp_index--) {
     if (temp[index] != compare[cmp_index])
       ASSERT_TRUE(0);
   }
@@ -601,11 +602,11 @@ UTEST(subprocess, read_stdout_async_small) {
   ASSERT_EQ(0, subprocess_create(commandLine, subprocess_option_enable_async,
                                  &process));
 
-  do {
+  while (subprocess_alive(&process)) {
     bytes_read = subprocess_read_stdout(&process, data + index,
                                         sizeof(data) - 1 - index);
     index += bytes_read;
-  } while (bytes_read != 0);
+  }
 
   ASSERT_EQ(13u, index);
 
@@ -631,11 +632,11 @@ UTEST(subprocess, read_stdout_async) {
   ASSERT_EQ(0, subprocess_create(commandLine, subprocess_option_enable_async,
                                  &process));
 
-  do {
+  while (subprocess_alive(&process)) {
     bytes_read = subprocess_read_stdout(&process, data + index,
                                         sizeof(data) - 1 - index);
     index += bytes_read;
-  } while (bytes_read != 0);
+  }
 
   ASSERT_EQ(212992u, index);
 
@@ -661,7 +662,57 @@ UTEST(subprocess, poll_stdout_async) {
   ASSERT_EQ(0, subprocess_create(commandLine, subprocess_option_enable_async,
                                  &process));
 
-  do {
+  while (subprocess_alive(&process)) {
+    bytes_read = subprocess_read_stdout(&process, data + index,
+                                        sizeof(data) - 1 - index);
+
+    // Send the control character to the subprocess to tell it to stop after
+    // we've read at least one thing from it's stdout (meaning the read was
+    // definitely async).
+    if (0 != index) {
+      fputc('s', subprocess_stdin(&process));
+      fflush(subprocess_stdin(&process));
+    }
+
+    index += bytes_read;
+  }
+
+  ASSERT_EQ(212992u, index);
+
+  for (index = 0; index < 16384; index++) {
+    const char *const helloWorld = "Hello, world!";
+    ASSERT_TRUE(0 == memcmp(data + (index * strlen(helloWorld)), helloWorld,
+                            strlen(helloWorld)));
+  }
+
+  ASSERT_EQ(0, subprocess_join(&process, &ret));
+  ASSERT_EQ(0, subprocess_destroy(&process));
+  ASSERT_EQ(ret, 0);
+}
+
+UTEST(subprocess, poll_stdout_async_wait_first) {
+  const char *const commandLine[] = {"./process_stdout_poll_wait_first",
+                                     "16384", 0};
+  struct subprocess_s process;
+  int ret = -1;
+  static char data[1048576 + 1] = {0};
+  unsigned index = 0;
+  unsigned bytes_read = 0;
+
+  ASSERT_EQ(0, subprocess_create(commandLine, subprocess_option_enable_async,
+                                 &process));
+
+  bytes_read =
+      subprocess_read_stdout(&process, data + index, sizeof(data) - 1 - index);
+
+  // We first have zero bytes read because we haven't wrote the control
+  // character to stdin.
+  ASSERT_EQ(0u, bytes_read);
+
+  fputc('s', subprocess_stdin(&process));
+  fflush(subprocess_stdin(&process));
+
+  while (subprocess_alive(&process)) {
     bytes_read = subprocess_read_stdout(&process, data + index,
                                         sizeof(data) - 1 - index);
 
@@ -674,7 +725,7 @@ UTEST(subprocess, poll_stdout_async) {
     }
 
     index += bytes_read;
-  } while (bytes_read != 0);
+  }
 
   ASSERT_EQ(212992u, index);
 
@@ -700,7 +751,7 @@ UTEST(subprocess, poll_stderr_async) {
   ASSERT_EQ(0, subprocess_create(commandLine, subprocess_option_enable_async,
                                  &process));
 
-  do {
+  while (subprocess_alive(&process)) {
     bytes_read = subprocess_read_stderr(&process, data + index,
                                         sizeof(data) - 1 - index);
 
@@ -713,7 +764,57 @@ UTEST(subprocess, poll_stderr_async) {
     }
 
     index += bytes_read;
-  } while (bytes_read != 0);
+  }
+
+  ASSERT_EQ(212992u, index);
+
+  for (index = 0; index < 16384; index++) {
+    const char *const helloWorld = "Hello, world!";
+    ASSERT_TRUE(0 == memcmp(data + (index * strlen(helloWorld)), helloWorld,
+                            strlen(helloWorld)));
+  }
+
+  ASSERT_EQ(0, subprocess_join(&process, &ret));
+  ASSERT_EQ(0, subprocess_destroy(&process));
+  ASSERT_EQ(ret, 0);
+}
+
+UTEST(subprocess, poll_stderr_async_wait_first) {
+  const char *const commandLine[] = {"./process_stderr_poll_wait_first",
+                                     "16384", 0};
+  struct subprocess_s process;
+  int ret = -1;
+  static char data[1048576 + 1] = {0};
+  unsigned index = 0;
+  unsigned bytes_read = 0;
+
+  ASSERT_EQ(0, subprocess_create(commandLine, subprocess_option_enable_async,
+                                 &process));
+
+  bytes_read =
+      subprocess_read_stderr(&process, data + index, sizeof(data) - 1 - index);
+
+  // We first have zero bytes read because we haven't wrote the control
+  // character to stdin.
+  ASSERT_EQ(0u, bytes_read);
+
+  fputc('s', subprocess_stdin(&process));
+  fflush(subprocess_stdin(&process));
+
+  while (subprocess_alive(&process)) {
+    bytes_read = subprocess_read_stderr(&process, data + index,
+                                        sizeof(data) - 1 - index);
+
+    // Send the control character to the subprocess to tell it to stop after
+    // we've read at least one thing from it's stderr (meaning the read was
+    // definitely async).
+    if (index == 0) {
+      fputc('s', subprocess_stdin(&process));
+      fflush(subprocess_stdin(&process));
+    }
+
+    index += bytes_read;
+  }
 
   ASSERT_EQ(212992u, index);
 

test/process_call_return_argc.c

@@ -29,7 +29,7 @@
 #include <direct.h>
 #define chdir(x) _chdir(x)
 #elif defined(__MINGW32__)
-#include <unistd.h>  // chdir
+#include <unistd.h> // chdir
 #endif
 
 int main(int argc, const char *const argv[]) {

test/process_return_lpcmdline.c

@@ -39,7 +39,7 @@ int main(int argc, const char *const argv[]) {
       printf("\"%s\"", argv[i]);
     else
       printf("%s", argv[i]);
-    if (i != (argc-1))
+    if (i != (argc - 1))
       printf(" ");
   }
 #endif

test/process_stderr_poll_wait_first.c

@@ -0,0 +1,44 @@
+// This is free and unencumbered software released into the public domain.
+//
+// Anyone is free to copy, modify, publish, use, compile, sell, or
+// distribute this software, either in source code form or as a compiled
+// binary, for any purpose, commercial or non-commercial, and by any
+// means.
+//
+// In jurisdictions that recognize copyright laws, the author or authors
+// of this software dedicate any and all copyright interest in the
+// software to the public domain. We make this dedication for the benefit
+// of the public at large and to the detriment of our heirs and
+// successors. We intend this dedication to be an overt act of
+// relinquishment in perpetuity of all present and future rights to this
+// software under copyright law.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+// IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+// OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+// ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+// OTHER DEALINGS IN THE SOFTWARE.
+//
+// For more information, please refer to <http://unlicense.org/>
+
+#include <stdio.h>
+#include <stdlib.h>
+
+int main(int argc, const char *const argv[]) {
+  int index;
+  const int max = atoi(argv[1]);
+
+  // We first wait for a signal on stdin to begin processing.
+  while (fgetc(stdin) != 's') {
+  }
+
+  do {
+    for (index = 0; index < max; index++) {
+      fprintf(stderr, "Hello, world!");
+    }
+  } while (fgetc(stdin) != 's');
+
+  return 0;
+}