mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-08-21 06:36:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into p2p-dialer-store-change

Browse Source
This commit is contained in:
tycho garen
2022-06-14 19:16:31 -04:00
parent 8195c97590 979a6a1b13
commit 246e6d9154
7 changed files with 52 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/p2p/peermanager.go
+1 -9
View File
@@ -15,7 +15,6 @@ import (
dbm "github.com/tendermint/tm-db"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/libs/log"
p2pproto "github.com/tendermint/tendermint/proto/tendermint/p2p"
"github.com/tendermint/tendermint/types"
)
@@ -149,8 +148,6 @@ type PeerManagerOptions struct {
// Peer Metrics
Metrics *Metrics
// Logger
Logger log.Logger
}
// Validate validates the options.
@@ -271,7 +268,6 @@ type PeerManager struct {
rand *rand.Rand
dialWaker *tmsync.Waker // wakes up DialNext() on relevant peer changes
evictWaker *tmsync.Waker // wakes up EvictNext() on relevant peer changes
logger log.Logger
mtx sync.Mutex
store *peerStore
@@ -307,7 +303,6 @@ func NewPeerManager(selfID types.NodeID, peerDB dbm.DB, options PeerManagerOptio
dialWaker: tmsync.NewWaker(),
evictWaker: tmsync.NewWaker(),
metrics: NopMetrics(),
logger: log.NewNopLogger(),
store: store,
dialing: map[types.NodeID]bool{},
@@ -322,9 +317,6 @@ func NewPeerManager(selfID types.NodeID, peerDB dbm.DB, options PeerManagerOptio
if options.Metrics != nil {
peerManager.metrics = options.Metrics
}
if options.Logger != nil {
peerManager.logger = options.Logger
}
if err = peerManager.configurePeers(); err != nil {
return nil, err
@@ -409,7 +401,7 @@ func (m *PeerManager) Add(address NodeAddress) (bool, error) {
return false, err
}
if address.NodeID == m.selfID {
return false, nil
return false, fmt.Errorf("can't add self (%v) to peer store", m.selfID)
}
m.mtx.Lock()
internal/p2p/peermanager_test.go
+6 -8
View File
@@ -265,9 +265,8 @@ func TestPeerManager_Add(t *testing.T) {
require.Error(t, err)
// Adding self should error
ok, err := peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID})
require.False(t, ok)
require.NoError(t, err)
_, err = peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID})
require.Error(t, err)
}
func TestPeerManager_DialNext(t *testing.T) {
@@ -843,14 +842,13 @@ func TestPeerManager_Dialed_Connected(t *testing.T) {
require.Error(t, peerManager.Dialed(b))
}
func TestPeerManager_Adding_Self(t *testing.T) {
func TestPeerManager_Dialed_Self(t *testing.T) {
peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
require.NoError(t, err)
// Ingesting self should not error.
ok, err := peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID})
require.False(t, ok)
require.NoError(t, err)
// Dialing self should error.
_, err = peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID})
require.Error(t, err)
}
func TestPeerManager_Dialed_MaxConnected(t *testing.T) {
internal/p2p/router.go
+10 -7
View File
@@ -470,14 +470,17 @@ func (r *Router) dialSleep(ctx context.Context) {
func (r *Router) acceptPeers(ctx context.Context, transport Transport) {
for {
conn, err := transport.Accept(ctx)
switch err {
case nil:
case io.EOF:
r.logger.Debug("stopping accept routine", "transport", transport)
switch {
case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
r.logger.Debug("stopping accept routine", "transport", transport, "err", "context canceled")
return
default:
case errors.Is(err, io.EOF):
r.logger.Debug("stopping accept routine", "transport", transport, "err", "EOF")
return
case err != nil:
// in this case we got an error from the net.Listener.
r.logger.Error("failed to accept connection", "transport", transport, "err", err)
return
continue
}
incomingIP := conn.RemoteEndpoint().IP
@@ -489,7 +492,7 @@ func (r *Router) acceptPeers(ctx context.Context, transport Transport) {
"close_err", closeErr,
)
return
continue
}
// Spawn a goroutine for the handshake, to avoid head-of-line blocking.
internal/p2p/router_test.go
+33 -63
View File
@@ -442,78 +442,48 @@ func TestRouter_AcceptPeers(t *testing.T) {
}
}
func TestRouter_AcceptPeers_Error(t *testing.T) {
t.Cleanup(leaktest.Check(t))
func TestRouter_AcceptPeers_Errors(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
for _, err := range []error{io.EOF, context.Canceled, context.DeadlineExceeded} {
t.Run(err.Error(), func(t *testing.T) {
t.Cleanup(leaktest.Check(t))
// Set up a mock transport that returns an error, which should prevent
// the router from calling Accept again.
mockTransport := &mocks.Transport{}
mockTransport.On("String").Maybe().Return("mock")
mockTransport.On("Accept", mock.Anything).Once().Return(nil, errors.New("boom"))
mockTransport.On("Close").Return(nil)
mockTransport.On("Listen", mock.Anything).Return(nil)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Set up and start the router.
peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
require.NoError(t, err)
// Set up a mock transport that returns io.EOF once, which should prevent
// the router from calling Accept again.
mockTransport := &mocks.Transport{}
mockTransport.On("String").Maybe().Return("mock")
mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
mockTransport.On("Close").Return(nil)
mockTransport.On("Listen", mock.Anything).Return(nil)
router, err := p2p.NewRouter(
log.NewNopLogger(),
p2p.NopMetrics(),
selfKey,
peerManager,
func() *types.NodeInfo { return &selfInfo },
mockTransport,
nil,
p2p.RouterOptions{},
)
require.NoError(t, err)
// Set up and start the router.
peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
require.NoError(t, err)
require.NoError(t, router.Start(ctx))
time.Sleep(time.Second)
router.Stop()
router, err := p2p.NewRouter(
log.NewNopLogger(),
p2p.NopMetrics(),
selfKey,
peerManager,
func() *types.NodeInfo { return &selfInfo },
mockTransport,
nil,
p2p.RouterOptions{},
)
require.NoError(t, err)
mockTransport.AssertExpectations(t)
}
require.NoError(t, router.Start(ctx))
time.Sleep(time.Second)
router.Stop()
func TestRouter_AcceptPeers_ErrorEOF(t *testing.T) {
t.Cleanup(leaktest.Check(t))
mockTransport.AssertExpectations(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
})
// Set up a mock transport that returns io.EOF once, which should prevent
// the router from calling Accept again.
mockTransport := &mocks.Transport{}
mockTransport.On("String").Maybe().Return("mock")
mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
mockTransport.On("Close").Return(nil)
mockTransport.On("Listen", mock.Anything).Return(nil)
// Set up and start the router.
peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
require.NoError(t, err)
router, err := p2p.NewRouter(
log.NewNopLogger(),
p2p.NopMetrics(),
selfKey,
peerManager,
func() *types.NodeInfo { return &selfInfo },
mockTransport,
nil,
p2p.RouterOptions{},
)
require.NoError(t, err)
require.NoError(t, router.Start(ctx))
time.Sleep(time.Second)
router.Stop()
mockTransport.AssertExpectations(t)
}
}
func TestRouter_AcceptPeers_HeadOfLineBlocking(t *testing.T) {
node/node.go
+1 -1
View File
@@ -203,7 +203,7 @@ func makeNode(
}
}
peerManager, peerCloser, err := createPeerManager(logger, cfg, dbProvider, nodeKey.ID, nodeMetrics.p2p)
peerManager, peerCloser, err := createPeerManager(cfg, dbProvider, nodeKey.ID, nodeMetrics.p2p)
closers = append(closers, peerCloser)
if err != nil {
return nil, combineCloseError(
node/seed.go
+1 -1
View File
@@ -67,7 +67,7 @@ func makeSeedNode(
// Setup Transport and Switch.
p2pMetrics := p2p.PrometheusMetrics(cfg.Instrumentation.Namespace, "chain_id", genDoc.ChainID)
peerManager, closer, err := createPeerManager(logger, cfg, dbProvider, nodeKey.ID, p2pMetrics)
peerManager, closer, err := createPeerManager(cfg, dbProvider, nodeKey.ID, p2pMetrics)
if err != nil {
return nil, combineCloseError(
fmt.Errorf("failed to create peer manager: %w", err),
node/setup.go
-2
View File
@@ -199,7 +199,6 @@ func createEvidenceReactor(
}
func createPeerManager(
logger log.Logger,
cfg *config.Config,
dbProvider config.DBProvider,
nodeID types.NodeID,
@@ -228,7 +227,6 @@ func createPeerManager(
maxUpgradeConns := uint16(4)
options := p2p.PeerManagerOptions{
Logger: logger.With("module", "peermanager"),
SelfAddress: selfAddr,
MaxConnected: maxConns,
MaxConnectedUpgrade: maxUpgradeConns,