[FIXED JENKINS-7871] Fixed a race condition in FilePath.copyTo

that results in missing tail bits when the master is heavily loaded.
jenkinsci · Mar 14, 2011 · c3fc099 · c3fc099
1 parent cbe2677
commit c3fc099
Show file tree

Hide file tree

Showing 6 changed files with 128 additions and 9 deletions.
diff --git a/changelog.html b/changelog.html
@@ -68,6 +68,9 @@
 <div id="rc" style="display:none;"><!--=BEGIN=-->
 <h3><a name=v1.402>What's new in 1.402</a> <!--=DATE=--></h3>
 <ul class=image>
+  <li class='major bug'>
+    Fixed a race condition in the remote data transfer that results in silent file copy failures.
+    (<a href="http://issues.jenkins-ci.org/browse/JENKINS-7871">issue 7871</a>)
   <li class=bug>
    Maven Plugin : Successful build ends with NPE 
     (<a href="http://issues.jenkins-ci.org/browse/JENKINS-8436">issue 8436</a>)

diff --git a/core/src/main/java/hudson/FilePath.java b/core/src/main/java/hudson/FilePath.java
@@ -1405,6 +1405,10 @@ public Void invoke(File f, VirtualChannel channel) throws IOException {
                 }
             }
         });
+
+        // make sure the write fully happens before we return.
+        if (channel!=null)
+            channel.syncLocalIO();
     }
 
     /**

diff --git a/core/src/test/java/hudson/FilePathTest.java b/core/src/test/java/hudson/FilePathTest.java
@@ -23,12 +23,24 @@
  */
 package hudson;
 
+import hudson.remoting.Future;
 import hudson.remoting.VirtualChannel;
+import hudson.util.IOException2;
 import hudson.util.NullStream;
 
 import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
+import org.apache.commons.io.output.CountingOutputStream;
 import org.apache.commons.io.output.NullOutputStream;
+import org.jvnet.hudson.test.Bug;
 
 /**
  * @author Kohsuke Kawaguchi
@@ -61,6 +73,89 @@ public void testCopyTo2() throws Exception {
         }
     }
 
+    /**
+     * As we moved the I/O handling to another thread, there's a race condition in
+     * {@link FilePath#copyTo(OutputStream)} &mdash; this method can return before
+     * all the writes are delivered to {@link OutputStream}.
+     *
+     * <p>
+     * To reproduce that problem, we use a large number of threads, so that we can
+     * maximize the chance of out-of-order execution, and make sure we are
+     * seeing the right byte count at the end.
+     *
+     * Also see JENKINS-7897
+     */
+    @Bug(7871)
+    public void testCopyTo3() throws Exception {
+        final File tmp = File.createTempFile("testCopyTo3","");
+
+        FileOutputStream os = new FileOutputStream(tmp);
+        final int size = 90000;
+        byte[] buf = new byte[size];
+        for (int i=0; i<buf.length; i++)
+            buf[i] = (byte)(i%256);
+        os.write(buf);
+        os.close();
+
+        ExecutorService es = Executors.newFixedThreadPool(100);
+        try {
+            List<java.util.concurrent.Future> r = new ArrayList<java.util.concurrent.Future>();
+            for (int i=0; i<100; i++) {
+                r.add(es.submit(new Callable<Object>() {
+                    public Object call() throws Exception {
+                        class Sink extends OutputStream {
+                            private Exception closed;
+                            private volatile int count;
+
+                            private void checkNotClosed() throws IOException2 {
+                                if (closed != null)
+                                    throw new IOException2(closed);
+                            }
+
+                            @Override
+                            public void write(int b) throws IOException {
+                                count++;
+                                checkNotClosed();
+                            }
+
+                            @Override
+                            public void write(byte[] b) throws IOException {
+                                count+=b.length;
+                                checkNotClosed();
+                            }
+
+                            @Override
+                            public void write(byte[] b, int off, int len) throws IOException {
+                                count+=len;
+                                checkNotClosed();
+                            }
+
+                            @Override
+                            public void close() throws IOException {
+                                closed = new Exception();
+                                if (size!=count)
+                                    fail();
+                            }
+                        }
+
+                        FilePath f = new FilePath(french, tmp.getPath());
+                        Sink sink = new Sink();
+                        f.copyTo(sink);
+                        assertEquals(size,sink.count);
+                        return null;
+                    }
+                }));
+            }
+
+            for (java.util.concurrent.Future f : r)
+                f.get();
+        } finally {
+            es.shutdown();
+        }
+    }
+
+
+
     public void testRepeatCopyRecursiveTo() throws Exception {
         // local->local copy used to return 0 if all files were "up to date"
         // should return number of files processed, whether or not they were copied or already current

diff --git a/remoting/src/main/java/hudson/remoting/Channel.java b/remoting/src/main/java/hudson/remoting/Channel.java
@@ -919,17 +919,22 @@ public void syncIO() throws IOException, InterruptedException {
         call(new IOSyncer());
     }
 
+    public void syncLocalIO() throws InterruptedException {
+        try {
+            pipeWriter.submit(new Runnable() {
+                public void run() {
+                    // noop
+                }
+            }).get();
+        } catch (ExecutionException e) {
+            throw new AssertionError(e); // impossible
+        }
+    }
+
     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
-            }
+            Channel.current().syncLocalIO();
+            return null;
         }
 
         private static final long serialVersionUID = 1L;

diff --git a/remoting/src/main/java/hudson/remoting/LocalChannel.java b/remoting/src/main/java/hudson/remoting/LocalChannel.java
@@ -98,4 +98,8 @@ public void join(long timeout) throws InterruptedException {
     public <T> T export(Class<T> intf, T instance) {
         return instance;
     }
+
+    public void syncLocalIO() throws InterruptedException {
+        // noop
+    }
 }
diff --git a/remoting/src/main/java/hudson/remoting/VirtualChannel.java b/remoting/src/main/java/hudson/remoting/VirtualChannel.java
@@ -103,4 +103,12 @@ public interface VirtualChannel {
      *      will invoke the same method on the given local <tt>instance</tt> object.
      */
     <T> T export( Class<T> type, T instance);
+
+    /**
+     * Blocks until all the I/O packets sent from remote is fully locally executed, then return.
+     *
+     * @since 1.402
+     */
+    void syncLocalIO() throws InterruptedException;
 }
+