p2p: base implementation of libp2p channel (#8476)

2026-08-28 03:46:33 +00:00 · 2022-05-09 10:34:02 -04:00
parent 2bc4f12aae
commit f1c9a56d57
2 changed files with 215 additions and 31 deletions
@@ -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
+}
@@ -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)