From f1c9a56d57e4b5b26680721c882bd2f90e5f008b Mon Sep 17 00:00:00 2001 From: Sam Kleinman Date: Mon, 9 May 2022 10:34:02 -0400 Subject: [PATCH] p2p: base implementation of libp2p channel (#8476) --- internal/p2p/channel.go | 144 ++++++++++++++++++++++++++++++++++++++++ internal/p2p/router.go | 102 +++++++++++++++++++--------- 2 files changed, 215 insertions(+), 31 deletions(-) diff --git a/internal/p2p/channel.go b/internal/p2p/channel.go index 30433b56d..2e6e3e070 100644 --- a/internal/p2p/channel.go +++ b/internal/p2p/channel.go @@ -2,10 +2,14 @@ package p2p import ( "context" + "errors" "fmt" "sync" "github.com/gogo/protobuf/proto" + "github.com/libp2p/go-libp2p-core/host" + "github.com/libp2p/go-libp2p-core/peer" + pubsub "github.com/libp2p/go-libp2p-pubsub" "github.com/tendermint/tendermint/types" ) @@ -214,3 +218,143 @@ func MergedChannelIterator(ctx context.Context, chs ...Channel) *ChannelIterator return iter } + +type libp2pChannelImpl struct { + chDesc *ChannelDescriptor + pubsub *pubsub.PubSub + host host.Host + topic *pubsub.Topic + chainID string + wrapper Wrapper +} + +func NewLibP2PChannel(chainID string, chDesc *ChannelDescriptor, ps *pubsub.PubSub, h host.Host) (Channel, error) { + ch := &libp2pChannelImpl{ + chDesc: chDesc, + pubsub: ps, + host: h, + chainID: chainID} + topic, err := ps.Join(ch.canonicalizedTopicName()) + if err != nil { + return nil, err + } + ch.topic = topic + + if w, ok := chDesc.MessageType.(Wrapper); ok { + ch.wrapper = w + } + + // TODO(tychoish) register handlers for + // request/response patterns + + return ch, nil +} + +func (ch *libp2pChannelImpl) String() string { + return fmt.Sprintf("Channel<%s>", ch.canonicalizedTopicName()) +} + +func (ch *libp2pChannelImpl) canonicalizedTopicName() string { + return fmt.Sprintf("%s.%s.%d", ch.chainID, ch.chDesc.Name, ch.chDesc.ID) +} + +func (ch *libp2pChannelImpl) Receive(ctx context.Context) *ChannelIterator { + iter := &ChannelIterator{ + pipe: make(chan Envelope), + } + + sub, err := ch.topic.Subscribe() + if err != nil { + return nil + } + + go func() { + defer close(iter.pipe) + for { + msg, err := sub.Next(ctx) + if err != nil { + // TODO: maybe signal to users that it + // was canceled, when we begin + // propagating errors out. + return + } + + payload := proto.Clone(ch.chDesc.MessageType) + if err := proto.Unmarshal(msg.Data, payload); err != nil { + // TODO: add error reporting abilities + // so we don't just miss these errors. + return + } + if wrapper, ok := payload.(Wrapper); ok { + if payload, err = wrapper.Unwrap(); err != nil { + return + } + } + + select { + case <-ctx.Done(): + return + case iter.pipe <- Envelope{ + From: types.NodeID(msg.From), + Message: payload, + ChannelID: ch.chDesc.ID, + }: + } + } + }() + + return nil +} + +func (ch *libp2pChannelImpl) Send(ctx context.Context, e Envelope) error { + if ch.wrapper != nil { + msg := proto.Clone(ch.wrapper) + if err := msg.(Wrapper).Wrap(e.Message); err != nil { + return err + } + + e.Message = msg + } + + e.From = types.NodeID(ch.host.ID()) + bz, err := proto.Marshal(e.Message) + if err != nil { + return err + } + + if e.Broadcast { + return ch.topic.Publish(ctx, bz) + } + + // TODO: remove this, likely. Checking to see if a topic has a + // peer is *probably* right, but maybe it's better to just try + // and connect to a peer directly (using whatever method) and + // go from there. + if !ch.topicHasPeer(peer.ID(e.To)) { + return fmt.Errorf("peer %q does not exist", e.To) + } + + // TODO: there's likely some tooling that exists for doing + // point-to-point messaging that we can leverage here, rather + // than implementing directly on-top of libp2p streams. + return errors.New("direct messages between peers not supported, yet") +} + +func (ch *libp2pChannelImpl) topicHasPeer(id peer.ID) bool { + for _, peer := range ch.pubsub.ListPeers(ch.canonicalizedTopicName()) { + if peer == id { + return true + } + } + return false +} + +func (ch *libp2pChannelImpl) SendError(ctx context.Context, pe PeerError) error { + // TODO: change handling of errors to peers. This problably + // shouldn't be handled as a property of the channel, and + // rather as part of some peer-info/network-management + // interface, but we can do it here for now, to ensure compatibility. + ch.pubsub.BlacklistPeer(peer.ID(pe.NodeID)) + + return nil +} diff --git a/internal/p2p/router.go b/internal/p2p/router.go index 12cae29dc..18e082ac0 100644 --- a/internal/p2p/router.go +++ b/internal/p2p/router.go @@ -202,6 +202,7 @@ type Router struct { chDescs []*ChannelDescriptor nodeInfoProducer func() *types.NodeInfo + chainID string legacy struct { peerManager *PeerManager @@ -225,6 +226,9 @@ type Router struct { network struct { host host.Host // network handle for ourselves ps *pubsub.PubSub + + mtx sync.Mutex + channels map[string]Channel } } @@ -303,47 +307,73 @@ type ChannelCreator func(context.Context, *ChannelDescriptor) (Channel, error) // wrapper message. The caller may provide a size to make the channel buffered, // which internally makes the inbound, outbound, and error channel buffered. func (r *Router) OpenChannel(ctx context.Context, chDesc *ChannelDescriptor) (Channel, error) { - r.legacy.channelMtx.Lock() - defer r.legacy.channelMtx.Unlock() + switch { + case r.options.UseLibP2P: + info := r.nodeInfoProducer() + ch, err := NewLibP2PChannel(info.Network, chDesc, r.options.NetworkPubSub, r.options.NetworkHost) + if err != nil { + return nil, err + } - if _, ok := r.legacy.channelQueues[chDesc.ID]; ok { - return nil, fmt.Errorf("channel %v already exists", chDesc.ID) - } - r.chDescs = append(r.chDescs, chDesc) + // TODO(tychoish): might be nice (though ultimately + // not particularly impactful(?)) to be able to get the + // canonical name for the channel without constructing + // it. - messageType := chDesc.MessageType + name := ch.String() + r.network.mtx.Lock() + defer r.network.mtx.Unlock() + if _, ok := r.network.channels[name]; ok { + // TODO(tychoish) actually maybe it would be ok to just + // return the existing channel. + return nil, fmt.Errorf("cannot construct channel %q more than once", name) + } + r.network.channels[name] = ch - queue := r.legacy.queueFactory(chDesc.RecvBufferCapacity) - outCh := make(chan Envelope, chDesc.RecvBufferCapacity) - errCh := make(chan PeerError, chDesc.RecvBufferCapacity) - channel := NewChannel(chDesc.ID, chDesc.Name, queue.dequeue(), outCh, errCh) + return ch, nil + default: + r.legacy.channelMtx.Lock() + defer r.legacy.channelMtx.Unlock() - var wrapper Wrapper - if w, ok := chDesc.MessageType.(Wrapper); ok { - wrapper = w - } + if _, ok := r.legacy.channelQueues[chDesc.ID]; ok { + return nil, fmt.Errorf("channel %v already exists", chDesc.ID) + } + r.chDescs = append(r.chDescs, chDesc) - r.legacy.channelQueues[chDesc.ID] = queue - r.legacy.channelMessages[chDesc.ID] = messageType + messageType := chDesc.MessageType - // add the channel to the nodeInfo if it's not already there. - r.nodeInfoProducer().AddChannel(uint16(chDesc.ID)) + queue := r.legacy.queueFactory(chDesc.RecvBufferCapacity) + outCh := make(chan Envelope, chDesc.RecvBufferCapacity) + errCh := make(chan PeerError, chDesc.RecvBufferCapacity) + channel := NewChannel(chDesc.ID, chDesc.Name, queue.dequeue(), outCh, errCh) - r.legacy.transport.AddChannelDescriptors([]*ChannelDescriptor{chDesc}) + var wrapper Wrapper + if w, ok := chDesc.MessageType.(Wrapper); ok { + wrapper = w + } - go func() { - defer func() { - r.legacy.channelMtx.Lock() - delete(r.legacy.channelQueues, chDesc.ID) - delete(r.legacy.channelMessages, chDesc.ID) - r.legacy.channelMtx.Unlock() - queue.close() + r.legacy.channelQueues[chDesc.ID] = queue + r.legacy.channelMessages[chDesc.ID] = messageType + + // add the channel to the nodeInfo if it's not already there. + r.nodeInfoProducer().AddChannel(uint16(chDesc.ID)) + + r.legacy.transport.AddChannelDescriptors([]*ChannelDescriptor{chDesc}) + + go func() { + defer func() { + r.legacy.channelMtx.Lock() + delete(r.legacy.channelQueues, chDesc.ID) + delete(r.legacy.channelMessages, chDesc.ID) + r.legacy.channelMtx.Unlock() + queue.close() + }() + + r.routeChannel(ctx, chDesc.ID, outCh, errCh, wrapper) }() - r.routeChannel(ctx, chDesc.ID, outCh, errCh, wrapper) - }() - - return channel, nil + return channel, nil + } } // routeChannel receives outbound channel messages and routes them to the @@ -976,6 +1006,12 @@ func (r *Router) setupQueueFactory(ctx context.Context) error { // OnStart implements service.Service. func (r *Router) OnStart(ctx context.Context) error { + if r.options.UseLibP2P { + return nil + } + + r.chainID = r.nodeInfoProducer().Network + if err := r.setupQueueFactory(ctx); err != nil { return err } @@ -998,6 +1034,10 @@ func (r *Router) OnStart(ctx context.Context) error { // here, since that would cause any reactor senders to panic, so it is the // sender's responsibility. func (r *Router) OnStop() { + if r.options.UseLibP2P { + return + } + // Close transport listeners (unblocks Accept calls). if err := r.legacy.transport.Close(); err != nil { r.logger.Error("failed to close transport", "err", err)