chore: add an old-style queue thread for intake (#11671)

Signed-off-by: Cody Littley <[email protected]>
hashgraph · Feb 26, 2024 · 08372c4 · 08372c4
1 parent 191fdc4
commit 08372c4
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 4 deletions.
diff --git a/platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/SwirldsPlatform.java b/platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/SwirldsPlatform.java
@@ -50,6 +50,7 @@
 import com.swirlds.common.threading.framework.config.QueueThreadConfiguration;
 import com.swirlds.common.threading.framework.config.QueueThreadMetricsConfiguration;
 import com.swirlds.common.threading.interrupt.InterruptableConsumer;
+import com.swirlds.common.threading.manager.AdHocThreadManager;
 import com.swirlds.common.threading.manager.ThreadManager;
 import com.swirlds.common.utility.AutoCloseableWrapper;
 import com.swirlds.common.utility.Clearable;
@@ -509,6 +510,16 @@ public class SwirldsPlatform implements Platform {
 
         platformWiring = components.add(new PlatformWiring(platformContext, time));
 
+        final QueueThread<GossipEvent> oldStyleIntakeQueue = new QueueThreadConfiguration<GossipEvent>(
+                        AdHocThreadManager.getStaticThreadManager())
+                .setCapacity(10_000)
+                .setThreadName("old_style_intake_queue")
+                .setComponent("platform")
+                .setMetricsConfiguration(new QueueThreadMetricsConfiguration(metrics).enableMaxSizeMetric())
+                .setHandler(event -> platformWiring.getGossipEventInput().put(event))
+                .build();
+        components.add(oldStyleIntakeQueue);
+
         savedStateController = new SavedStateController(stateConfig);
 
         final SignedStateMetrics signedStateMetrics = new SignedStateMetrics(platformContext.getMetrics());
@@ -664,7 +675,7 @@ public class SwirldsPlatform implements Platform {
                 selfId,
                 appVersion,
                 transactionPool,
-                platformWiring.getIntakeQueueSizeSupplier(),
+                oldStyleIntakeQueue::size,
                 platformStatusManager::getCurrentStatus,
                 latestReconnectRound::get);
 
@@ -724,8 +735,16 @@ public class SwirldsPlatform implements Platform {
                 shadowGraph,
                 emergencyRecoveryManager,
                 consensusRef,
-                platformWiring.getGossipEventInput()::put,
-                platformWiring.getIntakeQueueSizeSupplier(),
+                event -> {
+                    try {
+                        oldStyleIntakeQueue.put(event);
+                    } catch (final InterruptedException e) {
+                        logger.error(
+                                EXCEPTION.getMarker(), "Interrupted while adding event to old style intake queue", e);
+                        Thread.currentThread().interrupt();
+                    }
+                },
+                oldStyleIntakeQueue::size,
                 swirldStateManager,
                 latestCompleteState,
                 syncMetrics,

diff --git a/...lds-platform-core/src/main/java/com/swirlds/platform/wiring/PlatformSchedulersConfig.java b/...lds-platform-core/src/main/java/com/swirlds/platform/wiring/PlatformSchedulersConfig.java
@@ -86,7 +86,7 @@
 public record PlatformSchedulersConfig(
         @ConfigProperty(defaultValue = "1.0") double defaultPoolMultiplier,
         @ConfigProperty(defaultValue = "0") int defaultPoolConstant,
-        @ConfigProperty(defaultValue = "10000") int eventHasherUnhandledCapacity,
+        @ConfigProperty(defaultValue = "500") int eventHasherUnhandledCapacity,
         @ConfigProperty(defaultValue = "SEQUENTIAL") TaskSchedulerType internalEventValidatorSchedulerType,
         @ConfigProperty(defaultValue = "500") int internalEventValidatorUnhandledCapacity,
         @ConfigProperty(defaultValue = "SEQUENTIAL") TaskSchedulerType eventDeduplicatorSchedulerType,