Provide our own buffer pool. Now does no pooling.

4 files changed, 113 insertions, 6 deletions

diff --git a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactory.java b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactory.java
index 874d9ab7173..6787a336074 100644
--- a/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactory.java
+++ b/jdisc_http_service/src/main/java/com/yahoo/jdisc/http/server/jetty/ConnectorFactory.java
@@ -11,7 +11,10 @@ import com.yahoo.jdisc.http.SecretStore;
 import com.yahoo.jdisc.http.ssl.ReaderForPath;
 import com.yahoo.jdisc.http.ssl.SslKeyStore;
 import com.yahoo.jdisc.http.ssl.SslKeyStoreFactory;
+import com.yahoo.yolean.concurrent.ConcurrentResourcePool;
+import com.yahoo.yolean.concurrent.ResourceFactory;
 import org.eclipse.jetty.http.HttpVersion;
+import org.eclipse.jetty.io.ByteBufferPool;
 import org.eclipse.jetty.server.ConnectionFactory;
 import org.eclipse.jetty.server.ConnectorStatistics;
 import org.eclipse.jetty.server.HttpConfiguration;
@@ -28,6 +31,7 @@ import java.io.Reader;
 import java.lang.reflect.Field;
 import java.net.Socket;
 import java.net.SocketException;
+import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.FileChannel;
 import java.nio.channels.ServerSocketChannel;
@@ -250,6 +254,107 @@ public class ConnectorFactory {
     }
 
     public static class JDiscServerConnector extends ServerConnector {
+        public static class BufferPool implements ByteBufferPool {
+            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);
+            };
+            class NoPool extends Pool {
+                @Override
+                protected ByteBuffer aquireImpl(int size, boolean direct) {
+                    return newByteBuffer(size, direct);
+                }
+                @Override
+                protected void releaseImpl(ByteBuffer buf) {
+                    buf = null;
+                }
+            }
+            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 newByteBuffer(size, direct);
+                    }
+                }
+
+                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);
+                }
+            }
+            final int minSize2Pool;
+            final int maxSize2Pool;
+            final int numPools;
+            final NoPool noPooling = new NoPool();
+            Pool [] directPools;
+            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);
+            }
+        }
         public static final String REQUEST_ATTRIBUTE = JDiscServerConnector.class.getName();
         private final static Logger log = Logger.getLogger(JDiscServerConnector.class.getName());
         private final Metric.Context metricCtx;
@@ -264,7 +369,7 @@ public class ConnectorFactory {
                 final Server server,
                 final ServerSocketChannel channelOpenedByActivator,
                 final ConnectionFactory... factories) {
-            super(server, factories);
+            super(server, null, null, new BufferPool(), -1, -1, factories);
             this.channelOpenedByActivator = channelOpenedByActivator;
             this.tcpKeepAlive = config.tcpKeepAliveEnabled();
             this.tcpNoDelay = config.tcpNoDelay();
diff --git a/vespajlib/src/main/java/com/yahoo/collections/ConcurrentResourcePool.java b/yolean/src/main/java/com/yahoo/yolean/concurrent/ConcurrentResourcePool.java
index 23fed831d1f..5e7788e6e20 100644
--- a/vespajlib/src/main/java/com/yahoo/collections/ConcurrentResourcePool.java
+++ b/yolean/src/main/java/com/yahoo/yolean/concurrent/ConcurrentResourcePool.java
@@ -1,8 +1,8 @@
 // Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
-package com.yahoo.collections;
+package com.yahoo.yolean.concurrent;
+
+import com.yahoo.yolean.concurrent.ResourceFactory;
 
-import java.util.ArrayDeque;
-import java.util.Deque;
 import java.util.Iterator;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
diff --git a/vespajlib/src/main/java/com/yahoo/collections/ResourceFactory.java b/yolean/src/main/java/com/yahoo/yolean/concurrent/ResourceFactory.java
index d6fe032e0e3..68b505c1b03 100644
--- a/vespajlib/src/main/java/com/yahoo/collections/ResourceFactory.java
+++ b/yolean/src/main/java/com/yahoo/yolean/concurrent/ResourceFactory.java
@@ -1,5 +1,5 @@
 // Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
-package com.yahoo.collections;
+package com.yahoo.yolean.concurrent;
 
 /**
  * @author baldersheim
diff --git a/vespajlib/src/main/java/com/yahoo/collections/ResourcePool.java b/yolean/src/main/java/com/yahoo/yolean/concurrent/ResourcePool.java
index 112a77b069d..0bf4749fe63 100644
--- a/vespajlib/src/main/java/com/yahoo/collections/ResourcePool.java
+++ b/yolean/src/main/java/com/yahoo/yolean/concurrent/ResourcePool.java
@@ -1,5 +1,7 @@
 // Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
-package com.yahoo.collections;
+package com.yahoo.yolean.concurrent;
+
+import com.yahoo.yolean.concurrent.ResourceFactory;
 
 import java.util.ArrayDeque;
 import java.util.Deque;