network: wsnet with p2p backup meshing strategy (#6391)

Author: algorandskiy

network/connPerfMon.go

@@ -23,6 +23,7 @@ import (
 	"github.com/algorand/go-deadlock"
 
 	"github.com/algorand/go-algorand/crypto"
+	"github.com/algorand/go-algorand/logging"
 )
 
 //msgp:ignore pmStage
@@ -384,3 +385,127 @@ func (pm *connectionPerformanceMonitor) accumulateMessage(msg *IncomingMessage,
 		delete(msgBucket.messages, msgDigest)
 	}
 }
+
+type networkAdvanceMonitor struct {
+	// lastNetworkAdvance contains the last timestamp where the agreement protocol was able to make a notable progress.
+	// it used as a watchdog to help us detect connectivity issues ( such as cliques )
+	lastNetworkAdvance time.Time
+
+	mu deadlock.Mutex
+}
+
+func makeNetworkAdvanceMonitor() *networkAdvanceMonitor {
+	return &networkAdvanceMonitor{
+		lastNetworkAdvance: time.Now().UTC(),
+	}
+}
+
+func (m *networkAdvanceMonitor) lastAdvancedWithin(interval time.Duration) bool {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	// now < last + interval <=> now - last < interval
+	return time.Now().UTC().Before(m.lastNetworkAdvance.Add(interval))
+}
+
+func (m *networkAdvanceMonitor) updateLastAdvance() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.lastNetworkAdvance = time.Now().UTC()
+}
+
+type outgoingConnsCloser struct {
+	log                   logging.Logger
+	net                   outgoingDisconnectable
+	cliqueResolveInterval time.Duration
+	connPerfMonitor       *connectionPerformanceMonitor
+	netAdvMonitor         *networkAdvanceMonitor
+}
+
+type outgoingDisconnectable interface {
+	outgoingPeers() (peers []Peer)
+	numOutgoingPending() int
+	disconnect(badnode Peer, reason disconnectReason)
+	OnNetworkAdvance()
+}
+
+func makeOutgoingConnsCloser(log logging.Logger, net outgoingDisconnectable, connPerfMonitor *connectionPerformanceMonitor, cliqueResolveInterval time.Duration) *outgoingConnsCloser {
+	return &outgoingConnsCloser{
+		log:                   log,
+		net:                   net,
+		cliqueResolveInterval: cliqueResolveInterval,
+		connPerfMonitor:       connPerfMonitor,
+		netAdvMonitor:         makeNetworkAdvanceMonitor(),
+	}
+}
+
+// checkExistingConnectionsNeedDisconnecting check to see if existing connection need to be dropped due to
+// performance issues and/or network being stalled.
+func (cc *outgoingConnsCloser) checkExistingConnectionsNeedDisconnecting(targetConnCount int) bool {
+	// we already connected ( or connecting.. ) to  GossipFanout peers.
+	// get the actual peers.
+	outgoingPeers := cc.net.outgoingPeers()
+	if len(outgoingPeers) < targetConnCount {
+		// reset the performance monitor.
+		cc.connPerfMonitor.Reset([]Peer{})
+		return cc.checkNetworkAdvanceDisconnect()
+	}
+
+	if !cc.connPerfMonitor.ComparePeers(outgoingPeers) {
+		// different set of peers. restart monitoring.
+		cc.connPerfMonitor.Reset(outgoingPeers)
+	}
+
+	// same set of peers.
+	peerStat := cc.connPerfMonitor.GetPeersStatistics()
+	if peerStat == nil {
+		// performance metrics are not yet ready.
+		return cc.checkNetworkAdvanceDisconnect()
+	}
+
+	// update peers with the performance metrics we've gathered.
+	var leastPerformingPeer *wsPeer = nil
+	for _, stat := range peerStat.peerStatistics {
+		wsPeer := stat.peer.(*wsPeer)
+		wsPeer.peerMessageDelay = stat.peerDelay
+		cc.log.Infof("network performance monitor - peer '%s' delay %d first message portion %d%%", wsPeer.GetAddress(), stat.peerDelay, int(stat.peerFirstMessage*100))
+		if wsPeer.throttledOutgoingConnection && leastPerformingPeer == nil {
+			leastPerformingPeer = wsPeer
+		}
+	}
+	if leastPerformingPeer == nil {
+		return cc.checkNetworkAdvanceDisconnect()
+	}
+	cc.net.disconnect(leastPerformingPeer, disconnectLeastPerformingPeer)
+	cc.connPerfMonitor.Reset([]Peer{})
+
+	return true
+}
+
+// checkNetworkAdvanceDisconnect is using the lastNetworkAdvance indicator to see if the network is currently "stuck".
+// if it's seems to be "stuck", a randomly picked peer would be disconnected.
+func (cc *outgoingConnsCloser) checkNetworkAdvanceDisconnect() bool {
+	if cc.netAdvMonitor.lastAdvancedWithin(cc.cliqueResolveInterval) {
+		return false
+	}
+	outgoingPeers := cc.net.outgoingPeers()
+	if len(outgoingPeers) == 0 {
+		return false
+	}
+	if cc.net.numOutgoingPending() > 0 {
+		// we're currently trying to extend the list of outgoing connections. no need to
+		// disconnect any existing connection to free up room for another connection.
+		return false
+	}
+	var peer *wsPeer
+	disconnectPeerIdx := crypto.RandUint63() % uint64(len(outgoingPeers))
+	peer = outgoingPeers[disconnectPeerIdx].(*wsPeer)
+
+	cc.net.disconnect(peer, disconnectCliqueResolve)
+	cc.connPerfMonitor.Reset([]Peer{})
+	cc.net.OnNetworkAdvance()
+	return true
+}
+
+func (cc *outgoingConnsCloser) updateLastAdvance() {
+	cc.netAdvMonitor.updateLastAdvance()
+}

network/connPerfMon_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/algorand/go-algorand/crypto"
+	"github.com/algorand/go-algorand/logging"
 	"github.com/algorand/go-algorand/protocol"
 	"github.com/algorand/go-algorand/test/partitiontest"
 )
@@ -92,7 +93,7 @@ func BenchmarkConnMonitor(b *testing.B) {
 	}
 }
 
-func TestConnMonitorStageTiming(t *testing.T) {
+func TestConnMonitor_StageTiming(t *testing.T) {
 	partitiontest.PartitionTest(t)
 
 	peers := []Peer{&wsPeer{}, &wsPeer{}, &wsPeer{}, &wsPeer{}}
@@ -130,7 +131,7 @@ func TestConnMonitorStageTiming(t *testing.T) {
 	}
 
 }
-func TestBucketsPruning(t *testing.T) {
+func TestConnMonitor_BucketsPruning(t *testing.T) {
 	partitiontest.PartitionTest(t)
 
 	bucketsCount := 100
@@ -160,3 +161,82 @@ func TestBucketsPruning(t *testing.T) {
 		require.Equal(t, bucketsCount-i, len(perfMonitor.pendingMessagesBuckets))
 	}
 }
+
+type mockOutgoingNet struct {
+	peers            []Peer
+	pending          int
+	disconnectedPeer Peer
+	disconnectReason disconnectReason
+	advanceCalled    bool
+}
+
+func (m *mockOutgoingNet) outgoingPeers() (peers []Peer) { return m.peers }
+func (m *mockOutgoingNet) numOutgoingPending() int       { return m.pending }
+func (m *mockOutgoingNet) disconnect(badnode Peer, reason disconnectReason) {
+	m.disconnectedPeer = badnode
+	m.disconnectReason = reason
+}
+func (m *mockOutgoingNet) OnNetworkAdvance() { m.advanceCalled = true }
+
+func TestConnMonitor_CheckExistingConnections_ThrottledPeers(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	mon := makeConnectionPerformanceMonitor(nil)
+
+	p1 := &wsPeer{throttledOutgoingConnection: true}
+	mockNet := &mockOutgoingNet{peers: []Peer{p1}}
+	cc := makeOutgoingConnsCloser(logging.TestingLog(t), mockNet, mon, 100*time.Second)
+
+	res := cc.checkExistingConnectionsNeedDisconnecting(2)
+	require.False(t, res)
+	require.Nil(t, mockNet.disconnectedPeer)
+
+	p2 := &wsPeer{throttledOutgoingConnection: false} // not throttled
+	mockNet = &mockOutgoingNet{peers: []Peer{p1, p2}}
+	cc = makeOutgoingConnsCloser(logging.TestingLog(t), mockNet, mon, 100*time.Second)
+
+	mon.Reset(mockNet.peers)
+	mon.stage = pmStageStopped
+	mon.connectionDelay = map[Peer]int64{p1: 20, p2: 10}
+	mon.firstMessageCount = map[Peer]int64{p1: 1, p2: 2}
+	mon.msgCount = 3
+
+	res = cc.checkExistingConnectionsNeedDisconnecting(2)
+	require.True(t, res, "expected disconnect")
+	require.Equal(t, p1, mockNet.disconnectedPeer)
+	require.Equal(t, disconnectLeastPerformingPeer, mockNet.disconnectReason)
+}
+
+func TestConnMonitor_CheckExistingConnections_NoThrottledPeers(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	mon := makeConnectionPerformanceMonitor(nil)
+	p1 := &wsPeer{throttledOutgoingConnection: false}
+	p2 := &wsPeer{throttledOutgoingConnection: false}
+	mockNet := &mockOutgoingNet{peers: []Peer{p1, p2}}
+	cc := makeOutgoingConnsCloser(logging.TestingLog(t), mockNet, mon, 0)
+	mon.Reset(mockNet.peers)
+	mon.stage = pmStageStopped
+	mon.connectionDelay = map[Peer]int64{p1: 5, p2: 6}
+	mon.firstMessageCount = map[Peer]int64{p1: 1, p2: 1}
+	mon.msgCount = 2
+
+	res := cc.checkExistingConnectionsNeedDisconnecting(2)
+	require.True(t, res)
+	require.NotNil(t, mockNet.disconnectedPeer)
+	require.NotEqual(t, disconnectLeastPerformingPeer, mockNet.disconnectReason)
+}
+
+func TestNetworkAdvanceMonitor(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	m := makeNetworkAdvanceMonitor()
+
+	require.True(t, m.lastAdvancedWithin(500*time.Millisecond))
+
+	m.mu.Lock()
+	m.lastNetworkAdvance = time.Now().Add(-2 * time.Second)
+	m.mu.Unlock()
+	require.False(t, m.lastAdvancedWithin(500*time.Millisecond), "expected false after stale interval")
+
+	// update and verify within again
+	m.updateLastAdvance()
+	require.True(t, m.lastAdvancedWithin(500*time.Millisecond))
+}

network/hybridNetwork.go

@@ -52,13 +52,14 @@ func NewHybridP2PNetwork(log logging.Logger, cfg config.Local, datadir string, p
 	var childWsNetMeshCreator = meshCreator
 	var childP2PNetMeshCreator = meshCreator
 	var hybridMeshCreator MeshCreator = noopMeshCreator{}
+	noMeshCreatorAndHybridServer := meshCreator == nil && cfg.IsHybridServer()
 	_, isHybridMeshCreator := meshCreator.(hybridRelayMeshCreator)
-	if meshCreator == nil && cfg.IsHybridServer() || isHybridMeshCreator {
+	if noMeshCreatorAndHybridServer || isHybridMeshCreator {
 		// no mesh creator provided and this node is a listening/relaying node
 		// then override and use hybrid relay meshing
 		// or, if a hybrid relay meshing requested explicitly, do the same
 		childWsNetMeshCreator = noopMeshCreator{}
-		childP2PNetMeshCreator = noopMeshPubSubFilteredCreator{}
+		childP2PNetMeshCreator = noopMeshCreator{}
 		hybridMeshCreator = hybridRelayMeshCreator{}
 	}
 
@@ -77,6 +78,7 @@ func NewHybridP2PNetwork(log logging.Logger, cfg config.Local, datadir string, p
 	}
 
 	hybridMesh, err := hybridMeshCreator.create(
+		withTargetConnCount(cfg.GossipFanout),
 		withWebsocketNetwork(wsnet),
 		withP2PNetwork(p2pnet))
 	if err != nil {
@@ -187,7 +189,12 @@ func (n *HybridP2PNetwork) RegisterHTTPHandlerFunc(path string, handlerFunc func
 }
 
 // RequestConnectOutgoing implements GossipNode
-func (n *HybridP2PNetwork) RequestConnectOutgoing(replace bool, quit <-chan struct{}) {}
+func (n *HybridP2PNetwork) RequestConnectOutgoing(replace bool, quit <-chan struct{}) {
+	_ = n.runParallel(func(net GossipNode) error {
+		net.RequestConnectOutgoing(replace, quit)
+		return nil
+	})
+}
 
 // GetPeers implements GossipNode
 func (n *HybridP2PNetwork) GetPeers(options ...PeerOption) []Peer {

network/hybridNetwork_test.go

@@ -39,6 +39,7 @@ func TestHybridNetwork_DuplicateConn(t *testing.T) {
 
 	cfg := config.GetDefaultLocal()
 	cfg.EnableP2PHybridMode = true
+	cfg.DNSBootstrapID = ""
 	log := logging.TestingLog(t)
 	const p2pKeyDir = ""
 
@@ -208,6 +209,7 @@ func TestHybridNetwork_HybridRelayStrategy(t *testing.T) {
 
 	cfg := config.GetDefaultLocal()
 	cfg.EnableP2PHybridMode = true
+	cfg.DNSBootstrapID = ""
 	log := logging.TestingLog(t)
 
 	genesisInfo := GenesisInfo{genesisID, "net"}