network

package
v0.0.0-...-39cdb81 Latest Latest
Warning

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

Go to latest
Published: Jan 21, 2019 License: GPL-3.0 Imports: 21 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
  • swarm feeds 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 not organized separately for chunk types (Swarm feed updates v regular 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 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
)

Variables

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

bzzspec是通用群握手的规范

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

discovery spec是bzz discovery子协议的规范

View Source
var Pof = pot.DefaultPof(256)
View Source
var RequestTimeout = 10 * time.Second

考虑跳过对等机的时间。 也用于流传送。

Functions

func Label

func Label(e *entry) string

label是调试条目的短标记

func LogAddrs

func LogAddrs(nns [][]byte) string

func NewPeerPotMap

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

newpeerpotmap用键创建一个pot记录的映射*bzzaddr 作为地址的十六进制表示。 使用通过的卡德米利亚的邻里大小 仅用于测试 TODO移动到单独的测试工具文件

func NotifyDepth

func NotifyDepth(depth uint8, kad *Kademlia)

如果饱和深度更改,notifydepth将向所有连接发送消息

func NotifyPeer

func NotifyPeer(p *BzzAddr, k *Kademlia)

notifypeer通知所有对等方新添加的节点

Types

type Bzz

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

bzz是swarm协议包

func NewBzz

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

Newzz是Swarm协议的构造者 争论 *BZZ配置 *覆盖驱动程序 *对等存储

func (*Bzz) APIs

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

API返回BZZ提供的API *蜂箱 bzz实现node.service接口

func (*Bzz) GetOrCreateHandshake

func (b *Bzz) GetOrCreateHandshake(peerID enode.ID) (*HandshakeMsg, bool)

gethandshake返回peerid远程对等机发送的bzz handshake

func (*Bzz) NodeInfo

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

nodeinfo返回节点的覆盖地址

func (*Bzz) Protocols

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

协议返回Swarm提供的协议 bzz实现node.service接口 *握手/蜂窝 *发现

func (*Bzz) RunProtocol

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

runprotocol是swarm子协议的包装器 返回可分配给p2p.protocol run字段的p2p协议运行函数。 争论: *P2P协议规范 *以bzzpeer为参数运行函数 此运行函数用于在协议会话期间阻塞 返回时,会话终止,对等端断开连接。 协议等待BZZ握手被协商 bzzpeer上的覆盖地址是通过远程握手设置的。

func (*Bzz) UpdateLocalAddr

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

updateLocalAddr更新正在运行的节点的参考底图地址

type BzzAddr

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

bzzaddr实现peeraddr接口

func NewAddr

func NewAddr(node *enode.Node) *BzzAddr

newaddr从节点记录构造bzzaddr。

func RandomAddr

func RandomAddr() *BzzAddr

randomaddr是从公钥生成地址的实用方法

func (*BzzAddr) Address

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

地址实现覆盖中要使用的覆盖对等接口。

func (*BzzAddr) ID

func (a *BzzAddr) ID() enode.ID

ID返回参考底图中的节点标识符。

func (*BzzAddr) Over

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

over返回覆盖地址。

func (*BzzAddr) String

func (a *BzzAddr) String() string

字符串漂亮地打印地址

func (*BzzAddr) Under

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

在下返回参考底图地址。

func (*BzzAddr) Update

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

更新更新更新对等记录的底层地址

type BzzConfig

type BzzConfig struct {
	OverlayAddr  []byte //覆盖网络的基址
	UnderlayAddr []byte //节点的参考底图地址
	HiveParams   *HiveParams
	NetworkID    uint64
	LightNode    bool
}

bzzconfig捕获配置单元使用的配置参数

type BzzPeer

type BzzPeer struct {
	*protocols.Peer //表示联机对等机的连接
	*BzzAddr        //远程地址->实现addr interface=protocols.peer

	LightNode bool
	// contains filtered or unexported fields
}

bzz peer是协议的bzz协议视图。peer(本身是p2p.peer的扩展) 实现对等接口和所有接口对等实现:addr、overlaypeer

func NewBzzPeer

func NewBzzPeer(p *protocols.Peer) *BzzPeer

func (*BzzPeer) ID

func (p *BzzPeer) ID() enode.ID

ID返回对等方的参考线节点标识符。

type Fetcher

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

当在本地找不到块时,将创建提取程序。它启动一次请求处理程序循环,然后 在所有活动请求完成之前保持活动状态。这可能发生: 1。或者因为块被传递 2。或者因为请求者取消/超时 获取器在完成后自行销毁。 TODO:取消终止后的所有转发请求

func NewFetcher

func NewFetcher(addr storage.Address, rf RequestFunc, skipCheck bool) *Fetcher

new fetcher使用给定的请求函数为给定的块地址创建一个新的fetcher。

func (*Fetcher) Offer

func (f *Fetcher) Offer(ctx context.Context, source *enode.ID)

当上游对等端通过同步作为“offeredhashemsg”的一部分来提供区块,并且节点在本地没有区块时,调用offer。

func (*Fetcher) Request

func (f *Fetcher) Request(ctx context.Context, hopCount uint8)

当上游对等端作为“retrieverequestmsg”的一部分或通过filestore从本地请求请求请求块,并且节点在本地没有块时,调用请求。

type FetcherFactory

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

FetcherFactory是用请求函数初始化的,可以创建Fetcher

func NewFetcherFactory

func NewFetcherFactory(request RequestFunc, skipCheck bool) *FetcherFactory

NewFetcherFactory接受请求函数并跳过检查参数并创建FetcherFactory

func (*FetcherFactory) New

func (f *FetcherFactory) New(ctx context.Context, source storage.Address, peersToSkip *sync.Map) storage.NetFetcher

new为给定的块构造一个新的获取器。PeersToSkip中的所有对等机 不要求传递给定的块。PeersToSkip应该始终 包含主动请求此块的对等方,以确保 不要向他们要求回块。 创建的获取器将启动并返回。

type HandshakeMsg

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

func (*HandshakeMsg) String

func (bh *HandshakeMsg) String() string

字符串漂亮地打印了握手

type Health

type Health struct {
	KnowNN           bool     //节点是否知道其所有邻居
	CountKnowNN      int      //已知邻居数量
	MissingKnowNN    [][]byte //我们应该知道哪些邻居,但我们不知道
	ConnectNN        bool     //节点是否连接到其所有邻居
	CountConnectNN   int      //连接到的邻居数量
	MissingConnectNN [][]byte //我们应该和哪个邻居有联系,但我们没有
	Saturated        bool     //我们是否与所有我们想联系的同龄人建立了联系
	Hive             string
}

卡德米利亚的健康状况 仅用于测试

type Hive

type Hive struct {
	*HiveParams             //设置
	*Kademlia               //覆盖连接驱动程序
	Store       state.Store //存储接口,用于跨会话保存对等端
	// contains filtered or unexported fields
}

蜂巢管理群节点的网络连接

func NewHive

func NewHive(params *HiveParams, kad *Kademlia, store state.Store) *Hive

new hive构造新的hive hiveparams:配置参数 Kademlia:使用网络拓扑的连接驱动程序 Statestore:保存会话之间的对等点

func (*Hive) NodeInfo

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

p2p.server rpc接口使用nodeinfo函数显示 协议特定的节点信息

func (*Hive) PeerInfo

func (h *Hive) PeerInfo(id enode.ID) interface{}

p2p.server rpc接口使用peerinfo函数来显示 协议特定信息节点ID引用的任何连接的对等点

func (*Hive) Run

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

运行协议运行函数

func (*Hive) Start

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

启动星型配置单元,仅在启动时接收p2p.server 服务器用于基于其nodeid或enode url连接到对等机 在节点上运行的p2p.server上调用这些

func (*Hive) Stop

func (h *Hive) Stop() error

stop终止updateLoop并保存对等方

type HiveParams

type HiveParams struct {
	Discovery             bool  //如果不想发现
	PeersBroadcastSetSize uint8 //中继时要使用多少对等点
	MaxPeersPerRequest    uint8 //对等地址批的最大大小
	KeepAliveInterval     time.Duration
}

hiveparams保存配置选项以进行配置

func NewHiveParams

func NewHiveParams() *HiveParams

newhiveparams返回hive config,只使用

type KadParams

type KadParams struct {
	// 可调参数
	MaxProxDisplay    int   //表显示的行数
	NeighbourhoodSize int   //最近邻核最小基数
	MinBinSize        int   //一行中的最小对等数
	MaxBinSize        int   //修剪前一行中的最大对等数
	RetryInterval     int64 //对等机首次重新拨号前的初始间隔
	RetryExponent     int   //用指数乘以重试间隔
	MaxRetries        int   //重拨尝试的最大次数
	//制裁或阻止建议同伴的职能
	Reachable func(*BzzAddr) bool `json:"-"`
}

kadparams保存kademlia的配置参数

func NewKadParams

func NewKadParams() *KadParams

newkadparams返回带有默认值的params结构

type Kademlia

type Kademlia struct {
	*KadParams //Kademlia配置参数
	// contains filtered or unexported fields
}

Kademlia是一个活动对等端表和一个已知对等端数据库(节点记录)

func NewKademlia

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

newkademlia为基地址addr创建一个kademlia表 参数与参数相同 如果params为nil,则使用默认值

func (*Kademlia) AddrCountC

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

addrCountc返回发送新的 每次更改的地址计数值。 不从返回通道接收将阻止寄存器功能 地址计数值更改时。

func (*Kademlia) BaseAddr

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

baseaddr返回kademlia基地址

func (*Kademlia) EachAddr

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

用(base,po,f)调用的eachaddr是一个迭代器,将f应用于每个已知的对等端 从底部测量,接近顺序为O或更低 如果基为零,则使用Kademlia基地址

func (*Kademlia) EachConn

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

eachconn是一个带有args(base、po、f)的迭代器,将f应用于每个活动对等端 从基地测量,接近订单为po或更低 如果基为零,则使用Kademlia基地址

func (*Kademlia) Healthy

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

健康报告Kademlia连接性的健康状态

peerpot参数提供了网络的全知视图 结果健康对象是 问题中的卡德米利亚(接受者)的实际组成是什么,以及 当我们考虑到我们对网络的所有了解时,应该是什么情况呢?

仅用于测试

func (*Kademlia) NeighbourhoodDepth

func (k *Kademlia) NeighbourhoodDepth() (depth int)

Neighbourhooddepth返回壶的深度,请参见壶的深度

func (*Kademlia) NeighbourhoodDepthC

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

Neighbourhooddepthc返回发送新Kademlia的频道 每一次变化的邻里深度。 不从返回通道接收将阻塞功能 当邻近深度改变时。 托多:为什么要导出它,如果应该的话;为什么我们不能有多个订户?

func (*Kademlia) Off

func (k *Kademlia) Off(p *Peer)

关闭从活动对等中删除对等

func (*Kademlia) On

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

在上,将对等机作为Kademlia对等机插入活动对等机

func (*Kademlia) Register

func (k *Kademlia) Register(peers ...*BzzAddr) error

寄存器将每个地址作为kademlia对等记录输入 已知对等地址数据库

func (*Kademlia) String

func (k *Kademlia) String() string

字符串返回用ASCII显示的kademlia表+kaddb表

func (*Kademlia) SuggestPeer

func (k *Kademlia) SuggestPeer() (suggestedPeer *BzzAddr, saturationDepth int, changed bool)

SuggestPeer返回未连接的对等地址作为连接的对等建议

type Peer

type Peer struct {
	*BzzPeer
	// contains filtered or unexported fields
}

Peer包装BZZPeer并嵌入Kademlia覆盖连接驱动程序

func NewPeer

func NewPeer(p *BzzPeer, kad *Kademlia) *Peer

newpeer构造发现对等

func (*Peer) HandleMsg

func (d *Peer) HandleMsg(ctx context.Context, msg interface{}) error

handlemsg是委托传入消息的消息处理程序

func (*Peer) NotifyDepth

func (d *Peer) NotifyDepth(po uint8)

notifydepth向接收者发送一个子程序msg,通知他们 饱和深度的变化

func (*Peer) NotifyPeer

func (d *Peer) NotifyPeer(a *BzzAddr, po uint8)

如果出现以下情况,notifypeer将通知远程节点(收件人)有关对等机的信息: 对等方的采购订单在收件人的广告深度内 或者对方比自己更接近对方 除非在连接会话期间已通知

type PeerPot

type PeerPot struct {
	NNSet [][]byte
}

Peerpot保存有关预期最近邻居的信息 仅用于测试 TODO移动到单独的测试工具文件

type Request

type Request struct {
	Addr      storage.Address //块地址
	Source    *enode.ID       //请求方的节点ID(可以为零)
	SkipCheck bool            //是先提供块还是直接交付

	HopCount uint8 //转发请求数(跃点)
	// contains filtered or unexported fields
}

func NewRequest

func NewRequest(addr storage.Address, skipCheck bool, peersToSkip *sync.Map) *Request

NewRequest返回基于块地址跳过检查和 要跳过的对等映射。

func (*Request) SkipPeer

func (r *Request) SkipPeer(nodeID string) bool

如果不应请求具有nodeid的对等端传递块,则skippeer返回。 要跳过的对等点在每个请求和请求超时的一段时间内保持不变。 此函数用于delivery.requestfrompeers中的流包中以优化 请求块。

type RequestFunc

type RequestFunc func(context.Context, *Request) (*enode.ID, chan struct{}, error)

Directories

Path Synopsis
simulations

Jump to

Keyboard shortcuts

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