Issue #6728 - QUIC and HTTP/3

- Improvements to the thread model implementation.

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>

jetty-http3/http3-client/src/main/java/org/eclipse/jetty/http3/client/internal/ClientHTTP3Session.java

@@ -127,21 +127,21 @@ public class ClientHTTP3Session extends ClientProtocolSession
     }
 
     @Override
-    protected void onReadable(long readableStreamId)
+    protected boolean onReadable(long readableStreamId)
     {
         StreamType streamType = StreamType.from(readableStreamId);
         if (streamType == StreamType.CLIENT_BIDIRECTIONAL)
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("bidirectional stream #{} selected for read", readableStreamId);
-            super.onReadable(readableStreamId);
+            return super.onReadable(readableStreamId);
         }
         else
         {
             QuicStreamEndPoint streamEndPoint = getOrCreateStreamEndPoint(readableStreamId, this::configureUnidirectionalStreamEndPoint);
             if (LOG.isDebugEnabled())
                 LOG.debug("unidirectional stream #{} selected for read: {}", readableStreamId, streamEndPoint);
-            streamEndPoint.onReadable();
+            return streamEndPoint.onReadable();
         }
     }
 

jetty-http3/http3-server/src/main/java/org/eclipse/jetty/http3/server/internal/HttpChannelOverHTTP3.java

@@ -39,6 +39,7 @@ import org.slf4j.LoggerFactory;
 public class HttpChannelOverHTTP3 extends HttpChannel
 {
     private static final Logger LOG = LoggerFactory.getLogger(HttpChannelOverHTTP3.class);
+    private static final HttpInput.Content NULL_CONTENT = new NullContent();
 
     private static final HttpField SERVER_VERSION = new PreEncodedHttpField(HttpHeader.SERVER, HttpConfiguration.SERVER_VERSION);
     private static final HttpField POWERED_BY = new PreEncodedHttpField(HttpHeader.X_POWERED_BY, HttpConfiguration.SERVER_VERSION);
@@ -214,6 +215,9 @@ public class HttpChannelOverHTTP3 extends HttpChannel
     @Override
     public boolean needContent()
     {
+        if (content == NULL_CONTENT)
+            content = null;
+
         if (content != null)
             return true;
 
@@ -226,6 +230,8 @@ public class HttpChannelOverHTTP3 extends HttpChannel
     {
         if (content != null)
         {
+            if (content == NULL_CONTENT)
+                return null;
             HttpInput.Content result = content;
             if (!result.isSpecial())
                 content = null;
@@ -238,7 +244,10 @@ public class HttpChannelOverHTTP3 extends HttpChannel
         if (LOG.isDebugEnabled())
             LOG.debug("read {} on {}", data, this);
         if (data == null)
+        {
+            content = NULL_CONTENT;
             return null;
+        }
 
         content = new HttpInput.Content(data.getByteBuffer())
         {
@@ -296,4 +305,8 @@ public class HttpChannelOverHTTP3 extends HttpChannel
     {
         return false;
     }
+
+    private static class NullContent extends HttpInput.SpecialContent
+    {
+    }
 }

jetty-http3/http3-server/src/main/java/org/eclipse/jetty/http3/server/internal/ServerHTTP3Session.java

@@ -127,21 +127,21 @@ public class ServerHTTP3Session extends ServerProtocolSession
     }
 
     @Override
-    protected void onReadable(long readableStreamId)
+    protected boolean onReadable(long readableStreamId)
     {
         StreamType streamType = StreamType.from(readableStreamId);
         if (streamType == StreamType.CLIENT_BIDIRECTIONAL)
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("bidirectional stream #{} selected for read", readableStreamId);
-            super.onReadable(readableStreamId);
+            return super.onReadable(readableStreamId);
         }
         else
         {
             QuicStreamEndPoint streamEndPoint = getOrCreateStreamEndPoint(readableStreamId, this::configureUnidirectionalStreamEndPoint);
             if (LOG.isDebugEnabled())
                 LOG.debug("unidirectional stream #{} selected for read: {}", readableStreamId, streamEndPoint);
-            streamEndPoint.onReadable();
+            return streamEndPoint.onReadable();
         }
     }
 

jetty-quic/quic-client/src/main/java/org/eclipse/jetty/quic/client/ClientProtocolSession.java

@@ -50,14 +50,15 @@ public class ClientProtocolSession extends ProtocolSession
     }
 
     @Override
-    protected void onReadable(long readableStreamId)
+    protected boolean onReadable(long readableStreamId)
     {
         // On the client, we need a get-only semantic in case of reads.
         QuicStreamEndPoint streamEndPoint = getStreamEndPoint(readableStreamId);
         if (LOG.isDebugEnabled())
             LOG.debug("stream #{} selected for read: {}", readableStreamId, streamEndPoint);
         if (streamEndPoint != null)
-            streamEndPoint.onReadable();
+            return streamEndPoint.onReadable();
+        return false;
     }
 
     @Override

jetty-quic/quic-common/src/main/java/org/eclipse/jetty/quic/common/ProtocolSession.java

@@ -51,6 +51,8 @@ public abstract class ProtocolSession extends ContainerLifeCycle
 
     public void process()
     {
+        if (LOG.isDebugEnabled())
+            LOG.debug("processing {}", this);
         strategy.produce();
     }
 
@@ -87,15 +89,17 @@ public abstract class ProtocolSession extends ContainerLifeCycle
             streamEndPoint.onWritable();
     }
 
-    protected void processReadableStreams()
+    protected boolean processReadableStreams()
     {
         List<Long> readableStreamIds = session.getReadableStreamIds();
         if (LOG.isDebugEnabled())
             LOG.debug("readable stream ids: {}", readableStreamIds);
-        readableStreamIds.forEach(this::onReadable);
+        return readableStreamIds.stream()
+            .map(this::onReadable)
+            .reduce(false, (result, readable) -> result || readable);
     }
 
-    protected abstract void onReadable(long readableStreamId);
+    protected abstract boolean onReadable(long readableStreamId);
 
     public void configureProtocolEndPoint(QuicStreamEndPoint endPoint)
     {
@@ -167,18 +171,24 @@ public abstract class ProtocolSession extends ContainerLifeCycle
         {
             Runnable task = poll();
             if (LOG.isDebugEnabled())
-                LOG.debug("dequeued task {} on {}", task, ProtocolSession.this);
+                LOG.debug("dequeued existing task {} on {}", task, ProtocolSession.this);
             if (task != null)
                 return task;
 
-            processWritableStreams();
+            while (true)
-            processReadableStreams();
+            {
+                processWritableStreams();
+                boolean loop = processReadableStreams();
 
-            task = poll();
+                task = poll();
-            if (LOG.isDebugEnabled())
+                if (LOG.isDebugEnabled())
-                LOG.debug("dequeued produced task {} on {}", task, ProtocolSession.this);
+                    LOG.debug("dequeued produced task {} on {}", task, ProtocolSession.this);
-            if (task != null)
+                if (task != null)
-                return task;
+                    return task;
+
+                if (!loop)
+                    break;
+            }
 
             CloseInfo closeInfo = session.getRemoteCloseInfo();
             if (closeInfo != null)

jetty-quic/quic-common/src/main/java/org/eclipse/jetty/quic/common/QuicConnection.java

@@ -173,7 +173,11 @@ public abstract class QuicConnection extends AbstractConnection
         try
         {
             if (isFillInterested())
+            {
+                if (LOG.isDebugEnabled())
+                    LOG.debug("receiveAndProcess() idle");
                 return null;
+            }
 
             ByteBuffer cipherBuffer = byteBufferPool.acquire(getInputBufferSize(), isUseInputDirectByteBuffers());
             while (true)
@@ -253,7 +257,7 @@ public abstract class QuicConnection extends AbstractConnection
         catch (Throwable x)
         {
             if (LOG.isDebugEnabled())
-                LOG.debug("exception in receiveAndProcess()", x);
+                LOG.debug("receiveAndProcess() failure", x);
             // TODO: close?
             return null;
         }

jetty-quic/quic-common/src/main/java/org/eclipse/jetty/quic/common/QuicSession.java

@@ -349,6 +349,7 @@ public abstract class QuicSession extends ContainerLifeCycle
         processing.set(false);
     }
 
+    // TODO: this is ugly, is there a better solution?
     protected Runnable pollTask()
     {
         return null;

jetty-quic/quic-common/src/main/java/org/eclipse/jetty/quic/common/QuicStreamEndPoint.java

@@ -17,7 +17,6 @@ import java.io.IOException;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.IntStream;
 
 import org.eclipse.jetty.io.AbstractEndPoint;
@@ -41,7 +40,6 @@ public class QuicStreamEndPoint extends AbstractEndPoint
     private static final Logger LOG = LoggerFactory.getLogger(QuicStreamEndPoint.class);
     private static final ByteBuffer LAST_FLAG = ByteBuffer.allocate(0);
 
-    private final AtomicBoolean readable = new AtomicBoolean(true);
     private final QuicSession session;
     private final long streamId;
 
@@ -211,27 +209,32 @@ public class QuicStreamEndPoint extends AbstractEndPoint
         getWriteFlusher().completeWrite();
     }
 
-    public void onReadable()
+    /**
+     * @return whether this endPoint is interested in reads
+     */
+    public boolean onReadable()
     {
-        boolean expected = readable.compareAndExchange(true, false);
+        boolean interested = isFillInterested();
         if (LOG.isDebugEnabled())
-            LOG.debug("stream {} is readable, processing: {}", streamId, expected);
+            LOG.debug("stream {} is readable, processing: {}", streamId, interested);
-        if (expected)
+        if (interested)
             getFillInterest().fillable();
+        return interested;
     }
 
     @Override
     public void fillInterested(Callback callback)
     {
-        readable.set(true);
         super.fillInterested(callback);
+        getQuicSession().getProtocolSession().process();
     }
 
     @Override
     public boolean tryFillInterested(Callback callback)
     {
-        readable.set(true);
+        boolean result = super.tryFillInterested(callback);
-        return super.tryFillInterested(callback);
+        getQuicSession().getProtocolSession().process();
+        return result;
     }
 
     @Override

jetty-quic/quic-server/src/main/java/org/eclipse/jetty/quic/server/ServerProtocolSession.java

@@ -34,13 +34,13 @@ public class ServerProtocolSession extends ProtocolSession
     }
 
     @Override
-    protected void onReadable(long readableStreamId)
+    protected boolean onReadable(long readableStreamId)
     {
         // On the server, we need a get-or-create semantic in case of reads.
         QuicStreamEndPoint streamEndPoint = getOrCreateStreamEndPoint(readableStreamId, this::configureProtocolEndPoint);
         if (LOG.isDebugEnabled())
             LOG.debug("stream #{} selected for read: {}", readableStreamId, streamEndPoint);
-        streamEndPoint.onReadable();
+        return streamEndPoint.onReadable();
     }
 
     @Override