peers

README

Peer pool signals

Peer pool sends 3 types of signals.

Discovery started signal will be sent once discovery server is started. And every time node will have to re-start discovery server because peer number dropped too low.

{
  "type": "discovery.started",
  "event": null
}

Discovery stopped signal will be sent once discovery found max limit of peers for every registered topic.

{
  "type": "discovery.stopped",
  "event": null
}

Discovery summary signal will be sent every time new peer is added or removed from a cluster. It will contain a map with capability as a key and total numbers of peers with that capability as a value.

{
  "type": "discovery.summary",
  "event": [
    {
      "id": "339c84c816b5f17a622c8d7ab9498f9998e942a274f70794af934bf5d3d02e14db8ddca2170e4edccede29ea6d409b154c141c34c01006e76c95e17672a27454",
      "name": "peer-0/v1.0/darwin/go1.10.1",
      "caps": [
        "shh/6"
      ],
      "network": {
        "localAddress": "127.0.0.1:61049",
        "remoteAddress": "127.0.0.1:33732",
        "inbound": false,
        "trusted": false,
        "static": true
      },
      "protocols": {
        "shh": "unknown"
      }
    }
  ]
}

Or if we don't have any peers:

{
  "type": "discovery.summary",
  "event": []
}

Documentation

Index

• Constants
• Variables
• func SendDiscoverySummary(peers []*p2p.PeerInfo)
• func StartDiscv5(server *p2p.Server) (*discv5.Network, error)
• type Cache
  • func NewCache(db *leveldb.DB) *Cache
  • func (d *Cache) AddPeer(peer *discv5.Node, topic discv5.Topic) error
  • func (d *Cache) GetPeersRange(topic discv5.Topic, limit int) (nodes []*discv5.Node)
  • func (d *Cache) RemovePeer(peerID discv5.NodeID, topic discv5.Topic) error
• type Options
  • func NewDefaultOptions() *Options
• type PeerPool
  • func NewPeerPool(config map[discv5.Topic]params.Limits, cache *Cache, options *Options) *PeerPool
  • func (p *PeerPool) Start(server *p2p.Server) error
  • func (p *PeerPool) Stop()
• type PoolEvent
• type Register
  • func NewRegister(topics ...discv5.Topic) *Register
  • func (r *Register) Start(server *p2p.Server) error
  • func (r *Register) Stop()
• type TopicPool
  • func NewTopicPool(topic discv5.Topic, limits params.Limits, slowMode, fastMode time.Duration, ...) *TopicPool
  • func (t *TopicPool) AddPeerFromTable(server *p2p.Server) *discv5.Node
  • func (t *TopicPool) BelowMin() bool
  • func (t *TopicPool) ConfirmAdded(server *p2p.Server, nodeID discover.NodeID)
  • func (t *TopicPool) ConfirmDropped(server *p2p.Server, nodeID discover.NodeID) bool
  • func (t *TopicPool) MaxReached() bool
  • func (t *TopicPool) SearchRunning() bool
  • func (t *TopicPool) StartSearch(server *p2p.Server) error
  • func (t *TopicPool) StopSearch()

Constants

const (

	// DefaultFastSync is a recommended value for aggressive peers search.
	DefaultFastSync = 3 * time.Second
	// DefaultSlowSync is a recommended value for slow (background) peers search.
	DefaultSlowSync = 30 * time.Second
	// DefaultDiscV5Timeout is a timeout after which Discv5 is stopped.
	DefaultDiscV5Timeout = 3 * time.Minute
	// DefaultTopicFastModeTimeout is a timeout after which sync mode is switched to slow mode.
	DefaultTopicFastModeTimeout = 30 * time.Second
	// DefaultTopicStopSearchDelay is the default delay when stopping a topic search.
	DefaultTopicStopSearchDelay = 10 * time.Second
)

Variables

var (
	// ErrDiscv5NotRunning returned when pool is started but discover v5 is not running or not enabled.
	ErrDiscv5NotRunning = errors.New("Discovery v5 is not running")
)

Functions

func SendDiscoverySummary

func SendDiscoverySummary(peers []*p2p.PeerInfo)

SendDiscoverySummary sends discovery.summary signal.

func StartDiscv5

func StartDiscv5(server *p2p.Server) (*discv5.Network, error)

StartDiscv5 starts discv5 udp listener. This is done here to avoid patching p2p server, we can't hold a lock of course but no other sub-process should use discovery

Types

type Cache

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

Cache maintains list of peers that were discovered.

func NewCache

func NewCache(db *leveldb.DB) *Cache

NewCache returns instance of PeersDatabase

func (*Cache) AddPeer

func (d *Cache) AddPeer(peer *discv5.Node, topic discv5.Topic) error

AddPeer stores peer with a following key: <topic><peer ID>

func (*Cache) GetPeersRange

func (d *Cache) GetPeersRange(topic discv5.Topic, limit int) (nodes []*discv5.Node)

GetPeersRange returns peers for a given topic with a limit.

func (*Cache) RemovePeer

func (d *Cache) RemovePeer(peerID discv5.NodeID, topic discv5.Topic) error

RemovePeer deletes a peer from database.

type Options

type Options struct {
	FastSync time.Duration
	SlowSync time.Duration
	// After this time, Discovery is stopped even if max peers is not reached.
	DiscServerTimeout time.Duration
	// AllowStop allows stopping Discovery when reaching max peers or after timeout.
	AllowStop bool
	// TopicStopSearchDelay time stopSearch will be waiting for max cached peers to be
	// filled before really stopping the search.
	TopicStopSearchDelay time.Duration
}

Options is a struct with PeerPool configuration.

func NewDefaultOptions

func NewDefaultOptions() *Options

NewDefaultOptions returns a struct with default Options.

type PeerPool

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

PeerPool manages discovered peers and connects them to p2p server

func NewPeerPool

func NewPeerPool(config map[discv5.Topic]params.Limits, cache *Cache, options *Options) *PeerPool

NewPeerPool creates instance of PeerPool

func (*PeerPool) Start

func (p *PeerPool) Start(server *p2p.Server) error

Start creates topic pool for each topic in config and subscribes to server events.

func (*PeerPool) Stop

func (p *PeerPool) Stop()

Stop closes pool quit channel and all channels that are watched by search queries and waits till all goroutines will exit.

type PoolEvent

type PoolEvent string

PoolEvent is a type used to for peer pool events.

type Register

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

Register manages register topic queries

func NewRegister

func NewRegister(topics ...discv5.Topic) *Register

NewRegister creates instance of topic register

func (*Register) Start

func (r *Register) Start(server *p2p.Server) error

Start topic register query for every topic

func (*Register) Stop

func (r *Register) Stop()

Stop all register topic queries and waits for them to exit

type TopicPool

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

TopicPool manages peers for topic.

func NewTopicPool

func NewTopicPool(topic discv5.Topic, limits params.Limits, slowMode, fastMode time.Duration, cache *Cache) *TopicPool

NewTopicPool returns instance of TopicPool

func (*TopicPool) AddPeerFromTable

func (t *TopicPool) AddPeerFromTable(server *p2p.Server) *discv5.Node

AddPeerFromTable checks if there is a valid peer in local table and adds it to a server.

func (*TopicPool) BelowMin

func (t *TopicPool) BelowMin() bool

BelowMin returns true if current number of peers is below min limit.

func (*TopicPool) ConfirmAdded

func (t *TopicPool) ConfirmAdded(server *p2p.Server, nodeID discover.NodeID)

ConfirmAdded called when peer was added by p2p Server.

1. Skip a peer if it not in our peer table
2. Add a peer to a cache.
3. Disconnect a peer if it was connected after we reached max limit of peers. (we can't know in advance if peer will be connected, thats why we allow to overflow for short duration)
4. Switch search to slow mode if it is running.

func (*TopicPool) ConfirmDropped

func (t *TopicPool) ConfirmDropped(server *p2p.Server, nodeID discover.NodeID) bool

ConfirmDropped called when server receives drop event. 1. Skip peer if it is not in our peer table. 2. If disconnect request - we could drop that peer ourselves. 3. If connected number will drop below min limit - switch to fast mode. 4. Delete a peer from cache and peer table. Returns false if peer is not in our table or we requested removal of this peer. Otherwise peer is removed and true is returned.

func (*TopicPool) MaxReached

func (t *TopicPool) MaxReached() bool

MaxReached returns true if we connected with max number of peers.

func (*TopicPool) SearchRunning

func (t *TopicPool) SearchRunning() bool

SearchRunning returns true if search is running

func (*TopicPool) StartSearch

func (t *TopicPool) StartSearch(server *p2p.Server) error

StartSearch creates discv5 queries and runs a loop to consume found peers.

func (*TopicPool) StopSearch

func (t *TopicPool) StopSearch()

StopSearch stops the closes stop