Fixes retainability of special Chunks

documentation/jetty-documentation/src/main/java/org/eclipse/jetty/docs/programming/client/http/HTTPClientDocs.java

@@ -265,11 +265,7 @@ public void listeners() throws Exception
             .onResponseBegin(response -> { /* ... */ })
             .onResponseHeader((response, field) -> true)
             .onResponseHeaders(response -> { /* ... */ })
-            .onResponseContentAsync((response, chunk, demander) ->
-            {
-                chunk.release();
-                demander.run();
-            })
+            .onResponseContentAsync((response, chunk, demander) -> demander.run())
             .onResponseFailure((response, failure) -> { /* ... */ })
             .onResponseSuccess(response -> { /* ... */ })
             // Result hook.
@@ -505,12 +501,12 @@ private void forwardContent(Response response, Content.Source contentSource)
                     // nor demanded again until the demand callback is invoked.
                     return;
                 }
-                // Check if the chunk is the terminal one, in which case the
+                // Check if the chunk is last and empty, in which case the
                 // read/demand loop is done. Demanding again when the terminal
                 // chunk has been read will invoke the demand callback with
                 // the same terminal chunk, so this check must be present to
                 // avoid infinitely demanding and reading the terminal chunk.
-                if (chunk.isTerminal())
+                if (chunk.isLast() && !chunk.hasRemaining())
                 {
                     chunk.release();
                     return;

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/InputStreamResponseListener.java

@@ -102,31 +102,27 @@ public void onContent(Response response, Content.Chunk chunk, Runnable demander)
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("Skipped empty chunk {}", chunk);
-            chunk.release();
             demander.run();
             return;
         }
 
-        boolean closed;
         try (AutoLock.WithCondition l = lock.lock())
         {
-            closed = this.closed;
             if (!closed)
             {
                 if (LOG.isDebugEnabled())
                     LOG.debug("Queueing chunk {}", chunk);
+                if (chunk.canRetain())
+                    chunk.retain();
                 chunkCallbacks.add(new ChunkCallback(chunk, demander, response::abort));
                 l.signalAll();
+                return;
             }
         }
 
-        if (closed)
-        {
-            if (LOG.isDebugEnabled())
-                LOG.debug("InputStream closed, ignored chunk {}", chunk);
-            chunk.release();
-            response.abort(new AsynchronousCloseException());
-        }
+        if (LOG.isDebugEnabled())
+            LOG.debug("InputStream closed, ignored chunk {}", chunk);
+        response.abort(new AsynchronousCloseException());
     }
 
     @Override

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/Response.java

@@ -155,12 +155,10 @@ default void onContent(Response response, Content.Chunk chunk, Runnable demander
             try
             {
                 onContent(response, chunk.getByteBuffer());
-                chunk.release();
                 demander.run();
             }
             catch (Throwable x)
             {
-                chunk.release();
                 response.abort(x);
             }
         }
@@ -190,10 +188,22 @@ default void onContentSource(Response response, Content.Source contentSource)
                 response.abort(error.getCause());
                 return;
             }
-            if (chunk.isTerminal())
+            if (chunk.isLast() && !chunk.hasRemaining())
+            {
+                chunk.release();
                 return;
+            }
 
-            onContent(response, chunk, () -> contentSource.demand(demandCallback));
+            try
+            {
+                onContent(response, chunk, () -> contentSource.demand(demandCallback));
+                chunk.release();
+            }
+            catch (Throwable x)
+            {
+                chunk.release();
+                response.abort(x);
+            }
         }
     }
 

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/internal/HttpReceiver.java

@@ -574,7 +574,7 @@ protected Content.Chunk transform(Content.Chunk inputChunk)
                     return _chunk;
 
                 // Retain the input chunk because its ByteBuffer will be referenced by the Inflater.
-                if (retain && _chunk.hasRemaining())
+                if (retain && _chunk.canRetain())
                     _chunk.retain();
                 if (LOG.isDebugEnabled())
                     LOG.debug("decoding: {}", _chunk);

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/internal/ResponseNotifier.java

@@ -308,6 +308,7 @@ private void onDemandCallback()
             {
                 demultiplexerContentSource.onChunk(chunk);
             }
+            chunk.release();
         }
 
         private void registerFailure(Throwable failure)
@@ -368,6 +369,12 @@ public boolean isLast()
                     throw new UnsupportedOperationException();
                 }
 
+                @Override
+                public boolean canRetain()
+                {
+                    throw new UnsupportedOperationException();
+                }
+
                 @Override
                 public void retain()
                 {
@@ -383,7 +390,7 @@ public boolean release()
                 @Override
                 public String toString()
                 {
-                    return "ALREADY_READ_CHUNK";
+                    return "AlreadyReadChunk";
                 }
             };
             private final int index;
@@ -401,9 +408,18 @@ private void onChunk(Content.Chunk chunk)
                 if (LOG.isDebugEnabled())
                     LOG.debug("Registering content in multiplexed content source #{} that contains {}", index, currentChunk);
                 if (currentChunk == null || currentChunk == ALREADY_READ_CHUNK)
-                    this.chunk = chunk.slice();
+                {
+                    if (chunk.hasRemaining())
+                        chunk = Content.Chunk.asChunk(chunk.getByteBuffer().slice(), chunk.isLast(), chunk);
+                    // Retain the slice because it is stored for later reads.
+                    if (chunk.canRetain())
+                        chunk.retain();
+                    this.chunk = chunk;
+                }
                 else if (!currentChunk.isLast())
+                {
                     throw new IllegalStateException("Cannot overwrite chunk");
+                }
                 onDemandCallback();
             }
 
@@ -436,8 +452,8 @@ public Content.Chunk read()
                 }
 
                 Content.Chunk result = chunk;
-                if (result != null && !result.isTerminal())
-                    chunk = ALREADY_READ_CHUNK;
+                if (result != null)
+                    chunk = result.isLast() ? Content.Chunk.next(result) : ALREADY_READ_CHUNK;
                 if (LOG.isDebugEnabled())
                     LOG.debug("Content source #{} reading current chunk {}", index, result);
                 return result;

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/transport/internal/HttpReceiverOverHTTP.java

@@ -450,9 +450,10 @@ public boolean content(ByteBuffer buffer)
         if (chunk != null)
             throw new IllegalStateException("Content generated with unconsumed content left");
 
-        RetainableByteBuffer networkBuffer = this.networkBuffer;
+        // Retain the chunk because it is stored for later use.
         networkBuffer.retain();
-        chunk = Content.Chunk.from(buffer, false, networkBuffer);
+        chunk = Content.Chunk.asChunk(buffer, false, networkBuffer);
+
         if (LOG.isDebugEnabled())
             LOG.debug("Setting action to responseContentAvailable on {}", this);
         if (getAndSetAction(this::responseContentAvailable) != null)

jetty-core/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientAsyncContentTest.java

@@ -63,7 +63,6 @@ protected void service(Request request, org.eclipse.jetty.server.Response respon
             .scheme(scenario.getScheme())
             .onResponseContentAsync((response, chunk, demander) ->
             {
-                chunk.release();
                 contentCount.incrementAndGet();
                 demanderRef.set(demander);
                 contentLatch.get().countDown();

jetty-core/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientChunkedContentTest.java

@@ -115,7 +115,6 @@ public void testServerContentTerminalClientContentDelay() throws Exception
             client.newRequest("localhost", server.getLocalPort())
                 .onResponseContentAsync((response, chunk, demander) ->
                 {
-                    chunk.release();
                     if (demanderRef.compareAndSet(null, demander))
                         firstContentLatch.countDown();
                     else

jetty-core/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientTest.java

@@ -633,8 +633,7 @@ protected void service(org.eclipse.jetty.server.Request request, org.eclipse.jet
 
         AsyncRequestContent body = new AsyncRequestContent();
         body.write(false, BufferUtil.allocate(512), Callback.NOOP);
-        body.write(false, BufferUtil.allocate(512), Callback.NOOP);
-        body.write(Content.Chunk.from(new IOException("explicitly_thrown_by_test")), Callback.NOOP);
+        body.write(false, BufferUtil.allocate(512), Callback.from(() -> body.fail(new IOException("explicitly_thrown_by_test"))));
         CountDownLatch latch = new CountDownLatch(1);
         client.newRequest("localhost", connector.getLocalPort())
             .scheme(scenario.getScheme())
@@ -1412,7 +1411,6 @@ public boolean process(org.eclipse.jetty.server.Request request, org.eclipse.jet
                 @Override
                 public void onContent(Response response, Content.Chunk chunk, Runnable demander)
                 {
-                    chunk.release();
                     // Do not notify the callback yet.
                     demanderRef.set(demander);
                     contentLatch.countDown();

jetty-core/jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientUploadDuringServerShutdownTest.java

@@ -71,8 +71,7 @@ protected void service(Request request, Response response) throws Throwable
                         }
                         else
                         {
-                            if (chunk.hasRemaining())
-                                chunk.release();
+                            chunk.release();
                             if (chunk.isLast())
                                 break;
                             NanoTime.spinWait(TimeUnit.MICROSECONDS.toNanos(1));

jetty-core/jetty-client/src/test/java/org/eclipse/jetty/client/util/AsyncRequestContentTest.java

@@ -86,6 +86,7 @@ public void testWriteFlushDemandFlushBlocksUntilRead() throws Exception
 
         Content.Chunk chunk = content.read();
         assertNotNull(chunk);
+        chunk.release();
 
         // Flush should return.
         assertTrue(await(task, 5000));

jetty-core/jetty-fcgi/jetty-fcgi-client/src/main/java/org/eclipse/jetty/fcgi/client/transport/internal/HttpConnectionOverFCGI.java

@@ -484,11 +484,13 @@ public boolean onContent(int request, FCGI.StreamType stream, ByteBuffer buffer)
                     HttpChannelOverFCGI channel = HttpConnectionOverFCGI.this.channel;
                     if (channel != null)
                     {
-                        networkBuffer.retain();
-                        Content.Chunk chunk = Content.Chunk.from(buffer, false, networkBuffer);
-                        if (getAndSetAction(() -> channel.content(chunk)) != null)
-                            throw new IllegalStateException();
-                        return true;
+                        // No need to call networkBuffer.retain() here, since we know
+                        // that the action will be run before releasing the networkBuffer.
+                        // The receiver of the chunk decides whether to consume/retain it.
+                        Content.Chunk chunk = Content.Chunk.asChunk(buffer, false, networkBuffer);
+                        if (getAndSetAction(() -> channel.content(chunk)) == null)
+                            return true;
+                        throw new IllegalStateException();
                     }
                     else
                     {

jetty-core/jetty-fcgi/jetty-fcgi-client/src/main/java/org/eclipse/jetty/fcgi/client/transport/internal/HttpReceiverOverFCGI.java

@@ -98,6 +98,9 @@ void content(Content.Chunk chunk)
     {
         if (this.chunk != null)
             throw new IllegalStateException();
+        // Retain the chunk because it is stored for later reads.
+        if (chunk.canRetain())
+            chunk.retain();
         this.chunk = chunk;
         responseContentAvailable();
     }

jetty-core/jetty-fcgi/jetty-fcgi-proxy/src/test/java/org/eclipse/jetty/fcgi/proxy/FastCGIProxyHandlerTest.java

@@ -172,10 +172,6 @@ public boolean process(Request request, Response response, Callback callback)
                 {
                     response.abort(x);
                 }
-                finally
-                {
-                    chunk.release();
-                }
             })
             .path(proxyContext.getContextPath() + path);
         FutureResponseListener listener = new FutureResponseListener(request, length);

jetty-core/jetty-fcgi/jetty-fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/internal/HttpStreamOverFCGI.java

@@ -191,6 +191,9 @@ private void notifyContentAvailable()
 
     public void onContent(Content.Chunk chunk)
     {
+        // Retain the chunk because it is stored for later reads.
+        if (chunk.canRetain())
+            chunk.retain();
         _chunk = chunk;
     }
 

jetty-core/jetty-fcgi/jetty-fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/internal/ServerFCGIConnection.java

@@ -371,8 +371,10 @@ public boolean onContent(int request, FCGI.StreamType streamType, ByteBuffer buf
                 LOG.debug("Request {} {} content {} on {}", request, streamType, buffer, stream);
             if (stream != null)
             {
-                networkBuffer.retain();
-                stream.onContent(Content.Chunk.from(buffer, false, networkBuffer));
+                // No need to call networkBuffer.retain() here.
+                // The receiver of the chunk decides whether to consume/retain it.
+                Content.Chunk chunk = Content.Chunk.asChunk(buffer, false, networkBuffer);
+                stream.onContent(chunk);
                 // Signal that the content is processed asynchronously, to ensure backpressure.
                 return true;
             }

jetty-core/jetty-fcgi/jetty-fcgi-server/src/test/java/org/eclipse/jetty/fcgi/server/HttpClientTest.java

@@ -651,7 +651,6 @@ public boolean process(org.eclipse.jetty.server.Request request, org.eclipse.jet
             .scheme(scheme)
             .onResponseContentAsync((response, chunk, demander) ->
             {
-                chunk.release();
                 contentCount.incrementAndGet();
                 demanderRef.set(demander);
                 contentLatch.get().countDown();

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

@@ -222,9 +222,9 @@ else if ((_flags & 0x2) == 0x2)
 
                             if (buffer.hasRemaining())
                             {
-                                ByteBuffer chunk = buffer;
+                                ByteBuffer decoded = buffer;
                                 buffer = null;
-                                if (decodedChunk(chunk))
+                                if (decodedChunk(decoded))
                                     return;
                             }
                             else if (_inflater.needsInput())