Combined ByteBufferPool (#8171)

All `ByteBufferPool` can now be accessed as `RetainableByteBufferPools`. Users now need to configure only a single buffer pool and there is just the additional retained parameter that needs consideration. Default buffer pool has been changed to logarithmic, but we may wish to review that before next release. Default factor size has been increased to 4096.
jetty · Jul 4, 2022 · 2b817f0 · 2b817f0
1 parent 27a89b2
commit 2b817f0
Show file tree

Hide file tree

Showing 36 changed files with 617 additions and 311 deletions.
diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpReceiverOverHTTP.java b/jetty-client/src/main/java/org/eclipse/jetty/client/http/HttpReceiverOverHTTP.java
@@ -64,7 +64,7 @@ public HttpReceiverOverHTTP(HttpChannelOverHTTP channel)
  parser.setHeaderCacheCaseSensitive(httpTransport.isHeaderCacheCaseSensitive());
  }
 
- this.retainableByteBufferPool = RetainableByteBufferPool.findOrAdapt(httpClient, httpClient.getByteBufferPool());
+ this.retainableByteBufferPool = httpClient.getByteBufferPool().asRetainableByteBufferPool();
  }
 
  @Override

diff --git a/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTLSTest.java b/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTLSTest.java
@@ -22,9 +22,9 @@
 import java.net.SocketTimeoutException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
@@ -708,22 +708,61 @@ protected int networkFill(ByteBuffer input) throws IOException
  assertEquals(0, clientBytes.get());
  }
 
+ protected class TestRetained extends ArrayRetainableByteBufferPool
+ {
+ private final ByteBufferPool _pool;
+
+ public TestRetained(ByteBufferPool pool, int factor, int maxCapacity, int maxBucketSize, long retainedHeapMemory, long retainedDirectMemory)
+ {
+ super(0, factor, maxCapacity, maxBucketSize, retainedHeapMemory, retainedDirectMemory);
+ _pool = pool;
+ }
+
+ @Override
+ protected ByteBuffer allocate(int capacity)
+ {
+ return _pool.acquire(capacity, false);
+ }
+
+ @Override
+ protected ByteBuffer allocateDirect(int capacity)
+ {
+ return _pool.acquire(capacity, true);
+ }
+
+ @Override
+ protected void removed(RetainableByteBuffer retainedBuffer)
+ {
+ _pool.release(retainedBuffer.getBuffer());
+ }
+
+ @Override
+ public Pool<RetainableByteBuffer> poolFor(int capacity, boolean direct)
+ {
+ return super.poolFor(capacity, direct);
+ }
+ }
+
+ private class TestByteBufferPool extends ArrayByteBufferPool
+ {
+ @Override
+ protected RetainableByteBufferPool newRetainableByteBufferPool(int factor, int maxCapacity, int maxBucketSize, long retainedHeapMemory, long retainedDirectMemory)
+ {
+ return new TestRetained(this, factor, maxCapacity, maxBucketSize, retainedHeapMemory, retainedDirectMemory);
+ }
+ }
+
  @Test
  public void testEncryptedInputBufferRepooling() throws Exception
  {
  SslContextFactory.Server serverTLSFactory = createServerSslContextFactory();
  QueuedThreadPool serverThreads = new QueuedThreadPool();
  serverThreads.setName("server");
  server = new Server(serverThreads);
- var retainableByteBufferPool = new ArrayRetainableByteBufferPool()
- {
- @Override
- public Pool<RetainableByteBuffer> poolFor(int capacity, boolean direct)
- {
- return super.poolFor(capacity, direct);
- }
- };
- server.addBean(retainableByteBufferPool);
+
+ ArrayByteBufferPool byteBufferPool = new TestByteBufferPool();
+ RetainableByteBufferPool retainableByteBufferPool = byteBufferPool.asRetainableByteBufferPool();
+ server.addBean(byteBufferPool);
  HttpConfiguration httpConfig = new HttpConfiguration();
  httpConfig.addCustomizer(new SecureRequestCustomizer());
  HttpConnectionFactory http = new HttpConnectionFactory(httpConfig);
@@ -765,9 +804,12 @@ protected int networkFill(ByteBuffer input) throws IOException
 
  assertThrows(Exception.class, () -> client.newRequest("localhost", connector.getLocalPort()).scheme(HttpScheme.HTTPS.asString()).send());
 
- Pool<RetainableByteBuffer> bucket = retainableByteBufferPool.poolFor(16 * 1024 + 1, ssl.isDirectBuffersForEncryption());
+ Pool<RetainableByteBuffer> bucket = ((TestRetained)retainableByteBufferPool).poolFor(16 * 1024 + 1, connector.getConnectionFactory(HttpConnectionFactory.class).isUseInputDirectByteBuffers());
  assertEquals(1, bucket.size());
  assertEquals(1, bucket.getIdleCount());
+
+ long count = ssl.isDirectBuffersForDecryption() ? byteBufferPool.getDirectByteBufferCount() : byteBufferPool.getHeapByteBufferCount();
+ assertEquals(1, count);
  }
 
  @Test
@@ -777,7 +819,7 @@ public void testEncryptedOutputBufferRepooling() throws Exception
  QueuedThreadPool serverThreads = new QueuedThreadPool();
  serverThreads.setName("server");
  server = new Server(serverThreads);
- List<ByteBuffer> leakedBuffers = new ArrayList<>();
+ List<ByteBuffer> leakedBuffers = new CopyOnWriteArrayList<>();
  ArrayByteBufferPool byteBufferPool = new ArrayByteBufferPool()
  {
  @Override
@@ -834,6 +876,7 @@ protected boolean networkFlush(ByteBuffer output) throws IOException
 
  assertThrows(Exception.class, () -> client.newRequest("localhost", connector.getLocalPort()).scheme(HttpScheme.HTTPS.asString()).send());
 
+ byteBufferPool.asRetainableByteBufferPool().clear();
  await().atMost(5, TimeUnit.SECONDS).until(() -> leakedBuffers, is(empty()));
  }
 
@@ -845,7 +888,7 @@ public void testEncryptedOutputBufferRepoolingAfterNetworkFlushReturnsFalse(bool
  QueuedThreadPool serverThreads = new QueuedThreadPool();
  serverThreads.setName("server");
  server = new Server(serverThreads);
- List<ByteBuffer> leakedBuffers = new ArrayList<>();
+ List<ByteBuffer> leakedBuffers = new CopyOnWriteArrayList<>();
  ArrayByteBufferPool byteBufferPool = new ArrayByteBufferPool()
  {
  @Override
@@ -916,6 +959,7 @@ protected void service(String target, Request jettyRequest, HttpServletRequest r
 
  assertThrows(Exception.class, () -> client.newRequest("localhost", connector.getLocalPort()).scheme(HttpScheme.HTTPS.asString()).send());
 
+ byteBufferPool.asRetainableByteBufferPool().clear();
  await().atMost(5, TimeUnit.SECONDS).until(() -> leakedBuffers, is(empty()));
  }
 
@@ -927,7 +971,7 @@ public void testEncryptedOutputBufferRepoolingAfterNetworkFlushThrows(boolean cl
  QueuedThreadPool serverThreads = new QueuedThreadPool();
  serverThreads.setName("server");
  server = new Server(serverThreads);
- List<ByteBuffer> leakedBuffers = new ArrayList<>();
+ List<ByteBuffer> leakedBuffers = new CopyOnWriteArrayList<>();
  ArrayByteBufferPool byteBufferPool = new ArrayByteBufferPool()
  {
  @Override
@@ -998,6 +1042,7 @@ protected void service(String target, Request jettyRequest, HttpServletRequest r
 
  assertThrows(Exception.class, () -> client.newRequest("localhost", connector.getLocalPort()).scheme(HttpScheme.HTTPS.asString()).send());
 
+ byteBufferPool.asRetainableByteBufferPool().clear();
  await().atMost(5, TimeUnit.SECONDS).until(() -> leakedBuffers, is(empty()));
  }
 

diff --git a/.../fcgi-client/src/main/java/org/eclipse/jetty/fcgi/client/http/HttpConnectionOverFCGI.java b/.../fcgi-client/src/main/java/org/eclipse/jetty/fcgi/client/http/HttpConnectionOverFCGI.java
@@ -79,7 +79,7 @@ public HttpConnectionOverFCGI(EndPoint endPoint, HttpDestination destination, Pr
  this.parser = new ClientParser(new ResponseListener());
  requests.addLast(0);
  HttpClient client = destination.getHttpClient();
- this.networkByteBufferPool = RetainableByteBufferPool.findOrAdapt(client, client.getByteBufferPool());
+ this.networkByteBufferPool = client.getByteBufferPool().asRetainableByteBufferPool();
  }
 
  public HttpDestination getHttpDestination()

diff --git a/jetty-fcgi/fcgi-client/src/main/java/org/eclipse/jetty/fcgi/generator/ServerGenerator.java b/jetty-fcgi/fcgi-client/src/main/java/org/eclipse/jetty/fcgi/generator/ServerGenerator.java
@@ -120,7 +120,7 @@ public Result generateResponseContent(int request, ByteBuffer content, boolean l
  private ByteBuffer generateEndRequest(int request, boolean aborted)
  {
  request &= 0xFF_FF;
- ByteBuffer endRequestBuffer = acquire(8);
+ ByteBuffer endRequestBuffer = acquire(16);
  BufferUtil.clearToFill(endRequestBuffer);
  endRequestBuffer.putInt(0x01_03_00_00 + request);
  endRequestBuffer.putInt(0x00_08_00_00);

diff --git a/jetty-fcgi/fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/ServerFCGIConnection.java b/jetty-fcgi/fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/ServerFCGIConnection.java
@@ -50,7 +50,7 @@ public ServerFCGIConnection(Connector connector, EndPoint endPoint, HttpConfigur
  {
  super(endPoint, connector.getExecutor());
  this.connector = connector;
- this.networkByteBufferPool = RetainableByteBufferPool.findOrAdapt(connector, connector.getByteBufferPool());
+ this.networkByteBufferPool = connector.getByteBufferPool().asRetainableByteBufferPool();
  this.flusher = new Flusher(endPoint);
  this.configuration = configuration;
  this.sendStatus200 = sendStatus200;

diff --git a/...tp2-client/src/main/java/org/eclipse/jetty/http2/client/HTTP2ClientConnectionFactory.java b/...tp2-client/src/main/java/org/eclipse/jetty/http2/client/HTTP2ClientConnectionFactory.java
@@ -68,7 +68,7 @@ public Connection newConnection(EndPoint endPoint, Map<String, Object> context)
  parser.setMaxFrameLength(client.getMaxFrameLength());
  parser.setMaxSettingsKeys(client.getMaxSettingsKeys());
 
- RetainableByteBufferPool retainableByteBufferPool = RetainableByteBufferPool.findOrAdapt(client, byteBufferPool);
+ RetainableByteBufferPool retainableByteBufferPool = byteBufferPool.asRetainableByteBufferPool();
 
  HTTP2ClientConnection connection = new HTTP2ClientConnection(client, retainableByteBufferPool, executor, endPoint,
  parser, session, client.getInputBufferSize(), promise, listener);

diff --git a/...er/src/main/java/org/eclipse/jetty/http2/server/AbstractHTTP2ServerConnectionFactory.java b/...er/src/main/java/org/eclipse/jetty/http2/server/AbstractHTTP2ServerConnectionFactory.java
@@ -280,7 +280,7 @@ public Connection newConnection(Connector connector, EndPoint endPoint)
  parser.setMaxFrameLength(getMaxFrameLength());
  parser.setMaxSettingsKeys(getMaxSettingsKeys());
 
- RetainableByteBufferPool retainableByteBufferPool = RetainableByteBufferPool.findOrAdapt(connector, connector.getByteBufferPool());
+ RetainableByteBufferPool retainableByteBufferPool = connector.getByteBufferPool().asRetainableByteBufferPool();
 
  HTTP2Connection connection = new HTTP2ServerConnection(retainableByteBufferPool, connector.getExecutor(),
  endPoint, httpConfiguration, parser, session, getInputBufferSize(), listener);

diff --git a/...p3/http3-common/src/main/java/org/eclipse/jetty/http3/internal/HTTP3StreamConnection.java b/...p3/http3-common/src/main/java/org/eclipse/jetty/http3/internal/HTTP3StreamConnection.java
@@ -57,7 +57,7 @@ public abstract class HTTP3StreamConnection extends AbstractConnection
  public HTTP3StreamConnection(QuicStreamEndPoint endPoint, Executor executor, ByteBufferPool byteBufferPool, MessageParser parser)
  {
  super(endPoint, executor);
- this.buffers = RetainableByteBufferPool.findOrAdapt(null, byteBufferPool);
+ this.buffers = byteBufferPool.asRetainableByteBufferPool();
  this.parser = parser;
  parser.init(MessageListener::new);
  }