mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-08-16 12:16:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

p2p: add eviction metrics and cleanup dialing error handling (backport #8819) (#8820)

Browse Source
This commit is contained in:
mergify[bot]
2022-06-24 10:42:58 -04:00
committed by GitHub
parent 2df4c2b19d
commit 826f224c2d
6 changed files with 91 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/p2p/metrics.go
+10
View File
@@ -53,6 +53,9 @@ type Metrics struct {
// this node.
PeersConnectedIncoming metrics.Gauge
// Number of peers evicted by this node.
PeersEvicted metrics.Counter
// RouterPeerQueueRecv defines the time taken to read off of a peer's queue
// before sending on the connection.
RouterPeerQueueRecv metrics.Histogram
@@ -110,6 +113,12 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
Name: "peers_connected_success",
Help: "Number of successful peer connection attempts",
}, labels).With(labelsAndValues...),
PeersEvicted: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
Namespace: namespace,
Subsystem: MetricsSubsystem,
Name: "peers_evicted",
Help: "Number of connected peers evicted",
}, labels).With(labelsAndValues...),
PeersConnectedFailure: prometheus.NewCounterFrom(stdprometheus.CounterOpts{
Namespace: namespace,
Subsystem: MetricsSubsystem,
@@ -200,6 +209,7 @@ func NopMetrics() *Metrics {
PeersConnectedIncoming: discard.NewGauge(),
PeersConnectedOutgoing: discard.NewGauge(),
PeersInactivated: discard.NewGauge(),
PeersEvicted: discard.NewCounter(),
PeerReceiveBytesTotal: discard.NewCounter(),
PeerSendBytesTotal: discard.NewCounter(),
PeerPendingSendBytes: discard.NewGauge(),
internal/p2p/peermanager.go
+13 -14
View File
@@ -534,12 +534,13 @@ func (m *PeerManager) HasDialedMaxPeers() bool {
// returned peer.
func (m *PeerManager) DialNext(ctx context.Context) (NodeAddress, error) {
for {
address, err := m.TryDialNext()
if err != nil || (address != NodeAddress{}) {
return address, err
if address := m.TryDialNext(); (address != NodeAddress{}) {
return address, nil
}
select {
case <-m.dialWaker.Sleep():
continue
case <-ctx.Done():
return NodeAddress{}, ctx.Err()
}
@@ -548,21 +549,20 @@ func (m *PeerManager) DialNext(ctx context.Context) (NodeAddress, error) {
// TryDialNext is equivalent to DialNext(), but immediately returns an empty
// address if no peers or connection slots are available.
func (m *PeerManager) TryDialNext() (NodeAddress, error) {
func (m *PeerManager) TryDialNext() NodeAddress {
m.mtx.Lock()
defer m.mtx.Unlock()
// We allow dialing MaxConnected+MaxConnectedUpgrade peers. Including
// MaxConnectedUpgrade allows us to probe additional peers that have a
// higher score than any other peers, and if successful evict it.
if m.options.MaxConnected > 0 && len(m.connected)+len(m.dialing) >=
int(m.options.MaxConnected)+int(m.options.MaxConnectedUpgrade) {
return NodeAddress{}, nil
if m.options.MaxConnected > 0 && len(m.connected)+len(m.dialing) >= int(m.options.MaxConnected)+int(m.options.MaxConnectedUpgrade) {
return NodeAddress{}
}
cinfo := m.getConnectedInfo()
if m.options.MaxOutgoingConnections > 0 && cinfo.outgoing >= m.options.MaxOutgoingConnections {
return NodeAddress{}, nil
return NodeAddress{}
}
for _, peer := range m.store.Ranked() {
@@ -589,16 +589,16 @@ func (m *PeerManager) TryDialNext() (NodeAddress, error) {
if m.options.MaxConnected > 0 && len(m.connected) >= int(m.options.MaxConnected) {
upgradeFromPeer := m.findUpgradeCandidate(peer.ID, peer.Score())
if upgradeFromPeer == "" {
return NodeAddress{}, nil
return NodeAddress{}
}
m.upgrading[upgradeFromPeer] = peer.ID
}
m.dialing[peer.ID] = true
return addressInfo.Address, nil
return addressInfo.Address
}
}
return NodeAddress{}, nil
return NodeAddress{}
}
// DialFailed reports a failed dial attempt. This will make the peer available
@@ -706,8 +706,7 @@ func (m *PeerManager) Dialed(address NodeAddress) error {
return err
}
if upgradeFromPeer != "" && m.options.MaxConnected > 0 &&
len(m.connected) >= int(m.options.MaxConnected) {
if upgradeFromPeer != "" && m.options.MaxConnected > 0 && len(m.connected) >= int(m.options.MaxConnected) {
// Look for an even lower-scored peer that may have appeared since we
// started the upgrade.
if p, ok := m.store.Get(upgradeFromPeer); ok {
@@ -716,11 +715,11 @@ func (m *PeerManager) Dialed(address NodeAddress) error {
}
}
m.evict[upgradeFromPeer] = true
m.evictWaker.Wake()
}
m.metrics.PeersConnectedOutgoing.Add(1)
m.connected[peer.ID] = peerConnectionOutgoing
m.evictWaker.Wake()
return nil
}
internal/p2p/peermanager_test.go
+45 -87
View File
@@ -378,16 +378,14 @@ func TestPeerManager_DialNext_WakeOnDialFailed(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
// Add b. We shouldn't be able to dial it, due to MaxConnected.
added, err = peerManager.Add(b)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Zero(t, dial)
// Spawn a goroutine to fail a's dial attempt.
@@ -415,8 +413,7 @@ func TestPeerManager_DialNext_WakeOnDialFailedRetry(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
require.NoError(t, peerManager.DialFailed(dial))
failed := time.Now()
@@ -443,8 +440,7 @@ func TestPeerManager_DialNext_WakeOnDisconnected(t *testing.T) {
err = peerManager.Accepted(a.NodeID)
require.NoError(t, err)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Zero(t, dial)
go func() {
@@ -473,8 +469,7 @@ func TestPeerManager_TryDialNext_MaxConnected(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
require.NoError(t, peerManager.Dialed(a))
@@ -482,16 +477,14 @@ func TestPeerManager_TryDialNext_MaxConnected(t *testing.T) {
added, err = peerManager.Add(b)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, b, dial)
// At this point, adding c will not allow dialing it.
added, err = peerManager.Add(c)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Zero(t, dial)
}
@@ -520,7 +513,7 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
dial := peerManager.TryDialNext()
require.NoError(t, err)
require.Equal(t, a, dial)
require.NoError(t, peerManager.Dialed(a))
@@ -529,8 +522,7 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
added, err = peerManager.Add(b)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, b, dial)
// Even though we are at capacity, we should be allowed to dial c for an
@@ -538,8 +530,7 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
added, err = peerManager.Add(c)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, c, dial)
// However, since we're using all upgrade slots now, we can't add and dial
@@ -547,16 +538,14 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
added, err = peerManager.Add(d)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Zero(t, dial)
// We go through with c's upgrade.
require.NoError(t, peerManager.Dialed(c))
// Still can't dial d.
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Zero(t, dial)
// Now, if we disconnect a, we should be allowed to dial d because we have a
@@ -572,8 +561,7 @@ func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
added, err = peerManager.Add(e)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Zero(t, dial)
}
@@ -593,8 +581,7 @@ func TestPeerManager_TryDialNext_UpgradeReservesPeer(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
require.NoError(t, peerManager.Dialed(a))
@@ -602,8 +589,7 @@ func TestPeerManager_TryDialNext_UpgradeReservesPeer(t *testing.T) {
added, err = peerManager.Add(b)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, b, dial)
// Adding c and dialing it will fail, because a is the only connected
@@ -611,8 +597,7 @@ func TestPeerManager_TryDialNext_UpgradeReservesPeer(t *testing.T) {
added, err = peerManager.Add(c)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Empty(t, dial)
}
@@ -633,22 +618,19 @@ func TestPeerManager_TryDialNext_DialingConnected(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
// Adding a's TCP address will not dispense a, since it's already dialing.
added, err = peerManager.Add(aTCP)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Zero(t, dial)
// Marking a as dialed will still not dispense it.
require.NoError(t, peerManager.Dialed(a))
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Zero(t, dial)
// Adding b and accepting a connection from it will not dispense it either.
@@ -656,8 +638,7 @@ func TestPeerManager_TryDialNext_DialingConnected(t *testing.T) {
require.NoError(t, err)
require.True(t, added)
require.NoError(t, peerManager.Accepted(bID))
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Zero(t, dial)
}
@@ -683,16 +664,14 @@ func TestPeerManager_TryDialNext_Multiple(t *testing.T) {
// All addresses should be dispensed as long as dialing them has failed.
dial := []p2p.NodeAddress{}
for range addresses {
address, err := peerManager.TryDialNext()
require.NoError(t, err)
address := peerManager.TryDialNext()
require.NotZero(t, address)
require.NoError(t, peerManager.DialFailed(address))
dial = append(dial, address)
}
require.ElementsMatch(t, dial, addresses)
address, err := peerManager.TryDialNext()
require.NoError(t, err)
address := peerManager.TryDialNext()
require.Zero(t, address)
}
@@ -714,15 +693,14 @@ func TestPeerManager_DialFailed(t *testing.T) {
// Dialing and then calling DialFailed with a different address (same
// NodeID) should unmark as dialing and allow us to dial the other address
// again, but not register the failed address.
dial, err := peerManager.TryDialNext()
dial := peerManager.TryDialNext()
require.NoError(t, err)
require.Equal(t, a, dial)
require.NoError(t, peerManager.DialFailed(p2p.NodeAddress{
Protocol: "tcp", NodeID: aID, Hostname: "localhost"}))
require.Equal(t, []p2p.NodeAddress{a}, peerManager.Addresses(aID))
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, a, dial)
// Calling DialFailed on same address twice should be fine.
@@ -753,8 +731,7 @@ func TestPeerManager_DialFailed_UnreservePeer(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
require.NoError(t, peerManager.Dialed(a))
@@ -762,8 +739,7 @@ func TestPeerManager_DialFailed_UnreservePeer(t *testing.T) {
added, err = peerManager.Add(b)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, b, dial)
// Adding c and dialing it will fail, even though it could upgrade a and we
@@ -772,14 +748,12 @@ func TestPeerManager_DialFailed_UnreservePeer(t *testing.T) {
added, err = peerManager.Add(c)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Empty(t, dial)
// Failing b's dial will now make c available for dialing.
require.NoError(t, peerManager.DialFailed(b))
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, c, dial)
}
@@ -794,8 +768,7 @@ func TestPeerManager_Dialed_Connected(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
require.NoError(t, peerManager.Dialed(a))
@@ -805,8 +778,7 @@ func TestPeerManager_Dialed_Connected(t *testing.T) {
added, err = peerManager.Add(b)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, b, dial)
require.NoError(t, peerManager.Accepted(b.NodeID))
@@ -835,8 +807,7 @@ func TestPeerManager_Dialed_MaxConnected(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
// Marking b as dialed in the meanwhile (even without TryDialNext)
@@ -878,8 +849,7 @@ func TestPeerManager_Dialed_MaxConnectedUpgrade(t *testing.T) {
added, err = peerManager.Add(c)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, c, dial)
require.NoError(t, peerManager.Dialed(c))
@@ -923,8 +893,7 @@ func TestPeerManager_Dialed_Upgrade(t *testing.T) {
added, err = peerManager.Add(b)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, b, dial)
require.NoError(t, peerManager.Dialed(b))
@@ -933,8 +902,7 @@ func TestPeerManager_Dialed_Upgrade(t *testing.T) {
added, err = peerManager.Add(c)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Empty(t, dial)
// a should now be evicted.
@@ -977,8 +945,7 @@ func TestPeerManager_Dialed_UpgradeEvenLower(t *testing.T) {
added, err = peerManager.Add(c)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, c, dial)
// In the meanwhile, a disconnects and d connects. d is even lower-scored
@@ -1028,7 +995,7 @@ func TestPeerManager_Dialed_UpgradeNoEvict(t *testing.T) {
added, err = peerManager.Add(c)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
dial := peerManager.TryDialNext()
require.NoError(t, err)
require.Equal(t, c, dial)
@@ -1074,8 +1041,7 @@ func TestPeerManager_Accepted(t *testing.T) {
added, err = peerManager.Add(c)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, c, dial)
require.NoError(t, peerManager.Accepted(c.NodeID))
require.Error(t, peerManager.Dialed(c))
@@ -1084,8 +1050,7 @@ func TestPeerManager_Accepted(t *testing.T) {
added, err = peerManager.Add(d)
require.NoError(t, err)
require.True(t, added)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, d, dial)
require.NoError(t, peerManager.Dialed(d))
require.Error(t, peerManager.Accepted(d.NodeID))
@@ -1233,8 +1198,7 @@ func TestPeerManager_Accepted_UpgradeDialing(t *testing.T) {
added, err = peerManager.Add(b)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, b, dial)
// a has already been claimed as an upgrade of a, so accepting
@@ -1394,8 +1358,7 @@ func TestPeerManager_EvictNext_WakeOnUpgradeDialed(t *testing.T) {
added, err := peerManager.Add(b)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, b, dial)
require.NoError(t, peerManager.Dialed(b))
}()
@@ -1521,13 +1484,11 @@ func TestPeerManager_Disconnected(t *testing.T) {
// Disconnecting a dialing peer does not unmark it as dialing, to avoid
// dialing it multiple times in parallel.
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
peerManager.Disconnected(a.NodeID)
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Zero(t, dial)
}
@@ -1595,8 +1556,7 @@ func TestPeerManager_Subscribe(t *testing.T) {
require.Equal(t, p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusDown}, <-sub.Updates())
// Outbound connection with peer error and eviction.
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
require.Empty(t, sub.Updates())
@@ -1619,8 +1579,7 @@ func TestPeerManager_Subscribe(t *testing.T) {
require.Equal(t, p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusDown}, <-sub.Updates())
// Outbound connection with dial failure.
dial, err = peerManager.TryDialNext()
require.NoError(t, err)
dial = peerManager.TryDialNext()
require.Equal(t, a, dial)
require.Empty(t, sub.Updates())
@@ -1716,8 +1675,7 @@ func TestPeerManager_Close(t *testing.T) {
added, err := peerManager.Add(a)
require.NoError(t, err)
require.True(t, added)
dial, err := peerManager.TryDialNext()
require.NoError(t, err)
dial := peerManager.TryDialNext()
require.Equal(t, a, dial)
require.NoError(t, peerManager.DialFailed(a))
internal/p2p/router.go
+22 -18
View File
@@ -291,7 +291,7 @@ func NewRouter(
router := &Router{
logger: logger,
metrics: metrics,
metrics: NopMetrics(),
nodeInfo: nodeInfo,
privKey: privKey,
connTracker: newConnTracker(
@@ -312,6 +312,10 @@ func NewRouter(
router.BaseService = service.NewBaseService(logger, "router", router)
if metrics != nil {
router.metrics = metrics
}
qf, err := router.createQueueFactory()
if err != nil {
return nil, err
@@ -422,11 +426,7 @@ func (r *Router) routeChannel(
) {
for {
select {
case envelope, ok := <-outCh:
if !ok {
return
}
case envelope := <-outCh:
// Mark the envelope with the channel ID to allow sendPeer() to pass
// it on to Transport.SendMessage().
envelope.channelID = chID
@@ -503,20 +503,22 @@ func (r *Router) routeChannel(
}
}
case peerError, ok := <-errCh:
if !ok {
return
}
shouldEvict := peerError.Fatal || r.peerManager.HasMaxPeerCapacity()
case peerError := <-errCh:
maxPeerCapacity := r.peerManager.HasMaxPeerCapacity()
r.logger.Error("peer error",
"peer", peerError.NodeID,
"err", peerError.Err,
"evicting", shouldEvict,
"disconnecting", peerError.Fatal || maxPeerCapacity,
)
if shouldEvict {
if peerError.Fatal || maxPeerCapacity {
// if the error is fatal or all peer
// slots are in use, we can error
// (disconnect) from the peer.
r.peerManager.Errored(peerError.NodeID, peerError.Err)
} else {
// this just decrements the peer
// score.
r.peerManager.processPeerEvent(PeerUpdate{
NodeID: peerError.NodeID,
Status: PeerStatusBad,
@@ -697,9 +699,8 @@ LOOP:
case errors.Is(err, context.Canceled):
r.logger.Debug("stopping dial routine")
break LOOP
case err != nil:
r.logger.Error("failed to find next peer to dial", "err", err)
break LOOP
case address == NodeAddress{}:
continue LOOP
}
select {
@@ -709,7 +710,7 @@ LOOP:
// create connections too quickly.
r.dialSleep(ctx)
continue
continue LOOP
case <-ctx.Done():
close(addresses)
break LOOP
@@ -748,6 +749,7 @@ func (r *Router) connectPeer(ctx context.Context, address NodeAddress) {
if err := r.runWithPeerMutex(func() error { return r.peerManager.Dialed(address) }); err != nil {
r.logger.Error("failed to dial peer", "op", "outgoing/dialing", "peer", address.NodeID, "err", err)
r.peerManager.dialWaker.Wake()
conn.Close()
return
}
@@ -1035,6 +1037,8 @@ func (r *Router) evictPeers() {
queue, ok := r.peerQueues[peerID]
r.peerMtx.RUnlock()
r.metrics.PeersEvicted.Add(1)
if ok {
queue.close()
}
internal/p2p/router_test.go
-7
View File
@@ -133,13 +133,6 @@ func TestRouter_Channel_Basic(t *testing.T) {
require.NoError(t, err)
require.Contains(t, router.NodeInfo().Channels, chDesc2.ID)
// Closing the channel, then opening it again should be fine.
channel.Close()
time.Sleep(100 * time.Millisecond) // yes yes, but Close() is async...
channel, err = router.OpenChannel(chDesc, &p2ptest.Message{}, 0)
require.NoError(t, err)
// We should be able to send on the channel, even though there are no peers.
p2ptest.RequireSend(t, channel, p2p.Envelope{
To: types.NodeID(strings.Repeat("a", 40)),
test/e2e/runner/rpc.go
+1 -1
View File
@@ -111,7 +111,7 @@ func waitForHeight(ctx context.Context, testnet *e2e.Testnet, height int64) (*ty
if len(clients) == 0 {
return nil, nil, errors.New("unable to connect to any network nodes")
}
if time.Since(lastIncrease) >= time.Minute {
if time.Since(lastIncrease) >= 90*time.Second {
if lastHeight == 0 {
return nil, nil, errors.New("chain stalled at unknown height (most likely upon starting)")
}