[FIXED JENKINS-7809] expose child process stdout as InputStream.

I added a mode where one can launch a child process without the built-in pumping support, making it trivial to read until EOF in the distributed environments.

 This concludes the fix.
This commit is contained in:
Kohsuke Kawaguchi committed 2011-02-16 08:28:00 -08:00
1 parent e8971a5723
commit 96dd84b5d0
8 files changed
+395 -45

No files matched your search

+117 -14
View File
@@ -1,7 +1,7 @@
/*
* The MIT License
*
* Copyright (c) 2004-2009, Sun Microsystems, Inc., Kohsuke Kawaguchi
* Copyright (c) 2004-2009, Sun Microsystems, Inc., Kohsuke Kawaguchi, CloudBees, Inc.
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
@@ -23,11 +23,14 @@
*/
package hudson;
import hudson.Launcher.ProcStarter;
import hudson.model.TaskListener;
import hudson.remoting.Channel;
import hudson.util.IOException2;
import hudson.util.NullStream;
import hudson.util.StreamCopyThread;
import hudson.util.ProcessTree;
import org.apache.commons.io.input.NullInputStream;
import java.io.File;
import java.io.IOException;
@@ -71,7 +74,10 @@ public abstract class Proc {
public abstract void kill() throws IOException, InterruptedException;
/**
* Waits for the completion of the process and until we finish reading everything that the process has produced
* Waits for the completion of the process.
*
* Unless the caller opts to pump the streams via {@link #getStdout()} etc.,
* this method also blocks until we finish reading everything that the process has produced
* to stdout/stderr.
*
* <p>
@@ -85,6 +91,45 @@ public abstract class Proc {
*/
public abstract int join() throws IOException, InterruptedException;
/**
* Returns an {@link InputStream} to read from {@code stdout} of the child process.
* <p>
* When this method returns null, {@link Proc} will internally pump the output from
* the child process to your {@link OutputStream} of choosing.
*
* @return
* null unless {@link ProcStarter#readStdout()} is used to indicate
* that the caller intends to pump the stream by itself.
* @since 1.399
*/
public abstract InputStream getStdout();
/**
* Returns an {@link InputStream} to read from {@code stderr} of the child process.
* <p>
* When this method returns null, {@link Proc} will internally pump the output from
* the child process to your {@link OutputStream} of choosing.
*
* @return
* null unless {@link ProcStarter#readStderr()} is used to indicate
* that the caller intends to pump the stream by itself.
* @since 1.399
*/
public abstract InputStream getStderr();
/**
* Returns an {@link OutputStream} to write to {@code stdin} of the child process.
* <p>
* When this method returns null, {@link Proc} will internally pump the {@link InputStream}
* of your choosing to the child process.
*
* @return
* null unless {@link ProcStarter#writeStdin()} is used to indicate
* that the caller intends to pump the stream by itself.
* @since 1.399
*/
public abstract OutputStream getStdin();
private static final ExecutorService executor = Executors.newCachedThreadPool();
/**
* Like {@link #join} but can be given a maximum time to wait.
@@ -133,6 +178,9 @@ public abstract class Proc {
private final EnvVars cookie;
private final String name;
private final InputStream stdout,stderr;
private final OutputStream stdin;
public LocalProc(String cmd, Map<String,String> env, OutputStream out, File workDir) throws IOException {
this(cmd,Util.mapToEnv(env),out,workDir);
}
@@ -163,12 +211,12 @@ public abstract class Proc {
*/
public LocalProc(String[] cmd,String[] env,InputStream in,OutputStream out,OutputStream err,File workDir) throws IOException {
this( calcName(cmd),
stderr(environment(new ProcessBuilder(cmd),env).directory(workDir),err),
stderr(environment(new ProcessBuilder(cmd),env).directory(workDir), err==null || err== SELFPUMP_OUTPUT),
in, out, err );
}
private static ProcessBuilder stderr(ProcessBuilder pb, OutputStream stderr) {
if(stderr==null) pb.redirectErrorStream(true);
private static ProcessBuilder stderr(ProcessBuilder pb, boolean redirectError) {
if(redirectError) pb.redirectErrorStream(true);
return pb;
}
@@ -191,17 +239,39 @@ public abstract class Proc {
this.cookie = EnvVars.createCookie();
procBuilder.environment().putAll(cookie);
this.proc = procBuilder.start();
InputStream procInputStream = proc.getInputStream();
copier = new StreamCopyThread(name+": stdout copier", procInputStream, out);
copier.start();
if(in!=null)
new StdinCopyThread(name+": stdin copier",in,proc.getOutputStream()).start();
else
if (out==SELFPUMP_OUTPUT) {
stdout = procInputStream;
copier = null;
} else {
copier = new StreamCopyThread(name+": stdout copier", procInputStream, out);
copier.start();
stdout = null;
}
if (in == null) {
// nothing to feed to stdin
stdin = null;
proc.getOutputStream().close();
} else
if (in==SELFPUMP_INPUT) {
stdin = proc.getOutputStream();
} else {
new StdinCopyThread(name+": stdin copier",in,proc.getOutputStream()).start();
stdin = null;
}
InputStream procErrorStream = proc.getErrorStream();
if(err!=null) {
copier2 = new StreamCopyThread(name+": stderr copier", procErrorStream, err);
copier2.start();
if (err==SELFPUMP_OUTPUT) {
stderr = procErrorStream;
copier2 = null;
} else {
stderr = null;
copier2 = new StreamCopyThread(name+": stderr copier", procErrorStream, err);
copier2.start();
}
} else {
// the javadoc is unclear about what getErrorStream() returns when ProcessBuilder.redirectErrorStream(true),
//
@@ -212,9 +282,22 @@ public abstract class Proc {
procErrorStream.close();
}
copier2 = null;
stderr = null;
}
}
public InputStream getStdout() {
return stdout;
}
public InputStream getStderr() {
return stderr;
}
public OutputStream getStdin() {
return stdin;
}
/**
* Waits for the completion of the process.
*/
@@ -234,9 +317,9 @@ public abstract class Proc {
// see http://wiki.jenkins-ci.org/display/JENKINS/Spawning+processes+from+build
// problems like that shows up as infinite wait in join(), which confuses great many users.
// So let's do a timed wait here and try to diagnose the problem
copier.join(10*1000);
if (copier!=null) copier.join(10*1000);
if(copier2!=null) copier2.join(10*1000);
if(copier.isAlive() || (copier2!=null && copier2.isAlive())) {
if((copier!=null && copier.isAlive()) || (copier2!=null && copier2.isAlive())) {
// looks like handles are leaking.
// closing these handles should terminate the threads.
String msg = "Process leaked file descriptors. See http://wiki.jenkins-ci.org/display/JENKINS/Spawning+processes+from+build for more information";
@@ -335,10 +418,15 @@ public abstract class Proc {
}
return buf.toString();
}
public static final InputStream SELFPUMP_INPUT = new NullInputStream(0);
public static final OutputStream SELFPUMP_OUTPUT = new NullStream();
}
/**
* Remotely launched process via {@link Channel}.
*
* @deprecated as of 1.399
*/
public static final class RemoteProc extends Proc {
private final Future<Integer> process;
@@ -373,6 +461,21 @@ public abstract class Proc {
public boolean isAlive() throws IOException, InterruptedException {
return !process.isDone();
}
@Override
public InputStream getStdout() {
return null;
}
@Override
public InputStream getStderr() {
return null;
}
@Override
public OutputStream getStdin() {
return null;
}
}
private static final Logger LOGGER = Logger.getLogger(Proc.class.getName());