Merge pull request #158 from bturner/bturner-fix-start-exit-race-for-fast-processes

Prevent onStart/onExit race for fast-exiting processes.

Author: bturner

benchmark/pom.xml

@@ -25,7 +25,7 @@
         <dependency>
             <groupId>com.zaxxer</groupId>
             <artifactId>nuprocess</artifactId>
-            <version>2.0.8-SNAPSHOT</version>
+            <version>2.1.0-SNAPSHOT</version>
         </dependency>
 
         <dependency>

mvnvm.properties

@@ -1 +1 @@
-mvn_version=3.8.5
+mvn_version=3.9.6

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>com.zaxxer</groupId>
     <artifactId>nuprocess</artifactId>
-    <version>2.0.8-SNAPSHOT</version>
+    <version>2.1.0-SNAPSHOT</version>
     <packaging>bundle</packaging>
 
     <name>NuProcess</name>

src/main/java/com/zaxxer/nuprocess/internal/IEventProcessor.java

@@ -53,6 +53,18 @@ public interface IEventProcessor<T extends BasePosixProcess> extends Runnable
     */
    void registerProcess(T process);
 
+   /**
+    * Queues read handling for the process's STDOUT and STDERR streams.
+    * <p>
+    * Prior to 2.1, this was performed by {@link #registerProcess}, but it was moved to a separate method
+    * to help avoid a race condition when starting a new process where soft exit detection could result in
+    * a fast-running process exiting before start was called on its process handler.
+    *
+    * @param process the process from which STDOUT and STDERR should be read
+    * @since 2.1
+    */
+   void queueRead(T process);
+
    /**
     * Express that the client desires to write data into the STDIN stream as
     * soon as possible.

src/main/java/com/zaxxer/nuprocess/linux/LinuxProcess.java

@@ -75,9 +75,15 @@ public NuProcess start(List<String> command, String[] environment, Path cwd) {
 
          afterStart();
 
+         // Registration must happen prior to calling NuProcessHandler.onStart to allow handlers
+         // to call wantWrite (which calls myProcessor.queueWrite)
          registerProcess();
 
          callStart();
+
+         // Queueing read handling for stdout and stderr happens after start has been called
+         // to ensure fast-exiting processes don't call NuProcessHandler.onExit before onStart
+         myProcessor.queueRead(this);
       }
       catch (Exception e) {
          // TODO remove from event processor pid map?

src/main/java/com/zaxxer/nuprocess/linux/ProcessEpoll.java

@@ -54,6 +54,7 @@ class ProcessEpoll extends BaseEventProcessor<LinuxProcess>
       this.process = process;
 
       registerProcess(process);
+      queueRead(process);
       checkAndSetRunning();
    }
 
@@ -95,6 +96,32 @@ public void registerProcess(LinuxProcess process)
          fildesToProcessMap.put(stdinFd, process);
          fildesToProcessMap.put(stdoutFd, process);
          fildesToProcessMap.put(stderrFd, process);
+      }
+      finally {
+         if (stdinFd != Integer.MIN_VALUE) {
+            process.getStdin().release();
+         }
+         if (stdoutFd != Integer.MIN_VALUE) {
+            process.getStdout().release();
+         }
+         if (stderrFd != Integer.MIN_VALUE) {
+            process.getStderr().release();
+         }
+      }
+   }
+
+   @Override
+   public void queueRead(LinuxProcess process)
+   {
+      if (shutdown) {
+         return;
+      }
+
+      int stdoutFd = Integer.MIN_VALUE;
+      int stderrFd = Integer.MIN_VALUE;
+      try {
+         stdoutFd = process.getStdout().acquire();
+         stderrFd = process.getStderr().acquire();
 
          EpollEvent event = process.getEpollEvent();
          event.setEvents(LibEpoll.EPOLLIN);
@@ -114,9 +141,6 @@ public void registerProcess(LinuxProcess process)
          }
       }
       finally {
-         if (stdinFd != Integer.MIN_VALUE) {
-            process.getStdin().release();
-         }
          if (stdoutFd != Integer.MIN_VALUE) {
             process.getStdout().release();
          }

src/main/java/com/zaxxer/nuprocess/osx/OsxProcess.java

@@ -65,7 +65,11 @@ public NuProcess start(List<String> command, String[] environment, Path cwd) {
 
          afterStart();
 
+         // On macOS, registration can be immediately followed by queueing read handling for stdout
+         // and stderr without risking a racy exit because processes are launched suspended and are
+         // only resumed after NuProcessHandler.onStart is called
          registerProcess();
+         myProcessor.queueRead(this);
 
          callStart();
 

src/main/java/com/zaxxer/nuprocess/osx/ProcessKqueue.java

@@ -68,6 +68,7 @@ final class ProcessKqueue extends BaseEventProcessor<OsxProcess>
       this(-1);
 
       registerProcess(process);
+      queueRead(process);
       checkAndSetRunning();
    }
 
@@ -101,9 +102,18 @@ public void registerProcess(OsxProcess process)
       }
 
       int pid = process.getPid();
-      Pointer pidPointer = Pointer.createConstant(pid);
-
       pidToProcessMap.put(pid, process);
+   }
+
+   @Override
+   public void queueRead(OsxProcess process)
+   {
+      if (shutdown) {
+         return;
+      }
+
+      int pid = process.getPid();
+      Pointer pidPointer = Pointer.createConstant(pid);
 
       Integer stdinFd = null;
       Integer stdoutFd = null;
@@ -112,6 +122,7 @@ public void registerProcess(OsxProcess process)
          stdinFd = process.getStdin().acquire();
          stdoutFd = process.getStdout().acquire();
          stderrFd = process.getStderr().acquire();
+
          // We don't use the processEvents array here, since this method is not
          // called on the event processor thread.
          Kevent[] events = (Kevent[]) new Kevent().toArray(4);
@@ -125,7 +136,8 @@ public void registerProcess(OsxProcess process)
          events[3].EV_SET(stdinFd, Kevent.EVFILT_WRITE, Kevent.EV_ADD | Kevent.EV_DISABLE | Kevent.EV_RECEIPT, 0, 0L, pidPointer);
 
          registerEvents(events, 4);
-      } finally {
+      }
+      finally {
          if (stdinFd != null) {
             process.getStdin().release();
          }