56  * @author Mario Wolczko and Ross Knippel.

    57  * @author Konstantin Kladko (ported to Linux and Bsd)

    59  * @author Volker Simonis (ported to AIX)

   279 

   280 

   281     /**

   282      * Creates a process. Depending on the {@code mode} flag, this is done by

   283      * one of the following mechanisms:

   284      * <pre>

   285      *   1 - fork(2) and exec(2)

   286      *   2 - posix_spawn(3P)

   287      *   3 - vfork(2) and exec(2)

   288      *

   289      *  (4 - clone(2) and exec(2) - obsolete and currently disabled in native code)

   290      * </pre>

   291      * @param fds an array of three file descriptors.

   292      *        Indexes 0, 1, and 2 correspond to standard input,

   293      *        standard output and standard error, respectively.  On

   294      *        input, a value of -1 means to create a pipe to connect

   295      *        child and parent processes.  On output, a value which

   296      *        is not -1 is the parent pipe fd corresponding to the

   297      *        pipe which has been created.  An element of this array

   298      *        is -1 on input if and only if it is <em>not</em> -1 on

   299      *        output.

   300      * @return the pid of the subprocess

   301      */

   302     private native int forkAndExec(int mode, byte[] helperpath,

   303                                    byte[] prog,

   304                                    byte[] argBlock, int argc,

   305                                    byte[] envBlock, int envc,

   306                                    byte[] dir,

   307                                    int[] fds,

   308                                    boolean redirectErrorStream)

   309             throws IOException;

   310 

   311     /**

   312      * The thread pool of "process reaper" daemon threads.

   313      */

   314     private static final Executor processReaperExecutor =

   315             doPrivileged((PrivilegedAction<Executor>) () -> {

   316 

   317                 ThreadGroup tg = Thread.currentThread().getThreadGroup();

   318                 while (tg.getParent() != null) tg = tg.getParent();

   319                 ThreadGroup systemThreadGroup = tg;

   320 

   321                 ThreadFactory threadFactory = grimReaper -> {

   322                     // Our thread stack requirement is quite modest.

   323                     Thread t = new Thread(systemThreadGroup, grimReaper,

   324                             "process reaper", 32768);

   325                     t.setDaemon(true);

   326                     // A small attempt (probably futile) to avoid priority inversion

   327                     t.setPriority(Thread.MAX_PRIORITY);

   328                     return t;

   329                 };

   330 

   331                 return Executors.newCachedThreadPool(threadFactory);

   332             });

   333 

   334     private ProcessImpl(final byte[] prog,

   335                 final byte[] argBlock, final int argc,

   336                 final byte[] envBlock, final int envc,

   337                 final byte[] dir,

   338                 final int[] fds,

   339                 final boolean redirectErrorStream)

   340             throws IOException {

   341 

   342         pid = forkAndExec(launchMechanism.ordinal() + 1,

   343                 helperpath,

   344                 prog,

   345                 argBlock, argc,

   346                 envBlock, envc,

   347                 dir,

   348                 fds,

   349                 redirectErrorStream);

   350 

   351         try {

   352             doPrivileged((PrivilegedExceptionAction<Void>) () -> {

   353                 initStreams(fds);

   354                 return null;

   355             });

   356         } catch (PrivilegedActionException ex) {

   357             throw (IOException) ex.getException();

   358         }

   359     }

   360 

   361     static FileDescriptor newFileDescriptor(int fd) {

   362         FileDescriptor fileDescriptor = new FileDescriptor();

   363         fdAccess.set(fileDescriptor, fd);

   364         return fileDescriptor;

   365     }

   366 

   367     void initStreams(int[] fds) throws IOException {

   368         switch (platform) {

   369             case LINUX:

   370             case BSD:

   371                 stdin = (fds[0] == -1) ?

   372                         ProcessBuilder.NullOutputStream.INSTANCE :

   373                         new ProcessPipeOutputStream(fds[0]);

   374 

   375                 stdout = (fds[1] == -1) ?

   376                         ProcessBuilder.NullInputStream.INSTANCE :

   377                         new ProcessPipeInputStream(fds[1]);

   378 

   379                 stderr = (fds[2] == -1) ?

   380                         ProcessBuilder.NullInputStream.INSTANCE :

   381                         new ProcessPipeInputStream(fds[2]);

   382 

   383                 processReaperExecutor.execute(() -> {

   384                     int exitcode = waitForProcessExit(pid);

   385 

   386                     synchronized (this) {

   387                         this.exitcode = exitcode;

   388                         this.hasExited = true;

   389                         this.notifyAll();

   390                     }

   391 

   392                     if (stdout instanceof ProcessPipeInputStream)

   393                         ((ProcessPipeInputStream) stdout).processExited();

   394 

   395                     if (stderr instanceof ProcessPipeInputStream)

   396                         ((ProcessPipeInputStream) stderr).processExited();

   397 

   398                     if (stdin instanceof ProcessPipeOutputStream)

   399                         ((ProcessPipeOutputStream) stdin).processExited();

   400                 });

   401                 break;

   402 

   403             case SOLARIS:

   404                 stdin = (fds[0] == -1) ?

   405                         ProcessBuilder.NullOutputStream.INSTANCE :

   406                         new BufferedOutputStream(

   407                                 new FileOutputStream(newFileDescriptor(fds[0])));

   408 

   409                 stdout = (fds[1] == -1) ?

   410                         ProcessBuilder.NullInputStream.INSTANCE :

   411                         new BufferedInputStream(

   412                                 stdout_inner_stream =

   413                                         new DeferredCloseInputStream(

   414                                                 newFileDescriptor(fds[1])));

   415 

   416                 stderr = (fds[2] == -1) ?

   417                         ProcessBuilder.NullInputStream.INSTANCE :

   418                         new DeferredCloseInputStream(newFileDescriptor(fds[2]));

   419 

   420                 /*

   421                  * For each subprocess forked a corresponding reaper task

   422                  * is submitted.  That task is the only thread which waits

   423                  * for the subprocess to terminate and it doesn't hold any

   424                  * locks while doing so.  This design allows waitFor() and

   425                  * exitStatus() to be safely executed in parallel (and they

   426                  * need no native code).

   427                  */

   428                 processReaperExecutor.execute(() -> {

   429                     int exitcode = waitForProcessExit(pid);

   430 

   431                     synchronized (this) {

   432                         this.exitcode = exitcode;

   433                         this.hasExited = true;

   434                         this.notifyAll();

   435                     }

   436                 });

   437                 break;

   438 

   439             case AIX:

   440                 stdin = (fds[0] == -1) ?

   441                         ProcessBuilder.NullOutputStream.INSTANCE :

   442                         new ProcessPipeOutputStream(fds[0]);

   443 

   444                 stdout = (fds[1] == -1) ?

   445                         ProcessBuilder.NullInputStream.INSTANCE :

   446                         new DeferredCloseProcessPipeInputStream(fds[1]);

   447 

   448                 stderr = (fds[2] == -1) ?

   449                         ProcessBuilder.NullInputStream.INSTANCE :

   450                         new DeferredCloseProcessPipeInputStream(fds[2]);

   451 

   452                 processReaperExecutor.execute(() -> {

   453                     int exitcode = waitForProcessExit(pid);

   454 

   455                     synchronized (this) {

   456                         this.exitcode = exitcode;

   457                         this.hasExited = true;

   458                         this.notifyAll();

   459                     }

   460 

   461                     if (stdout instanceof DeferredCloseProcessPipeInputStream)

   462                         ((DeferredCloseProcessPipeInputStream) stdout).processExited();

   463 

   464                     if (stderr instanceof DeferredCloseProcessPipeInputStream)

   465                         ((DeferredCloseProcessPipeInputStream) stderr).processExited();

   466 

   467                     if (stdin instanceof ProcessPipeOutputStream)

   468                         ((ProcessPipeOutputStream) stdin).processExited();

   469                 });

   470                 break;

   471 

   472             default: throw new AssertionError("Unsupported platform: " + platform);

   473         }

   474     }

   475 

   476     public OutputStream getOutputStream() {

   477         return stdin;

   478     }

   479 

   480     public InputStream getInputStream() {

   481         return stdout;

   482     }

   483 

   484     public InputStream getErrorStream() {

   485         return stderr;

   486     }

   487 

   488     public synchronized int waitFor() throws InterruptedException {

   489         while (!hasExited) {

   490             wait();

   491         }

   492         return exitcode;

   493     }

   494 

   495     @Override

   496     public synchronized boolean waitFor(long timeout, TimeUnit unit)

   497             throws InterruptedException

   498     {

   499         if (hasExited) return true;

   500         if (timeout <= 0) return false;

   501 

   502         long remainingNanos = unit.toNanos(timeout);

   503         long deadline = System.nanoTime() + remainingNanos;

   504 

   505         do {

   506             // Round up to next millisecond

   507             wait(TimeUnit.NANOSECONDS.toMillis(remainingNanos + 999_999L));

   508             if (hasExited) {

   509                 return true;

   510             }

   511             remainingNanos = deadline - System.nanoTime();

   512         } while (remainingNanos > 0);

   513         return hasExited;

   514     }

   515 

   516     public synchronized int exitValue() {

   517         if (!hasExited) {

   518             throw new IllegalThreadStateException("process hasn't exited");

   519         }

   520         return exitcode;

   521     }

   522 

   523     private static native void destroyProcess(int pid, boolean force);

   524 

   525     private void destroy(boolean force) {

   526         switch (platform) {

   527             case LINUX:

   528             case BSD:

   529             case AIX:

   530                 // There is a risk that pid will be recycled, causing us to

   531                 // kill the wrong process!  So we only terminate processes

   532                 // that appear to still be running.  Even with this check,

   533                 // there is an unavoidable race condition here, but the window

   534                 // is very small, and OSes try hard to not recycle pids too

   535                 // soon, so this is quite safe.

   536                 synchronized (this) {

   537                     if (!hasExited)

   538                         destroyProcess(pid, force);

   539                 }

   540                 try { stdin.close();  } catch (IOException ignored) {}

   541                 try { stdout.close(); } catch (IOException ignored) {}

   542                 try { stderr.close(); } catch (IOException ignored) {}

   543                 break;

   544 

   545             case SOLARIS:

   546                 // There is a risk that pid will be recycled, causing us to

   547                 // kill the wrong process!  So we only terminate processes

   548                 // that appear to still be running.  Even with this check,

   549                 // there is an unavoidable race condition here, but the window

   550                 // is very small, and OSes try hard to not recycle pids too

   551                 // soon, so this is quite safe.

   552                 synchronized (this) {

   553                     if (!hasExited)

   554                         destroyProcess(pid, force);

   555                     try {

   556                         stdin.close();

   557                         if (stdout_inner_stream != null)

   558                             stdout_inner_stream.closeDeferred(stdout);

   559                         if (stderr instanceof DeferredCloseInputStream)

   560                             ((DeferredCloseInputStream) stderr)

   561                                     .closeDeferred(stderr);

   562                     } catch (IOException e) {

   563                         // ignore

   564                     }

   565                 }

   566                 break;

   567 

   568             default: throw new AssertionError("Unsupported platform: " + platform);

   569         }

   570     }

   571 

   572     public void destroy() {

   573         destroy(false);

   574     }

   575 

   576     @Override

   577     public Process destroyForcibly() {

   578         destroy(true);

   579         return this;

   580     }

   581 

   582     @Override

   583     public long getPid() {

   584         return pid;

   585     }

   586 

   587     @Override

   588     public synchronized boolean isAlive() {

   589         return !hasExited;

   590     }

   591 

   592     private static native void init();

   593 

   594     static {

   595         init();

   596     }

   597 

   598     /**

   599      * A buffered input stream for a subprocess pipe file descriptor

   600      * that allows the underlying file descriptor to be reclaimed when

   601      * the process exits, via the processExited hook.

   602      *

   603      * This is tricky because we do not want the user-level InputStream to be

   604      * closed until the user invokes close(), and we need to continue to be

   605      * able to read any buffered data lingering in the OS pipe buffer.

   606      */

   607     private static class ProcessPipeInputStream extends BufferedInputStream {

   608         private final Object closeLock = new Object();

   609 

   610         ProcessPipeInputStream(int fd) {

   611             super(new FileInputStream(newFileDescriptor(fd)));

   612         }

   613         private static byte[] drainInputStream(InputStream in)

   614                 throws IOException {

   615             int n = 0;

   616             int j;

   617             byte[] a = null;

   618             while ((j = in.available()) > 0) {

   619                 a = (a == null) ? new byte[j] : Arrays.copyOf(a, n + j);

   620                 n += in.read(a, n, j);

   621             }

   622             return (a == null || n == a.length) ? a : Arrays.copyOf(a, n);

   623         }

   624 

   625         /** Called by the process reaper thread when the process exits. */

   626         synchronized void processExited() {

   627             synchronized (closeLock) {

   628                 try {

   629                     InputStream in = this.in;

   630                     // this stream is closed if and only if: in == null

   631                     if (in != null) {

   632                         byte[] stragglers = drainInputStream(in);

   633                         in.close();

   634                         this.in = (stragglers == null) ?

   635                                 ProcessBuilder.NullInputStream.INSTANCE :

   636                                 new ByteArrayInputStream(stragglers);

   637                     }

   638                 } catch (IOException ignored) {}

   639             }

   640         }

   641 

   642         @Override

   643         public void close() throws IOException {

   644             // BufferedInputStream#close() is not synchronized unlike most other

   645             // methods. Synchronizing helps avoid race with processExited().

   646             synchronized (closeLock) {

   647                 super.close();

   648             }

   649         }

   650     }

   651 

   652     /**

   653      * A buffered output stream for a subprocess pipe file descriptor

   654      * that allows the underlying file descriptor to be reclaimed when

   655      * the process exits, via the processExited hook.

   656      */

   657     private static class ProcessPipeOutputStream extends BufferedOutputStream {

   658         ProcessPipeOutputStream(int fd) {

   659             super(new FileOutputStream(newFileDescriptor(fd)));

   660         }

   661 

   662         /** Called by the process reaper thread when the process exits. */

   663         synchronized void processExited() {

   664             OutputStream out = this.out;

   665             if (out != null) {

   666                 try {

   667                     out.close();

   668                 } catch (IOException ignored) {

   669                     // We know of no reason to get an IOException, but if

   670                     // we do, there's nothing else to do but carry on.

   671                 }

   672                 this.out = ProcessBuilder.NullOutputStream.INSTANCE;

   673             }

   674         }

   675     }

   676 

   677     // A FileInputStream that supports the deferment of the actual close

   678     // operation until the last pending I/O operation on the stream has

   679     // finished.  This is required on Solaris because we must close the stdin

   680     // and stdout streams in the destroy method in order to reclaim the

   681     // underlying file descriptors.  Doing so, however, causes any thread

   682     // currently blocked in a read on one of those streams to receive an

   683     // IOException("Bad file number"), which is incompatible with historical

   684     // behavior.  By deferring the close we allow any pending reads to see -1

   685     // (EOF) as they did before.

   686     //

   687     private static class DeferredCloseInputStream extends FileInputStream

   688     {

   689         DeferredCloseInputStream(FileDescriptor fd) {

   690             super(fd);

   691         }

   692 

   693         private Object lock = new Object();     // For the following fields

   694         private boolean closePending = false;

   695         private int useCount = 0;

   696         private InputStream streamToClose;

   697 

   698         private void raise() {

   699             synchronized (lock) {

   700                 useCount++;

   701             }

   702         }

   703 

   704         private void lower() throws IOException {

   705             synchronized (lock) {

   706                 useCount--;

   707                 if (useCount == 0 && closePending) {

   708                     streamToClose.close();

   709                 }

   710             }

   711         }

   712 

   713         // stc is the actual stream to be closed; it might be this object, or

   714         // it might be an upstream object for which this object is downstream.

   715         //

   716         private void closeDeferred(InputStream stc) throws IOException {

   717             synchronized (lock) {

   718                 if (useCount == 0) {

   719                     stc.close();

   720                 } else {

   721                     closePending = true;

   722                     streamToClose = stc;

   723                 }

   724             }

   725         }

   726 

   727         public void close() throws IOException {

   728             synchronized (lock) {

   729                 useCount = 0;

   730                 closePending = false;

   731             }

   732             super.close();

   733         }

   734 

   735         public int read() throws IOException {

   736             raise();

   737             try {

   738                 return super.read();

   739             } finally {

   740                 lower();

   741             }

   742         }

   743 

   744         public int read(byte[] b) throws IOException {

   745             raise();

   746             try {

   747                 return super.read(b);

   748             } finally {

   749                 lower();

   750             }

   751         }

   752 

   753         public int read(byte[] b, int off, int len) throws IOException {

   754             raise();

   755             try {

   756                 return super.read(b, off, len);

   757             } finally {

   758                 lower();

   759             }

   760         }

   761 

   762         public long skip(long n) throws IOException {

   763             raise();

   764             try {

   765                 return super.skip(n);

   766             } finally {

   767                 lower();

   768             }

   769         }

   770 

   771         public int available() throws IOException {

   772             raise();

   773             try {

   774                 return super.available();

   775             } finally {

   776                 lower();

   777             }

   778         }

   779     }

   780 

   781     /**

   782      * A buffered input stream for a subprocess pipe file descriptor

   783      * that allows the underlying file descriptor to be reclaimed when

   784      * the process exits, via the processExited hook.

   785      *

   786      * This is tricky because we do not want the user-level InputStream to be

   787      * closed until the user invokes close(), and we need to continue to be

   788      * able to read any buffered data lingering in the OS pipe buffer.

   789      *

   790      * On AIX this is especially tricky, because the 'close()' system call

   791      * will block if another thread is at the same time blocked in a file

   792      * operation (e.g. 'read()') on the same file descriptor. We therefore

   793      * combine 'ProcessPipeInputStream' approach used on Linux and Bsd

   794      * with the DeferredCloseInputStream approach used on Solaris. This means

   795      * that every potentially blocking operation on the file descriptor

   796      * increments a counter before it is executed and decrements it once it

   797      * finishes. The 'close()' operation will only be executed if there are

   798      * no pending operations. Otherwise it is deferred after the last pending

   799      * operation has finished.

   800      *

   801      */

   802     private static class DeferredCloseProcessPipeInputStream

   803             extends BufferedInputStream {

   804 

   805         private final Object closeLock = new Object();

   806         private int useCount = 0;

   807         private boolean closePending = false;

   808 

   809         DeferredCloseProcessPipeInputStream(int fd) {

   810             super(new FileInputStream(newFileDescriptor(fd)));

   811         }

   812 

   813         private InputStream drainInputStream(InputStream in)

   814                 throws IOException {

   815             int n = 0;

   816             int j;

   817             byte[] a = null;

   818             synchronized (closeLock) {

   819                 if (buf == null) // asynchronous close()?

   820                     return null; // discard

   821                 j = in.available();

   822             }

   823             while (j > 0) {

   824                 a = (a == null) ? new byte[j] : Arrays.copyOf(a, n + j);

   825                 synchronized (closeLock) {

   826                     if (buf == null) // asynchronous close()?

   827                         return null; // discard

   828                     n += in.read(a, n, j);

   829                     j = in.available();

   830                 }

   831             }

   832             return (a == null) ?

   833                     ProcessBuilder.NullInputStream.INSTANCE :

   834                     new ByteArrayInputStream(n == a.length ? a : Arrays.copyOf(a, n));

   835         }

   836 

   837         /** Called by the process reaper thread when the process exits. */

   838         synchronized void processExited() {

   839             try {

   840                 InputStream in = this.in;

   841                 if (in != null) {

   842                     InputStream stragglers = drainInputStream(in);

   843                     in.close();

   844                     this.in = stragglers;

   845                 }

   846             } catch (IOException ignored) { }

   847         }

   848 

   849         private void raise() {

   850             synchronized (closeLock) {

   851                 useCount++;

   852             }

   853         }

   854 

   855         private void lower() throws IOException {

   856             synchronized (closeLock) {

   857                 useCount--;

   858                 if (useCount == 0 && closePending) {

   859                     closePending = false;

   860                     super.close();

   861                 }

   862             }

   863         }

   864 

   865         @Override

   866         public int read() throws IOException {

   867             raise();

   868             try {

   869                 return super.read();

   870             } finally {

   871                 lower();

   872             }

   873         }

   874 

   875         @Override

   876         public int read(byte[] b) throws IOException {

   877             raise();

   878             try {

   879                 return super.read(b);

   880             } finally {

   881                 lower();

   882             }

   883         }

   884 

   885         @Override

   886         public int read(byte[] b, int off, int len) throws IOException {

   887             raise();

   888             try {

   889                 return super.read(b, off, len);

   890             } finally {

   891                 lower();

   892             }

   893         }

   894 

   895         @Override

   896         public long skip(long n) throws IOException {

   897             raise();

   898             try {

   899                 return super.skip(n);

   900             } finally {

   901                 lower();

   902             }

   903         }

   904 

   905         @Override

   906         public int available() throws IOException {

   907             raise();

   908             try {

   909                 return super.available();

   910             } finally {

   911                 lower();

   912             }

   913         }

   914 

   915         @Override

   916         public void close() throws IOException {

   917             // BufferedInputStream#close() is not synchronized unlike most other

   918             // methods. Synchronizing helps avoid racing with drainInputStream().

   919             synchronized (closeLock) {

   920                 if (useCount == 0) {

   921                     super.close();

   922                 }

   923                 else {

   924                     closePending = true;

   925                 }

   926             }

   927         }

   928     }