mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-20 15:04:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

p2p: make p2p.Channel an interface (#8967)

Browse Source 
This is (#8446) pulled from the `main/libp2p` branch but without any
of the libp2p content, and is perhaps the easiest first step to enable
pluggability at the peer layer, and makes it possible hoist shims
(including for, say 0.34) into tendermint without touching the reactors.
This commit is contained in:
Sam Kleinman
2022-07-11 20:22:40 +00:00
committed by GitHub
parent 6902fa9282
commit d5fb82e414
23 changed files with 157 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/statesync/dispatcher.go
+2 -2
View File
@@ -26,14 +26,14 @@ var (
// NOTE: It is not the responsibility of the dispatcher to verify the light blocks.
type Dispatcher struct {
// the channel with which to send light block requests on
requestCh *p2p.Channel
requestCh p2p.Channel
mtx sync.Mutex
// all pending calls that have been dispatched and are awaiting an answer
calls map[types.NodeID]chan *types.LightBlock
}
func NewDispatcher(requestChannel *p2p.Channel) *Dispatcher {
func NewDispatcher(requestChannel p2p.Channel) *Dispatcher {
return &Dispatcher{
requestCh: requestChannel,
calls: make(map[types.NodeID]chan *types.LightBlock),
internal/statesync/dispatcher_test.go
+2 -2
View File
@@ -24,13 +24,13 @@ type channelInternal struct {
Error chan p2p.PeerError
}
func testChannel(size int) (*channelInternal, *p2p.Channel) {
func testChannel(size int) (*channelInternal, p2p.Channel) {
in := &channelInternal{
In: make(chan p2p.Envelope, size),
Out: make(chan p2p.Envelope, size),
Error: make(chan p2p.PeerError, size),
}
return in, p2p.NewChannel(0, in.In, in.Out, in.Error)
return in, p2p.NewChannel(0, "test", in.In, in.Out, in.Error)
}
func TestDispatcherBasic(t *testing.T) {
internal/statesync/reactor.go
+9 -9
View File
@@ -305,7 +305,7 @@ func (r *Reactor) OnStart(ctx context.Context) error {
return nil
}
go r.processChannels(ctx, map[p2p.ChannelID]*p2p.Channel{
go r.processChannels(ctx, map[p2p.ChannelID]p2p.Channel{
SnapshotChannel: snapshotCh,
ChunkChannel: chunkCh,
LightBlockChannel: blockCh,
@@ -611,7 +611,7 @@ func (r *Reactor) backfill(
// handleSnapshotMessage handles envelopes sent from peers on the
// SnapshotChannel. It returns an error only if the Envelope.Message is unknown
// for this channel. This should never be called outside of handleMessage.
func (r *Reactor) handleSnapshotMessage(ctx context.Context, envelope *p2p.Envelope, snapshotCh *p2p.Channel) error {
func (r *Reactor) handleSnapshotMessage(ctx context.Context, envelope *p2p.Envelope, snapshotCh p2p.Channel) error {
logger := r.logger.With("peer", envelope.From)
switch msg := envelope.Message.(type) {
@@ -683,7 +683,7 @@ func (r *Reactor) handleSnapshotMessage(ctx context.Context, envelope *p2p.Envel
// handleChunkMessage handles envelopes sent from peers on the ChunkChannel.
// It returns an error only if the Envelope.Message is unknown for this channel.
// This should never be called outside of handleMessage.
func (r *Reactor) handleChunkMessage(ctx context.Context, envelope *p2p.Envelope, chunkCh *p2p.Channel) error {
func (r *Reactor) handleChunkMessage(ctx context.Context, envelope *p2p.Envelope, chunkCh p2p.Channel) error {
switch msg := envelope.Message.(type) {
case *ssproto.ChunkRequest:
r.logger.Debug(
@@ -772,7 +772,7 @@ func (r *Reactor) handleChunkMessage(ctx context.Context, envelope *p2p.Envelope
return nil
}
func (r *Reactor) handleLightBlockMessage(ctx context.Context, envelope *p2p.Envelope, blockCh *p2p.Channel) error {
func (r *Reactor) handleLightBlockMessage(ctx context.Context, envelope *p2p.Envelope, blockCh p2p.Channel) error {
switch msg := envelope.Message.(type) {
case *ssproto.LightBlockRequest:
r.logger.Info("received light block request", "height", msg.Height)
@@ -829,7 +829,7 @@ func (r *Reactor) handleLightBlockMessage(ctx context.Context, envelope *p2p.Env
return nil
}
func (r *Reactor) handleParamsMessage(ctx context.Context, envelope *p2p.Envelope, paramsCh *p2p.Channel) error {
func (r *Reactor) handleParamsMessage(ctx context.Context, envelope *p2p.Envelope, paramsCh p2p.Channel) error {
switch msg := envelope.Message.(type) {
case *ssproto.ParamsRequest:
r.logger.Debug("received consensus params request", "height", msg.Height)
@@ -878,7 +878,7 @@ func (r *Reactor) handleParamsMessage(ctx context.Context, envelope *p2p.Envelop
// handleMessage handles an Envelope sent from a peer on a specific p2p Channel.
// It will handle errors and any possible panics gracefully. A caller can handle
// any error returned by sending a PeerError on the respective channel.
func (r *Reactor) handleMessage(ctx context.Context, envelope *p2p.Envelope, chans map[p2p.ChannelID]*p2p.Channel) (err error) {
func (r *Reactor) handleMessage(ctx context.Context, envelope *p2p.Envelope, chans map[p2p.ChannelID]p2p.Channel) (err error) {
defer func() {
if e := recover(); e != nil {
err = fmt.Errorf("panic in processing message: %v", e)
@@ -912,12 +912,12 @@ func (r *Reactor) handleMessage(ctx context.Context, envelope *p2p.Envelope, cha
// encountered during message execution will result in a PeerError being sent on
// the respective channel. When the reactor is stopped, we will catch the signal
// and close the p2p Channel gracefully.
func (r *Reactor) processChannels(ctx context.Context, chanTable map[p2p.ChannelID]*p2p.Channel) {
// make sure that the iterator gets cleaned up in case of error
func (r *Reactor) processChannels(ctx context.Context, chanTable map[p2p.ChannelID]p2p.Channel) {
// make sure tht the iterator gets cleaned up in case of error
ctx, cancel := context.WithCancel(ctx)
defer cancel()
chs := make([]*p2p.Channel, 0, len(chanTable))
chs := make([]p2p.Channel, 0, len(chanTable))
for key := range chanTable {
chs = append(chs, chanTable[key])
}
internal/statesync/reactor_test.go
+9 -5
View File
@@ -40,22 +40,22 @@ type reactorTestSuite struct {
conn *clientmocks.Client
stateProvider *mocks.StateProvider
snapshotChannel *p2p.Channel
snapshotChannel p2p.Channel
snapshotInCh chan p2p.Envelope
snapshotOutCh chan p2p.Envelope
snapshotPeerErrCh chan p2p.PeerError
chunkChannel *p2p.Channel
chunkChannel p2p.Channel
chunkInCh chan p2p.Envelope
chunkOutCh chan p2p.Envelope
chunkPeerErrCh chan p2p.PeerError
blockChannel *p2p.Channel
blockChannel p2p.Channel
blockInCh chan p2p.Envelope
blockOutCh chan p2p.Envelope
blockPeerErrCh chan p2p.PeerError
paramsChannel *p2p.Channel
paramsChannel p2p.Channel
paramsInCh chan p2p.Envelope
paramsOutCh chan p2p.Envelope
paramsPeerErrCh chan p2p.PeerError
@@ -102,6 +102,7 @@ func setup(
rts.snapshotChannel = p2p.NewChannel(
SnapshotChannel,
"snapshot",
rts.snapshotInCh,
rts.snapshotOutCh,
rts.snapshotPeerErrCh,
@@ -109,6 +110,7 @@ func setup(
rts.chunkChannel = p2p.NewChannel(
ChunkChannel,
"chunk",
rts.chunkInCh,
rts.chunkOutCh,
rts.chunkPeerErrCh,
@@ -116,6 +118,7 @@ func setup(
rts.blockChannel = p2p.NewChannel(
LightBlockChannel,
"lightblock",
rts.blockInCh,
rts.blockOutCh,
rts.blockPeerErrCh,
@@ -123,6 +126,7 @@ func setup(
rts.paramsChannel = p2p.NewChannel(
ParamsChannel,
"params",
rts.paramsInCh,
rts.paramsOutCh,
rts.paramsPeerErrCh,
@@ -133,7 +137,7 @@ func setup(
cfg := config.DefaultStateSyncConfig()
chCreator := func(ctx context.Context, desc *p2p.ChannelDescriptor) (*p2p.Channel, error) {
chCreator := func(ctx context.Context, desc *p2p.ChannelDescriptor) (p2p.Channel, error) {
switch desc.ID {
case SnapshotChannel:
return rts.snapshotChannel, nil
internal/statesync/stateprovider.go
+2 -2
View File
@@ -208,7 +208,7 @@ type stateProviderP2P struct {
sync.Mutex // light.Client is not concurrency-safe
lc *light.Client
initialHeight int64
paramsSendCh *p2p.Channel
paramsSendCh p2p.Channel
paramsRecvCh chan types.ConsensusParams
}
@@ -220,7 +220,7 @@ func NewP2PStateProvider(
initialHeight int64,
providers []lightprovider.Provider,
trustOptions light.TrustOptions,
paramsSendCh *p2p.Channel,
paramsSendCh p2p.Channel,
logger log.Logger,
) (StateProvider, error) {
if len(providers) < 2 {
internal/statesync/syncer.go
+2 -2
View File
@@ -56,8 +56,8 @@ type syncer struct {
stateProvider StateProvider
conn abciclient.Client
snapshots *snapshotPool
snapshotCh *p2p.Channel
chunkCh *p2p.Channel
snapshotCh p2p.Channel
chunkCh p2p.Channel
tempDir string
fetchers int32
retryTimeout time.Duration