p2p

Documentation

Index

• Constants
• Variables
• func EventId(channel channels.Channel, payload []byte) (hash.Hash, error)
• func MessageType(decodedPayload interface{}) string
• type ConnectionGater
• type Connector
• type GetTimeNow
• type IDTranslator
• type LibP2PNode
• type Network
  • func NewNetwork(param *NetworkParameters) (*Network, error)
  • func (n *Network) Identities() flow.IdentityList
  • func (n *Network) Identity(pid peer.ID) (*flow.Identity, bool)
  • func (n *Network) MulticastOnChannel(channel channels.Channel, message interface{}, num uint, ...) error
  • func (n *Network) PublishOnChannel(channel channels.Channel, message interface{}, targetIDs ...flow.Identifier) error
  • func (n *Network) Receive(nodeID flow.Identifier, msg *message.Message, decodedMsgPayload interface{}) error
  • func (n *Network) Register(channel channels.Channel, messageProcessor network.MessageProcessor) (network.Conduit, error)
  • func (n *Network) RegisterBlobService(channel channels.Channel, ds datastore.Batching, ...) (network.BlobService, error)
  • func (n *Network) RegisterPingService(pingProtocol protocol.ID, provider network.PingInfoProvider) (network.PingService, error)
  • func (n *Network) Topology() flow.IdentityList
  • func (n *Network) UnRegisterChannel(channel channels.Channel) error
  • func (n *Network) UnicastOnChannel(channel channels.Channel, message interface{}, targetID flow.Identifier) error
• type NetworkOptFunction
  • func WithConduitFactory(f network.ConduitFactory) NetworkOptFunction
• type NetworkParameters
• type PeerFilter
• type PeerManager
• type PeerManagerFactoryFunc
• type PeersProvider
• type RateLimiter
• type RateLimiterOpt
  • func WithGetTimeNowFunc(now GetTimeNow) RateLimiterOpt
• type SubscriptionProvider
• type TopicProvider

Constants

const (
	// DefaultReceiveCacheSize represents size of receive cache that keeps hash of incoming messages
	// for sake of deduplication.
	DefaultReceiveCacheSize = 10e4
)

Variables

var (
	// ErrInvalidId indicates that the given ID (either `peer.ID` or `flow.Identifier`) has an invalid format.
	ErrInvalidId = errors.New("empty peer ID")

	// ErrUnknownId indicates that the given ID (either `peer.ID` or `flow.Identifier`) is unknown.
	ErrUnknownId = errors.New("unknown ID")
)

var ErrNetworkShutdown = errors.New("network has already shutdown")

var NotEjectedFilter = filter.Not(filter.Ejected)

NotEjectedFilter is an identity filter that, when applied to the identity table at a given snapshot, returns all nodes that we should communicate with over the networking layer.

NOTE: The protocol state includes nodes from the previous/next epoch that should be included in network communication. We omit any nodes that have been ejected.

Functions

func EventId

func EventId(channel channels.Channel, payload []byte) (hash.Hash, error)

func MessageType

func MessageType(decodedPayload interface{}) string

Types

type ConnectionGater

type ConnectionGater interface {
	InterceptPeerDial(p peer.ID) (allow bool)

	InterceptAddrDial(peer.ID, multiaddr.Multiaddr) (allow bool)

	InterceptAccept(network.ConnMultiaddrs) (allow bool)

	InterceptSecured(network.Direction, peer.ID, network.ConnMultiaddrs) (allow bool)

	InterceptUpgraded(network.Conn) (allow bool, reason control.DisconnectReason)
}

ConnectionGater is a copy of the libp2p ConnectionGater interface: https://github.com/libp2p/go-libp2p/blob/master/core/connmgr/gater.go#L54 We use it here to generate a mock for testing through testify mock.

type Connector

type Connector interface {
	// UpdatePeers connects to the given peer.IDs. It also disconnects from any other peers with which it may have
	// previously established connection.
	// UpdatePeers implementation should be idempotent such that multiple calls to connect to the same peer should not
	// create multiple connections
	UpdatePeers(ctx context.Context, peerIDs peer.IDSlice)
}

Connector connects to peer and disconnects from peer using the underlying networking library

type GetTimeNow

type GetTimeNow func() time.Time

GetTimeNow callback used to get the current time. This allows us to improve testing by manipulating the current time as opposed to using time.Now directly.

type IDTranslator

type IDTranslator interface {
	// GetPeerID returns the peer ID for the given `flow.Identifier`.
	// During normal operations, the following error returns are expected
	//  * ErrUnknownId if the given Identifier is unknown
	//  * ErrInvalidId if the given Identifier has an invalid format.
	//    This error return is only possible, when the Identifier is generated from some
	//    other input (e.g. the node's public key). While the Flow protocol itself makes
	//    no assumptions about the structure of the `Identifier`, protocol extensions
	//    (e.g. for Observers) might impose a strict relationship between both flow
	//    Identifier and peer ID, in which case they can use this error.
	// TODO: implementations do not fully adhere to this convention on error returns
	GetPeerID(flow.Identifier) (peer.ID, error)

	// GetFlowID returns the `flow.Identifier` for the given `peer.ID`.
	// During normal operations, the following error returns are expected
	//  * ErrUnknownId if the given Identifier is unknown
	//  * ErrInvalidId if the given Identifier has an invalid format.
	//    This error return is only possible, when the Identifier is generated from some
	//    other input (e.g. the node's public key). While the Flow protocol itself makes
	//    no assumptions about the structure of the `Identifier`, protocol extensions
	//    (e.g. for Observers) might impose a strict relationship between both flow
	//    Identifier and peer ID, in which case they can use this error.
	// TODO: implementations do not fully adhere to this convention on error returns
	GetFlowID(peer.ID) (flow.Identifier, error)
}

IDTranslator provides an interface for converting from Flow ID's to LibP2P peer ID's and vice versa.

type LibP2PNode

type LibP2PNode interface {
	module.ReadyDoneAware
	// Start the libp2p node.
	Start(ctx irrecoverable.SignalerContext)
	// Stop terminates the libp2p node.
	Stop() error
	// AddPeer adds a peer to this node by adding it to this node's peerstore and connecting to it.
	AddPeer(ctx context.Context, peerInfo peer.AddrInfo) error
	// RemovePeer closes the connection with the peer.
	RemovePeer(peerID peer.ID) error
	// GetPeersForProtocol returns slice peer IDs for the specified protocol ID.
	GetPeersForProtocol(pid protocol.ID) peer.IDSlice
	// CreateStream returns an existing stream connected to the peer if it exists, or creates a new stream with it.
	CreateStream(ctx context.Context, peerID peer.ID) (libp2pnet.Stream, error)
	// GetIPPort returns the IP and Port the libp2p node is listening on.
	GetIPPort() (string, string, error)
	// RoutingTable returns the node routing table
	RoutingTable() *kbucket.RoutingTable
	// ListPeers returns list of peer IDs for peers subscribed to the topic.
	ListPeers(topic string) []peer.ID
	// Subscribe subscribes the node to the given topic and returns the subscription
	Subscribe(topic channels.Topic, topicValidator pubsub.ValidatorEx) (*pubsub.Subscription, error)
	// UnSubscribe cancels the subscriber and closes the topic.
	UnSubscribe(topic channels.Topic) error
	// Publish publishes the given payload on the topic.
	Publish(ctx context.Context, topic channels.Topic, data []byte) error
	// Host returns pointer to host object of node.
	Host() host.Host
	// WithDefaultUnicastProtocol overrides the default handler of the unicast manager and registers all preferred protocols.
	WithDefaultUnicastProtocol(defaultHandler libp2pnet.StreamHandler, preferred []unicast.ProtocolName) error
	// WithPeersProvider sets the PeersProvider for the peer manager.
	// If a peer manager factory is set, this method will set the peer manager's PeersProvider.
	WithPeersProvider(peersProvider PeersProvider)
	// PeerManagerComponent returns the component interface of the peer manager.
	PeerManagerComponent() component.Component
	// RequestPeerUpdate requests an update to the peer connections of this node using the peer manager.
	RequestPeerUpdate()
	// IsConnected returns true is address is a direct peer of this node else false.
	IsConnected(peerID peer.ID) (bool, error)
	// SetRouting sets the node's routing implementation.
	// SetRouting may be called at most once.
	SetRouting(r routing.Routing)
	// Routing returns node routing object.
	Routing() routing.Routing
	// SetPubSub sets the node's pubsub implementation.
	// SetPubSub may be called at most once.
	SetPubSub(ps *pubsub.PubSub)
	// SetComponentManager sets the component manager for the node.
	// SetComponentManager may be called at most once.
	SetComponentManager(cm *component.ComponentManager)
}

LibP2PNode represents a flow libp2p node. It provides the network layer with the necessary interface to control the underlying libp2p node. It is essentially the flow wrapper around the libp2p node, and allows us to define different types of libp2p nodes that can operate in different ways by overriding these methods.

type Network

type Network struct {
	sync.RWMutex
	*component.ComponentManager
	// contains filtered or unexported fields
}

Network represents the overlay network of our peer-to-peer network, including the protocols for handshakes, authentication, gossiping and heartbeats.

func NewNetwork

func NewNetwork(param *NetworkParameters) (*Network, error)

NewNetwork creates a new naive overlay network, using the given middleware to communicate to direct peers, using the given codec for serialization, and using the given state & cache interfaces to track volatile information. csize determines the size of the cache dedicated to keep track of received messages

func (*Network) Identities

func (n *Network) Identities() flow.IdentityList

func (*Network) Identity

func (n *Network) Identity(pid peer.ID) (*flow.Identity, bool)

func (*Network) MulticastOnChannel

func (n *Network) MulticastOnChannel(channel channels.Channel, message interface{}, num uint, targetIDs ...flow.Identifier) error

MulticastOnChannel unreliably sends the specified event over the channel to randomly selected 'num' number of recipients selected from the specified targetIDs.

func (*Network) PublishOnChannel

func (n *Network) PublishOnChannel(channel channels.Channel, message interface{}, targetIDs ...flow.Identifier) error

PublishOnChannel sends the message in an unreliable way to the given recipients. In this context, unreliable means that the message is published over a libp2p pub-sub channel and can be read by any node subscribed to that channel. The selector could be used to optimize or restrict delivery.

func (*Network) Receive

func (n *Network) Receive(nodeID flow.Identifier, msg *message.Message, decodedMsgPayload interface{}) error

func (*Network) Register

func (n *Network) Register(channel channels.Channel, messageProcessor network.MessageProcessor) (network.Conduit, error)

Register will register the given engine with the given unique engine engineID, returning a conduit to directly submit messages to the message bus of the engine.

func (*Network) RegisterBlobService

func (n *Network) RegisterBlobService(channel channels.Channel, ds datastore.Batching, opts ...network.BlobServiceOption) (network.BlobService, error)

RegisterBlobService registers a BlobService on the given channel. The returned BlobService can be used to request blobs from the network.

func (*Network) RegisterPingService

func (n *Network) RegisterPingService(pingProtocol protocol.ID, provider network.PingInfoProvider) (network.PingService, error)

func (*Network) Topology

func (n *Network) Topology() flow.IdentityList

func (*Network) UnRegisterChannel

func (n *Network) UnRegisterChannel(channel channels.Channel) error

UnRegisterChannel unregisters the engine for the specified channel. The engine will no longer be able to send or receive messages from that channel.

func (*Network) UnicastOnChannel

func (n *Network) UnicastOnChannel(channel channels.Channel, message interface{}, targetID flow.Identifier) error

UnicastOnChannel sends the message in a reliable way to the given recipient. It uses 1-1 direct messaging over the underlying network to deliver the message. It returns an error if unicasting fails.

type NetworkOptFunction

type NetworkOptFunction func(*Network)

func WithConduitFactory

func WithConduitFactory(f network.ConduitFactory) NetworkOptFunction

type NetworkParameters

type NetworkParameters struct {
	Logger              zerolog.Logger
	Codec               network.Codec
	Me                  module.Local
	MiddlewareFactory   func() (network.Middleware, error)
	Topology            network.Topology
	SubscriptionManager network.SubscriptionManager
	Metrics             module.NetworkMetrics
	IdentityProvider    module.IdentityProvider
	ReceiveCache        *netcache.ReceiveCache
	Options             []NetworkOptFunction
}

type PeerFilter

type PeerFilter func(peer.ID) error

type PeerManager

type PeerManager interface {
	component.Component

	// RequestPeerUpdate requests an update to the peer connections of this node.
	// If a peer update has already been requested (either as a periodic request or an on-demand
	// request) and is outstanding, then this call is a no-op.
	RequestPeerUpdate()

	// ForceUpdatePeers initiates an update to the peer connections of this node immediately
	ForceUpdatePeers(context.Context)

	// SetPeersProvider sets the peer managers's peers provider and may be called at most once
	SetPeersProvider(PeersProvider)
}

PeerManager adds and removes connections to peers periodically and on request

type PeerManagerFactoryFunc

type PeerManagerFactoryFunc func(host host.Host, peersProvider PeersProvider, logger zerolog.Logger) (PeerManager, error)

PeerManagerFactoryFunc is a factory function type for generating a PeerManager instance using the given host, peersProvider and logger

type PeersProvider

type PeersProvider func() peer.IDSlice

type RateLimiter

type RateLimiter interface {
	// Allow returns true if a message should be allowed to be processed.
	Allow(peerID peer.ID, msg *message.Message) bool

	// IsRateLimited returns true if a peer is rate limited.
	IsRateLimited(peerID peer.ID) bool

	// SetTimeNowFunc allows users to override the underlying time module used.
	SetTimeNowFunc(now GetTimeNow)

	// Stop sends cleanup signal to underlying rate limiters and rate limited peers maps. After the rate limiter
	// is stopped it can not be reused.
	Stop()

	// Start starts cleanup loop for underlying rate limiters and rate limited peers maps.
	Start()
}

RateLimiter unicast rate limiter interface

type RateLimiterOpt

type RateLimiterOpt func(limiter RateLimiter)

func WithGetTimeNowFunc

func WithGetTimeNowFunc(now GetTimeNow) RateLimiterOpt

type SubscriptionProvider

type SubscriptionProvider interface {
	// GetSubscribedTopics returns all the subscriptions of a peer within the pubsub network.
	// Note that the current peer must be subscribed to the topic for it to the same topics in order
	// to query for other peers, e.g., if current peer has subscribed to topics A and B, and peer1
	// has subscribed to topics A, B, and C, then GetSubscribedTopics(peer1) will return A and B. Since this peer
	// has not subscribed to topic C, it will not be able to query for other peers subscribed to topic C.
	GetSubscribedTopics(pid peer.ID) []string
}

SubscriptionProvider provides a list of topics a peer is subscribed to.

type TopicProvider

type TopicProvider interface {
	// GetTopics returns all the topics within the pubsub network that the current peer has subscribed to.
	GetTopics() []string

	// ListPeers returns all the peers subscribed to a topic.
	// Note that the current peer must be subscribed to the topic for it to query for other peers.
	// If the current peer is not subscribed to the topic, an empty list is returned.
	// For example, if current peer has subscribed to topics A and B, then ListPeers only return
	// subscribed peers for topics A and B, and querying for topic C will return an empty list.
	ListPeers(topic string) []peer.ID
}

TopicProvider provides a low-level abstraction for pubsub to perform topic-related queries. This abstraction is provided to encapsulate the pubsub implementation details from the rest of the codebase.