Factored out GCFlushedOutputStream, and reverting 4bef56f so that DelayBufferedOutputStream is again a BufferedOutputStream.

jglick · jglick · commit b944d25027d8 · 2018-11-19T14:24:12.000-05:00
diff --git a/src/main/java/org/jenkinsci/plugins/workflow/log/BufferedBuildListener.java b/src/main/java/org/jenkinsci/plugins/workflow/log/BufferedBuildListener.java
@@ -74,7 +74,7 @@ private static final class Replacement implements SerializableOnlyOverRemoting {
         }
 
         private Object readResolve() throws IOException {
-            return new BufferedBuildListener(new DelayBufferedOutputStream(ros, tuning));
+            return new BufferedBuildListener(new GCFlushedOutputStream(new DelayBufferedOutputStream(ros, tuning)));
         }
 
     }
diff --git a/src/main/java/org/jenkinsci/plugins/workflow/log/DelayBufferedOutputStream.java b/src/main/java/org/jenkinsci/plugins/workflow/log/DelayBufferedOutputStream.java
@@ -28,9 +28,7 @@
 import java.io.FilterOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
-import java.lang.ref.PhantomReference;
 import java.lang.ref.Reference;
-import java.lang.ref.ReferenceQueue;
 import java.lang.ref.WeakReference;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
@@ -41,10 +39,8 @@
 /**
  * Buffered output stream which is guaranteed to deliver content after some time even if idle and the buffer does not fill up.
  * The automatic “flushing” does <em>not</em> flush the underlying stream, for example via {@code ProxyOutputStream.Flush}.
- * Also the stream will be flushed before garbage collection.
- * Otherwise it is similar to {@link BufferedOutputStream}.
  */
-final class DelayBufferedOutputStream extends FilterOutputStream {
+final class DelayBufferedOutputStream extends BufferedOutputStream {
 
     private static final Logger LOGGER = Logger.getLogger(DelayBufferedOutputStream.class.getName());
 
@@ -58,45 +54,6 @@ private Tuning() {}
         static final Tuning DEFAULT = new Tuning();
     }
 
-    /**
-     * The interesting state of the buffered stream, kept as a separate object so that {@link FlushRef} can hold on to it.
-     */
-    private static final class Buffer {
-
-        final OutputStream out;
-        private final byte[] dat;
-        private int pos;
-
-        Buffer(OutputStream out, int size) {
-            this.out = out;
-            dat = new byte[size];
-        }
-
-        synchronized void drain() throws IOException {
-            if (pos == 0) {
-                return;
-            }
-            out.write(dat, 0, pos);
-            pos = 0;
-        }
-
-        void write(int b) throws IOException {
-            assert Thread.holdsLock(this);
-            if (pos == dat.length) {
-                drain();
-            }
-            dat[pos++] = (byte) b;
-        }
-
-        synchronized void write(byte[] b, int off, int len) throws IOException {
-            for (int i = 0; i < len; i++) {
-                write(b[off + i]);
-            }
-        }
-
-    }
-
-    private final Buffer buf;
     private final Tuning tuning;
     private long recurrencePeriod;
 
@@ -105,97 +62,86 @@ synchronized void write(byte[] b, int off, int len) throws IOException {
     }
 
     DelayBufferedOutputStream(OutputStream out, Tuning tuning) {
-        super(out);
-        buf = new Buffer(out, tuning.bufferSize);
+        super(new FlushControlledOutputStream(out), tuning.bufferSize);
         this.tuning = tuning;
         recurrencePeriod = tuning.minRecurrencePeriod;
-        FlushRef.register(this);
         reschedule();
     }
 
-    @Override public void write(int b) throws IOException {
-        synchronized (buf) {
-            buf.write(b);
-        }
-    }
-
-    @Override public void write(byte[] b, int off, int len) throws IOException {
-        buf.write(b, off, len);
-    }
-
-    @Override public void flush() throws IOException {
-        buf.drain();
-        super.flush();
-    }
-
     private void reschedule() {
-        Timer.get().schedule(new Drain(this), recurrencePeriod, TimeUnit.MILLISECONDS);
+        Timer.get().schedule(new Flush(this), recurrencePeriod, TimeUnit.MILLISECONDS);
         recurrencePeriod = Math.min((long) (recurrencePeriod * tuning.recurrencePeriodBackoff), tuning.maxRecurrencePeriod);
     }
 
-    void drainAndReschedule() {
+    /** We can only call {@link BufferedOutputStream#flushBuffer} via {@link #flush}, but we do not wish to flush the underlying stream, only write out the buffer. */
+    private void flushBuffer() throws IOException {
+        ThreadLocal<Boolean> enableFlush = ((FlushControlledOutputStream) out).enableFlush;
+        boolean orig = enableFlush.get();
+        enableFlush.set(false);
+        try {
+            flush();
+        } finally {
+            enableFlush.set(orig);
+        }
+    }
+
+    void flushAndReschedule() {
         // TODO as an optimization, avoid flushing the buffer if it was recently flushed anyway due to filling up
         try {
-            buf.drain();
+            flushBuffer();
         } catch (IOException x) {
             LOGGER.log(Level.FINE, null, x);
         }
         reschedule();
     }
 
-    private static final class Drain implements Runnable {
+    @Override public String toString() {
+        return "DelayBufferedOutputStream[" + out + "]";
+    }
+
+    private static final class Flush implements Runnable {
 
         /** Since there is no explicit close event, just keep flushing periodically until the stream is collected. */
         private final Reference<DelayBufferedOutputStream> osr;
 
-        Drain(DelayBufferedOutputStream os) {
+        Flush(DelayBufferedOutputStream os) {
             osr = new WeakReference<>(os);
         }
 
         @Override public void run() {
             DelayBufferedOutputStream os = osr.get();
             if (os != null) {
-                os.drainAndReschedule();
+                os.flushAndReschedule();
             }
         }
 
     }
 
-    /**
-     * Flushes streams prior to garbage collection.
-     * In Java 9+ could use {@code java.util.Cleaner} instead.
-     */
-    private static final class FlushRef extends PhantomReference<DelayBufferedOutputStream> {
-
-        private static final ReferenceQueue<DelayBufferedOutputStream> rq = new ReferenceQueue<>();
-
-        static {
-            Timer.get().scheduleWithFixedDelay(() -> {
-                while (true) {
-                    FlushRef ref = (FlushRef) rq.poll();
-                    if (ref == null) {
-                        break;
-                    }
-                    LOGGER.log(Level.FINE, "flushing {0} from a DelayBufferedOutputStream", ref.buf.out);
-                    try {
-                        ref.buf.drain();
-                        ref.buf.out.flush();
-                    } catch (IOException x) {
-                        LOGGER.log(Level.WARNING, null, x);
-                    }
-                }
-            }, 0, 10, TimeUnit.SECONDS);
+    /** @see DelayBufferedOutputStream#flushBuffer */
+    private static final class FlushControlledOutputStream extends FilterOutputStream {
+
+        private final ThreadLocal<Boolean> enableFlush = new ThreadLocal<Boolean>() {
+            @Override protected Boolean initialValue() {
+                return true;
+            }
+        };
+
+        FlushControlledOutputStream(OutputStream out) {
+            super(out);
         }
 
-        static void register(DelayBufferedOutputStream dbos) {
-            new FlushRef(dbos, rq).enqueue();
+        @Override public void write(byte[] b, int off, int len) throws IOException {
+            out.write(b, off, len); // super method writes one byte at a time!
         }
 
-        private final Buffer buf;
+        @Override public void flush() throws IOException {
+            if (enableFlush.get()) {
+                super.flush();
+            }
+        }
 
-        private FlushRef(DelayBufferedOutputStream dbos, ReferenceQueue<DelayBufferedOutputStream> rq) {
-            super(dbos, rq);
-            this.buf = dbos.buf;
+        @Override public String toString() {
+            return "FlushControlledOutputStream[" + out + "]";
         }
 
     }
diff --git a/src/main/java/org/jenkinsci/plugins/workflow/log/FileLogStorage.java b/src/main/java/org/jenkinsci/plugins/workflow/log/FileLogStorage.java
@@ -85,7 +85,7 @@ private FileLogStorage(File log) {
     private synchronized void open() throws IOException {
         if (os == null) {
             os = new FileOutputStream(log, true);
-            bos = new DelayBufferedOutputStream(os);
+            bos = new GCFlushedOutputStream(new DelayBufferedOutputStream(os));
             if (index.isFile()) {
                 try (BufferedReader r = Files.newBufferedReader(index.toPath(), StandardCharsets.UTF_8)) {
                     // TODO would be faster to scan the file backwards for the penultimate \n, then convert the byte sequence from there to EOF to UTF-8 and set lastId accordingly
diff --git a/src/main/java/org/jenkinsci/plugins/workflow/log/GCFlushedOutputStream.java b/src/main/java/org/jenkinsci/plugins/workflow/log/GCFlushedOutputStream.java
@@ -0,0 +1,98 @@
+/*
+ * The MIT License
+ *
+ * Copyright 2018 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
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+package org.jenkinsci.plugins.workflow.log;
+
+import java.io.BufferedOutputStream;
+import java.io.FilterOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.ref.PhantomReference;
+import java.lang.ref.ReferenceQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import jenkins.util.Timer;
+
+/**
+ * A stream which will be flushed before garbage collection.
+ * {@link BufferedOutputStream} does not do this automatically.
+ */
+final class GCFlushedOutputStream extends FilterOutputStream {
+    
+    private static final Logger LOGGER = Logger.getLogger(GCFlushedOutputStream.class.getName());
+
+    GCFlushedOutputStream(OutputStream out) {
+        super(out);
+        FlushRef.register(this, out);
+    }
+
+    @Override public void write(byte[] b, int off, int len) throws IOException {
+        out.write(b, off, len); // super method is surprising
+    }
+
+    @Override public String toString() {
+        return "GCFlushedOutputStream[" + out + "]";
+    }
+
+    /**
+     * Flushes streams prior to garbage collection.
+     * ({@link BufferedOutputStream} does not do this automatically.)
+     * TODO Java 9+ could use java.util.Cleaner
+     */
+    private static final class FlushRef extends PhantomReference<GCFlushedOutputStream> {
+
+        private static final ReferenceQueue<GCFlushedOutputStream> rq = new ReferenceQueue<>();
+
+        static {
+            Timer.get().scheduleWithFixedDelay(() -> {
+                while (true) {
+                    FlushRef ref = (FlushRef) rq.poll();
+                    if (ref == null) {
+                        break;
+                    }
+                    LOGGER.log(Level.FINE, "flushing {0}", ref.out);
+                    try {
+                        ref.out.flush();
+                    } catch (IOException x) {
+                        LOGGER.log(Level.WARNING, null, x);
+                    }
+                }
+            }, 0, 10, TimeUnit.SECONDS);
+        }
+
+        static void register(GCFlushedOutputStream fos, OutputStream out) {
+            new FlushRef(fos, out, rq).enqueue();
+        }
+
+        private final OutputStream out;
+
+        private FlushRef(GCFlushedOutputStream fos, OutputStream out, ReferenceQueue<GCFlushedOutputStream> rq) {
+            super(fos, rq);
+            this.out = out;
+        }
+
+    }
+
+}

Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,7 @@ private static final class Replacement implements SerializableOnlyOverRemoting {`
`74`	`74`	`}`
`75`	`75`
`76`	`76`	`private Object readResolve() throws IOException {`
`77`		`- return new BufferedBuildListener(new DelayBufferedOutputStream(ros, tuning));`
	`77`	`+ return new BufferedBuildListener(new GCFlushedOutputStream(new DelayBufferedOutputStream(ros, tuning)));`
`78`	`78`	`}`
`79`	`79`
`80`	`80`	`}`