Merge pull request #40619 from mkouba/ws-next-conlim-perf-01

WebSockets Next: improve the concurrency limiter
quarkusio · May 14, 2024 · 7f8a039 · 7f8a039
2 parents 7a871cf + 6cb225a
commit 7f8a039
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 26 deletions.
diff --git a/...ets-next/runtime/src/main/java/io/quarkus/websockets/next/runtime/ConcurrencyLimiter.java b/...ets-next/runtime/src/main/java/io/quarkus/websockets/next/runtime/ConcurrencyLimiter.java
@@ -25,7 +25,8 @@ class ConcurrencyLimiter {
     ConcurrencyLimiter(WebSocketConnectionBase connection) {
         this.connection = connection;
         this.uncompleted = new AtomicLong();
-        this.queueCounter = new AtomicLong();
+        // Counter is only used for debugging
+        this.queueCounter = LOG.isDebugEnabled() ? new AtomicLong() : null;
         this.queue = Queues.createMpscQueue();
     }
 
@@ -51,7 +52,7 @@ void run(Context context, Runnable action) {
             LOG.debugf("Run action: %s", connection);
             action.run();
         } else {
-            long queueIndex = queueCounter.incrementAndGet();
+            long queueIndex = queueCounter != null ? queueCounter.incrementAndGet() : 0l;
             LOG.debugf("Action queued as %s: %s", queueIndex, connection);
             queue.offer(new Action(queueIndex, action, context));
             // We need to make sure that at least one completion is in flight
@@ -76,38 +77,31 @@ void failure(Throwable t) {
             try {
                 promise.fail(t);
             } finally {
-                if (uncompleted.decrementAndGet() == 0) {
-                    return;
-                }
-                Action queuedAction = queue.poll();
-                assert queuedAction != null;
-                LOG.debugf("Run action %s from queue: %s", queuedAction.queueIndex, connection);
-                queuedAction.context.runOnContext(new Handler<Void>() {
-                    @Override
-                    public void handle(Void event) {
-                        queuedAction.runnable.run();
-                    }
-                });
+                tryNext();
             }
         }
 
         void complete() {
             try {
                 promise.complete();
             } finally {
-                if (uncompleted.decrementAndGet() == 0) {
-                    return;
-                }
-                Action queuedAction = queue.poll();
-                assert queuedAction != null;
-                LOG.debugf("Run action %s from queue: %s", queuedAction.queueIndex, connection);
-                queuedAction.context.runOnContext(new Handler<Void>() {
-                    @Override
-                    public void handle(Void event) {
-                        queuedAction.runnable.run();
-                    }
-                });
+                tryNext();
+            }
+        }
+
+        private void tryNext() {
+            if (uncompleted.decrementAndGet() == 0) {
+                return;
             }
+            Action queuedAction = queue.poll();
+            assert queuedAction != null;
+            LOG.debugf("Run action %s from queue: %s", queuedAction.queueIndex, connection);
+            queuedAction.context.runOnContext(new Handler<Void>() {
+                @Override
+                public void handle(Void event) {
+                    queuedAction.runnable.run();
+                }
+            });
         }
     }
 

diff --git a/...ext/runtime/src/main/java/io/quarkus/websockets/next/runtime/WebSocketConnectionBase.java b/...ext/runtime/src/main/java/io/quarkus/websockets/next/runtime/WebSocketConnectionBase.java
@@ -93,6 +93,10 @@ public Uni<Void> close() {
     }
 
     public Uni<Void> close(CloseReason reason) {
+        if (isClosed()) {
+            LOG.warnf("Connection already closed: %s", this);
+            return Uni.createFrom().voidItem();
+        }
         return UniHelper.toUni(webSocket().close((short) reason.getCode(), reason.getMessage()));
     }