Issue #9965 - make multiple websocket demand throw ISE

Signed-off-by: Lachlan Roberts <lachlan@webtide.com>
jetty · Jun 27, 2023 · 0709946 · 0709946
1 parent 2b4e896
commit 0709946
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 19 deletions.
diff --git a/...ocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/WebSocketConnection.java b/...ocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/WebSocketConnection.java
@@ -17,6 +17,7 @@
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
+import java.nio.channels.ReadPendingException;
 import java.security.SecureRandom;
 import java.util.Objects;
 import java.util.Random;
@@ -31,7 +32,6 @@
 import org.eclipse.jetty.io.RetainableByteBuffer;
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.Callback;
-import org.eclipse.jetty.util.MathUtils;
 import org.eclipse.jetty.util.component.Dumpable;
 import org.eclipse.jetty.util.thread.AutoLock;
 import org.eclipse.jetty.util.thread.Scheduler;
@@ -61,7 +61,7 @@ public class WebSocketConnection extends AbstractConnection implements Connectio
     private final WebSocketCoreSession coreSession;
     private final Flusher flusher;
     private final Random random;
-    private long demand;
+    private Boolean demand;
     private boolean fillingAndParsing;
     private final LongAdder messagesIn = new LongAdder();
     private final LongAdder bytesIn = new LongAdder();
@@ -351,10 +351,12 @@ public void demand()
             if (LOG.isDebugEnabled())
                 LOG.debug("demand {} d={} fp={} {}", demand, fillingAndParsing, networkBuffer, this);
 
-            if (demand < 0)
-                return;
-
-            demand = MathUtils.cappedAdd(demand, 1);
+            if (demand != null)
+            {
+                if (demand)
+                    throw new ReadPendingException();
+                demand = true;
+            }
 
             if (!fillingAndParsing)
             {
@@ -379,13 +381,12 @@ public boolean moreDemand()
 
             if (!fillingAndParsing)
                 throw new IllegalStateException();
-            if (demand != 0) //if demand was canceled, this creates synthetic demand in order to read until EOF
+            if (demand == null) // If demand was canceled, this creates synthetic demand in order to read until EOF.
                 return true;
 
             fillingAndParsing = false;
             if (!networkBuffer.hasRemaining())
                 releaseNetworkBuffer();
-
             return false;
         }
     }
@@ -397,14 +398,13 @@ public boolean meetDemand()
             if (LOG.isDebugEnabled())
                 LOG.debug("meetDemand d={} fp={} {} {}", demand, fillingAndParsing, networkBuffer, this);
 
-            if (demand == 0)
+            if (demand == Boolean.FALSE)
                 throw new IllegalStateException();
             if (!fillingAndParsing)
                 throw new IllegalStateException();
 
-            if (demand > 0)
-                demand--;
-
+            if (demand != null)
+                demand = false;
             return true;
         }
     }
@@ -415,7 +415,7 @@ public void cancelDemand()
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("cancelDemand d={} fp={} {} {}", demand, fillingAndParsing, networkBuffer, this);
-            demand = -1;
+            demand = null;
         }
     }
 

diff --git a/...ket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/DemandingFlusher.java b/...ket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/DemandingFlusher.java
@@ -13,7 +13,8 @@
 
 package org.eclipse.jetty.websocket.core.util;
 
-import java.util.concurrent.atomic.AtomicLong;
+import java.nio.channels.ReadPendingException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.eclipse.jetty.util.Callback;
@@ -41,7 +42,7 @@ public abstract class DemandingFlusher extends IteratingCallback implements Dema
     private static final Throwable SENTINEL_CLOSE_EXCEPTION = new StaticException("Closed");
 
     private final IncomingFrames _emitFrame;
-    private final AtomicLong _demand = new AtomicLong();
+    private final AtomicBoolean _demand = new AtomicBoolean();
     private final AtomicReference<Throwable> _failure = new AtomicReference<>();
     private DemandChain _nextDemand;
 
@@ -77,7 +78,8 @@ public DemandingFlusher(IncomingFrames emitFrame)
     @Override
     public void demand()
     {
-        _demand.incrementAndGet();
+        if (!_demand.compareAndSet(false, true))
+            throw new ReadPendingException();
         iterate();
     }
 
@@ -139,8 +141,8 @@ public void failFlusher(Throwable t)
      */
     public void emitFrame(Frame frame, Callback callback)
     {
-        if (_demand.decrementAndGet() < 0)
-            throw new IllegalStateException("Negative Demand");
+        if (_demand.compareAndSet(true, false))
+            throw new IllegalStateException("Demand Already Fulfilled");
         _emitFrame.onFrame(frame, callback);
     }
 
@@ -153,7 +155,7 @@ protected Action process() throws Throwable
             if (failure != null)
                 throw failure;
 
-            if (_demand.get() <= 0)
+            if (!_demand.get())
                 break;
 
             if (_needContent)