Merge pull request #1223 from yahoo/bjorncs/jetty-jdisc-improvements

Bjorncs/jetty jdisc improvements

4 files changed, 21 insertions, 137 deletions

diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/BufferPool.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/BufferPool.java
deleted file mode 100644
index 26605435f66..00000000000
--- a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/BufferPool.java
+++ /dev/null
@@ -1,122 +0,0 @@
-// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
-package com.yahoo.jdisc.http.server.jetty;
-
-import com.yahoo.yolean.concurrent.ConcurrentResourcePool;
-import com.yahoo.yolean.concurrent.ResourceFactory;
-import org.eclipse.jetty.io.ByteBufferPool;
-import org.eclipse.jetty.util.BufferUtil;
-
-import java.nio.ByteBuffer;
-
-
-/**
- * @author <a href="mailto:balder@yahoo-inc.com">Henning</a>
- * @since 6.41
- * TODO: Add tests, currently uses no pooling to chase down nasty jetty bug.
- */
-
-class BufferPool implements ByteBufferPool {
-    private static ByteBuffer allocate(int capacity, boolean direct) {
-        return direct ? BufferUtil.allocateDirect(capacity) : BufferUtil.allocate(capacity);
-    }
-    private static abstract class Pool {
-        ByteBuffer aquire(int size, boolean direct) {
-            return aquireImpl(size, direct);
-        }
-        void release(ByteBuffer buf) { releaseImpl(buf); }
-        protected abstract ByteBuffer aquireImpl(int size, boolean direct);
-        protected abstract void releaseImpl(ByteBuffer buf);
-    };
-    private static class NoPool extends Pool {
-        @Override
-        protected ByteBuffer aquireImpl(int size, boolean direct) {
-            return BufferPool.allocate(size, direct);
-        }
-        @Override
-        protected void releaseImpl(ByteBuffer buf) {
-            buf = null;
-        }
-    }
-    private static class ConcurrentPool extends Pool {
-        class ByteBufferFactory extends ResourceFactory<ByteBuffer> {
-            final int size;
-            final boolean direct;
-            ByteBufferFactory(int size, boolean direct) {
-                this.size = size;
-                this.direct = direct;
-            }
-
-            @Override
-            public ByteBuffer create() {
-                return BufferPool.allocate(size, direct);
-            }
-        }
-
-        private final ConcurrentResourcePool<ByteBuffer> pool;
-
-        ConcurrentPool(int size, boolean direct) {
-            pool = new ConcurrentResourcePool<>(new ByteBufferFactory(size, direct));
-        }
-        @Override
-        protected ByteBuffer aquireImpl(int size, boolean direct) {
-            return pool.alloc();
-        }
-
-        @Override
-        protected void releaseImpl(ByteBuffer buf) {
-            pool.free(buf);
-        }
-    }
-
-    private final int minSize2Pool;
-    private final int maxSize2Pool;
-    private final int numPools;
-    private final NoPool noPooling = new NoPool();
-    private final Pool [] directPools;
-    private final Pool [] heapPools;
-    BufferPool() {
-        // No pooling.
-        this(Integer.MAX_VALUE, 0, 0);
-    }
-    BufferPool(int minSize2Pool, int maxSize2Pool, int numPools) {
-        this.minSize2Pool = minSize2Pool;
-        this.maxSize2Pool = maxSize2Pool;
-        this.numPools = numPools;
-        if ((numPools > 0) && (maxSize2Pool > minSize2Pool)) {
-            directPools = new Pool[numPools];
-            heapPools = new Pool[numPools];
-            int size = (maxSize2Pool - minSize2Pool) / numPools;
-            for (int i = 0; i < numPools; i++) {
-                int maxSize = minSize2Pool + size*(i+1);
-                directPools[i] = new ConcurrentPool(maxSize, true);
-                heapPools[i] = new ConcurrentPool(maxSize, false);
-            }
-        } else {
-            directPools = null;
-            heapPools = null;
-        }
-    }
-
-    @Override
-    public ByteBuffer acquire(int size, boolean direct) {
-        return selectPool(size, direct).aquire(size, direct);
-    }
-
-    @Override
-    public void release(ByteBuffer buffer) {
-        selectPool(buffer.capacity(), buffer.isDirect()).release(buffer);
-
-    }
-    private Pool selectPool(int size, boolean direct) {
-        if (poolable(size)) {
-            int offsetSize = size - minSize2Pool;
-            int index = offsetSize * numPools / (maxSize2Pool - minSize2Pool);
-            return direct ? directPools[index] : heapPools[index];
-        } else {
-            return noPooling;
-        }
-    }
-    private boolean poolable(int size) {
-        return (size > minSize2Pool) && (size <= maxSize2Pool);
-    }
-}
diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/HttpRequestDispatch.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/HttpRequestDispatch.java
index 8626cda6045..55b31d6e696 100644
--- a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/HttpRequestDispatch.java
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/HttpRequestDispatch.java
@@ -159,8 +159,7 @@ class HttpRequestDispatch {
                         servletInputStream,
                         requestContentChannel,
                         jDiscContext.janitor,
-                        metricReporter,
-                        servletResponseController);
+                        metricReporter);
 
         servletInputStream.setReadListener(servletRequestReader);
         return servletRequestReader;
diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletOutputStreamWriter.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletOutputStreamWriter.java
index 5f248db9fe3..eb782737a0c 100644
--- a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletOutputStreamWriter.java
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletOutputStreamWriter.java
@@ -42,7 +42,6 @@ public class ServletOutputStreamWriter {
 
     private static final Logger log = Logger.getLogger(ServletOutputStreamWriter.class.getName());
 
-    // TODO: This reference is not guaranteed to be unique; ByteBuffer.allocate(0) MAY in principle return a singleton!
     // If so, application code could fake a close by writing such a byte buffer.
     // The problem can be solved by filtering out zero-length byte buffers from application code.
     // Other ways to express this are also possible, e.g. with a 'closed' state checked when queue goes empty.
@@ -78,25 +77,37 @@ public class ServletOutputStreamWriter {
     }
 
     public void registerWriteListener() {
-        synchronized (monitor) {
-            assertStateIs(state, State.NOT_STARTED);
-            outputStream.setWriteListener(writeListener);
-        }
+        outputStream.setWriteListener(writeListener);
     }
 
     public void sendErrorContentAndCloseAsync(ByteBuffer errorContent) {
+        boolean thisThreadShouldWrite;
+
         synchronized (monitor) {
             // Assert that no content has been written as it is too late to write error response if the response is committed.
             switch (state) {
                 case NOT_STARTED:
+                    queueErrorContent_holdingLock(errorContent);
+                    state = State.WAITING_FOR_WRITE_POSSIBLE_CALLBACK;
+                    thisThreadShouldWrite = false;
+                    break;
                 case WAITING_FOR_FIRST_BUFFER:
-                    writeBuffer(errorContent, null);
-                    close(null);
-                    return;
+                    queueErrorContent_holdingLock(errorContent);
+                    state = State.WRITING_BUFFERS;
+                    thisThreadShouldWrite = true;
+                    break;
                 default:
                     throw createAndLogAssertionError("Invalid state: " + state);
             }
         }
+        if (thisThreadShouldWrite) {
+            writeBuffersInQueueToOutputStream();
+        }
+    }
+
+    private void queueErrorContent_holdingLock(ByteBuffer errorContent) {
+        responseContentQueue.addLast(new ResponseContentPart(errorContent, null));
+        responseContentQueue.addLast(new ResponseContentPart(CLOSE_STREAM_BUFFER, null));
     }
 
     public void writeBuffer(ByteBuffer buf, CompletionHandler handler) {
@@ -129,7 +140,6 @@ public class ServletOutputStreamWriter {
         }
 
         if (thisThreadShouldWrite) {
-            // TODO: Consider refactoring to avoid multiple monitor entry-exit.
             writeBuffersInQueueToOutputStream();
         }
     }
diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletRequestReader.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletRequestReader.java
index afb16659b9e..0d0e80f349f 100644
--- a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletRequestReader.java
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ServletRequestReader.java
@@ -41,7 +41,6 @@ class ServletRequestReader implements ReadListener {
 
     private final ServletInputStream servletInputStream;
     private final ContentChannel requestContentChannel;
-    private final ServletResponseController responseController;
 
     private final Executor executor;
     private final MetricReporter metricReporter;
@@ -93,8 +92,7 @@ class ServletRequestReader implements ReadListener {
             ServletInputStream servletInputStream,
             ContentChannel requestContentChannel,
             Executor executor,
-            MetricReporter metricReporter,
-            ServletResponseController responseController) {
+            MetricReporter metricReporter) {
 
         Preconditions.checkNotNull(servletInputStream);
         Preconditions.checkNotNull(requestContentChannel);
@@ -105,7 +103,6 @@ class ServletRequestReader implements ReadListener {
         this.requestContentChannel = requestContentChannel;
         this.executor = executor;
         this.metricReporter = metricReporter;
-        this.responseController = responseController;
     }
 
     @Override