Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Use legacy intake pipeline by default #10120

Merged
merged 4 commits into from
Nov 28, 2023
Merged

Use legacy intake pipeline by default #10120

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2c75963
Use legacy intake pipeline by default
litt3 Nov 27, 2023
0c06739
Merge branch 'develop' into 10119-use-legacy-intake
litt3 Nov 27, 2023
1f8687d
Fix NPE
litt3 Nov 27, 2023
68e0574
Move where platform wiring is instantiated
litt3 Nov 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion platform-sdk/swirlds-common/src/main/java/com/swirlds/common/config/EventConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,4 +72,4 @@ public record EventConfig(
@ConfigProperty(defaultValue = "/opt/hgcapp/eventsStreams") String eventsLogDir,
@ConfigProperty(defaultValue = "true") boolean enableEventStreaming,
@ConfigProperty(defaultValue = "8") int prehandlePoolSize,
@ConfigProperty(defaultValue = "false") boolean useLegacyIntake) {}
@ConfigProperty(defaultValue = "true") boolean useLegacyIntake) {}
3 changes: 1 addition & 2 deletions platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/SwirldsPlatform.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -699,9 +699,9 @@ public class SwirldsPlatform implements Platform {
final EventValidator eventValidator = new EventValidator(
eventValidators, eventIntake::addUnlinkedEvent, eventIntakePhaseTimer, intakeEventCounter);

platformWiring = new PlatformWiring(platformContext, time);
if (eventConfig.useLegacyIntake()) {
intakeHandler = eventValidator::validateEvent;
platformWiring = null;
} else {
final InternalEventValidator internalEventValidator = new InternalEventValidator(
platformContext, time, currentAddressBook.getSize() == 1, intakeEventCounter);
Expand All @@ -727,7 +727,6 @@ public class SwirldsPlatform implements Platform {
preConsensusEventHandler::preconsensusEvent,
intakeEventCounter);

platformWiring = new PlatformWiring(platformContext, time);
platformWiring.bind(
internalEventValidator,
eventDeduplicator,
Expand Down
20 changes: 13 additions & 7 deletions ...m-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/wiring/PlatformWiring.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import com.swirlds.base.state.Startable;
import com.swirlds.base.state.Stoppable;
import com.swirlds.base.time.Time;
import com.swirlds.common.config.EventConfig;
import com.swirlds.common.context.PlatformContext;
import com.swirlds.common.threading.interrupt.InterruptableConsumer;
import com.swirlds.common.utility.Clearable;
Expand All @@ -35,12 +36,14 @@
import com.swirlds.platform.event.validation.EventSignatureValidator;
import com.swirlds.platform.event.validation.InternalEventValidator;
import edu.umd.cs.findbugs.annotations.NonNull;
import java.util.Objects;
import java.util.function.Consumer;

/**
* Encapsulates wiring for {@link com.swirlds.platform.SwirldsPlatform}.
*/
public class PlatformWiring implements Startable, Stoppable, Clearable {
private final PlatformContext platformContext;
private final WiringModel model;

private final InternalEventValidatorWiring internalEventValidatorWiring;
Expand All @@ -57,6 +60,7 @@ public class PlatformWiring implements Startable, Stoppable, Clearable {
* @param time provides wall clock time
*/
public PlatformWiring(@NonNull final PlatformContext platformContext, @NonNull final Time time) {
this.platformContext = Objects.requireNonNull(platformContext);
model = WiringModel.create(platformContext, time);

final PlatformSchedulers schedulers = PlatformSchedulers.create(platformContext, model);
Expand Down Expand Up @@ -214,14 +218,16 @@ private void flushAll() {
*/
@Override
public void clear() {
// pause the orphan buffer to break the cycle, and flush the pause through
orphanBufferWiring.pauseInput().inject(true);
orphanBufferWiring.flushRunnable().run();
if (!platformContext.getConfiguration().getConfigData(EventConfig.class).useLegacyIntake()) {
// pause the orphan buffer to break the cycle, and flush the pause through
orphanBufferWiring.pauseInput().inject(true);
orphanBufferWiring.flushRunnable().run();

// now that no cycles exist, flush all the wiring objects
flushAll();
// now that no cycles exist, flush all the wiring objects
flushAll();

// once everything has been flushed through the system, it's safe to unpause the orphan buffer
orphanBufferWiring.pauseInput().inject(false);
// once everything has been flushed through the system, it's safe to unpause the orphan buffer
orphanBufferWiring.pauseInput().inject(false);
}
}
}
Loading