fix: 22553: Reconnect teacher may experience high virtual pipeline backpressure

Signed-off-by: Artem Ananev <[email protected]>

Author: artemananiev

platform-sdk/swirlds-common/src/main/java/com/swirlds/common/merkle/synchronization/TeachingSynchronizer.java

@@ -111,6 +111,39 @@ public TeachingSynchronizer(
         this.reconnectConfig = Objects.requireNonNull(reconnectConfig, "reconnectConfig must not be null");
     }
 
+    /**
+     * Create a new teaching synchronizer.
+     *
+     * @param threadManager   responsible for managing thread lifecycles
+     * @param in              the input stream
+     * @param out             the output stream
+     * @param view            the teacher tree view, used to access all tree nodes
+     * @param breakConnection a method that breaks the connection. Used iff an exception is encountered. Prevents
+     *                        deadlock if there is a thread stuck on a blocking IO operation that will never finish due
+     *                        to a failure.
+     * @param reconnectConfig reconnect configuration from platform
+     */
+    public TeachingSynchronizer(
+            @NonNull final Time time,
+            @NonNull final ThreadManager threadManager,
+            @NonNull final MerkleDataInputStream in,
+            @NonNull final MerkleDataOutputStream out,
+            @NonNull final TeacherTreeView<?> view,
+            @Nullable final Runnable breakConnection,
+            @NonNull final ReconnectConfig reconnectConfig) {
+
+        this.time = Objects.requireNonNull(time);
+        this.threadManager = Objects.requireNonNull(threadManager, "threadManager must not be null");
+        inputStream = Objects.requireNonNull(in, "in must not be null");
+        outputStream = Objects.requireNonNull(out, "out must not be null");
+
+        subtrees = new LinkedList<>();
+        subtrees.add(new TeacherSubtree(null, view));
+
+        this.breakConnection = breakConnection;
+        this.reconnectConfig = Objects.requireNonNull(reconnectConfig, "reconnectConfig must not be null");
+    }
+
     /**
      * Perform synchronization in the role of the teacher.
      */

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/reconnect/ReconnectStatePeerProtocol.java

@@ -24,6 +24,7 @@
 import com.swirlds.platform.network.protocol.PeerProtocol;
 import com.swirlds.platform.network.protocol.ReservedSignedStateResult;
 import com.swirlds.platform.state.signed.ReservedSignedState;
+import com.swirlds.platform.state.signed.SignedState;
 import com.swirlds.state.MerkleNodeState;
 import com.swirlds.state.StateLifecycleManager;
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -367,18 +368,26 @@ private void learner(final Connection connection) {
      * @param connection the connection to use for the reconnect
      */
     private void teacher(final Connection connection) {
-        try (final ReservedSignedState state = teacherState) {
-            new ReconnectStateTeacher(
-                            platformContext,
-                            time,
-                            threadManager,
-                            connection,
-                            reconnectSocketTimeout,
-                            connection.getSelfId(),
-                            connection.getOtherId(),
-                            state.get().getRound(),
-                            reconnectMetrics)
-                    .execute(state.get());
+        try {
+            final SignedState state = teacherState.get();
+            final ReconnectStateTeacher teacher;
+            try {
+                teacher = new ReconnectStateTeacher(
+                        platformContext,
+                        time,
+                        threadManager,
+                        connection,
+                        reconnectSocketTimeout,
+                        connection.getSelfId(),
+                        connection.getOtherId(),
+                        state.getRound(),
+                        state,
+                        reconnectMetrics);
+            } finally {
+                // The teacher now has all the information needed to teach. Time to release the original state
+                teacherState.close();
+            }
+            teacher.execute();
         } finally {
             teacherThrottle.reconnectAttemptFinished();
             teacherState = null;

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/reconnect/ReconnectStateTeacher.java

@@ -6,27 +6,33 @@
 import static com.swirlds.platform.reconnect.ReconnectStateLearner.endReconnectHandshake;
 import static com.swirlds.platform.state.service.PlatformStateUtils.getInfoString;
 
+import com.hedera.hapi.node.state.roster.Roster;
 import com.swirlds.base.time.Time;
 import com.swirlds.common.context.PlatformContext;
 import com.swirlds.common.io.streams.MerkleDataInputStream;
 import com.swirlds.common.io.streams.MerkleDataOutputStream;
 import com.swirlds.common.merkle.synchronization.TeachingSynchronizer;
 import com.swirlds.common.merkle.synchronization.config.ReconnectConfig;
+import com.swirlds.common.merkle.synchronization.views.TeacherTreeView;
 import com.swirlds.common.threading.manager.ThreadManager;
 import com.swirlds.config.api.Configuration;
 import com.swirlds.logging.legacy.payload.ReconnectFinishPayload;
 import com.swirlds.logging.legacy.payload.ReconnectStartPayload;
 import com.swirlds.platform.config.StateConfig;
 import com.swirlds.platform.metrics.ReconnectMetrics;
 import com.swirlds.platform.network.Connection;
+import com.swirlds.platform.state.signed.SigSet;
 import com.swirlds.platform.state.signed.SignedState;
+import com.swirlds.state.merkle.VirtualMapState;
+import com.swirlds.virtualmap.VirtualMap;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.io.IOException;
 import java.net.SocketException;
 import java.time.Duration;
 import java.util.Objects;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.hiero.base.crypto.Hash;
 import org.hiero.consensus.model.node.NodeId;
 import org.hiero.consensus.roster.RosterUtils;
 
@@ -41,6 +47,12 @@ public class ReconnectStateTeacher {
     private final Connection connection;
     private final Duration reconnectSocketTimeout;
 
+    private final TeacherTreeView<Long> teacherView;
+    private final SigSet signatures;
+    private final long signingWeight;
+    private final Roster roster;
+    private final Hash hash;
+
     private final NodeId selfId;
     private final NodeId otherId;
     private final long lastRoundReceived;
@@ -76,6 +88,7 @@ public ReconnectStateTeacher(
             @NonNull final NodeId selfId,
             @NonNull final NodeId otherId,
             final long lastRoundReceived,
+            @NonNull final SignedState signedState,
             @NonNull final ReconnectMetrics statistics) {
 
         this.platformContext = Objects.requireNonNull(platformContext);
@@ -89,6 +102,19 @@ public ReconnectStateTeacher(
         this.lastRoundReceived = lastRoundReceived;
         this.statistics = Objects.requireNonNull(statistics);
         this.configuration = Objects.requireNonNull(platformContext.getConfiguration());
+
+        signatures = signedState.getSigSet();
+        signingWeight = signedState.getSigningWeight();
+        roster = signedState.getRoster();
+        hash = signedState.getState().getHash();
+        if (!(signedState.getState() instanceof VirtualMapState virtualMapState)) {
+            throw new UnsupportedOperationException("Reconnects are only supported for VirtualMap states");
+        }
+        final ReconnectConfig reconnectConfig = configuration.getConfigData(ReconnectConfig.class);
+        // The teacher view will be closed by TeacherSynchronizer in reconnect() below
+        teacherView = ((VirtualMap) virtualMapState.getRoot()).buildTeacherView(reconnectConfig);
+
+        logReconnectStart(signedState);
     }
 
     /**
@@ -133,8 +159,7 @@ private void resetSocketTimeout() throws ReconnectStateException {
      * @throws ReconnectStateException thrown when current thread is interrupted, or when any I/O related errors occur, or
      *                            when there is an error in the underlying protocol
      */
-    public void execute(final SignedState signedState) throws ReconnectStateException {
-
+    public void execute() throws ReconnectStateException {
         // If the connection object to be used here has been disconnected on another thread, we can
         // not reconnect with this connection.
         if (!connection.connected()) {
@@ -145,12 +170,11 @@ public void execute(final SignedState signedState) throws ReconnectStateExceptio
                     connection.getOtherId());
             return;
         }
-        logReconnectStart(signedState);
         increaseSocketTimeout();
 
         try {
-            sendSignatures(signedState);
-            reconnect(signedState);
+            sendSignatures();
+            reconnect();
             endReconnectHandshake(connection);
         } catch (final InterruptedException e) {
             Thread.currentThread().interrupt();
@@ -197,7 +221,7 @@ private void logReconnectFinish() {
      *
      * @throws InterruptedException thrown if the current thread is interrupted
      */
-    private void reconnect(final SignedState signedState) throws InterruptedException, IOException {
+    private void reconnect() throws InterruptedException, IOException {
         logger.info(RECONNECT.getMarker(), "Starting synchronization in the role of the sender.");
         statistics.incrementSenderStartTimes();
 
@@ -206,12 +230,11 @@ private void reconnect(final SignedState signedState) throws InterruptedExceptio
 
         final ReconnectConfig reconnectConfig = configuration.getConfigData(ReconnectConfig.class);
         final TeachingSynchronizer synchronizer = new TeachingSynchronizer(
-                platformContext.getConfiguration(),
                 time,
                 threadManager,
                 new MerkleDataInputStream(connection.getDis()),
                 new MerkleDataOutputStream(connection.getDos()),
-                signedState.getState().getRoot(),
+                teacherView,
                 connection::disconnect,
                 reconnectConfig);
 
@@ -227,19 +250,19 @@ private void reconnect(final SignedState signedState) throws InterruptedExceptio
      *
      * @throws IOException thrown when any I/O related errors occur
      */
-    private void sendSignatures(final SignedState signedState) throws IOException {
+    private void sendSignatures() throws IOException {
         final StringBuilder sb = new StringBuilder();
         sb.append("Sending signatures from nodes ");
-        formattedList(sb, signedState.getSigSet().iterator());
+        formattedList(sb, signatures.iterator());
         sb.append(" (signing weight = ")
-                .append(signedState.getSigningWeight())
+                .append(signingWeight)
                 .append("/")
-                .append(RosterUtils.computeTotalWeight(signedState.getRoster()))
+                .append(RosterUtils.computeTotalWeight(roster))
                 .append(") for state hash ")
-                .append(signedState.getState().getHash());
+                .append(hash);
 
         logger.info(RECONNECT.getMarker(), sb);
-        connection.getDos().writeSerializable(signedState.getSigSet(), true);
+        connection.getDos().writeSerializable(signatures, true);
         connection.getDos().flush();
     }
 }

platform-sdk/swirlds-platform-core/src/test/java/com/swirlds/platform/reconnect/ReconnectTest.java

@@ -137,8 +137,9 @@ private void executeReconnect(final ReconnectMetrics reconnectMetrics) throws In
                     final ReconnectStateTeacher sender = buildSender(
                             new DummyConnection(
                                     platformContext, pairedStreams.getTeacherInput(), pairedStreams.getTeacherOutput()),
+                            signedState,
                             reconnectMetrics);
-                    sender.execute(signedState);
+                    sender.execute();
                 } catch (final IOException ex) {
                     ex.printStackTrace();
                 }
@@ -156,7 +157,8 @@ private void executeReconnect(final ReconnectMetrics reconnectMetrics) throws In
     }
 
     private ReconnectStateTeacher buildSender(
-            final SocketConnection connection, final ReconnectMetrics reconnectMetrics) throws IOException {
+            final SocketConnection connection, final SignedState signedState, final ReconnectMetrics reconnectMetrics)
+            throws IOException {
 
         final PlatformContext platformContext =
                 TestPlatformContextBuilder.create().build();
@@ -173,6 +175,7 @@ private ReconnectStateTeacher buildSender(
                 selfId,
                 otherId,
                 lastRoundReceived,
+                signedState,
                 reconnectMetrics);
     }