network

package
v2.1.1+incompatible Latest Latest
Warning

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

 Go to latest Published: Aug 22, 2018 License: GPL-3.0 Imports: 19 Imported by: 0

README

Streaming

Streaming is a new protocol of the swarm bzz bundle of protocols. This protocol provides the basic logic for chunk-based data flow. It implements simple retrieve requests and delivery using priority queue. A data exchange stream is a directional flow of chunks between peers. The source of datachunks is the upstream, the receiver is called the downstream peer. Each streaming protocol defines an outgoing streamer and an incoming streamer, the former installing on the upstream, the latter on the downstream peer.

Subscribe on StreamerPeer launches an incoming streamer that sends a subscribe msg upstream. The streamer on the upstream peer handles the subscribe msg by installing the relevant outgoing streamer . The modules now engage in a process of upstream sending a sequence of hashes of chunks downstream (OfferedHashesMsg). The downstream peer evaluates which hashes are needed and get it delivered by sending back a msg (WantedHashesMsg).

Historical syncing is supported - currently not the right abstraction -- state kept across sessions by saving a series of intervals after their last batch actually arrived.

Live streaming is also supported, by starting session from the first item after the subscription.

Provable data exchange. In case a stream represents a swarm document's data layer or higher level chunks, streaming up to a certain index is always provable. It saves on sending intermediate chunks.

Using the streamer logic, various stream types are easy to implement:

  • light node requests:
    • url lookup with offset
    • document download
    • document upload
  • syncing
    • live session syncing
    • historical syncing
  • simple retrieve requests and deliveries
  • mutable resource updates streams
  • receipting for finger pointing

Syncing

Syncing is the process that makes sure storer nodes end up storing all and only the chunks that are requested from them.

Requirements

  • eventual consistency: so each chunk historical should be syncable
  • since the same chunk can and will arrive from many peers, (network traffic should be optimised, only one transfer of data per chunk)
  • explicit request deliveries should be prioritised higher than recent chunks received during the ongoing session which in turn should be higher than historical chunks.
  • insured chunks should get receipted for finger pointing litigation, the receipts storage should be organised efficiently, upstream peer should also be able to find these receipts for a deleted chunk easily to refute their challenge.
  • syncing should be resilient to cut connections, metadata should be persisted that keep track of syncing state across sessions, historical syncing state should survive restart
  • extra data structures to support syncing should be kept at minimum
  • syncing is organized separately for chunk types (resource update v content chunk)
  • various types of streams should have common logic abstracted

Syncing is now entirely mediated by the localstore, ie., no processes or memory leaks due to network contention. When a new chunk is stored, its chunk hash is index by proximity bin

peers syncronise by getting the chunks closer to the downstream peer than to the upstream one. Consequently peers just sync all stored items for the kad bin the receiving peer falls into. The special case of nearest neighbour sets is handled by the downstream peer indicating they want to sync all kademlia bins with proximity equal to or higher than their depth.

This sync state represents the initial state of a sync connection session. Retrieval is dictated by downstream peers simply using a special streamer protocol.

Syncing chunks created during the session by the upstream peer is called live session syncing while syncing of earlier chunks is historical syncing.

Once the relevant chunk is retrieved, downstream peer looks up all hash segments in its localstore and sends to the upstream peer a message with a a bitvector to indicate missing chunks (e.g., for chunk k, hash with chunk internal index which case ) new items. In turn upstream peer sends the relevant chunk data alongside their index.

On sending chunks there is a priority queue system. If during looking up hashes in its localstore, downstream peer hits on an open request then a retrieve request is sent immediately to the upstream peer indicating that no extra round of checks is needed. If another peers syncer hits the same open request, it is slightly unsafe to not ask that peer too: if the first one disconnects before delivering or fails to deliver and therefore gets disconnected, we should still be able to continue with the other. The minimum redundant traffic coming from such simultaneous eventualities should be sufficiently rare not to warrant more complex treatment.

Session syncing involves downstream peer to request a new state on a bin from upstream. using the new state, the range (of chunks) between the previous state and the new one are retrieved and chunks are requested identical to the historical case. After receiving all the missing chunks from the new hashes, downstream peer will request a new range. If this happens before upstream peer updates a new state, we say that session syncing is live or the two peers are in sync. In general the time interval passed since downstream peer request up to the current session cursor is a good indication of a permanent (probably increasing) lag.

If there is no historical backlog, and downstream peer has an acceptable 'last synced' tag, then it is said to be fully synced with the upstream peer. If a peer is fully synced with all its storer peers, it can advertise itself as globally fully synced.

The downstream peer persists the record of the last synced offset. When the two peers disconnect and reconnect syncing can start from there. This situation however can also happen while historical syncing is not yet complete. Effectively this means that the peer needs to persist a record of an arbitrary array of offset ranges covered.

Delivery requests

once the appropriate ranges of the hashstream are retrieved and buffered, downstream peer just scans the hashes, looks them up in localstore, if not found, create a request entry. The range is referenced by the chunk index. Alongside the name (indicating the stream, e.g., content chunks for bin 6) and the range downstream peer sends a 128 long bitvector indicating which chunks are needed. Newly created requests are satisfied bound together in a waitgroup which when done, will promptt sending the next one. to be able to do check and storage concurrently, we keep a buffer of one, we start with two batches of hashes. If there is nothing to give, upstream peers SetNextBatch is blocking. Subscription ends with an unsubscribe. which removes the syncer from the map.

Canceling requests (for instance the late chunks of an erasure batch) should be a chan closed on the request

Simple request is also a subscribe different streaming protocols are different p2p protocols with same message types. the constructor is the Run function itself. which takes a streamerpeer as argument

provable streams

The swarm hash over the hash stream has many advantages. It implements a provable data transfer and provide efficient storage for receipts in the form of inclusion proofs useable for finger pointing litigation. When challenged on a missing chunk, upstream peer will provide an inclusion proof of a chunk hash against the state of the sync stream. In order to be able to generate such an inclusion proof, upstream peer needs to store the hash index (counting consecutive hash-size segments) alongside the chunk data and preserve it even when the chunk data is deleted until the chunk is no longer insured. if there is no valid insurance on the files the entry may be deleted. As long as the chunk is preserved, no takeover proof will be needed since the node can respond to any challenge. However, once the node needs to delete an insured chunk for capacity reasons, a receipt should be available to refute the challenge by finger pointing to a downstream peer. As part of the deletion protocol then, hashes of insured chunks to be removed are pushed to an infinite stream for every bin.

Downstream peer on the other hand needs to make sure that they can only be finger pointed about a chunk they did receive and store. For this the check of a state should be exhaustive. If historical syncing finishes on one state, all hashes before are covered, no surprises. In other words historical syncing this process is self verifying. With session syncing however, it is not enough to check going back covering the range from old offset to new. Continuity (i.e., that the new state is extension of the old) needs to be verified: after downstream peer reads the range into a buffer, it appends the buffer the last known state at the last known offset and verifies the resulting hash matches the latest state. Past intervals of historical syncing are checked via the the session root. Upstream peer signs the states, downstream peers can use as handover proofs. Downstream peers sign off on a state together with an initial offset.

Once historical syncing is complete and the session does not lag, downstream peer only preserves the latest upstream state and store the signed version.

Upstream peer needs to keep the latest takeover states: each deleted chunk's hash should be covered by takeover proof of at least one peer. If historical syncing is complete, upstream peer typically will store only the latest takeover proof from downstream peer. Crucially, the structure is totally independent of the number of peers in the bin, so it scales extremely well.

implementation

The simplest protocol just involves upstream peer to prefix the key with the kademlia proximity order (say 0-15 or 0-31) and simply iterate on index per bin when syncing with a peer.

priority queues are used for sending chunks so that user triggered requests should be responded to first, session syncing second, and historical with lower priority. The request on chunks remains implemented as a dataless entry in the memory store. The lifecycle of this object should be more carefully thought through, ie., when it fails to retrieve it should be removed.

Documentation

Index

Constants

View Source 
const (
	DefaultNetworkID = 3
	// ProtocolMaxMsgSize maximum allowed message size
	ProtocolMaxMsgSize = 10 * 1024 * 1024
)

Variables

View Source 
var BzzSpec = &protocols.Spec{
	Name:       "bzz",
	Version:    6,
	MaxMsgSize: 10 * 1024 * 1024,
	Messages: []interface{}{
		HandshakeMsg{},
	},
}

BzzSpec is the spec of the generic swarm handshake

View Source 
var DiscoverySpec = &protocols.Spec{
	Name:       "hive",
	Version:    5,
	MaxMsgSize: 10 * 1024 * 1024,
	Messages: []interface{}{
		peersMsg{},
		subPeersMsg{},
	},
}

DiscoverySpec is the spec for the bzz discovery subprotocols

Functions

func Label added in v1.8.15 

func Label(e *entry) string

Label is a short tag for the entry for debug

func LogAddrs added in v1.8.15 

func LogAddrs(nns [][]byte) string

func NewNodeIDFromAddr added in v1.8.15 

func NewNodeIDFromAddr(addr Addr) discover.NodeID

NewNodeIDFromAddr transforms the underlay address to an adapters.NodeID

func NewPeerPotMap added in v1.8.15 

func NewPeerPotMap(kadMinProxSize int, addrs [][]byte) map[string]*PeerPot

NewPeerPotMap creates a map of pot record of OverlayAddr with keys as hexadecimal representations of the address.

func NotifyDepth added in v1.8.15 

func NotifyDepth(depth uint8, h Overlay)

NotifyDepth sends a message to all connections if depth of saturation is changed

func NotifyPeer added in v1.8.15 

func NotifyPeer(p OverlayAddr, k Overlay)

NotifyPeer informs all peers about a newly added node

func ToOverlayAddr added in v1.8.15 

func ToOverlayAddr(id []byte) []byte

ToOverlayAddr creates an overlayaddress from a byte slice

Types

type Addr added in v1.8.15 

type Addr interface {
	OverlayPeer
	Over() []byte
	Under() []byte
	String() string
	Update(OverlayAddr) OverlayAddr
}

Addr interface that peerPool needs

type Bzz 

type Bzz struct {
	*Hive
	NetworkID uint64
	LightNode bool
	// contains filtered or unexported fields
}

Bzz is the swarm protocol bundle

func NewBzz added in v1.8.15 

func NewBzz(config *BzzConfig, kad Overlay, store state.Store, streamerSpec *protocols.Spec, streamerRun func(*BzzPeer) error) *Bzz

NewBzz is the swarm protocol constructor arguments * bzz config * overlay driver * peer store

func (*Bzz) APIs added in v1.8.15 

func (b *Bzz) APIs() []rpc.API

APIs returns the APIs offered by bzz * hive Bzz implements the node.Service interface

func (*Bzz) GetHandshake added in v1.8.15 

func (b *Bzz) GetHandshake(peerID discover.NodeID) (*HandshakeMsg, bool)

GetHandshake returns the bzz handhake that the remote peer with peerID sent

func (*Bzz) NodeInfo added in v1.8.15 

func (b *Bzz) NodeInfo() interface{}

NodeInfo returns the node's overlay address

func (*Bzz) Protocols added in v1.8.15 

func (b *Bzz) Protocols() []p2p.Protocol

Protocols return the protocols swarm offers Bzz implements the node.Service interface * handshake/hive * discovery

func (*Bzz) RunProtocol added in v1.8.15 

func (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*BzzPeer) error) func(*p2p.Peer, p2p.MsgReadWriter) error

RunProtocol is a wrapper for swarm subprotocols returns a p2p protocol run function that can be assigned to p2p.Protocol#Run field arguments:

  • p2p protocol spec
  • run function taking BzzPeer as argument this run function is meant to block for the duration of the protocol session on return the session is terminated and the peer is disconnected

the protocol waits for the bzz handshake is negotiated the overlay address on the BzzPeer is set from the remote handshake

func (*Bzz) UpdateLocalAddr added in v1.8.15 

func (b *Bzz) UpdateLocalAddr(byteaddr []byte) *BzzAddr

UpdateLocalAddr updates underlayaddress of the running node

type BzzAddr added in v1.8.15 

type BzzAddr struct {
	OAddr []byte
	UAddr []byte
}

BzzAddr implements the PeerAddr interface

func NewAddrFromNodeID added in v1.8.15 

func NewAddrFromNodeID(id discover.NodeID) *BzzAddr

NewAddrFromNodeID constucts a BzzAddr from a discover.NodeID the overlay address is derived as the hash of the nodeID

func NewAddrFromNodeIDAndPort added in v1.8.15 

func NewAddrFromNodeIDAndPort(id discover.NodeID, host net.IP, port uint16) *BzzAddr

NewAddrFromNodeIDAndPort constucts a BzzAddr from a discover.NodeID and port uint16 the overlay address is derived as the hash of the nodeID

func RandomAddr added in v1.8.15 

func RandomAddr() *BzzAddr

RandomAddr is a utility method generating an address from a public key

func ToAddr added in v1.8.15 

func ToAddr(pa OverlayPeer) *BzzAddr

ToAddr returns the serialisable version of u

func (*BzzAddr) Address added in v1.8.15 

func (a *BzzAddr) Address() []byte

Address implements OverlayPeer interface to be used in Overlay

func (*BzzAddr) ID added in v1.8.15 

func (a *BzzAddr) ID() discover.NodeID

ID returns the nodeID from the underlay enode address

func (*BzzAddr) Over added in v1.8.15 

func (a *BzzAddr) Over() []byte

Over returns the overlay address

func (*BzzAddr) String added in v1.8.15 

func (a *BzzAddr) String() string

String pretty prints the address

func (*BzzAddr) Under added in v1.8.15 

func (a *BzzAddr) Under() []byte

Under returns the underlay address

func (*BzzAddr) Update added in v1.8.15 

func (a *BzzAddr) Update(na OverlayAddr) OverlayAddr

Update updates the underlay address of a peer record

type BzzConfig added in v1.8.15 

type BzzConfig struct {
	OverlayAddr  []byte // base address of the overlay network
	UnderlayAddr []byte // node's underlay address
	HiveParams   *HiveParams
	NetworkID    uint64
	LightNode    bool
}

BzzConfig captures the config params used by the hive

type BzzPeer added in v1.8.15 

type BzzPeer struct {
	*protocols.Peer // represents the connection for online peers

	*BzzAddr // remote address -> implements Addr interface = protocols.Peer

	LightNode bool
	// contains filtered or unexported fields
}

BzzPeer is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer) implements the Peer interface and all interfaces Peer implements: Addr, OverlayPeer

func NewBzzTestPeer added in v1.8.15 

func NewBzzTestPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer

func (*BzzPeer) LastActive added in v1.8.15 

func (p *BzzPeer) LastActive() time.Time

LastActive returns the time the peer was last active

func (*BzzPeer) Off added in v1.8.15 

func (p *BzzPeer) Off() OverlayAddr

Off returns the overlay peer record for offline persistence

type Conn added in v1.8.15 

type Conn interface {
	ID() discover.NodeID                                                                  // the key that uniquely identifies the Node for the peerPool
	Handshake(context.Context, interface{}, func(interface{}) error) (interface{}, error) // can send messages
	Send(context.Context, interface{}) error                                              // can send messages
	Drop(error)                                                                           // disconnect this peer
	Run(func(context.Context, interface{}) error) error                                   // the run function to run a protocol
	Off() OverlayAddr
}

Conn interface represents an live peer connection

type HandshakeMsg added in v1.8.15 

type HandshakeMsg struct {
	Version   uint64
	NetworkID uint64
	Addr      *BzzAddr
	LightNode bool
	// contains filtered or unexported fields
}
Handshake

* Version: 8 byte integer version of the protocol * NetworkID: 8 byte integer network identifier * Addr: the address advertised by the node including underlay and overlay connecctions

func (*HandshakeMsg) String added in v1.8.15 

func (bh *HandshakeMsg) String() string

String pretty prints the handshake

type Health added in v1.8.15 

type Health struct {
	KnowNN     bool     // whether node knows all its nearest neighbours
	GotNN      bool     // whether node is connected to all its nearest neighbours
	CountNN    int      // amount of nearest neighbors connected to
	CulpritsNN [][]byte // which known NNs are missing
	Full       bool     // whether node has a peer in each kademlia bin (where there is such a peer)
	Hive       string
}

Health state of the Kademlia

type Hive 

type Hive struct {
	*HiveParams             // settings
	Overlay                 // the overlay connectiviy driver
	Store       state.Store // storage interface to save peers across sessions
	// contains filtered or unexported fields
}

Hive manages network connections of the swarm node

func NewHive 

func NewHive(params *HiveParams, overlay Overlay, store state.Store) *Hive

NewHive constructs a new hive HiveParams: config parameters Overlay: connectivity driver using a network topology StateStore: to save peers across sessions

func (*Hive) NodeInfo added in v1.8.15 

func (h *Hive) NodeInfo() interface{}

NodeInfo function is used by the p2p.server RPC interface to display protocol specific node information

func (*Hive) PeerInfo added in v1.8.15 

func (h *Hive) PeerInfo(id discover.NodeID) interface{}

PeerInfo function is used by the p2p.server RPC interface to display protocol specific information any connected peer referred to by their NodeID

func (*Hive) Run added in v1.8.15 

func (h *Hive) Run(p *BzzPeer) error

Run protocol run function

func (*Hive) Start 

func (h *Hive) Start(server *p2p.Server) error

Start stars the hive, receives p2p.Server only at startup server is used to connect to a peer based on its NodeID or enode URL these are called on the p2p.Server which runs on the node

func (*Hive) Stop 

func (h *Hive) Stop() error

Stop terminates the updateloop and saves the peers

type HiveParams 

type HiveParams struct {
	Discovery             bool  // if want discovery of not
	PeersBroadcastSetSize uint8 // how many peers to use when relaying
	MaxPeersPerRequest    uint8 // max size for peer address batches
	KeepAliveInterval     time.Duration
}

HiveParams holds the config options to hive

func NewHiveParams added in v1.8.15 

func NewHiveParams() *HiveParams

NewHiveParams returns hive config with only the

type KadParams added in v1.8.15 

type KadParams struct {
	// adjustable parameters
	MaxProxDisplay int   // number of rows the table shows
	MinProxBinSize int   // nearest neighbour core minimum cardinality
	MinBinSize     int   // minimum number of peers in a row
	MaxBinSize     int   // maximum number of peers in a row before pruning
	RetryInterval  int64 // initial interval before a peer is first redialed
	RetryExponent  int   // exponent to multiply retry intervals with
	MaxRetries     int   // maximum number of redial attempts
	// function to sanction or prevent suggesting a peer
	Reachable func(OverlayAddr) bool
}

KadParams holds the config params for Kademlia

func NewKadParams added in v1.8.15 

func NewKadParams() *KadParams

NewKadParams returns a params struct with default values

type Kademlia added in v1.8.15 

type Kademlia struct {
	*KadParams // Kademlia configuration parameters
	// contains filtered or unexported fields
}

Kademlia is a table of live peers and a db of known peers (node records)

func NewKademlia added in v1.8.15 

func NewKademlia(addr []byte, params *KadParams) *Kademlia

NewKademlia creates a Kademlia table for base address addr with parameters as in params if params is nil, it uses default values

func (*Kademlia) AddrCountC added in v1.8.15 

func (k *Kademlia) AddrCountC() <-chan int

AddrCountC returns the channel that sends a new address count value on each change. Not receiving from the returned channel will block Register function when address count value changes.

func (*Kademlia) BaseAddr added in v1.8.15 

func (k *Kademlia) BaseAddr() []byte

BaseAddr return the kademlia base address

func (*Kademlia) EachAddr added in v1.8.15 

func (k *Kademlia) EachAddr(base []byte, o int, f func(OverlayAddr, int, bool) bool)

EachAddr called with (base, po, f) is an iterator applying f to each known peer that has proximity order po or less as measured from the base if base is nil, kademlia base address is used

func (*Kademlia) EachBin added in v1.8.15 

func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(conn OverlayConn, po int) bool)

func (*Kademlia) EachConn added in v1.8.15 

func (k *Kademlia) EachConn(base []byte, o int, f func(OverlayConn, int, bool) bool)

EachConn is an iterator with args (base, po, f) applies f to each live peer that has proximity order po or less as measured from the base if base is nil, kademlia base address is used

func (*Kademlia) Healthy added in v1.8.15 

func (k *Kademlia) Healthy(pp *PeerPot) *Health

Healthy reports the health state of the kademlia connectivity returns a Health struct

func (*Kademlia) NeighbourhoodDepthC added in v1.8.15 

func (k *Kademlia) NeighbourhoodDepthC() <-chan int

NeighbourhoodDepthC returns the channel that sends a new kademlia neighbourhood depth on each change. Not receiving from the returned channel will block On function when the neighbourhood depth is changed.

func (*Kademlia) Off added in v1.8.15 

func (k *Kademlia) Off(p OverlayConn)

Off removes a peer from among live peers

func (*Kademlia) On added in v1.8.15 

func (k *Kademlia) On(p OverlayConn) (uint8, bool)

On inserts the peer as a kademlia peer into the live peers

func (*Kademlia) Register added in v1.8.15 

func (k *Kademlia) Register(peers []OverlayAddr) error

Register enters each OverlayAddr as kademlia peer record into the database of known peer addresses

func (*Kademlia) String added in v1.8.15 

func (k *Kademlia) String() string

String returns kademlia table + kaddb table displayed with ascii

func (*Kademlia) SuggestPeer added in v1.8.15 

func (k *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool)

SuggestPeer returns a known peer for the lowest proximity bin for the lowest bincount below depth naturally if there is an empty row it returns a peer for that

type Overlay added in v1.8.15 

type Overlay interface {
	// suggest peers to connect to
	SuggestPeer() (OverlayAddr, int, bool)
	// register and deregister peer connections
	On(OverlayConn) (depth uint8, changed bool)
	Off(OverlayConn)
	// register peer addresses
	Register([]OverlayAddr) error
	// iterate over connected peers
	EachConn([]byte, int, func(OverlayConn, int, bool) bool)
	// iterate over known peers (address records)
	EachAddr([]byte, int, func(OverlayAddr, int, bool) bool)
	// pretty print the connectivity
	String() string
	// base Overlay address of the node itself
	BaseAddr() []byte
	// connectivity health check used for testing
	Healthy(*PeerPot) *Health
}

Overlay is the interface for kademlia (or other topology drivers)

type OverlayAddr added in v1.8.15 

type OverlayAddr interface {
	OverlayPeer
	Update(OverlayAddr) OverlayAddr // returns the updated version of the original
}

OverlayAddr represents a kademlia peer record

type OverlayConn added in v1.8.15 

type OverlayConn interface {
	OverlayPeer
	Drop(error)       // call to indicate a peer should be expunged
	Off() OverlayAddr // call to return a persitent OverlayAddr
}

OverlayConn represents a connected peer

type OverlayPeer added in v1.8.15 

type OverlayPeer interface {
	Address() []byte
}

OverlayPeer interface captures the common aspect of view of a peer from the Overlay topology driver

type Peer added in v1.8.15 

type Peer interface {
	Addr                   // the address of a peer
	Conn                   // the live connection (protocols.Peer)
	LastActive() time.Time // last time active
}

Peer interface represents an live peer connection

type PeerPot added in v1.8.15 

type PeerPot struct {
	NNSet     [][]byte
	EmptyBins []int
}

PeerPot keeps info about expected nearest neighbours and empty bins used for testing only

Source Files

View all Source files

Directories

Path Synopsis
You can run this simulation using
 You can run this simulation using
discovery
intervals
testing

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.