13486 tentative backpressure fix

Signed-off-by: Oleg Mazurov <[email protected]>

platform-sdk/swirlds-common/src/main/java/com/swirlds/common/config/BasicCommonConfig.java

@@ -18,6 +18,7 @@
 
 import com.swirlds.config.api.ConfigData;
 import com.swirlds.config.api.ConfigProperty;
+import java.time.Duration;
 
 /**
  * Basic configuration data record. This record contains all general config properties that can not be defined for a
@@ -36,4 +37,5 @@
 @ConfigData
 public record BasicCommonConfig(
         @ConfigProperty(defaultValue = "true") boolean showInternalStats,
-        @ConfigProperty(defaultValue = "false") boolean verboseStatistics) {}
+        @ConfigProperty(defaultValue = "false") boolean verboseStatistics,
+        @ConfigProperty(defaultValue = "1ms") Duration backPressureSleepInterval) {}

platform-sdk/swirlds-common/src/main/java/com/swirlds/common/wiring/counters/BackpressureBlocker.java

@@ -90,15 +90,15 @@ public boolean block() throws InterruptedException {
      */
     @Override
     public boolean isReleasable() {
-        final long resultingCount = count.incrementAndGet();
-        if (resultingCount <= capacity) {
-            // We didn't violate capacity by incrementing the count, so we're done.
-            return true;
-        } else {
-            // We may have violated capacity restrictions by incrementing the count.
-            // Decrement count and take the slow pathway.
-            count.decrementAndGet();
-            return false;
+        for (; ; ) {
+            final long resultingCount = count.get();
+            if (resultingCount < capacity) {
+                if (count.compareAndSet(resultingCount, resultingCount + 1)) {
+                    return true;
+                }
+            } else {
+                return false;
+            }
         }
     }
 }

platform-sdk/swirlds-common/src/main/java/com/swirlds/common/wiring/counters/BackpressureObjectCounter.java

@@ -77,16 +77,16 @@ public BackpressureObjectCounter(
      */
     @Override
     public void onRamp() {
-        final long resultingCount = count.incrementAndGet();
-        if (resultingCount <= capacity) {
-            // We didn't violate capacity by incrementing the count, so we're done.
-            return;
-        } else {
-            // We may have violated capacity restrictions by incrementing the count.
-            // Decrement count and take the slow pathway.
-            count.decrementAndGet();
+        for (; ; ) {
+            final long resultingCount = count.get();
+            if (resultingCount < capacity) {
+                if (count.compareAndSet(resultingCount, resultingCount + 1)) {
+                    return;
+                }
+            } else {
+                break;
+            }
         }
-
         // Slow case. Capacity wasn't reserved, so we need to block.
 
         while (true) {

platform-sdk/swirlds-common/src/main/java/com/swirlds/common/wiring/schedulers/builders/internal/AbstractTaskSchedulerBuilder.java

@@ -21,6 +21,7 @@
 import static com.swirlds.common.wiring.schedulers.builders.TaskSchedulerType.NO_OP;
 import static com.swirlds.logging.legacy.LogMarker.EXCEPTION;
 
+import com.swirlds.common.config.BasicCommonConfig;
 import com.swirlds.common.context.PlatformContext;
 import com.swirlds.common.wiring.counters.BackpressureObjectCounter;
 import com.swirlds.common.wiring.counters.MultiObjectCounter;
@@ -68,7 +69,7 @@ public abstract class AbstractTaskSchedulerBuilder<OUT> implements TaskScheduler
     protected boolean unhandledTaskMetricEnabled = false;
     protected boolean busyFractionMetricEnabled = false;
 
-    protected Duration sleepDuration = Duration.ofNanos(100);
+    protected Duration sleepDuration;
 
     protected final PlatformContext platformContext;
 
@@ -101,6 +102,10 @@ public AbstractTaskSchedulerBuilder(
         }
         this.name = name;
         this.pool = Objects.requireNonNull(defaultPool);
+        sleepDuration = platformContext
+                .getConfiguration()
+                .getConfigData(BasicCommonConfig.class)
+                .backPressureSleepInterval();
     }
 
     /**