p2p

package
v0.13.1-patch.7 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Feb 1, 2021 License: AGPL-3.0 Imports: 50 Imported by: 10

Documentation

Overview

Package libp2p encapsulates the libp2p library

Index

Constants

View Source
const (
	// A unique Libp2p protocol ID prefix for Flow (https://docs.libp2p.io/concepts/protocols/)
	// All nodes communicate with each other using this protocol id suffixed with the id of the root block
	FlowLibP2PProtocolIDPrefix = "/flow/push/"

	// Maximum time to wait for a ping reply from a remote node
	PingTimeoutSecs = time.Second * 4
)
View Source
const (
	NoOp communicationMode = iota
	OneToOne
	OneToK
)
View Source
const (
	// defines maximum message size in publish and multicast modes
	DefaultMaxPubSubMsgSize = 1 << 21 // 2 mb

	// defines maximum message size in unicast mode
	DefaultMaxUnicastMsgSize = 5 * DefaultMaxPubSubMsgSize // 10 mb

	// maximum time to wait for a unicast request to complete
	DefaultUnicastTimeout = 2 * time.Second
)

Variables

View Source
var NetworkingSetFilter = filter.Not(filter.Ejected)

NetworkingSetFilter 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.

View Source
var PeerUpdateInterval = 1 * time.Minute

PeerUpdateInterval is how long the peer manager waits in between attempts to update peer connections

Functions

func ConnectednessToString

func ConnectednessToString(connectedness network.Connectedness) (string, bool)

ConnectednessToString reverse translates libp2p network connectedness to string

func CountStream

func CountStream(host host.Host, targetID peer.ID, protocol core.ProtocolID, dir network.Direction) int

CountStream finds total number of outbound stream to the target id

func DirectionToString

func DirectionToString(direction network.Direction) (string, bool)

DirectionToString reverse translates libp2p network direction to string

func FindOutboundStream

func FindOutboundStream(host host.Host, targetID peer.ID, protocol core.ProtocolID) (network.Stream, bool)

FindOutboundStream finds an existing outbound stream to the target id if it exists by querying libp2p

func IPPortFromMultiAddress

func IPPortFromMultiAddress(addrs ...multiaddr.Multiaddr) (string, string, error)

IPPortFromMultiAddress returns the IP/hostname and the port for the given multi-addresses associated with a libp2p host

func MultiAddressStr added in v0.13.0

func MultiAddressStr(ip, port string) string

MultiAddressStr receives a node ip and port and returns its corresponding Libp2p MultiAddressStr in string format in current implementation IP part of the node address is either an IP or a dns4. https://docs.libp2p.io/concepts/addressing/

func PeerAddressInfo added in v0.13.0

func PeerAddressInfo(identity flow.Identity) (peer.AddrInfo, error)

PeerAddressInfo generates the libp2p peer.AddrInfo for the given Flow.Identity. A node in flow is defined by a flow.Identity while it is defined by a peer.AddrInfo in libp2p. flow.Identity ---> peer.AddrInfo

|-- Address          --->   |-- []multiaddr.Multiaddr
|-- NetworkPublicKey --->   |-- ID

Types

type ChannelSubscriptionManager

type ChannelSubscriptionManager struct {
	// contains filtered or unexported fields
}

ChannelSubscriptionManager manages the engine to channelID subscription

func NewChannelSubscriptionManager

func NewChannelSubscriptionManager(mw network.Middleware) *ChannelSubscriptionManager

func (*ChannelSubscriptionManager) GetChannelIDs

func (sm *ChannelSubscriptionManager) GetChannelIDs() []string

GetChannelIDs returns list of topics this subscription manager has an engine registered for.

func (*ChannelSubscriptionManager) GetEngine

func (sm *ChannelSubscriptionManager) GetEngine(channelID string) (network.Engine, error)

func (*ChannelSubscriptionManager) Register

func (sm *ChannelSubscriptionManager) Register(channelID string, engine network.Engine) error

func (*ChannelSubscriptionManager) Unregister

func (sm *ChannelSubscriptionManager) Unregister(channelID string) error

type CloseFunc

type CloseFunc func(channelID string) error

CloseFunc is a function that unsubscribes the conduit from the channel

type Conduit

type Conduit struct {
	// contains filtered or unexported fields
}

Conduit is a helper of the overlay layer which functions as an accessor for sending messages within a single engine process. It sends all messages to what can be considered a bus reserved for that specific engine.

func (*Conduit) Close

func (c *Conduit) Close() error

func (*Conduit) Multicast

func (c *Conduit) Multicast(event interface{}, num uint, targetIDs ...flow.Identifier) error

Multicast unreliably sends the specified event to the specified number of recipients selected from the specified subset. The recipients are selected randomly from targetIDs

func (*Conduit) Publish

func (c *Conduit) Publish(event interface{}, targetIDs ...flow.Identifier) error

Publish sends an event to the network layer for unreliable delivery to subscribers of the given event on the network layer. It uses a publish-subscribe layer and can thus not guarantee that the specified recipients received the event.

func (*Conduit) Submit

func (c *Conduit) Submit(event interface{}, targetIDs ...flow.Identifier) error

Submit will submit an event for delivery on the engine bus that is reserved for events of the engine it was initialized with.

func (*Conduit) Unicast

func (c *Conduit) Unicast(event interface{}, targetID flow.Identifier) error

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

type ConnManager

type ConnManager struct {
	connmgr.NullConnMgr // a null conn mgr provided by libp2p to allow implementing only the functions needed
	// contains filtered or unexported fields
}

ConnManager provides an implementation of Libp2p's ConnManager interface (https://godoc.org/github.com/libp2p/go-libp2p-core/connmgr#ConnManager) It is called back by libp2p when certain events occur such as opening/closing a stream, opening/closing connection etc. This implementation updates networking metrics when a peer connection is added or removed

func NewConnManager

func NewConnManager(log zerolog.Logger, metrics module.NetworkMetrics) ConnManager

func (ConnManager) Connected

func (c ConnManager) Connected(n network.Network, con network.Conn)

called by libp2p when a connection opened

func (ConnManager) Disconnected

func (c ConnManager) Disconnected(n network.Network, con network.Conn)

called by libp2p when a connection closed

func (ConnManager) ListenCloseNotifee

func (c ConnManager) ListenCloseNotifee(n network.Network, m multiaddr.Multiaddr)

called by libp2p when network stops listening on an addr * This is never called back by libp2p currently and may be a bug on their side

func (ConnManager) ListenNotifee

func (c ConnManager) ListenNotifee(n network.Network, m multiaddr.Multiaddr)

called by libp2p when network starts listening on an addr

func (ConnManager) Notifee

func (c ConnManager) Notifee() network.Notifiee

type Connector

type Connector interface {

	// ConnectPeers connect to the given flow.Identities and returns a map of identifiers which failed.
	// ConnectPeer implementation should be idempotent such that multiple calls to connect to the same peer should not
	// return an error or create multiple connections
	ConnectPeers(ctx context.Context, ids flow.IdentityList) map[flow.Identifier]error

	// DisconnectPeers disconnect from the given flow.Identities and returns a map of identifiers which failed.
	// DisconnectPeers implementation should be idempotent such that multiple calls to connect to the same peer should
	// not return an error
	DisconnectPeers(ctx context.Context, ids flow.IdentityList) map[flow.Identifier]error
}

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

type LibP2PFactoryFunc added in v0.13.0

type LibP2PFactoryFunc func() (*Node, error)

LibP2PFactoryFunc is a factory function type for generating libp2p Node instances.

func DefaultLibP2PNodeFactory added in v0.13.0

func DefaultLibP2PNodeFactory(log zerolog.Logger, me flow.Identifier, address string, flowKey fcrypto.PrivateKey, rootBlockID string,
	maxPubSubMsgSize int, metrics module.NetworkMetrics) (LibP2PFactoryFunc, error)

DefaultLibP2PNodeFactory is a factory function that receives a middleware instance and generates a libp2p Node by invoking its factory with proper parameters.

type Middleware

type Middleware struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Middleware handles the input & output on the direct connections we have to our neighbours on the peer-to-peer network.

func NewMiddleware

func NewMiddleware(log zerolog.Logger,
	libP2PNodeFactory LibP2PFactoryFunc,
	flowID flow.Identifier,
	metrics module.NetworkMetrics,
	maxUnicastMsgSize int,
	maxPubSubMsgSize int,
	rootBlockID string,
	validators ...network.MessageValidator) *Middleware

NewMiddleware creates a new middleware instance with the given config and using the given codec to encode/decode messages to our peers.

func (*Middleware) GetIPPort

func (m *Middleware) GetIPPort() (string, string, error)

GetIPPort returns the ip address and port number associated with the middleware

func (*Middleware) IsConnected

func (m *Middleware) IsConnected(identity flow.Identity) (bool, error)

IsConnected returns true if this node is connected to the node with id nodeID.

func (*Middleware) Me

func (m *Middleware) Me() flow.Identifier

Me returns the flow identifier of the this middleware

func (*Middleware) Ping

func (m *Middleware) Ping(targetID flow.Identifier) (time.Duration, error)

Ping pings the target node and returns the ping RTT or an error

func (*Middleware) Publish

func (m *Middleware) Publish(msg *message.Message, channelID string) error

Publish publishes msg on the channel. It models a distributed broadcast where the message is meant for all or a many nodes subscribing to the channel ID. It does not guarantee the delivery though, and operates on a best effort.

func (*Middleware) Send deprecated

func (m *Middleware) Send(channelID string, msg *message.Message, targetIDs ...flow.Identifier) error

Send sends the message to the set of target ids If there is only one target NodeID, then a direct 1-1 connection is used by calling middleware.SendDirect Otherwise, middleware.Publish is used, which uses the PubSub method of communication.

Deprecated: Send exists for historical compatibility, and should not be used on new developments. It is planned to be cleaned up in near future. Proper utilization of Dispatch or Publish are recommended instead.

func (*Middleware) SendDirect

func (m *Middleware) SendDirect(msg *message.Message, targetID flow.Identifier) error

SendDirect sends msg on a 1-1 direct connection to the target ID. It models a guaranteed delivery asynchronous direct one-to-one connection on the underlying network. No intermediate node on the overlay is utilized as the router.

Dispatch should be used whenever guaranteed delivery to a specific target is required. Otherwise, Publish is a more efficient candidate.

func (*Middleware) Start

func (m *Middleware) Start(ov network.Overlay) error

Start will start the middleware.

func (*Middleware) Stop

func (m *Middleware) Stop()

Stop will end the execution of the middleware and wait for it to end.

func (*Middleware) Subscribe

func (m *Middleware) Subscribe(channelID string) error

Subscribe will subscribe the middleware for a topic with the fully qualified channel ID name

func (*Middleware) Unsubscribe

func (m *Middleware) Unsubscribe(channelID string) error

Unsubscribe will unsubscribe the middleware for a topic with the fully qualified channel ID name

func (*Middleware) UpdateAllowList

func (m *Middleware) UpdateAllowList() error

UpdateAllowList fetches the most recent identity of the nodes from overlay and updates the underlying libp2p node.

type MulticastFunc

type MulticastFunc func(channelID string, event interface{}, num uint, targetIDs ...flow.Identifier) error

MulticastFunc is a function that unreliably sends the event in the underlying network to randomly chosen subset of nodes from targetIDs

type Network

type Network struct {
	sync.RWMutex
	// 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

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) Done

func (n *Network) Done() <-chan struct{}

Done returns a channel that will close when shutdown is complete.

func (*Network) Identity

func (n *Network) Identity() (map[flow.Identifier]flow.Identity, error)

Identity returns a map of all flow.Identifier to flow identity by querying the flow state

func (*Network) Ready

func (n *Network) Ready() <-chan struct{}

Ready returns a channel that will close when the network stack is ready.

func (*Network) Receive

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

func (*Network) Register

func (n *Network) Register(channelID string, engine network.Engine) (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) SetIDs

func (n *Network) SetIDs(ids flow.IdentityList) error

SetIDs updates the identity list cached by the network layer

func (*Network) Topology

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

Topology returns the identities of a uniform subset of nodes in protocol state using the topology provided earlier. Independent invocations of Topology on different nodes collectively constructs a connected network graph.

type Node

type Node struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Node is a wrapper around LibP2P host.

func NewLibP2PNode added in v0.13.0

func NewLibP2PNode(logger zerolog.Logger,
	id flow.Identifier,
	address string,
	conMgr ConnManager,
	key fcrypto.PrivateKey,
	allowList bool,
	rootBlockID string,
	psOption ...pubsub.Option) (*Node, error)

func (*Node) AddPeer

func (n *Node) AddPeer(ctx context.Context, identity flow.Identity) error

AddPeer adds a peer to this node by adding it to this node's peerstore and connecting to it

func (*Node) CreateStream

func (n *Node) CreateStream(ctx context.Context, identity flow.Identity) (libp2pnet.Stream, error)

CreateStream returns an existing stream connected to identity, if it exists or adds one to identity as a peer and creates a new stream with it.

func (*Node) GetIPPort

func (n *Node) GetIPPort() (string, string, error)

GetIPPort returns the IP and Port the libp2p node is listening on.

func (*Node) Host

func (n *Node) Host() host.Host

Host returns pointer to host object of node.

func (*Node) IsConnected

func (n *Node) IsConnected(identity flow.Identity) (bool, error)

IsConnected returns true is address is a direct peer of this node else false

func (*Node) Ping

func (n *Node) Ping(ctx context.Context, identity flow.Identity) (time.Duration, error)

Ping pings a remote node and returns the time it took to ping the remote node if successful or the error

func (*Node) Publish

func (n *Node) Publish(ctx context.Context, topic string, data []byte) error

Publish publishes the given payload on the topic

func (*Node) RemovePeer

func (n *Node) RemovePeer(ctx context.Context, identity flow.Identity) error

RemovePeer closes the connection with the identity.

func (*Node) SetStreamHandler added in v0.13.0

func (n *Node) SetStreamHandler(handler libp2pnet.StreamHandler)

SetStreamHandler sets the stream handler of libp2p host of the node.

func (*Node) Stop

func (n *Node) Stop() (chan struct{}, error)

Stop stops the libp2p node.

func (*Node) Subscribe

func (n *Node) Subscribe(ctx context.Context, topic string) (*pubsub.Subscription, error)

Subscribe subscribes the node to the given topic and returns the subscription Currently only one subscriber is allowed per topic. NOTE: A node will receive its own published messages.

func (*Node) UnSubscribe

func (n *Node) UnSubscribe(topic string) error

UnSubscribe cancels the subscriber and closes the topic.

func (*Node) UpdateAllowList added in v0.13.0

func (n *Node) UpdateAllowList(identities flow.IdentityList) error

UpdateAllowList allows the peer allow list to be updated.

type NodeIDRefresher

type NodeIDRefresher struct {
	// contains filtered or unexported fields
}

NodeIDRefresher derives the latest list of flow identities with which the network should be communicating based on identity table changes in the protocol state.

func NewNodeIDRefresher

func NewNodeIDRefresher(logger zerolog.Logger, state protocol.ReadOnlyState, callBack func(list flow.IdentityList) error) *NodeIDRefresher

func (*NodeIDRefresher) OnIdentityTableChanged

func (listener *NodeIDRefresher) OnIdentityTableChanged()

OnIdentityTableChanged updates the networking layer's list of nodes to connect to when the identity table changes in the protocol state.

type PeerManager

type PeerManager struct {
	// contains filtered or unexported fields
}

PeerManager adds and removes connections to peers periodically and on request

func NewPeerManager

func NewPeerManager(ctx context.Context, logger zerolog.Logger, idsProvider func() (flow.IdentityList, error),
	connector Connector) *PeerManager

NewPeerManager creates a new peer manager which calls the idsProvider callback to get a list of peers to connect to and it uses the connector to actually connect or disconnect from peers.

func (*PeerManager) Done

func (pm *PeerManager) Done() <-chan struct{}

func (*PeerManager) Ready

func (pm *PeerManager) Ready() <-chan struct{}

Ready kicks off the ambient periodic connection updates.

func (*PeerManager) RequestPeerUpdate

func (pm *PeerManager) RequestPeerUpdate()

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.

type PublishFunc

type PublishFunc func(channelID string, event interface{}, targetIDs ...flow.Identifier) error

PublishFunc is a function that broadcasts the specified event to all participants on the given channel.

type RcvCache

type RcvCache struct {
	// contains filtered or unexported fields
}

RcvCache implements an LRU cache of the received eventIDs that delivered to their engines

type RcvCacheEntry

type RcvCacheEntry struct {
	// contains filtered or unexported fields
}

RcvCacheEntry represents an entry for the RcvCache

type SubmitFunc

type SubmitFunc func(channelID string, event interface{}, targetIDs ...flow.Identifier) error

SubmitFunc is a function that submits the given event for the given engine to the overlay network, which should take care of delivering it to the given recipients.

type UnicastFunc

type UnicastFunc func(channelID string, event interface{}, targetID flow.Identifier) error

UnicastFunc is a function that reliably sends the event via reliable 1-1 direct connection in the underlying network to the target ID.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL