From 27ebc81ab252d9cf642b3dbe44325630aeea8a53 Mon Sep 17 00:00:00 2001 From: Marco Munizaga Date: Fri, 12 Aug 2022 13:05:20 -0700 Subject: [PATCH 1/2] Prefer peers with no streams when closing connections in connmgr --- p2p/net/connmgr/connmgr.go | 37 +++++++++++------------ p2p/net/connmgr/connmgr_test.go | 52 ++++++++++++++++++++++++++++----- 2 files changed, 63 insertions(+), 26 deletions(-) diff --git a/p2p/net/connmgr/connmgr.go b/p2p/net/connmgr/connmgr.go index b73e3a71d5..c596c9efd8 100644 --- a/p2p/net/connmgr/connmgr.go +++ b/p2p/net/connmgr/connmgr.go @@ -243,19 +243,11 @@ type peerInfo struct { type peerInfos []peerInfo -func (p peerInfos) SortByValue() { - sort.Slice(p, func(i, j int) bool { - left, right := p[i], p[j] - // temporary peers are preferred for pruning. - if left.temp != right.temp { - return left.temp - } - // otherwise, compare by value. - return left.value < right.value - }) -} - -func (p peerInfos) SortByValueAndStreams() { +// SortByValueAndStreams sorts peerInfos by their value and stream count. It +// will sort peers with no streams before those with streams (all else being +// equal). If `sortByMoreStreams` is true it will sort peers with more streams +// before those with fewer streams. This is useful to prioritize freeing memory. +func (p peerInfos) SortByValueAndStreams(sortByMoreStreams bool) { sort.Slice(p, func(i, j int) bool { left, right := p[i], p[j] // temporary peers are preferred for pruning. @@ -278,12 +270,21 @@ func (p peerInfos) SortByValueAndStreams() { } leftIncoming, leftStreams := incomingAndStreams(left.conns) rightIncoming, rightStreams := incomingAndStreams(right.conns) + // prefer closing inactive connections (no streams open) + if rightStreams != leftStreams && (leftStreams == 0 || rightStreams == 0) { + return leftStreams < rightStreams + } // incoming connections are preferred for pruning if leftIncoming != rightIncoming { return leftIncoming } - // prune connections with a higher number of streams first - return rightStreams < leftStreams + + if sortByMoreStreams { + // prune connections with a higher number of streams first + return rightStreams < leftStreams + } else { + return leftStreams < rightStreams + } }) } @@ -368,7 +369,7 @@ func (cm *BasicConnMgr) getConnsToCloseEmergency(target int) []network.Conn { cm.plk.RUnlock() // Sort peers according to their value. - candidates.SortByValueAndStreams() + candidates.SortByValueAndStreams(true) selected := make([]network.Conn, 0, target+10) for _, inf := range candidates { @@ -398,7 +399,7 @@ func (cm *BasicConnMgr) getConnsToCloseEmergency(target int) []network.Conn { } cm.plk.RUnlock() - candidates.SortByValueAndStreams() + candidates.SortByValueAndStreams(true) for _, inf := range candidates { if target <= 0 { break @@ -459,7 +460,7 @@ func (cm *BasicConnMgr) getConnsToClose() []network.Conn { } // Sort peers according to their value. - candidates.SortByValue() + candidates.SortByValueAndStreams(false) target := ncandidates - cm.cfg.lowWater diff --git a/p2p/net/connmgr/connmgr_test.go b/p2p/net/connmgr/connmgr_test.go index d70ec363e1..356ecbad7e 100644 --- a/p2p/net/connmgr/connmgr_test.go +++ b/p2p/net/connmgr/connmgr_test.go @@ -802,7 +802,7 @@ func TestPeerInfoSorting(t *testing.T) { p1 := peerInfo{id: peer.ID("peer1")} p2 := peerInfo{id: peer.ID("peer2"), temp: true} pis := peerInfos{p1, p2} - pis.SortByValue() + pis.SortByValueAndStreams(false) require.Equal(t, pis, peerInfos{p2, p1}) }) @@ -810,31 +810,67 @@ func TestPeerInfoSorting(t *testing.T) { p1 := peerInfo{id: peer.ID("peer1"), value: 40} p2 := peerInfo{id: peer.ID("peer2"), value: 20} pis := peerInfos{p1, p2} - pis.SortByValue() + pis.SortByValueAndStreams(false) require.Equal(t, pis, peerInfos{p2, p1}) }) - t.Run("in a memory emergency, starts with incoming connections", func(t *testing.T) { + t.Run("prefer peers with no streams", func(t *testing.T) { + p1 := peerInfo{id: peer.ID("peer1"), + conns: map[network.Conn]time.Time{ + &mockConn{stats: network.ConnStats{NumStreams: 0}}: time.Now(), + }, + } + p2 := peerInfo{id: peer.ID("peer2"), + conns: map[network.Conn]time.Time{ + &mockConn{stats: network.ConnStats{NumStreams: 1}}: time.Now(), + }, + } + pis := peerInfos{p2, p1} + pis.SortByValueAndStreams(false) + require.Equal(t, pis, peerInfos{p1, p2}) + }) + + t.Run("in a memory emergency, starts with incoming connections and higher streams", func(t *testing.T) { incoming := network.ConnStats{} incoming.Direction = network.DirInbound outgoing := network.ConnStats{} outgoing.Direction = network.DirOutbound + + outgoingSomeStreams := network.ConnStats{Stats: network.Stats{Direction: network.DirOutbound}, NumStreams: 1} + outgoingMoreStreams := network.ConnStats{Stats: network.Stats{Direction: network.DirOutbound}, NumStreams: 2} p1 := peerInfo{ id: peer.ID("peer1"), conns: map[network.Conn]time.Time{ - &mockConn{stats: outgoing}: time.Now(), + &mockConn{stats: outgoingSomeStreams}: time.Now(), }, } p2 := peerInfo{ id: peer.ID("peer2"), + conns: map[network.Conn]time.Time{ + &mockConn{stats: outgoingSomeStreams}: time.Now(), + &mockConn{stats: incoming}: time.Now(), + }, + } + p3 := peerInfo{ + id: peer.ID("peer3"), conns: map[network.Conn]time.Time{ &mockConn{stats: outgoing}: time.Now(), &mockConn{stats: incoming}: time.Now(), }, } - pis := peerInfos{p1, p2} - pis.SortByValueAndStreams() - require.Equal(t, pis, peerInfos{p2, p1}) + p4 := peerInfo{ + id: peer.ID("peer4"), + conns: map[network.Conn]time.Time{ + &mockConn{stats: outgoingMoreStreams}: time.Now(), + &mockConn{stats: incoming}: time.Now(), + }, + } + pis := peerInfos{p1, p2, p3, p4} + pis.SortByValueAndStreams(true) + // p3 is first because it is inactive (no streams). + // p4 is second because it has the most streams and we priortize killing + // connections with the higher number of streams. + require.Equal(t, pis, peerInfos{p3, p4, p2, p1}) }) t.Run("in a memory emergency, starts with connections that have many streams", func(t *testing.T) { @@ -852,7 +888,7 @@ func TestPeerInfoSorting(t *testing.T) { }, } pis := peerInfos{p1, p2} - pis.SortByValueAndStreams() + pis.SortByValueAndStreams(true) require.Equal(t, pis, peerInfos{p2, p1}) }) } From be64439a9dbcbbf1f99c37d89b8728a1991e6f16 Mon Sep 17 00:00:00 2001 From: Marco Munizaga Date: Wed, 17 Aug 2022 13:17:45 -0700 Subject: [PATCH 2/2] Implement Stat for test connection --- p2p/net/connmgr/connmgr_test.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/p2p/net/connmgr/connmgr_test.go b/p2p/net/connmgr/connmgr_test.go index 356ecbad7e..f95b8c5eaf 100644 --- a/p2p/net/connmgr/connmgr_test.go +++ b/p2p/net/connmgr/connmgr_test.go @@ -42,6 +42,15 @@ func (c *tconn) RemotePeer() peer.ID { return c.peer } +func (c *tconn) Stat() network.ConnStats { + return network.ConnStats{ + Stats: network.Stats{ + Direction: network.DirOutbound, + }, + NumStreams: 1, + } +} + func (c *tconn) RemoteMultiaddr() ma.Multiaddr { addr, err := ma.NewMultiaddr("/ip4/127.0.0.1/udp/1234") if err != nil {