Cancel write semantic

jetty-core/jetty-fcgi/jetty-fcgi-client/src/main/java/org/eclipse/jetty/fcgi/generator/Flusher.java

@@ -13,15 +13,19 @@
 
 package org.eclipse.jetty.fcgi.generator;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Queue;
+import java.util.concurrent.CountDownLatch;
 
 import org.eclipse.jetty.io.ByteBufferPool;
 import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.io.RuntimeIOException;
 import org.eclipse.jetty.util.Callback;
+import org.eclipse.jetty.util.ExceptionUtil;
 import org.eclipse.jetty.util.IteratingCallback;
 import org.eclipse.jetty.util.thread.AutoLock;
 import org.slf4j.Logger;
@@ -33,14 +37,39 @@ public class Flusher
 
     private final AutoLock lock = new AutoLock();
     private final Queue<Entry> queue = new ArrayDeque<>();
-    private final IteratingCallback flushCallback = new FlushCallback();
+    private final FlushCallback flushCallback = new FlushCallback();
     private final EndPoint endPoint;
 
     public Flusher(EndPoint endPoint)
     {
         this.endPoint = endPoint;
     }
 
+    public Callback cancel(Throwable cause)
+    {
+        // Cancel the IteratingCallback and take the nest Callback
+        CancelSendException cancelSendException = new CancelSendException(cause);
+        if (!flushCallback.abort(cancelSendException))
+            return Callback.NOOP;
+
+        // We now know that we aborted this ICB with the CSE above, so onAbort will eventually be called
+        // in a serialized context and the callback will be set on the CSE.
+
+        // If a write operation has been scheduled cancel it and fail its callback, otherwise complete ourselves
+        Callback writeCallback = endPoint.cancelWrite(cause);
+
+        if (writeCallback == null)
+            // There was no write in operation, so we can complete the CSE ourselves
+            cancelSendException.complete();
+        else
+            // The write was cancelled and the callback is this ICB, so failing it will call onCompleted
+            writeCallback.failed(cause);
+
+        // wait for the cancellation to be complete and the callback to be set by onAbort.
+        // This should never block indefinitely, as onAborted only waits for active states like PROCESSING to complete.
+        return cancelSendException.join();
+    }
+
     public void flush(ByteBufferPool.Accumulator accumulator, Callback callback)
     {
         offer(new Entry(accumulator, callback));
@@ -73,6 +102,44 @@ public void shutdown()
         }));
     }
 
+    private static class CancelSendException extends IOException
+    {
+        private final CountDownLatch _complete = new CountDownLatch(2);
+        private Callback _callback;
+
+        public CancelSendException(Throwable cause)
+        {
+            super(cause);
+        }
+
+        public void complete()
+        {
+            _complete.countDown();
+        }
+
+        public Callback join()
+        {
+            try
+            {
+                _complete.await();
+            }
+            catch (InterruptedException x)
+            {
+                Throwable cause = getCause();
+                ExceptionUtil.addSuppressedIfNotAssociated(cause, x);
+                throw new RuntimeIOException(cause);
+            }
+
+            return _callback;
+        }
+
+        public void setCallback(Callback callback)
+        {
+            _callback = callback;
+            _complete.countDown();
+        }
+    }
+
     private class FlushCallback extends IteratingCallback
     {
         private Entry active;
@@ -94,6 +161,28 @@ protected Action process() throws Exception
             return Action.SCHEDULED;
         }
 
+        @Override
+        protected void onAborted(Throwable cause)
+        {
+            if (cause instanceof CancelSendException cancelSend)
+                cancelSend.setCallback(resetCallback());
+        }
+
+        private Callback resetCallback()
+        {
+            Flusher.Entry entry = active;
+            active = null;
+            return Callback.from(entry.callback, entry::release);
+        }
+
+        @Override
+        protected void onCompleted(Throwable causeOrNull)
+        {
+            if (causeOrNull instanceof CancelSendException cancelSendException)
+                cancelSendException.complete();
+            super.onCompleted(causeOrNull);
+        }
+
         @Override
         protected void onCompleteSuccess()
         {

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

@@ -250,6 +250,12 @@ public void send(MetaData.Request request, MetaData.Response response, boolean l
         }
     }
 
+    @Override
+    public Runnable cancelSend(Throwable cause, Callback appCallback)
+    {
+        return () -> Callback.combine(_connection.getFlusher().cancel(cause), appCallback).failed(cause);
+    }
+
     private void commit(MetaData.Response info, boolean head, boolean last, ByteBuffer content, Callback callback)
     {
         if (LOG.isDebugEnabled())

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

@@ -736,6 +736,25 @@ void reset(HTTP2Stream stream, ResetFrame frame, Callback callback)
         }, callback), frame);
     }
 
+    public void flush(Callback callback)
+    {
+        Entry entry = new Entry(new FlushFrame(), null, callback)
+        {
+            @Override
+            public int getFrameBytesGenerated()
+            {
+                return 0;
+            }
+
+            @Override
+            public boolean generate(RetainableByteBuffer.Mutable accumulator)
+            {
+                return true;
+            }
+        };
+        frame(entry, true);
+    }
+
     /**
      * <p>Invoked internally and by applications to send a GO_AWAY frame to the other peer.</p>
      *
@@ -1336,6 +1355,8 @@ public boolean shouldBeDropped()
                 case WINDOW_UPDATE:
                 case PREFACE:
                 case DISCONNECT:
+                case FAILURE:
+                case FLUSH:
                     return false;
                 // Frames of this type follow the logic below.
                 case DATA:
@@ -2518,4 +2539,12 @@ protected boolean onExpired(HTTP2Stream stream)
             return false;
         }
     }
+
+    private static class FlushFrame extends Frame
+    {
+        public FlushFrame()
+        {
+            super(FrameType.FLUSH);
+        }
+    }
 }

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

@@ -166,7 +166,7 @@ public void reset(ResetFrame frame, Callback callback)
         Throwable resetFailure = null;
         try (AutoLock ignored = lock.lock())
         {
-            if (isReset())
+            if (localReset)
             {
                 resetFailure = failure;
             }
@@ -180,6 +180,8 @@ public void reset(ResetFrame frame, Callback callback)
         session.dataConsumed(this, flowControlLength);
         if (resetFailure != null)
         {
+            if (LOG.isDebugEnabled())
+                LOG.debug("failing callback immediately as stream {} already is locally reset", this, resetFailure);
             close();
             session.removeStream(this);
             callback.failed(resetFailure);