[FIXED JENKINS-7809] fixed a race condition in obtaining the tail of the...

[FIXED JENKINS-7809] fixed a race condition in obtaining the tail of the output from remote process.

changelog.html

@@ -61,6 +61,9 @@ Upcoming changes</a>
   <li class='major bug'>
     On IBM JDKs, Jenkins incorrectly ended up closing stdout to read from forked processes.
     (<a href="http://issues.jenkins-ci.org/browse/JENKINS-8420">issue 8420</a>)
+  <li class=bug>
+    Fixed a race condition in obtaining the tail of the output from remote process.
+    (<a href="http://issues.jenkins-ci.org/browse/JENKINS-7809">issue 7809</a>)
   <li class=bug>
     Jenkins was unable to kill/list up native processses on 64bit Mac JVMs.
   <li class=bug>

core/src/main/java/hudson/Launcher.java

@@ -795,6 +795,13 @@ public abstract class Launcher {
                 return p.join();
             } catch (InterruptedException e) {
                 return -1;
+            } finally {
+                // make sure I/O is delivered to the remote before we return
+                try {
+                    Channel.current().syncIO();
+                } catch (Throwable _) {
+                    // this includes a failure to sync, slave.jar too old, etc
+                }
             }
         }
 

core/src/main/java/hudson/Proc.java

@@ -24,6 +24,7 @@
 package hudson;
 
 import hudson.model.TaskListener;
+import hudson.remoting.Channel;
 import hudson.util.IOException2;
 import hudson.util.StreamCopyThread;
 import hudson.util.ProcessTree;
@@ -337,7 +338,7 @@ public abstract class Proc {
     }
 
     /**
-     * Retemoly launched process via {@link Channel}.
+     * Remotely launched process via {@link Channel}.
      */
     public static final class RemoteProc extends Proc {
         private final Future<Integer> process;

remoting/src/main/java/hudson/remoting/Channel.java

@@ -908,6 +908,32 @@ public class Channel implements VirtualChannel, IChannel {
                 ForwarderFactory.create(forwardHost, forwardPort));
     }
 
+    /**
+     * Blocks until all the I/O packets sent before this gets fully executed by the remote side, then return.
+     *
+     * @throws IOException
+     *      If the remote doesn't support this operation, or if sync fails for other reasons.
+     */
+    public void syncIO() throws IOException, InterruptedException {
+        call(new IOSyncer());
+    }
+
+    private static final class IOSyncer implements Callable<Object, InterruptedException> {
+        public Object call() throws InterruptedException {
+            try {
+                return Channel.current().pipeWriter.submit(new Runnable() {
+                    public void run() {
+                        // noop
+                    }
+                }).get();
+            } catch (ExecutionException e) {
+                throw new AssertionError(e); // impossible
+            }
+        }
+
+        private static final long serialVersionUID = 1L;
+    }
+
     @Override
     public String toString() {
         return super.toString()+":"+name;

test/src/test/java/hudson/ProcTest.java

+package hudson;
+
+import hudson.Launcher.RemoteLauncher;
+import hudson.Proc.RemoteProc;
+import hudson.remoting.Callable;
+import hudson.remoting.Future;
+import hudson.remoting.Pipe;
+import hudson.remoting.VirtualChannel;
+import hudson.slaves.DumbSlave;
+import hudson.util.IOUtils;
+import hudson.util.StreamTaskListener;
+import org.jvnet.hudson.test.Bug;
+import org.jvnet.hudson.test.HudsonTestCase;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.Charset;
+
+/**
+ * @author Kohsuke Kawaguchi
+ */
+public class ProcTest extends HudsonTestCase {
+    /**
+     * Makes sure that the output flushing and {@link RemoteProc#join()} is synced.
+     */
+    @Bug(7809)
+    public void testRemoteProcOutputSync() throws Exception {
+        DumbSlave s = createSlave();
+        s.toComputer().connect(false).get();
+        VirtualChannel ch=null;
+        while (ch==null) {
+            ch = s.toComputer().getChannel();
+            Thread.sleep(100);
+        }
+
+        // keep the pipe fairly busy
+        final Pipe p = Pipe.createRemoteToLocal();
+        for (int i=0; i<10; i++)
+            ch.callAsync(new ChannelFiller(p.getOut()));
+        new Thread() {
+            @Override
+            public void run() {
+                try {
+                    IOUtils.drain(p.getIn());
+                } catch (IOException e) {
+                }
+            }
+        }.start();
+
+        RemoteLauncher launcher = new RemoteLauncher(StreamTaskListener.NULL, ch, true);
+
+        String str="";
+        for (int i=0; i<256; i++)
+            str += "oxox";
+
+        for (int i=0; i<1000; i++) {
+            ByteArrayOutputStream baos = new ByteArrayOutputStream();
+            launcher.launch().cmds("echo",str).stdout(baos).join();
+            assertEquals(str, baos.toString().trim());
+        }
+
+        ch.close();
+    }
+
+    private static class ChannelFiller implements Callable<Void,IOException> {
+        private final OutputStream o;
+
+        private ChannelFiller(OutputStream o) {
+            this.o = o;
+        }
+
+        public Void call() throws IOException {
+            while (!Thread.interrupted()) {
+                o.write(new byte[256]);
+            }
+            return null;
+        }
+    }
+}