From 88c8f4b351778c4328f46af08fe17514535fa4d9 Mon Sep 17 00:00:00 2001
From: Andrzej Cichocki <andrzej.cichocki@r3.com>
Date: Fri, 30 Jun 2017 10:52:24 +0100
Subject: [PATCH] Avoid BFT printStackTraces when cluster is starting up (#899)

---
 .../node/services/transactions/BFTSMaRt.kt    |  7 ++--
 .../services/transactions/BFTSMaRtConfig.kt   | 34 ++++++++++++++++++-
 2 files changed, 38 insertions(+), 3 deletions(-)

diff --git a/node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRt.kt b/node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRt.kt
index d95f768a4a..f910c0ef1e 100644
--- a/node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRt.kt
+++ b/node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRt.kt
@@ -179,8 +179,11 @@ object BFTSMaRt {
         // TODO: Use Requery with proper DB schema instead of JDBCHashMap.
         // Must be initialised before ServiceReplica is started
         private val commitLog = services.database.transaction { JDBCHashMap<StateRef, UniquenessProvider.ConsumingTx>(tableName) }
-        @Suppress("LeakingThis")
-        private val replica = CordaServiceReplica(replicaId, config.path, this)
+        private val replica = run {
+            config.waitUntilReplicaWillNotPrintStackTrace(replicaId)
+            @Suppress("LeakingThis")
+            CordaServiceReplica(replicaId, config.path, this)
+        }
 
         fun dispose() {
             replica.dispose()
diff --git a/node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRtConfig.kt b/node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRtConfig.kt
index 0643e4aa6a..c11952c202 100644
--- a/node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRtConfig.kt
+++ b/node/src/main/kotlin/net/corda/node/services/transactions/BFTSMaRtConfig.kt
@@ -2,10 +2,15 @@ package net.corda.node.services.transactions
 
 import com.google.common.net.HostAndPort
 import net.corda.core.div
+import net.corda.core.utilities.debug
+import net.corda.core.utilities.loggerFor
 import java.io.FileWriter
 import java.io.PrintWriter
 import java.net.InetAddress
+import java.net.Socket
+import java.net.SocketException
 import java.nio.file.Files
+import java.util.concurrent.TimeUnit.MILLISECONDS
 
 /**
  * BFT SMaRt can only be configured via files in a configHome directory.
@@ -14,6 +19,7 @@ import java.nio.file.Files
  */
 class BFTSMaRtConfig(private val replicaAddresses: List<HostAndPort>, debug: Boolean = false) : PathManager<BFTSMaRtConfig>(Files.createTempDirectory("bft-smart-config")) {
     companion object {
+        private val log = loggerFor<BFTSMaRtConfig>()
         internal val portIsClaimedFormat = "Port %s is claimed by another replica: %s"
     }
 
@@ -47,12 +53,38 @@ class BFTSMaRtConfig(private val replicaAddresses: List<HostAndPort>, debug: Boo
         }
     }
 
+    fun waitUntilReplicaWillNotPrintStackTrace(contextReplicaId: Int) {
+        // A replica will printStackTrace until all lower-numbered replicas are listening.
+        // But we can't probe a replica without it logging EOFException when our probe succeeds.
+        // So to keep logging to a minimum we only check the previous replica:
+        val peerId = contextReplicaId - 1
+        if (peerId < 0) return
+        // The printStackTrace we want to avoid is in replica-replica communication code:
+        val address = BFTSMaRtPort.FOR_REPLICAS.ofReplica(replicaAddresses[peerId])
+        log.debug { "Waiting for replica $peerId to start listening on: $address" }
+        while (!address.isListening()) MILLISECONDS.sleep(200)
+        log.debug { "Replica $peerId is ready for P2P." }
+    }
+
     private fun replicaPorts(replicaId: Int): List<HostAndPort> {
         val base = replicaAddresses[replicaId]
-        return (0..1).map { HostAndPort.fromParts(base.host, base.port + it) }
+        return BFTSMaRtPort.values().map { it.ofReplica(base) }
     }
 }
 
+private enum class BFTSMaRtPort(private val off: Int) {
+    FOR_CLIENTS(0),
+    FOR_REPLICAS(1);
+
+    fun ofReplica(base: HostAndPort) = HostAndPort.fromParts(base.host, base.port + off)
+}
+
+private fun HostAndPort.isListening() = try {
+    Socket(host, port).use { true } // Will cause one error to be logged in the replica on success.
+} catch (e: SocketException) {
+    false
+}
+
 fun maxFaultyReplicas(clusterSize: Int) = (clusterSize - 1) / 3
 fun minCorrectReplicas(clusterSize: Int) = (2 * clusterSize + 3) / 3
 fun minClusterSize(maxFaultyReplicas: Int) = maxFaultyReplicas * 3 + 1