09448 Implement intake with new framework

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

@@ -35,7 +35,7 @@
  *                                          because syncs that started before the throttle engages can grow the queue to
  *                                          very large sizes on larger networks.
  * @param randomEventProbability            The probability that after a sync, a node will create an event with a random
- *                                          other parent. The probability is is 1 in X, where X is the value of
+ *                                          other parent. The probability is 1 in X, where X is the value of
  *                                          randomEventProbability. A value of 0 means that a node will not create any
  *                                          random events.
  *                                          <p>
@@ -57,6 +57,7 @@
  * @param eventsLogDir                      eventStream files will be generated in this directory.
  * @param enableEventStreaming              enable stream event to server.
  * @param prehandlePoolSize                 the size of the thread pool used for prehandling transactions
+ * @param useLegacyIntake                   if true then use the legacy intake monolith, if false then use the new intake pipeline
  */
 @ConfigData("event")
 public record EventConfig(
@@ -70,4 +71,5 @@ public record EventConfig(
         @ConfigProperty(defaultValue = "5") long eventsLogPeriod,
         @ConfigProperty(defaultValue = "/opt/hgcapp/eventsStreams") String eventsLogDir,
         @ConfigProperty(defaultValue = "true") boolean enableEventStreaming,
-        @ConfigProperty(defaultValue = "8") int prehandlePoolSize) {}
+        @ConfigProperty(defaultValue = "8") int prehandlePoolSize,
+        @ConfigProperty(defaultValue = "false") boolean useLegacyIntake) {}

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

@@ -0,0 +1,107 @@
+/*
+ * Copyright (C) 2022-2023 Hedera Hashgraph, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.swirlds.common.config;
+
+import com.swirlds.common.wiring.schedulers.builders.TaskSchedulerType;
+import com.swirlds.config.api.ConfigData;
+import com.swirlds.config.api.ConfigProperty;
+
+/**
+ * Contains configuration values for the platform schedulers.
+ *
+ * @param internalEventValidatorSchedulerType      the internal event validator scheduler type
+ * @param internalEventValidatorUnhandledCapacity  number of unhandled events allowed in the internal event validator scheduler
+ * @param eventDeduplicatorSchedulerType           the event deduplicator scheduler type
+ * @param eventDeduplicatorUnhandledCapacity       number of unhandled tasks allowed in the event deduplicator scheduler
+ * @param eventSignatureValidatorSchedulerType     the event signature validator scheduler type
+ * @param eventSignatureValidatorUnhandledCapacity number of unhandled tasks allowed in the event signature validator scheduler
+ * @param orphanBufferSchedulerType                the orphan buffer scheduler type
+ * @param orphanBufferUnhandledCapacity            number of unhandled tasks allowed in the orphan buffer scheduler
+ * @param inOrderLinkerSchedulerType               the in-order linker scheduler type
+ * @param inOrderLinkerUnhandledCapacity           number of unhandled tasks allowed in the in-order linker scheduler
+ * @param linkedEventIntakeSchedulerType           the linked event intake scheduler type
+ * @param linkedEventIntakeUnhandledCapacity       number of unhandled tasks allowed in the linked event intake scheduler
+ */
+@ConfigData("platformSchedulers")
+public record PlatformSchedulersConfig(
+        @ConfigProperty(defaultValue = "SEQUENTIAL") String internalEventValidatorSchedulerType,
+        @ConfigProperty(defaultValue = "500") int internalEventValidatorUnhandledCapacity,
+        @ConfigProperty(defaultValue = "SEQUENTIAL") String eventDeduplicatorSchedulerType,
+        @ConfigProperty(defaultValue = "500") int eventDeduplicatorUnhandledCapacity,
+        @ConfigProperty(defaultValue = "SEQUENTIAL") String eventSignatureValidatorSchedulerType,
+        @ConfigProperty(defaultValue = "500") int eventSignatureValidatorUnhandledCapacity,
+        @ConfigProperty(defaultValue = "SEQUENTIAL") String orphanBufferSchedulerType,
+        @ConfigProperty(defaultValue = "500") int orphanBufferUnhandledCapacity,
+        @ConfigProperty(defaultValue = "SEQUENTIAL") String inOrderLinkerSchedulerType,
+        @ConfigProperty(defaultValue = "500") int inOrderLinkerUnhandledCapacity,
+        @ConfigProperty(defaultValue = "SEQUENTIAL") String linkedEventIntakeSchedulerType,
+        @ConfigProperty(defaultValue = "500") int linkedEventIntakeUnhandledCapacity) {
+
+    /**
+     * Get the internal event validator scheduler type
+     *
+     * @return the internal event validator scheduler type
+     */
+    public TaskSchedulerType getInternalEventValidatorSchedulerType() {
+        return TaskSchedulerType.valueOf(internalEventValidatorSchedulerType);
+    }
+
+    /**
+     * Get the event deduplicator scheduler type
+     *
+     * @return the event deduplicator scheduler type
+     */
+    public TaskSchedulerType getEventDeduplicatorSchedulerType() {
+        return TaskSchedulerType.valueOf(eventDeduplicatorSchedulerType);
+    }
+
+    /**
+     * Get the event signature validator scheduler type
+     *
+     * @return the event signature validator scheduler type
+     */
+    public TaskSchedulerType getEventSignatureValidatorSchedulerType() {
+        return TaskSchedulerType.valueOf(eventSignatureValidatorSchedulerType);
+    }
+
+    /**
+     * Get the orphan buffer scheduler type
+     *
+     * @return the orphan buffer scheduler type
+     */
+    public TaskSchedulerType getOrphanBufferSchedulerType() {
+        return TaskSchedulerType.valueOf(orphanBufferSchedulerType);
+    }
+
+    /**
+     * Get the in-order linker scheduler type
+     *
+     * @return the in-order linker scheduler type
+     */
+    public TaskSchedulerType getInOrderLinkerSchedulerType() {
+        return TaskSchedulerType.valueOf(inOrderLinkerSchedulerType);
+    }
+
+    /**
+     * Get the linked event intake scheduler type
+     *
+     * @return the linked event intake scheduler type
+     */
+    public TaskSchedulerType getLinkedEventIntakeSchedulerType() {
+        return TaskSchedulerType.valueOf(linkedEventIntakeSchedulerType);
+    }
+}

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/SwirldsPlatform.java

@@ -33,7 +33,6 @@
 import com.swirlds.base.state.Startable;
 import com.swirlds.base.time.Time;
 import com.swirlds.base.utility.Pair;
-import com.swirlds.common.config.BasicConfig;
 import com.swirlds.common.config.ConsensusConfig;
 import com.swirlds.common.config.EventConfig;
 import com.swirlds.common.config.StateConfig;
@@ -77,6 +76,7 @@
 import com.swirlds.common.threading.framework.QueueThread;
 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.ThreadManager;
 import com.swirlds.common.utility.AutoCloseableWrapper;
 import com.swirlds.common.utility.Clearable;
@@ -88,6 +88,7 @@
 import com.swirlds.logging.legacy.LogMarker;
 import com.swirlds.logging.legacy.payload.FatalErrorPayload;
 import com.swirlds.platform.components.EventIntake;
+import com.swirlds.platform.components.LinkedEventIntake;
 import com.swirlds.platform.components.SavedStateController;
 import com.swirlds.platform.components.appcomm.AppCommunicationComponent;
 import com.swirlds.platform.components.state.DefaultStateManagementComponent;
@@ -106,9 +107,12 @@
 import com.swirlds.platform.event.EventUtils;
 import com.swirlds.platform.event.GossipEvent;
 import com.swirlds.platform.event.creation.AsyncEventCreationManager;
+import com.swirlds.platform.event.deduplication.EventDeduplicator;
 import com.swirlds.platform.event.linking.EventLinker;
+import com.swirlds.platform.event.linking.InOrderLinker;
 import com.swirlds.platform.event.linking.OrphanBufferingLinker;
 import com.swirlds.platform.event.linking.ParentFinder;
+import com.swirlds.platform.event.orphan.OrphanBuffer;
 import com.swirlds.platform.event.preconsensus.AsyncPreconsensusEventWriter;
 import com.swirlds.platform.event.preconsensus.NoOpPreconsensusEventWriter;
 import com.swirlds.platform.event.preconsensus.PreconsensusEventFileManager;
@@ -117,11 +121,14 @@
 import com.swirlds.platform.event.preconsensus.PreconsensusEventStreamSequencer;
 import com.swirlds.platform.event.preconsensus.PreconsensusEventWriter;
 import com.swirlds.platform.event.preconsensus.SyncPreconsensusEventWriter;
+import com.swirlds.platform.event.validation.AddressBookUpdate;
 import com.swirlds.platform.event.validation.AncientValidator;
 import com.swirlds.platform.event.validation.EventDeduplication;
+import com.swirlds.platform.event.validation.EventSignatureValidator;
 import com.swirlds.platform.event.validation.EventValidator;
 import com.swirlds.platform.event.validation.GossipEventValidator;
 import com.swirlds.platform.event.validation.GossipEventValidators;
+import com.swirlds.platform.event.validation.InternalEventValidator;
 import com.swirlds.platform.event.validation.SignatureValidator;
 import com.swirlds.platform.event.validation.StaticValidators;
 import com.swirlds.platform.event.validation.TransactionSizeValidator;
@@ -173,6 +180,7 @@
 import com.swirlds.platform.system.Shutdown;
 import com.swirlds.platform.threading.PauseAndLoad;
 import com.swirlds.platform.util.PlatformComponents;
+import com.swirlds.platform.wiring.PlatformWiring;
 import com.swirlds.platform.wiring.SignedStateFileManagerWiring;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
@@ -232,9 +240,10 @@
     private final EventLinker eventLinker;
 
     /**
-     * Validates events and passes valid events further down the pipeline.
+     * Validates events and passes valid events further down the intake pipeline.
      */
-    private final EventValidator eventValidator;
+    private final InterruptableConsumer<GossipEvent> intakeHandler;
+
     /** Contains all validators for events */
     private final GossipEventValidators eventValidators;
 
@@ -323,6 +332,11 @@
     /** Manages emergency recovery */
     private final EmergencyRecoveryManager emergencyRecoveryManager;
 
+    /**
+     * Encapsulated wiring for the platform.
+     */
+    private final PlatformWiring platformWiring;
+
     /**
      * the browser gives the Platform what app to run. There can be multiple Platforms on one computer.
      *
@@ -666,8 +680,6 @@
                 preConsensusEventHandler::preconsensusEvent,
                 intakeEventCounter);
 
-        final BasicConfig basicConfig = platformContext.getConfiguration().getConfigData(BasicConfig.class);
-
         final List<GossipEventValidator> validators = new ArrayList<>();
         // it is very important to discard ancient events, otherwise the deduplication will not work, since it
         // doesn't track ancient events
@@ -685,14 +697,55 @@
                 time));
 
         eventValidators = new GossipEventValidators(validators);
-        eventValidator = new EventValidator(
+
+        final EventValidator eventValidator = new EventValidator(
                 eventValidators, eventIntake::addUnlinkedEvent, eventIntakePhaseTimer, intakeEventCounter);
 
+        if (eventConfig.useLegacyIntake()) {
+            intakeHandler = eventValidator::validateEvent;
+            platformWiring = null;
+        } else {
+            final InternalEventValidator internalEventValidator = new InternalEventValidator(
+                    platformContext, time, currentAddressBook.getSize() == 1, intakeEventCounter);
+            final EventDeduplicator eventDeduplicator =
+                    new EventDeduplicator(platformContext, intakeEventCounter, eventIntakeMetrics);
+            final EventSignatureValidator eventSignatureValidator = new EventSignatureValidator(
+                    platformContext,
+                    time,
+                    CryptoStatic::verifySignature,
+                    appVersion,
+                    initialState.getState().getPlatformState().getPreviousAddressBook(),
+                    currentAddressBook,
+                    intakeEventCounter);
+            final OrphanBuffer orphanBuffer = new OrphanBuffer(platformContext, intakeEventCounter);
+            final InOrderLinker inOrderLinker = new InOrderLinker(intakeEventCounter);
+            final LinkedEventIntake linkedEventIntake = new LinkedEventIntake(
+                    platformContext,
+                    threadManager,
+                    time,
+                    consensusRef::get,
+                    eventObserverDispatcher,
+                    shadowGraph,
+                    preConsensusEventHandler::preconsensusEvent,
+                    intakeEventCounter);
+
+            platformWiring = new PlatformWiring(platformContext, time);
+            platformWiring.bind(
+                    internalEventValidator,
+                    eventDeduplicator,
+                    eventSignatureValidator,
+                    orphanBuffer,
+                    inOrderLinker,
+                    linkedEventIntake);
+
+            intakeHandler = platformWiring.getEventInput();
+        }
+
         intakeQueue = components.add(new QueueThreadConfiguration<GossipEvent>(threadManager)
                 .setNodeId(selfId)
                 .setComponent(PLATFORM_THREAD_POOL_NAME)
                 .setThreadName("event-intake")
-                .setHandler(eventValidator::validateEvent)
+                .setHandler(intakeHandler)
                 .setCapacity(eventConfig.eventIntakeQueueSize())
                 .setLogAfterPauseDuration(threadConfig.logStackTracePauseDuration())
                 .setMetricsConfiguration(new QueueThreadMetricsConfiguration(metrics).enableMaxSizeMetric())
@@ -761,7 +814,12 @@
 
             loadStateIntoConsensus(initialState);
             loadStateIntoEventCreator(initialState);
-            eventLinker.loadFromSignedState(initialState);
+
+            if (eventConfig.useLegacyIntake()) {
+                eventLinker.loadFromSignedState(initialState);
+            } else {
+                platformWiring.updateMinimumGenerationNonAncient(initialMinimumGenerationNonAncient);
+            }
 
             // We don't want to invoke these callbacks until after we are starting up.
             final long round = initialState.getRound();
@@ -787,7 +845,8 @@
                         Pair.of(gossip, "gossip"),
                         Pair.of(preConsensusEventHandler, "preConsensusEventHandler"),
                         Pair.of(consensusRoundHandler, "consensusRoundHandler"),
-                        Pair.of(transactionPool, "transactionPool")));
+                        Pair.of(transactionPool, "transactionPool"),
+                        Pair.of(platformWiring, "platformWiring")));
 
         if (platformContext.getConfiguration().getConfigData(ThreadConfig.class).jvmAnchor()) {
             components.add(new JvmAnchor(threadManager));
@@ -1001,14 +1060,26 @@
             // from the ones we had before the reconnect
             intakeQueue.pause();
             try {
-                eventValidators.replaceValidator(
-                        SignatureValidator.VALIDATOR_NAME,
-                        new SignatureValidator(
-                                signedState.getState().getPlatformState().getPreviousAddressBook(),
-                                signedState.getState().getPlatformState().getAddressBook(),
-                                appVersion,
-                                CryptoStatic::verifySignature,
-                                Time.getCurrent()));
+                if (platformContext
+                        .getConfiguration()
+                        .getConfigData(EventConfig.class)
+                        .useLegacyIntake()) {
+                    eventValidators.replaceValidator(
+                            SignatureValidator.VALIDATOR_NAME,
+                            new SignatureValidator(
+                                    signedState.getState().getPlatformState().getPreviousAddressBook(),
+                                    signedState.getState().getPlatformState().getAddressBook(),
+                                    appVersion,
+                                    CryptoStatic::verifySignature,
+                                    Time.getCurrent()));
+                } else {
+                    platformWiring
+                            .getAddressBookUpdateInput()
+                            .accept(new AddressBookUpdate(
+                                    signedState.getState().getPlatformState().getPreviousAddressBook(),
+                                    signedState.getState().getPlatformState().getAddressBook()));
+                    platformWiring.updateMinimumGenerationNonAncient(signedState.getMinRoundGeneration());
+                }
             } finally {
                 intakeQueue.resume();
             }
@@ -1176,7 +1247,7 @@
                     Time.getCurrent(),
                     preconsensusEventFileManager,
                     preconsensusEventWriter,
-                    eventValidator,
+                    intakeHandler,
                     intakeQueue,
                     consensusRoundHandler,
                     stateHashSignQueue,

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/components/LinkedEventIntake.java

@@ -122,14 +122,16 @@
      * Add an event to the hashgraph
      *
      * @param event an event to be added
+     * @return a list of rounds that came to consensus as a result of adding the event
      */
-    public void addEvent(@NonNull final EventImpl event) {
+    @NonNull
+    public List<ConsensusRound> addEvent(@NonNull final EventImpl event) {
         Objects.requireNonNull(event);
 
         try {
             if (event.getGeneration() < consensusSupplier.get().getMinGenerationNonAncient()) {
                 // ancient events *may* be discarded, and stale events *must* be discarded
-                return;
+                return List.of();
             }
 
             dispatcher.preConsensusEvent(event);
@@ -153,6 +155,8 @@
                 // with no consensus events, so we check the diff in generations to look for stale events
                 handleStale(minGenNonAncientBeforeAdding);
             }
+
+            return Objects.requireNonNullElseGet(consensusRounds, List::of);
         } finally {
             intakeEventCounter.eventExitedIntakePipeline(event.getBaseEvent().getSenderId());
         }