make CAS less aggressive, more fair, as per Sam's recommendation

Author: marregui

hivemq-edge/src/main/java/com/hivemq/fsm/ProtocolAdapterFSM.java

@@ -25,6 +25,7 @@
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
 import java.util.function.Consumer;
 
 public abstract class ProtocolAdapterFSM implements Consumer<ProtocolAdapterState.ConnectionStatus> {
@@ -120,15 +121,24 @@ public void stopAdapter() {
     }
 
     public boolean transitionAdapterState(final @NotNull AdapterStateEnum newState) {
+        int retryCount = 0;
         while (true) {
             final var currentState = adapterState.get();
             if (canTransition(currentState, newState)) {
                 if (adapterState.compareAndSet(currentState, newState)) {
+                    final State snapshotState = new State(newState, northboundState.get(), southboundState.get());
                     log.debug("Adapter state transition from {} to {} for adapter {}", currentState, newState, adapterId);
-                    notifyListenersAboutStateTransition(currentState());
+                    notifyListenersAboutStateTransition(snapshotState);
                     return true;
                 }
-                // CAS failed due to concurrent modification, retry
+                retryCount++;
+                if (retryCount > 3) {
+                    // progressive backoff: 1μs, 2μs, 4μs, 8μs, ..., capped at 100μs
+                    // reduces CPU consumption and cache line contention under high load
+                    final long backoffNanos = Math.min(1_000L * (1L << (retryCount - 4)), 100_000L);
+                    LockSupport.parkNanos(backoffNanos);
+                }
+                // Fast retry for attempts 1-3 (optimizes for low contention case)
             } else {
                 // Transition not allowed from current state
                 throw new IllegalStateException("Cannot transition adapter state to " + newState);
@@ -137,15 +147,23 @@ public boolean transitionAdapterState(final @NotNull AdapterStateEnum newState)
     }
 
     public boolean transitionNorthboundState(final @NotNull StateEnum newState) {
+        int retryCount = 0;
         while (true) {
             final var currentState = northboundState.get();
             if (canTransition(currentState, newState)) {
                 if (northboundState.compareAndSet(currentState, newState)) {
+                    final State snapshotState = new State(adapterState.get(), newState, southboundState.get());
                     log.debug("Northbound state transition from {} to {} for adapter {}", currentState, newState, adapterId);
-                    notifyListenersAboutStateTransition(currentState());
+                    notifyListenersAboutStateTransition(snapshotState);
                     return true;
                 }
-                // CAS failed due to concurrent modification, retry
+                retryCount++;
+                if (retryCount > 3) {
+                    // progressive backoff: 1μs, 2μs, 4μs, 8μs, ..., capped at 100μs
+                    final long backoffNanos = Math.min(1_000L * (1L << (retryCount - 4)), 100_000L);
+                    LockSupport.parkNanos(backoffNanos);
+                }
+                // Fast retry for attempts 1-3 (optimizes for low contention case)
             } else {
                 // Transition not allowed from current state
                 throw new IllegalStateException("Cannot transition northbound state to " + newState);
@@ -154,15 +172,23 @@ public boolean transitionNorthboundState(final @NotNull StateEnum newState) {
     }
 
     public boolean transitionSouthboundState(final @NotNull StateEnum newState) {
+        int retryCount = 0;
         while (true) {
             final var currentState = southboundState.get();
             if (canTransition(currentState, newState)) {
                 if (southboundState.compareAndSet(currentState, newState)) {
+                    final State snapshotState = new State(adapterState.get(), northboundState.get(), newState);
                     log.debug("Southbound state transition from {} to {} for adapter {}", currentState, newState, adapterId);
-                    notifyListenersAboutStateTransition(currentState());
+                    notifyListenersAboutStateTransition(snapshotState);
                     return true;
                 }
-                // CAS failed due to concurrent modification, retry
+                retryCount++;
+                if (retryCount > 3) {
+                    // progressive backoff: 1μs, 2μs, 4μs, 8μs, ..., capped at 100μs
+                    final long backoffNanos = Math.min(1_000L * (1L << (retryCount - 4)), 100_000L);
+                    LockSupport.parkNanos(backoffNanos);
+                }
+                // Fast retry for attempts 1-3 (optimizes for low contention case)
             } else {
                 // Transition not allowed from current state
                 throw new IllegalStateException("Cannot transition southbound state to " + newState);

hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterWrapper.java

@@ -84,6 +84,8 @@ public class ProtocolAdapterWrapper extends ProtocolAdapterFSM {
     private @Nullable CompletableFuture<Void> currentStartFuture;
     private @Nullable CompletableFuture<Void> currentStopFuture;
     private @Nullable Consumer<ProtocolAdapterState.ConnectionStatus> connectionStatusListener;
+    private volatile boolean startOperationInProgress;
+    private volatile boolean stopOperationInProgress;
 
     public ProtocolAdapterWrapper(
             final @NotNull ProtocolAdapterMetricsService protocolAdapterMetricsService,
@@ -164,22 +166,22 @@ public boolean startSouthbound() {
     public @NotNull CompletableFuture<Void> startAsync(final @NotNull ModuleServices moduleServices) {
         operationLock.lock();
         try {
-            // validate state
-            if (currentStartFuture != null && !currentStartFuture.isDone()) {
+            if (startOperationInProgress) {
                 log.info("Start operation already in progress for adapter '{}'", getId());
                 return currentStartFuture;
             }
             if (adapterState.get() == AdapterStateEnum.STARTED) {
                 log.info("Adapter '{}' is already started, returning success", getId());
                 return CompletableFuture.completedFuture(null);
             }
-            if (currentStopFuture != null && !currentStopFuture.isDone()) {
+            if (stopOperationInProgress) {
                 log.warn("Cannot start adapter '{}' while stop operation is in progress", getId());
                 return CompletableFuture.failedFuture(new IllegalStateException("Cannot start adapter '" +
                         adapter.getId() +
                         "' while stop operation is in progress"));
             }
 
+            startOperationInProgress = true;
             lastStartAttemptTime = System.currentTimeMillis();
             currentStartFuture =
                     CompletableFuture.supplyAsync(startProtocolAdapter(moduleServices), sharedAdapterExecutor)
@@ -196,6 +198,7 @@ public boolean startSouthbound() {
                                 }
                                 operationLock.lock();
                                 try {
+                                    startOperationInProgress = false;
                                     currentStartFuture = null;
                                 } finally {
                                     operationLock.unlock();
@@ -210,8 +213,7 @@ public boolean startSouthbound() {
     public @NotNull CompletableFuture<Void> stopAsync() {
         operationLock.lock();
         try {
-            // validate state
-            if (currentStopFuture != null && !currentStopFuture.isDone()) {
+            if (stopOperationInProgress) {
                 log.info("Stop operation already in progress for adapter '{}'", getId());
                 return currentStopFuture;
             }
@@ -220,13 +222,14 @@ public boolean startSouthbound() {
                 log.info("Adapter '{}' is already stopped, returning success", getId());
                 return CompletableFuture.completedFuture(null);
             }
-            if (currentStartFuture != null && !currentStartFuture.isDone()) {
+            if (startOperationInProgress) {
                 log.warn("Cannot stop adapter '{}' while start operation is in progress", getId());
                 return CompletableFuture.failedFuture(new IllegalStateException("Cannot stop adapter '" +
                         adapter.getId() +
                         "' while start operation is in progress"));
             }
 
+            stopOperationInProgress = true;
             log.debug("Adapter '{}': Creating stop operation future", getId());
             currentStopFuture = CompletableFuture.supplyAsync(this::stopProtocolAdapter, sharedAdapterExecutor)
                     .thenCompose(Function.identity())
@@ -244,6 +247,7 @@ public boolean startSouthbound() {
                         }
                         operationLock.lock();
                         try {
+                            stopOperationInProgress = false;
                             currentStopFuture = null;
                         } finally {
                             operationLock.unlock();
@@ -342,9 +346,10 @@ public boolean isBatchPolling() {
     }
 
     private void cleanupConnectionStatusListener() {
-        if (connectionStatusListener != null) {
-            protocolAdapterState.setConnectionStatusListener(CONNECTION_STATUS_NOOP_CONSUMER);
+        final Consumer<ProtocolAdapterState.ConnectionStatus> listenerToClean = connectionStatusListener;
+        if (listenerToClean != null) {
             connectionStatusListener = null;
+            protocolAdapterState.setConnectionStatusListener(CONNECTION_STATUS_NOOP_CONSUMER);
         }
     }