Experiment for aborting callbacks

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

@@ -545,6 +545,10 @@ protected Action process() throws Throwable
         @Override
         protected void onSuccess()
         {
+            // TODO This logic should be moved to process() and/or onCompleteSuccess()
+            //      Anything executed here is not mutually excluded from other threads in process() and/or onCompleteSuccess() and/or onCompleteFailure()
+            //      nor is there a memory barrier.
+            //      So, for example, two threads might try to release and null the chunk field at the same time.
             boolean proceed = true;
             if (committed)
             {
@@ -590,6 +594,17 @@ else if (expect100)
             }
         }
 
+        @Override
+        protected void onAborted(Throwable cause)
+        {
+            // TODO Review this
+            internalAbort(exchange, cause);
+            Promise<Boolean> promise = abort;
+            abort = null;
+            if (promise != null)
+                promise.succeeded(true);
+        }
+
         @Override
         protected void onCompleteFailure(Throwable x)
         {
@@ -603,6 +618,7 @@ protected void onCompleteFailure(Throwable x)
             internalAbort(exchange, x);
 
             Promise<Boolean> promise = abort;
+            abort = null;
             if (promise != null)
                 promise.succeeded(true);
         }

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

@@ -1236,7 +1236,7 @@ public String toString()
         );
     }
 
-    public abstract static class Entry extends Callback.Nested
+    public abstract static class Entry extends Callback.Wrapper
     {
         protected final Frame frame;
         protected final HTTP2Stream stream;
@@ -2009,7 +2009,14 @@ private void onWriteFailure(Throwable x)
 
         private void sendGoAwayAndTerminate(GoAwayFrame frame, GoAwayFrame eventFrame)
         {
-            sendGoAway(frame, Callback.from(() -> terminate(eventFrame)));
+            sendGoAway(frame, Callback.from(
+                t ->
+                {
+                    terminate(eventFrame);
+                    return true;
+                },
+                () -> terminate(eventFrame),
+                t -> terminate(eventFrame)));
         }
 
         private void sendGoAway(GoAwayFrame frame, Callback callback)

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

@@ -335,6 +335,27 @@ protected void onCompleteSuccess()
         throw new IllegalStateException();
     }
 
+    @Override
+    protected void onAborted(Throwable cause)
+    {
+        Set<HTTP2Session.Entry> abortEntries;
+        Set<HTTP2Session.Entry> failEntries;
+        try (AutoLock ignored = lock.lock())
+        {
+            if (LOG.isDebugEnabled())
+                LOG.debug("aborted {} {}", cause, this);
+
+            // We can fail non-pending entries, as they are not being written yet
+            failEntries = new HashSet<>(entries);
+            entries.clear();
+
+            // We can only abort pending entries
+            abortEntries = new HashSet<>(pendingEntries);
+        }
+        abortEntries.forEach(entry -> entry.abort(cause));
+        failEntries.forEach(entry -> entry.failed(cause));
+    }
+
     @Override
     protected void onCompleteFailure(Throwable x)
     {

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

@@ -642,7 +642,7 @@ public void failed(Throwable x)
         _stream.reset(new ResetFrame(_stream.getId(), errorCode.code), Callback.NOOP);
     }
 
-    private class SendTrailers extends Callback.Nested
+    private class SendTrailers extends Callback.Wrapper
     {
         private final HttpFields trailers;
 

jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/content/BufferedContentSink.java

@@ -138,7 +138,7 @@ else if (BufferUtil.hasContent(currentBuffer))
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("flushing aggregated buffer {}", aggregatedBuffer);
-            _flusher.offer(false, aggregatedBuffer.getByteBuffer(), new Callback.Nested(Callback.from(aggregatedBuffer::release))
+            _flusher.offer(false, aggregatedBuffer.getByteBuffer(), new Callback.Wrapper(Callback.from(aggregatedBuffer::release))
             {
                 @Override
                 public void succeeded()
@@ -200,7 +200,7 @@ else if (flush)
 
             if (BufferUtil.hasContent(currentBuffer))
             {
-                _flusher.offer(false, aggregatedBuffer.getByteBuffer(), new Callback.Nested(Callback.from(aggregatedBuffer::release))
+                _flusher.offer(false, aggregatedBuffer.getByteBuffer(), new Callback.Wrapper(Callback.from(aggregatedBuffer::release))
                 {
                     @Override
                     public void succeeded()

jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/HttpStream.java

@@ -221,6 +221,12 @@ public void succeeded()
             getWrapped().succeeded();
         }
 
+        @Override
+        public boolean abort(Throwable cause)
+        {
+            return getWrapped().abort(cause);
+        }
+
         @Override
         public void failed(Throwable x)
         {

jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/ConnectHandler.java

@@ -726,7 +726,11 @@ private class ProxyIteratingCallback extends IteratingCallback
             @Override
             protected Action process()
             {
-                buffer = bufferPool.acquire(getInputBufferSize(), true);
+                if (buffer == null)
+                    buffer = bufferPool.acquire(getInputBufferSize(), true);
+                else
+                    buffer.clear();
+
                 try
                 {
                     ByteBuffer byteBuffer = buffer.getByteBuffer();
@@ -736,18 +740,18 @@ protected Action process()
                         write(connection.getEndPoint(), byteBuffer, this);
                         return Action.SCHEDULED;
                     }
-                    else if (filled == 0)
+
+                    buffer.release();
+                    buffer = null;
+
+                    if (filled == 0)
                     {
-                        buffer.release();
                         fillInterested();
                         return Action.IDLE;
                     }
-                    else
-                    {
-                        buffer.release();
-                        connection.getEndPoint().shutdownOutput();
-                        return Action.SUCCEEDED;
-                    }
+
+                    connection.getEndPoint().shutdownOutput();
+                    return Action.SUCCEEDED;
                 }
                 catch (IOException x)
                 {
@@ -768,12 +772,19 @@ protected void onSuccess()
             }
 
             @Override
-            protected void onCompleteFailure(Throwable x)
+            protected void onAborted(Throwable cause)
+            {
+                disconnect(cause);
+            }
+
+            @Override
+            protected void onCompleteFailure(Throwable cause)
             {
                 if (LOG.isDebugEnabled())
-                    LOG.debug("Failed to write {} bytes {}", filled, TunnelConnection.this, x);
+                    LOG.debug("Failed to write {} bytes {}", filled, TunnelConnection.this, cause);
                 buffer.release();
-                disconnect(x);
+                buffer = null;
+                disconnect(cause);
             }
 
             private void disconnect(Throwable x)

jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/ContextResponse.java

@@ -18,7 +18,6 @@
 import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Response;
 import org.eclipse.jetty.util.Callback;
-import org.eclipse.jetty.util.thread.Invocable;
 
 public class ContextResponse extends Response.Wrapper
 {
@@ -33,24 +32,24 @@ public ContextResponse(ContextHandler.ScopedContext context, Request request, Re
     @Override
     public void write(boolean last, ByteBuffer content, Callback callback)
     {
-        Callback contextCallback = new Callback()
+        Callback contextCallback = new Callback.Wrapper(callback)
         {
             @Override
-            public void succeeded()
+            public boolean abort(Throwable cause)
             {
-                _context.run(callback::succeeded, getRequest());
+                return _context.test(callback::abort, cause, getRequest());
             }
 
             @Override
-            public void failed(Throwable x)
+            public void failed(Throwable cause)
             {
-                _context.accept(callback::failed, x, getRequest());
+                _context.accept(callback::failed, cause, getRequest());
             }
 
             @Override
-            public InvocationType getInvocationType()
+            public void succeeded()
             {
-                return Invocable.getInvocationType(callback);
+                _context.run(callback::succeeded, getRequest());;
             }
         };
         super.write(last, content, contextCallback);

jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/GracefulHandler.java

@@ -21,7 +21,6 @@
 import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Response;
 import org.eclipse.jetty.util.Callback;
-import org.eclipse.jetty.util.CountingCallback;
 import org.eclipse.jetty.util.annotation.ManagedAttribute;
 import org.eclipse.jetty.util.component.Graceful;
 import org.slf4j.Logger;
@@ -100,7 +99,11 @@ public boolean handle(Request request, Response response, Callback callback) thr
         {
             boolean handled = super.handle(request, response, shutdownCallback);
             if (!handled)
-                shutdownCallback.completed();
+            {
+                _requests.decrement();
+                if (isShutdown())
+                    _shutdown.check();
+            }
             return handled;
         }
         catch (Throwable t)
@@ -123,25 +126,26 @@ public CompletableFuture<Void> shutdown()
         return _shutdown.shutdown();
     }
 
-    private class ShutdownTrackingCallback extends CountingCallback
+    private class ShutdownTrackingCallback extends Callback.Nested
     {
         final Request request;
         final Response response;
 
         public ShutdownTrackingCallback(Request request, Response response, Callback callback)
         {
-            super(callback, 1);
+            super(callback);
             this.request = request;
             this.response = response;
             _requests.increment();
         }
 
         @Override
-        public void completed()
+        protected void onCompleted(Throwable cause)
         {
             _requests.decrement();
             if (isShutdown())
                 _shutdown.check();
+            super.onCompleted(cause);
         }
     }
 }

jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/StateTrackingHandler.java

@@ -582,7 +582,7 @@ public void dump(Appendable out, String indent) throws IOException
         }
     }
 
-    private class HandlerCallback extends Callback.Nested implements Runnable, Dumpable
+    private class HandlerCallback extends Callback.Wrapper implements Runnable, Dumpable
     {
         private final AtomicBoolean completed = new AtomicBoolean();
         private final Request request;
@@ -803,7 +803,7 @@ public void write(boolean last, ByteBuffer byteBuffer, Callback callback)
             }
         }
 
-        private class WriteCallback extends Callback.Nested implements Dumpable
+        private class WriteCallback extends Callback.Wrapper implements Dumpable
         {
             private final AtomicBoolean callbackCompleted = new AtomicBoolean();
             private final Thread writeThread;

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

@@ -1160,7 +1160,7 @@ private Runnable lockedFailWrite(Throwable x)
                 _writeFailure = x;
             else
                 ExceptionUtil.addSuppressedIfNotAssociated(_writeFailure, x);
-            return () -> HttpChannelState.failed(writeCallback, x);
+            return () -> ExceptionUtil.call(x, writeCallback::abort);
         }
 
         public long getContentBytesWritten()
@@ -1275,7 +1275,7 @@ else if (last && !(totalWritten == 0 && HttpMethod.HEAD.is(_request.getMethod())
                 if (writeFailure != null)
                 {
                     Throwable failure = writeFailure;
-                    httpChannelState._writeInvoker.run(() -> HttpChannelState.failed(callback, failure));
+                    httpChannelState._writeInvoker.run(() -> ExceptionUtil.call(failure, callback::failed));
                     return;
                 }
 
@@ -1343,7 +1343,7 @@ public void failed(Throwable x)
                 httpChannel.lockedStreamSendCompleted(false);
             }
             if (callback != null)
-                httpChannel._writeInvoker.run(() -> HttpChannelState.failed(callback, x));
+                httpChannel._writeInvoker.run(() -> ExceptionUtil.call(x, callback::failed));
         }
 
         @Override
@@ -1460,7 +1460,7 @@ public String toString()
         }
     }
 
-    private static class ChannelCallback implements Callback
+    private static class ChannelCallback extends Callback.Abstract
     {
         private final ChannelRequest _request;
         private Throwable _completedBy;
@@ -1474,7 +1474,7 @@ private ChannelCallback(ChannelRequest request)
          * Called when the {@link Handler} (or it's delegates) succeeds the request handling.
          */
         @Override
-        public void succeeded()
+        public void onSucceeded()
         {
             // Called when the request/response cycle is completing successfully.
             HttpStream stream;
@@ -1556,7 +1556,7 @@ else if (LOG.isDebugEnabled())
          * @param failure The reason for the failure.
          */
         @Override
-        public void failed(Throwable failure)
+        public void onFailed(Throwable failure)
         {
             try
             {
@@ -1736,7 +1736,7 @@ public void succeeded()
             }
             else
             {
-                HttpChannelState.failed(httpChannelState._handlerInvoker, failure);
+                ExceptionUtil.call(failure, ((Callback)httpChannelState._handlerInvoker)::failed);
             }
         }
 
@@ -1759,7 +1759,7 @@ public void failed(Throwable x)
                 httpChannelState._response._status = _errorResponse._status;
             }
             ExceptionUtil.addSuppressedIfNotAssociated(failure, x);
-            HttpChannelState.failed(httpChannelState._handlerInvoker, failure);
+            ExceptionUtil.call(failure, ((Callback)httpChannelState._handlerInvoker)::failed);
         }
 
         @Override
@@ -1928,25 +1928,4 @@ public void onComplianceViolation(ComplianceViolation.Event event)
             }
         }
     }
-
-    /**
-     * Invoke a callback failure, handling any {@link Throwable} thrown
-     * by adding the passed {@code failure} as a suppressed with
-     * {@link ExceptionUtil#addSuppressedIfNotAssociated(Throwable, Throwable)}.
-     * @param callback The callback to fail
-     * @param failure The failure
-     * @throws RuntimeException If thrown, will have the {@code failure} added as a suppressed.
-     */
-    private static void failed(Callback callback, Throwable failure)
-    {
-        try
-        {
-            callback.failed(failure);
-        }
-        catch (Throwable t)
-        {
-            ExceptionUtil.addSuppressedIfNotAssociated(t, failure);
-            throw t;
-        }
-    }
 }