EagerContentHandler. #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(), 0);
             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(1024, 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> savedBuffer = 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");
+            saveBuffer(networkBuffer);
         }
 
         @Override
@@ -328,13 +353,15 @@ public Runnable produce()
                 return null;
 
             boolean interested = false;
-            acquireNetworkBuffer();
+            int filled = 0;
+            networkBuffer = acquireBuffer();
             try
             {
                 boolean parse = networkBuffer.hasRemaining();
 
                 while (true)
                 {
+                    boolean compact = true;
                     if (parse)
                     {
                         while (networkBuffer.hasRemaining())
@@ -349,17 +376,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());
+                    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 +421,63 @@ else if (filled == 0)
             }
             finally
             {
-                releaseNetworkBuffer();
+                if (networkBuffer.isRetained() && !shutdown)
+                {
+                    saveBuffer(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 = savedBuffer.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 saveBuffer(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 (savedBuffer.compareAndSet(null, buffer))
+            {
+                if (LOG.isDebugEnabled())
+                    LOG.debug("Saved {}", buffer);
+            }
+            else
+            {
+                if (savedBuffer.get() == STOPPED)
+                {
+                    if (LOG.isDebugEnabled())
+                        LOG.debug("Released in save {}", 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 = savedBuffer.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);
         }
     }

jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/AbstractConnection.java

@@ -142,7 +142,7 @@ public boolean isFillInterested()
     protected void onFillInterestedFailed(Throwable cause)
     {
         if (LOG.isDebugEnabled())
-            LOG.debug("{} onFillInterestedFailed {}", this, cause);
+            LOG.debug("onFillInterestedFailed {}", this, cause);
         if (_endPoint.isOpen())
         {
             boolean close = true;

jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/ArrayByteBufferPool.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.IntUnaryOperator;
 import java.util.stream.Collectors;
@@ -66,6 +67,7 @@ public class ArrayByteBufferPool implements ByteBufferPool, Dumpable
     private final long _maxDirectMemory;
     private final IntUnaryOperator _bucketIndexFor;
     private final AtomicBoolean _evictor = new AtomicBoolean(false);
+    private final AtomicLong _reserved = new AtomicLong();
     private boolean _statisticsEnabled;
 
     /**
@@ -175,6 +177,12 @@ private long maxMemory(long maxMemory)
         return maxMemory;
     }
 
+    @ManagedAttribute("The current number of allocated bytes reserved to be added to the pool once released")
+    public long getReserved()
+    {
+        return _reserved.get();
+    }
+
     @ManagedAttribute("Whether statistics are enabled")
     public boolean isStatisticsEnabled()
     {
@@ -214,6 +222,7 @@ public RetainableByteBuffer.Mutable acquire(int size, boolean direct)
         if (entry == null)
         {
             ByteBuffer buffer = BufferUtil.allocate(bucket.getCapacity(), direct);
+            _reserved.addAndGet(buffer.capacity());
             return new ReservedBuffer(buffer, bucket);
         }
 
@@ -249,6 +258,7 @@ public boolean releaseAndRemove(RetainableByteBuffer buffer)
 
     private void reserve(RetainedBucket bucket, ByteBuffer byteBuffer)
     {
+        _reserved.addAndGet(-byteBuffer.capacity());
         bucket.recordRelease();
 
         // Try to reserve an entry to put the buffer into the pool.

jetty-core/jetty-server/src/main/config/etc/jetty-delayed.xml

@@ -9,6 +9,7 @@
   <Call name="insertHandler">
     <Arg>
       <New id="DelayedHandler" class="org.eclipse.jetty.server.handler.DelayedHandler">
+        <Set name="maxRetainedContent" property="jetty.delayed.maxRetainedContent"/>
       </New>
     </Arg>
   </Call>

jetty-core/jetty-server/src/main/config/etc/jetty.xml

@@ -71,6 +71,7 @@
       <Set name="generateRedirectBody" property="jetty.httpConfig.generateRedirectBody"/>
       <Set name="useInputDirectByteBuffers" property="jetty.httpConfig.useInputDirectByteBuffers"/>
       <Set name="useOutputDirectByteBuffers" property="jetty.httpConfig.useOutputDirectByteBuffers"/>
+      <Set name="minInputBufferSpace" property="jetty.httpConfig.minInputBufferSpace"/>
     </New>
 
     <!-- =========================================================== -->

jetty-core/jetty-server/src/main/config/modules/delay-until-content.mod

@@ -1,10 +1,10 @@
 
 [description]
 Applies DelayedHandler to entire server.
-Delays request handling until any body content has arrived, to minimize blocking.
+Delays request handling until body content has arrived, to minimize blocking.
 For form data and multipart, the handling is delayed until the entire request body has
-been asynchronously read. For all other content types, the delay is until the first byte
-has arrived.
+been asynchronously read. For all other content types, the delay is for up to a limited size,
+which by default is one input buffer.
 
 [tags]
 server
@@ -18,3 +18,8 @@ threadlimit
 [xml]
 etc/jetty-delayed.xml
 
+
+[ini-template]
+#tag::documentation[]
+## The maximum bytes to retain whilst delaying content; or 0 for no delay; or -1 (default) for a default value.
+# jetty.delayed.maxRetainedContent=-1