Improve DelayedHandler. #9051

jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/MimeTypes.java

@@ -691,7 +691,12 @@ public static MimeTypes.Type getMimeTypeFromContentType(HttpField field)
         if (field instanceof MimeTypes.ContentTypeField contentTypeField)
             return contentTypeField.getMimeType();
 
-        return MimeTypes.CACHE.get(field.getValue());
+        String contentType = field.getValue();
+        int semicolon = contentType.indexOf(';');
+        if (semicolon >= 0)
+            contentType = contentType.substring(0, semicolon).trim();
+
+        return MimeTypes.CACHE.get(contentType);
     }
 
     /**

jetty-core/jetty-http2/jetty-http2-client/src/main/java/org/eclipse/jetty/http2/client/HTTP2ClientConnectionFactory.java

@@ -83,7 +83,7 @@ private static class HTTP2ClientConnection extends HTTP2Connection implements Ca
 
         private HTTP2ClientConnection(HTTP2Client client, EndPoint endpoint, HTTP2ClientSession session, Promise<Session> sessionPromise, Session.Listener listener)
         {
-            super(client.getByteBufferPool(), client.getExecutor(), endpoint, session, client.getInputBufferSize());
+            super(client.getByteBufferPool(), client.getExecutor(), endpoint, session, client.getInputBufferSize(), -1);
             this.client = client;
             this.promise = sessionPromise;
             this.listener = listener;

jetty-core/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Connection.java

@@ -20,6 +20,7 @@
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.eclipse.jetty.http2.api.Stream;
 import org.eclipse.jetty.http2.frames.DataFrame;
@@ -58,16 +59,23 @@ public class HTTP2Connection extends AbstractConnection implements Parser.Listen
     private final ByteBufferPool bufferPool;
     private final HTTP2Session session;
     private final int bufferSize;
+    private final int minBufferSpace;
     private final ExecutionStrategy strategy;
     private boolean useInputDirectByteBuffers;
     private boolean useOutputDirectByteBuffers;
 
     protected HTTP2Connection(ByteBufferPool bufferPool, Executor executor, EndPoint endPoint, HTTP2Session session, int bufferSize)
+    {
+        this(bufferPool, executor, endPoint, session, bufferSize, -1);
+    }
+
+    protected HTTP2Connection(ByteBufferPool bufferPool, Executor executor, EndPoint endPoint, HTTP2Session session, int bufferSize, int minBufferSpace)
     {
         super(endPoint, executor);
         this.bufferPool = bufferPool;
         this.session = session;
         this.bufferSize = bufferSize;
+        this.minBufferSpace = minBufferSpace < 0 ? Math.min(1500, bufferSize) : minBufferSpace;
         this.strategy = new AdaptiveExecutionStrategy(producer, executor);
         LifeCycle.start(strategy);
     }
@@ -147,6 +155,7 @@ public void onClose(Throwable cause)
         super.onClose(cause);
 
         LifeCycle.stop(strategy);
+        producer.stop();
     }
 
     @Override
@@ -157,12 +166,20 @@ public void onFillable()
         produce();
     }
 
-    private int fill(EndPoint endPoint, ByteBuffer buffer)
+    private int fill(EndPoint endPoint, ByteBuffer buffer, boolean compact)
     {
+        int padding = 0;
         try
         {
             if (endPoint.isInputShutdown())
                 return -1;
+
+            if (!compact)
+            {
+                // Add padding content to avoid compaction
+                padding = buffer.limit();
+                buffer.position(0);
+            }
             return endPoint.fill(buffer);
         }
         catch (IOException x)
@@ -171,6 +188,11 @@ private int fill(EndPoint endPoint, ByteBuffer buffer)
                 LOG.debug("Could not read from {}", endPoint, x);
             return -1;
         }
+        finally
+        {
+            if (!compact && padding > 0)
+                buffer.position(padding);
+        }
     }
 
     @Override
@@ -303,16 +325,19 @@ public void onConnectionFailure(int error, String reason)
 
     protected class HTTP2Producer implements ExecutionStrategy.Producer
     {
+        private static final RetainableByteBuffer.Mutable STOPPED = new RetainableByteBuffer.NonRetainableByteBuffer(BufferUtil.EMPTY_BUFFER);
         private final Callback fillableCallback = new FillableCallback();
+        private final AtomicReference<RetainableByteBuffer.Mutable> heldBuffer = new AtomicReference<>();
         private RetainableByteBuffer.Mutable networkBuffer;
         private boolean shutdown;
         private boolean failed;
 
         private void setInputBuffer(ByteBuffer byteBuffer)
         {
-            acquireNetworkBuffer();
+            RetainableByteBuffer.Mutable networkBuffer = acquireBuffer();
             if (!networkBuffer.append(byteBuffer))
-                LOG.warn("overflow");
+                throw new IllegalStateException("overflow");
+            holdBuffer(networkBuffer);
         }
 
         @Override
@@ -328,13 +353,14 @@ public Runnable produce()
                 return null;
 
             boolean interested = false;
-            acquireNetworkBuffer();
+            networkBuffer = acquireBuffer();
             try
             {
                 boolean parse = networkBuffer.hasRemaining();
 
                 while (true)
                 {
+                    boolean compact = true;
                     if (parse)
                     {
                         while (networkBuffer.hasRemaining())
@@ -349,17 +375,30 @@ public Runnable produce()
                             LOG.debug("Dequeued new task {}", task);
                         if (task != null)
                             return task;
+                    }
 
-                        // If more references than 1 (ie not just us), don't refill into buffer and risk compaction.
-                        if (networkBuffer.isRetained())
-                            reacquireNetworkBuffer();
+                    // If the application has retained the content chunks then we must not overwrite content.
+                    if (networkBuffer.isRetained())
+                    {
+                        // If there is sufficient space available, we can top up the buffer rather than allocate a new one
+                        if (minBufferSpace > 0 && BufferUtil.space(networkBuffer.getByteBuffer()) >= minBufferSpace)
+                        {
+                            // do not compact the buffer
+                            compact = false;
+                        }
+                        else
+                        {
+                            // otherwise reacquire the buffer and fill into the new buffer.
+                            if (LOG.isDebugEnabled())
+                                LOG.debug("Released retained {}", networkBuffer);
+                            networkBuffer.release();
+                            networkBuffer = acquireBuffer();
+                        }
                     }
 
-                    // Here we know that this.networkBuffer is not retained by
-                    // application code: either it has been released, or it's a new one.
-                    int filled = fill(getEndPoint(), networkBuffer.getByteBuffer());
+                    int filled = fill(getEndPoint(), networkBuffer.getByteBuffer(), compact);
                     if (LOG.isDebugEnabled())
-                        LOG.debug("Filled {} bytes in {}", filled, networkBuffer);
+                        LOG.debug("Filled {} bytes compacted {} in {}", filled, compact, networkBuffer);
 
                     if (filled > 0)
                     {
@@ -381,50 +420,63 @@ else if (filled == 0)
             }
             finally
             {
-                releaseNetworkBuffer();
+                if (networkBuffer.isRetained() && !shutdown)
+                {
+                    holdBuffer(networkBuffer);
+                }
+                else
+                {
+                    if (LOG.isDebugEnabled())
+                        LOG.debug("Released after process {}", networkBuffer);
+                    networkBuffer.release();
+                }
+                networkBuffer = null;
                 if (interested)
                     getEndPoint().fillInterested(fillableCallback);
             }
         }
 
-        private void acquireNetworkBuffer()
+        private RetainableByteBuffer.Mutable acquireBuffer()
         {
-            if (networkBuffer == null)
-            {
-                networkBuffer = bufferPool.acquire(bufferSize, isUseInputDirectByteBuffers()).asMutable();
-                if (LOG.isDebugEnabled())
-                    LOG.debug("Acquired {}", networkBuffer);
-            }
+            RetainableByteBuffer.Mutable buffer = heldBuffer.getAndSet(null);
+            if (buffer == null)
+                buffer = bufferPool.acquire(bufferSize, isUseInputDirectByteBuffers()).asMutable();
+            if (LOG.isDebugEnabled())
+                LOG.debug("Acquired {}", buffer);
+            return buffer;
         }
 
-        private void reacquireNetworkBuffer()
+        private void holdBuffer(RetainableByteBuffer.Mutable buffer)
         {
-            RetainableByteBuffer.Mutable currentBuffer = networkBuffer;
-            if (currentBuffer == null)
-                throw new IllegalStateException();
-
-            if (currentBuffer.hasRemaining())
-                throw new IllegalStateException();
-
-            currentBuffer.release();
-            networkBuffer = bufferPool.acquire(bufferSize, isUseInputDirectByteBuffers());
-            if (LOG.isDebugEnabled())
-                LOG.debug("Reacquired {}<-{}", currentBuffer, networkBuffer);
+            if (heldBuffer.compareAndSet(null, buffer))
+            {
+                if (LOG.isDebugEnabled())
+                    LOG.debug("Held {}", buffer);
+            }
+            else
+            {
+                if (heldBuffer.get() == STOPPED)
+                {
+                    if (LOG.isDebugEnabled())
+                        LOG.debug("Released instead of holding {}", buffer);
+                    buffer.release();
+                }
+                else
+                {
+                    throw new IllegalStateException("Buffer already saved");
+                }
+            }
         }
 
-        private void releaseNetworkBuffer()
+        private void stop()
         {
-            RetainableByteBuffer.Mutable currentBuffer = networkBuffer;
-            if (currentBuffer == null)
-                throw new IllegalStateException();
-
-            if (currentBuffer.hasRemaining() && !shutdown && !failed)
-                throw new IllegalStateException();
-
-            currentBuffer.release();
-            networkBuffer = null;
-            if (LOG.isDebugEnabled())
-                LOG.debug("Released {}", currentBuffer);
+            RetainableByteBuffer.Mutable buffer = heldBuffer.getAndSet(STOPPED);
+            if (buffer != null)
+            {
+                if (LOG.isDebugEnabled())
+                    LOG.debug("Released in stop {}", buffer);
+                buffer.release();
+            }
         }
 
         @Override

jetty-core/jetty-http2/jetty-http2-server/src/main/java/org/eclipse/jetty/http2/server/internal/HTTP2ServerConnection.java

@@ -76,7 +76,7 @@ public class HTTP2ServerConnection extends HTTP2Connection implements Connection
 
     public HTTP2ServerConnection(Connector connector, EndPoint endPoint, HttpConfiguration httpConfig, HTTP2ServerSession session, ServerSessionListener listener)
     {
-        super(connector.getByteBufferPool(), connector.getExecutor(), endPoint, session, httpConfig.getInputBufferSize());
+        super(connector.getByteBufferPool(), connector.getExecutor(), endPoint, session, httpConfig.getInputBufferSize(), httpConfig.getMinInputBufferSpace());
         this.connector = connector;
         this.listener = listener;
         this.httpConfig = httpConfig;

jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/AbstractTest.java

@@ -143,6 +143,8 @@ protected MetaData.Request newRequest(String method, String path, HttpFields fie
     @AfterEach
     public void dispose() throws Exception
     {
+        // Stop the client so that all connections are closed and any saved buffers are released
+        LifeCycle.stop(httpClient);
         try
         {
             if (serverBufferPool != null)
@@ -152,7 +154,6 @@ public void dispose() throws Exception
         }
         finally
         {
-            LifeCycle.stop(httpClient);
             LifeCycle.stop(server);
         }
     }