forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 09:23:48 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/network: rewrite phase 1 - connectivity

Browse source
This commit is contained in:
zelig 2017-09-18 16:59:06 +02:00 committed by Lewis Marshall
parent 5c3bd0667e
commit 7abb7648fa
11 changed files with 2817 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

196
swarm/network/discovery.go Normal file
View file

@ -0,0 +1,196 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network
import (
"fmt"
"sync"
"github.com/ethereum/go-ethereum/pot"
)
// discovery bzz extension for requesting and relaying node address records
// discPeer wraps bzzPeer and embeds an Overlay connectivity driver
type discPeer struct {
*bzzPeer
overlay Overlay
sentPeers bool // whether we already sent peer closer to this address
mtx sync.Mutex
peers map[string]bool // tracks node records sent to the peer
depth uint8 // the proximity order advertised by remote as depth of saturation
}
// NewDiscovery constructs a discovery peer
func newDiscovery(p *bzzPeer, o Overlay) *discPeer {
d := &discPeer{
overlay: o,
bzzPeer: p,
peers: make(map[string]bool),
}
// record remote as seen so we never send a peer its own record
d.seen(d)
return d
}
// HandleMsg is the message handler that delegates incoming messages
func (d *discPeer) HandleMsg(msg interface{}) error {
switch msg := msg.(type) {
case *peersMsg:
return d.handlePeersMsg(msg)
case *subPeersMsg:
return d.handleSubPeersMsg(msg)
default:
return fmt.Errorf("unknown message type: %T", msg)
}
}
// NotifyDepth sends a message to all connections if depth of saturation is changed
func NotifyDepth(depth uint8, h Overlay) {
f := func(val OverlayConn, po int, _ bool) bool {
dp, ok := val.(*discPeer)
if ok {
go dp.NotifyDepth(depth)
}
return true
}
h.EachConn(nil, 255, f)
}
// NotifyPeer informs all peers about a newly added node
func NotifyPeer(p OverlayAddr, k Overlay) {
f := func(val OverlayConn, po int, _ bool) bool {
dp, ok := val.(*discPeer)
if ok {
go dp.NotifyPeer(p, uint8(po))
}
return true
}
k.EachConn(p.Address(), 255, f)
}
// NotifyPeer notifies the remote node about
func (d *discPeer) NotifyPeer(a OverlayAddr, po uint8) error {
// immediately return
if (po < d.depth && pot.ProxCmp(d.localAddr, d, a) != 1) || d.seen(a) {
return nil
}
// log.Trace(fmt.Sprintf("%08x peer %08x notified of peer %08x", d.localAddr.Over()[:4], d.Address()[:4], a.Address()[:4]))
resp := &peersMsg{
Peers: []*BzzAddr{ToAddr(a)}, // perhaps the PeerAddr interface is unnecessary generalization
}
return d.Send(resp)
}
// NotifyDepth sends a subPeers Msg to the receiver notifying them about
// a change in the depth of saturation
func (d *discPeer) NotifyDepth(po uint8) error {
// log.Trace(fmt.Sprintf("%08x peer %08x notified of new depth %v", d.localAddr.Over()[:4], d.Address()[:4], po))
return d.Send(&subPeersMsg{Depth: po})
}
/*
peersMsg is the message to pass peer information
It is always a response to a peersRequestMsg
The encoding of a peer address is identical the devp2p base protocol peers
messages: [IP, Port, NodeID],
Note that a node's DPA address is not the NodeID but the hash of the NodeID.
TODO:
To mitigate against spurious peers messages, requests should be remembered
and correctness of responses should be checked
If the proxBin of peers in the response is incorrect the sender should be
disconnected
*/
// peersMsg encapsulates an array of peer addresses
// used for communicating about known peers
// relevant for bootstrapping connectivity and updating peersets
type peersMsg struct {
Peers []*BzzAddr
}
// String pretty prints a peersMsg
func (msg peersMsg) String() string {
return fmt.Sprintf("%T: %v", msg, msg.Peers)
}
// handlePeersMsg called by the protocol when receiving peerset (for target address)
// list of nodes ([]PeerAddr in peersMsg) is added to the overlay db using the
// Register interface method
func (d *discPeer) handlePeersMsg(msg *peersMsg) error {
// register all addresses
if len(msg.Peers) == 0 {
return nil
}
for _, a := range msg.Peers {
d.seen(a)
NotifyPeer(a, d.overlay)
}
return d.overlay.Register(toOverlayAddrs(msg.Peers...))
}
// subPeers msg is communicating the depth/sharpness/focus of the overlay table of a peer
type subPeersMsg struct {
Depth uint8
}
// String returns the pretty printer
func (msg subPeersMsg) String() string {
return fmt.Sprintf("%T: request peers > PO%02d. ", msg, msg.Depth)
}
func (d *discPeer) handleSubPeersMsg(msg *subPeersMsg) error {
if !d.sentPeers {
d.depth = msg.Depth
var peers []*BzzAddr
d.overlay.EachConn(d.Over(), 255, func(p OverlayConn, po int, isproxbin bool) bool {
if pob, _ := pof(d, d.localAddr, 0); pob > po {
return false
}
if !d.seen(p) {
peers = append(peers, ToAddr(p.Off()))
}
return true
})
if len(peers) > 0 {
// log.Debug(fmt.Sprintf("%08x: %v peers sent to %v", d.overlay.BaseAddr(), len(peers), d))
go d.Send(&peersMsg{Peers: peers})
}
}
d.sentPeers = true
return nil
}
// seen takes an Overlay peer and checks if it was sent to a peer already
// if not, marks the peer as sent
func (d *discPeer) seen(p OverlayPeer) bool {
d.mtx.Lock()
defer d.mtx.Unlock()
k := string(p.Address())
if d.peers[k] {
return true
}
d.peers[k] = true
return false
}

57
swarm/network/discovery_test.go Normal file
View file

@ -0,0 +1,57 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network
import (
"fmt"
"testing"
"github.com/ethereum/go-ethereum/log"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
)
/***
*
* - after connect, that outgoing subpeersmsg is sent
*
*/
func TestDiscovery(t *testing.T) {
addr := RandomAddr()
to := NewKademlia(addr.OAddr, NewKadParams())
run := func(p *bzzPeer) error {
dp := newDiscovery(p, to)
to.On(p)
defer to.Off(p)
log.Trace(fmt.Sprintf("kademlia on %v", p))
return p.Run(dp.HandleMsg)
}
s := newBzzBaseTester(t, 1, addr, DiscoverySpec, run)
defer s.Stop()
s.TestExchanges(p2ptest.Exchange{
Label: "outgoing SubPeersMsg",
Expects: []p2ptest.Expect{
p2ptest.Expect{
Code: 3,
Msg: &subPeersMsg{Depth: 0},
Peer: s.ProtocolTester.IDs[0],
},
},
})
}

240
swarm/network/hive.go Normal file
View file

@ -0,0 +1,240 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network
import (
"encoding/json"
"fmt"
"math/rand"
"sync"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
)
/*
Hive is the logistic manager of the swarm
When the hive is started, a forever loop is launched that
asks the Overlay Topology driver (e.g., generic kademlia nodetable)
to suggest peers to bootstrap connectivity
*/
// Overlay is the interface for kademlia (or other topology drivers)
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 // used by the
// 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
}
// HiveParams holds the config options to hive
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
}
// NewHiveParams returns hive config with only the
func NewHiveParams() *HiveParams {
return &HiveParams{
Discovery: true,
PeersBroadcastSetSize: 3,
MaxPeersPerRequest: 5,
KeepAliveInterval: 1000 * time.Millisecond,
}
}
// Hive manages network connections of the swarm node
type Hive struct {
*HiveParams // settings
Overlay // the overlay connectiviy driver
store StateStore // storage interface to save peers across sessions
addPeer func(*discover.Node) // server callback to connect to a peer
// bookkeeping
lock sync.Mutex
ticker *time.Ticker
}
// NewHive constructs a new hive
// HiveParams: config parameters
// Overlay: connectivity driver using a network topology
// StateStore: to save peers across sessions
func NewHive(params *HiveParams, overlay Overlay, store StateStore) *Hive {
return &Hive{
HiveParams: params,
Overlay: overlay,
store: store,
}
}
// 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 (h *Hive) Start(server *p2p.Server) error {
log.Trace(fmt.Sprintf("%08x hive starting", h.BaseAddr()[:4]))
// if state store is specified, load peers to prepopulate the overlay address book
if h.store != nil {
if err := h.loadPeers(); err != nil {
return err
}
}
// assigns the p2p.Server#AddPeer function to connect to peers
h.addPeer = server.AddPeer
// ticker to keep the hive alive
h.ticker = time.NewTicker(h.KeepAliveInterval)
// this loop is doing bootstrapping and maintains a healthy table
go h.connect()
return nil
}
// Stop terminates the updateloop and saves the peers
func (h *Hive) Stop() error {
log.Info(fmt.Sprintf("%08x hive stopping, saving peers", h.BaseAddr()[:4]))
h.ticker.Stop()
if h.store != nil {
return h.savePeers()
}
log.Info(fmt.Sprintf("%08x hive stopped, dropping peers", h.BaseAddr()[:4]))
h.EachConn(nil, 255, func(p OverlayConn, _ int, _ bool) bool {
log.Info(fmt.Sprintf("%08x dropping peer %08x", h.BaseAddr()[:4], p.Address()[:4]))
p.Drop(nil)
return true
})
log.Info(fmt.Sprintf("%08x all peers dropped", h.BaseAddr()[:4]))
return nil
}
// connect is a forever loop
// at each iteration, ask the overlay driver to suggest the most preferred peer to connect to
// as well as advertises saturation depth if needed
func (h *Hive) connect() {
time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
for range h.ticker.C {
addr, depth, changed := h.SuggestPeer()
if h.Discovery && changed {
NotifyDepth(uint8(depth), h)
}
if addr == nil {
continue
}
under, err := discover.ParseNode(string(addr.(Addr).Under()))
if err != nil {
log.Warn(fmt.Sprintf("%08x unable to connect to bee %08x: invalid node URL: %v", h.BaseAddr()[:4], addr.Address()[:4], err))
continue
}
log.Trace(fmt.Sprintf("%08x attempt to connect to bee %08x", h.BaseAddr()[:4], addr.Address()[:4]))
h.addPeer(under)
}
}
// Run protocol run function
func (h *Hive) Run(p *bzzPeer) error {
dp := newDiscovery(p, h)
depth, changed := h.On(dp)
// if we want discovery, advertise changed depth of depth
if h.Discovery && changed {
NotifyDepth(depth, h)
}
NotifyPeer(p.Off(), h)
defer h.Off(dp)
return dp.Run(dp.HandleMsg)
}
// NodeInfo function is used by the p2p.server RPC interface to display
// protocol specific node information
func (h *Hive) NodeInfo() interface{} {
return h.String()
}
// PeerInfo function is used by the p2p.server RPC interface to display
// protocol specific information any connected peer referred to by their NodeID
func (h *Hive) PeerInfo(id discover.NodeID) interface{} {
return NewAddrFromNodeID(id)
}
// ToAddr returns the serialisable version of u
func ToAddr(pa OverlayPeer) *BzzAddr {
if addr, ok := pa.(*BzzAddr); ok {
return addr
}
if p, ok := pa.(*discPeer); ok {
return p.BzzAddr
}
return pa.(*bzzPeer).BzzAddr
}
// loadPeers, savePeer implement persistence callback/
func (h *Hive) loadPeers() error {
data, err := h.store.Load("peers")
if err != nil {
return err
}
if data == nil {
return nil
}
var as []*BzzAddr
if err := json.Unmarshal(data, &as); err != nil {
return err
}
return h.Register(toOverlayAddrs(as...))
}
// toOverlayAddrs transforms an array of BzzAddr to OverlayAddr
func toOverlayAddrs(as ...*BzzAddr) (oas []OverlayAddr) {
for _, a := range as {
oas = append(oas, OverlayAddr(a))
}
return
}
// savePeers, savePeer implement persistence callback/
func (h *Hive) savePeers() error {
var peers []*BzzAddr
h.Overlay.EachAddr(nil, 256, func(pa OverlayAddr, i int, _ bool) bool {
if pa == nil {
log.Warn(fmt.Sprintf("empty addr: %v", i))
return true
}
peers = append(peers, ToAddr(pa))
return true
})
data, err := json.Marshal(peers)
if err != nil {
return fmt.Errorf("could not encode peers: %v", err)
}
if err := h.store.Save("peers", data); err != nil {
return fmt.Errorf("could not save peers: %v", err)
}
return nil
}

56
swarm/network/hive_test.go Normal file
View file

@ -0,0 +1,56 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network
import (
"testing"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
)
func newHiveTester(t *testing.T, params *HiveParams) (*bzzTester, *Hive) {
// setup
addr := RandomAddr() // tested peers peer address
to := NewKademlia(addr.OAddr, NewKadParams())
pp := NewHive(params, to, nil) // hive
return newBzzBaseTester(t, 1, addr, DiscoverySpec, pp.Run), pp
}
func TestRegisterAndConnect(t *testing.T) {
params := NewHiveParams()
s, pp := newHiveTester(t, params)
id := s.IDs[0]
raddr := NewAddrFromNodeID(id)
pp.Register([]OverlayAddr{OverlayAddr(raddr)})
// start the hive and wait for the connection
pp.Start(s.Server)
defer pp.Stop()
// retrieve and broadcast
s.TestExchanges(p2ptest.Exchange{
Label: "getPeersMsg message",
Expects: []p2ptest.Expect{
p2ptest.Expect{
Code: 2,
Msg: &subPeersMsg{0},
Peer: id,
},
},
})
}

702
swarm/network/kademlia.go Normal file
View file

@ -0,0 +1,702 @@
// Copyright 2017 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network
import (
"bytes"
"fmt"
"math/rand"
"strings"
"sync"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/pot"
)
/*
Taking the proximity order relative to a fix point x classifies the points in
the space (n byte long byte sequences) into bins. Items in each are at
most half as distant from x as items in the previous bin. Given a sample of
uniformly distributed items (a hash function over arbitrary sequence) the
proximity scale maps onto series of subsets with cardinalities on a negative
exponential scale.
It also has the property that any two item belonging to the same bin are at
most half as distant from each other as they are from x.
If we think of random sample of items in the bins as connections in a network of
interconnected nodes then relative proximity can serve as the basis for local
decisions for graph traversal where the task is to find a route between two
points. Since in every hop, the finite distance halves, there is
a guaranteed constant maximum limit on the number of hops needed to reach one
node from the other.
*/
var pof = pot.DefaultPof(256)
// KadParams holds the config params for Kademlia
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 int // initial interval before a peer is first redialed
RetryExponent int // exponent to multiply retry intervals with
MaxRetries int // maximum number of redial attempts
PruneInterval int // interval between peer pruning cycles
// function to sanction or prevent suggesting a peer
Reachable func(OverlayAddr) bool
}
// NewKadParams returns a params struct with default values
func NewKadParams() *KadParams {
return &KadParams{
MaxProxDisplay: 16,
MinProxBinSize: 2,
MinBinSize: 2,
MaxBinSize: 4,
RetryInterval: 4200000000, // 4.2 sec
MaxRetries: 42,
RetryExponent: 2,
PruneInterval: 0, // TODO:
}
}
// Kademlia is a table of live peers and a db of known peers (node records)
type Kademlia struct {
lock sync.RWMutex
*KadParams // Kademlia configuration parameters
base []byte // immutable baseaddress of the table
addrs *pot.Pot // pots container for known peer addresses
conns *pot.Pot // pots container for live peer connections
depth uint8 // stores the last current depth of saturation
}
// NewKademlia creates a Kademlia table for base address addr
// with parameters as in params
// if params is nil, it uses default values
func NewKademlia(addr []byte, params *KadParams) *Kademlia {
if params == nil {
params = NewKadParams()
}
return &Kademlia{
base: addr,
KadParams: params,
addrs: pot.NewPot(nil, 0),
conns: pot.NewPot(nil, 0),
}
}
// OverlayPeer interface captures the common aspect of view of a peer from the Overlay
// topology driver
type OverlayPeer interface {
Address() []byte
}
// OverlayConn represents a connected peer
type OverlayConn interface {
OverlayPeer
Drop(error) // call to indicate a peer should be expunged
Off() OverlayAddr // call to return a persitent OverlayAddr
}
// OverlayAddr represents a kademlia peer record
type OverlayAddr interface {
OverlayPeer
Update(OverlayAddr) OverlayAddr // returns the updated version of the original
}
// entry represents a Kademlia table entry (an extension of OverlayPeer)
type entry struct {
OverlayPeer
seenAt time.Time
retries int
}
// newEntry creates a kademlia peer from an OverlayPeer interface
func newEntry(p OverlayPeer) *entry {
return &entry{
OverlayPeer: p,
seenAt: time.Now(),
}
}
// Bin is the binary (bitvector) serialisation of the entry address
func (e *entry) Bin() string {
return pot.ToBin(e.addr().Address())
}
// Label is a short tag for the entry for debug
func Label(e *entry) string {
return fmt.Sprintf("%s (%d)", e.Hex()[:4], e.retries)
}
// Hex is the hexadecimal serialisation of the entry address
func (e *entry) Hex() string {
return fmt.Sprintf("%x", e.addr().Address())
}
// String is the short tag for the entry
func (e *entry) String() string {
return fmt.Sprintf("%s (%d)", e.Hex()[:8], e.retries)
}
// addr returns the kad peer record (OverlayAddr) corresponding to the entry
func (e *entry) addr() OverlayAddr {
a, _ := e.OverlayPeer.(OverlayAddr)
return a
}
// conn returns the connected peer (OverlayPeer) corresponding to the entry
func (e *entry) conn() OverlayConn {
c, _ := e.OverlayPeer.(OverlayConn)
return c
}
// Register enters each OverlayAddr as kademlia peer record into the
// database of known peer addresses
func (k *Kademlia) Register(peers []OverlayAddr) error {
k.lock.Lock()
defer k.lock.Unlock()
var known, size int
for _, p := range peers {
// error if self received, peer should know better
// and should be punished for this
if bytes.Equal(p.Address(), k.base) {
return fmt.Errorf("add peers: %x is self", k.base)
}
var found bool
k.addrs, _, found, _ = pot.Swap(k.addrs, p, pof, func(v pot.Val) pot.Val {
// if not found
if v == nil {
// insert new offline peer into conns
return newEntry(p)
}
// found among known peers, do nothing
return v
})
if found {
known++
}
size++
}
// log.Trace(fmt.Sprintf("%x registered %v peers, %v known, total: %v", k.BaseAddr()[:4], size, known, k.addrs.Size()))
return nil
}
// 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
func (k *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
k.lock.RLock()
defer k.lock.RUnlock()
minsize := k.MinBinSize
depth := k.neighbourhoodDepth()
// if there is a callable neighbour within the current proxBin, connect
// this makes sure nearest neighbour set is fully connected
var ppo int
k.addrs.EachNeighbour(k.base, pof, func(val pot.Val, po int) bool {
if po < depth {
return false
}
a = k.callable(val)
ppo = po
return a == nil
})
if a != nil {
log.Trace(fmt.Sprintf("%08x candidate nearest neighbour found: %v (%v)", k.BaseAddr()[:4], a, ppo))
return a, 0, false
}
// log.Trace(fmt.Sprintf("%08x no candidate nearest neighbours to connect to (Depth: %v, minProxSize: %v) %#v", k.BaseAddr()[:4], depth, k.MinProxBinSize, a))
var bpo []int
prev := -1
k.conns.EachBin(k.base, pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
prev++
for ; prev < po; prev++ {
bpo = append(bpo, prev)
minsize = 0
}
if size < minsize {
bpo = append(bpo, po)
minsize = size
}
return size > 0 && po < depth
})
// all buckets are full, ie., minsize == k.MinBinSize
if len(bpo) == 0 {
// log.Debug(fmt.Sprintf("%08x: all bins saturated", k.BaseAddr()[:4]))
return nil, 0, false
}
// as long as we got candidate peers to connect to
// dont ask for new peers (want = false)
// try to select a candidate peer
// find the first callable peer
nxt := bpo[0]
k.addrs.EachBin(k.base, pof, nxt, func(po, _ int, f func(func(pot.Val, int) bool) bool) bool {
// for each bin (up until depth) we find callable candidate peers
if po >= depth {
return false
}
f(func(val pot.Val, _ int) bool {
a = k.callable(val)
return a == nil
})
return false
})
// found a candidate
if a != nil {
return a, 0, false
}
// no candidate peer found, request for the short bin
var changed bool
if uint8(nxt) < k.depth {
k.depth = uint8(nxt)
changed = true
}
return a, nxt, changed
}
// On inserts the peer as a kademlia peer into the live peers
func (k *Kademlia) On(p OverlayConn) (uint8, bool) {
k.lock.Lock()
defer k.lock.Unlock()
e := newEntry(p)
var ins bool
k.conns, _, _, _ = pot.Swap(k.conns, p, pof, func(v pot.Val) pot.Val {
// if not found live
if v == nil {
ins = true
// insert new online peer into conns
return e
}
// found among live peers, do nothing
return v
})
if ins {
// insert new online peer into addrs
k.addrs, _, _, _ = pot.Swap(k.addrs, p, pof, func(v pot.Val) pot.Val {
return e
})
}
log.Trace(k.string())
// calculate if depth of saturation changed
depth := uint8(k.saturation(k.MinBinSize))
var changed bool
if depth != k.depth {
changed = true
k.depth = depth
}
return k.depth, changed
}
// Off removes a peer from among live peers
func (k *Kademlia) Off(p OverlayConn) {
k.lock.Lock()
defer k.lock.Unlock()
var del bool
k.addrs, _, _, _ = pot.Swap(k.addrs, p, pof, func(v pot.Val) pot.Val {
// v cannot be nil, must check otherwise we overwrite entry
if v == nil {
panic(fmt.Sprintf("connected peer not found %v", p))
}
del = true
return newEntry(p.Off())
})
if del {
k.conns, _, _, _ = pot.Swap(k.conns, p, pof, func(_ pot.Val) pot.Val {
// v cannot be nil, but no need to check
return nil
})
}
}
// 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 (k *Kademlia) EachConn(base []byte, o int, f func(OverlayConn, int, bool) bool) {
k.lock.RLock()
defer k.lock.RUnlock()
k.eachConn(base, o, f)
}
func (k *Kademlia) eachConn(base []byte, o int, f func(OverlayConn, int, bool) bool) {
if len(base) == 0 {
base = k.base
}
depth := k.neighbourhoodDepth()
k.conns.EachNeighbour(base, pof, func(val pot.Val, po int) bool {
if po > o {
return true
}
return f(val.(*entry).conn(), po, po >= depth)
})
}
// 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 (k *Kademlia) EachAddr(base []byte, o int, f func(OverlayAddr, int, bool) bool) {
k.lock.RLock()
defer k.lock.RUnlock()
k.eachAddr(base, o, f)
}
func (k *Kademlia) eachAddr(base []byte, o int, f func(OverlayAddr, int, bool) bool) {
if len(base) == 0 {
base = k.base
}
depth := k.neighbourhoodDepth()
k.addrs.EachNeighbour(base, pof, func(val pot.Val, po int) bool {
if po > o {
return true
}
return f(val.(*entry).addr(), po, po >= depth)
})
}
// neighbourhoodDepth returns the proximity order that defines the distance of
// the nearest neighbour set with cardinality >= MinProxBinSize
// if there is altogether less than MinProxBinSize peers it returns 0
// caller must hold the lock
func (k *Kademlia) neighbourhoodDepth() (depth int) {
if k.conns.Size() < k.MinProxBinSize {
return 0
}
var size int
f := func(v pot.Val, i int) bool {
size++
depth = i
return size < k.MinProxBinSize
}
k.conns.EachNeighbour(k.base, pof, f)
return depth
}
// callable when called with val,
func (k *Kademlia) callable(val pot.Val) OverlayAddr {
e := val.(*entry)
// not callable if peer is live or exceeded maxRetries
if e.conn() != nil || e.retries > k.MaxRetries {
return nil
}
// calculate the allowed number of retries based on time lapsed since last seen
timeAgo := int(time.Since(e.seenAt))
div := k.RetryExponent
div += (150000 - rand.Intn(300000)) * div / 1000000
var retries int
for delta := timeAgo; delta > k.RetryInterval; delta /= div {
retries++
}
// this is never called concurrently, so safe to increment
// peer can be retried again
if retries < e.retries {
log.Trace(fmt.Sprintf("%08x: %v long time since last try (at %v) needed before retry %v, wait only warrants %v", k.BaseAddr()[:4], e, timeAgo, e.retries, retries))
return nil
}
// function to sanction or prevent suggesting a peer
if k.Reachable != nil && !k.Reachable(e.addr()) {
log.Trace(fmt.Sprintf("%08x: peer %v is temporarily not callable", k.BaseAddr()[:4], e))
return nil
}
e.retries++
log.Trace(fmt.Sprintf("%08x: peer %v is callable", k.BaseAddr()[:4], e))
return e.addr()
}
// BaseAddr return the kademlia base addres
func (k *Kademlia) BaseAddr() []byte {
return k.base
}
// String returns kademlia table + kaddb table displayed with ascii
func (k *Kademlia) String() string {
k.lock.RLock()
defer k.lock.RUnlock()
return k.string()
}
// String returns kademlia table + kaddb table displayed with ascii
func (k *Kademlia) string() string {
wsrow := " "
var rows []string
rows = append(rows, "=========================================================================")
rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %x", time.Now().UTC().Format(time.UnixDate), k.BaseAddr()[:3]))
rows = append(rows, fmt.Sprintf("population: %d (%d), MinProxBinSize: %d, MinBinSize: %d, MaxBinSize: %d", k.conns.Size(), k.addrs.Size(), k.MinProxBinSize, k.MinBinSize, k.MaxBinSize))
liverows := make([]string, k.MaxProxDisplay)
peersrows := make([]string, k.MaxProxDisplay)
depth := k.neighbourhoodDepth()
rest := k.conns.Size()
k.conns.EachBin(k.base, pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
var rowlen int
if po >= k.MaxProxDisplay {
po = k.MaxProxDisplay - 1
}
row := []string{fmt.Sprintf("%2d", size)}
rest -= size
f(func(val pot.Val, vpo int) bool {
e := val.(*entry)
row = append(row, fmt.Sprintf("%x", e.Address()[:2]))
rowlen++
return rowlen < 4
})
r := strings.Join(row, " ")
r = r + wsrow
liverows[po] = r[:31]
return true
})
k.addrs.EachBin(k.base, pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
var rowlen int
if po >= k.MaxProxDisplay {
po = k.MaxProxDisplay - 1
}
if size < 0 {
panic("wtf")
}
row := []string{fmt.Sprintf("%2d", size)}
// we are displaying live peers too
f(func(val pot.Val, vpo int) bool {
row = append(row, Label(val.(*entry)))
rowlen++
return rowlen < 4
})
peersrows[po] = strings.Join(row, " ")
return true
})
for i := 0; i < k.MaxProxDisplay; i++ {
if i == depth {
rows = append(rows, fmt.Sprintf("============ DEPTH: %d ==========================================", i))
}
left := liverows[i]
right := peersrows[i]
if len(left) == 0 {
left = " 0 "
}
if len(right) == 0 {
right = " 0"
}
rows = append(rows, fmt.Sprintf("%03d %v | %v", i, left, right))
}
rows = append(rows, "=========================================================================")
return "\n" + strings.Join(rows, "\n")
}
// Prune implements a forever loop reacting to a ticker time channel given
// as the first argument
// the loop quits if the channel is closed
// it checks each kademlia bin and if the peer count is higher than
// the MaxBinSize parameter it drops the oldest n peers such that
// the bin is reduced to MinBinSize peers thus leaving slots to newly
// connecting peers
func (k *Kademlia) Prune(c <-chan time.Time) {
go func() {
for range c {
k.lock.RLock()
conns := k.conns
k.lock.RUnlock()
total := 0
conns.EachBin(k.base, pof, 0, func(po, size int, f func(func(pot.Val, int) bool) bool) bool {
extra := size - k.MinBinSize
if size > k.MaxBinSize {
n := 0
f(func(v pot.Val, po int) bool {
v.(*entry).conn().Drop(fmt.Errorf("bucket full"))
n++
return n < extra
})
total += extra
}
return true
})
log.Trace(fmt.Sprintf("pruned %v peers", total))
}
}()
}
// PeerPot keeps info about expected nearest neighbours and empty bins
// used for testing only
type PeerPot struct {
NNSet [][]byte
EmptyBins []int
}
// NewPeerPot just creates a new pot record OverlayAddr
func NewPeerPot(kadMinProxSize int, ids []discover.NodeID, addrs [][]byte) map[discover.NodeID]*PeerPot {
// create a table of all nodes for health check
np := pot.NewPot(nil, 0)
for _, addr := range addrs {
np, _, _ = pot.Add(np, addr, pof)
}
ppmap := make(map[discover.NodeID]*PeerPot)
for i, id := range ids {
pl := 256
prev := 256
var emptyBins []int
var nns [][]byte
np.EachNeighbour(addrs[i], pof, func(val pot.Val, po int) bool {
a := val.([]byte)
if po == 256 {
return true
}
if pl == 256 || pl == po {
nns = append(nns, a)
}
if pl == 256 && len(nns) >= kadMinProxSize {
pl = po
prev = po
}
if prev < pl {
for j := prev; j > po; j-- {
emptyBins = append(emptyBins, j)
}
}
prev = po - 1
return true
})
for j := prev; j >= 0; j-- {
emptyBins = append(emptyBins, j)
}
log.Trace(fmt.Sprintf("%x NNS: %s", addrs[i][:4], logNNS(nns)))
ppmap[id] = &PeerPot{nns, emptyBins}
}
return ppmap
}
// saturation returns the lowest proximity order that the bin for that order
// has less than n peers
func (k *Kademlia) saturation(n int) int {
prev := -1
k.addrs.EachBin(k.base, pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
prev++
return prev == po && size >= n
})
depth := k.neighbourhoodDepth()
if depth < prev {
return depth
}
return prev
}
func (k *Kademlia) full(emptyBins []int) (full bool) {
prev := 0
e := len(emptyBins)
k.conns.EachBin(k.base, pof, 0, func(po, _ int, _ func(func(val pot.Val, i int) bool) bool) bool {
for i := prev; e > 0 && i < po; i++ {
e--
if emptyBins[e] != i {
log.Trace(fmt.Sprintf("%08x po: %d, i: %d, e: %d, emptybins: %v", k.BaseAddr()[:4], po, i, e, logEmptyBins(emptyBins)))
if emptyBins[e] < i {
panic("incorrect peerpot")
}
return false
}
}
prev = po + 1
return true
})
return e == 0
}
func (k *Kademlia) knowNearestNeighbours(peers [][]byte) bool {
pm := make(map[string]bool)
k.eachAddr(nil, 255, func(p OverlayAddr, po int, nn bool) bool {
if !nn {
return false
}
pk := fmt.Sprintf("%x", p.Address())
pm[pk] = true
return true
})
for _, p := range peers {
pk := fmt.Sprintf("%x", p)
if !pm[pk] {
log.Trace(fmt.Sprintf("%08x: known nearest neighbour %s not found", k.BaseAddr()[:4], pk[:8]))
return false
}
}
return true
}
func (k *Kademlia) gotNearestNeighbours(peers [][]byte) bool {
pm := make(map[string]bool)
k.eachConn(nil, 255, func(p OverlayConn, po int, nn bool) bool {
if !nn {
return false
}
pk := fmt.Sprintf("%x", p.Address())
pm[pk] = true
return true
})
for _, p := range peers {
pk := fmt.Sprintf("%x", p)
if !pm[pk] {
log.Trace(fmt.Sprintf("%08x: ExpNN: %s not found", k.BaseAddr()[:4], pk[:8]))
return false
}
}
return true
}
// Health state of the Kademlia
type Health struct {
KnowNN bool // whether node knows all its nearest neighbours
GotNN bool // whether node is connected to all its nearest neighbours
Full bool // whether node has a peer in each kademlia bin (where there is such a peer)
Hive string
}
// Healthy reports the health state of the kademlia connectivity
// returns a Health struct
func (k *Kademlia) Healthy(pp *PeerPot) *Health {
k.lock.RLock()
defer k.lock.RUnlock()
gotnn := k.gotNearestNeighbours(pp.NNSet)
knownn := k.knowNearestNeighbours(pp.NNSet)
full := k.full(pp.EmptyBins)
log.Trace(fmt.Sprintf("%08x: healthy: knowNNs: %v, gotNNs: %v, full: %v\n%v", k.BaseAddr()[:4], knownn, gotnn, full, k.string()))
return &Health{knownn, gotnn, full, k.string()}
}
func logNNS(nns [][]byte) string {
var nnsa []string
for _, nn := range nns {
nnsa = append(nnsa, fmt.Sprintf("%08x", nn[:4]))
}
return strings.Join(nnsa, ", ")
}
func logEmptyBins(ebs []int) string {
var ebss []string
for _, eb := range ebs {
ebss = append(ebss, fmt.Sprintf("%d", eb))
}
return strings.Join(ebss, ", ")
}

407
swarm/network/kademlia_test.go Normal file
View file

@ -0,0 +1,407 @@
// Copyright 2017 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network
import (
"fmt"
"os"
"testing"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/pot"
)
func init() {
h := log.LvlFilterHandler(log.LvlWarn, log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
log.Root().SetHandler(h)
}
func testKadPeerAddr(s string) *BzzAddr {
a := pot.NewAddressFromString(s)
return &BzzAddr{OAddr: a, UAddr: a}
}
type testDropPeer struct {
Peer
dropc chan error
}
type dropError struct {
error
addr string
}
func (d *testDropPeer) Drop(err error) {
err2 := &dropError{err, binStr(d)}
d.dropc <- err2
}
type testKademlia struct {
*Kademlia
Discovery bool
dropc chan error
}
func newTestKademlia(b string) *testKademlia {
params := NewKadParams()
params.MinBinSize = 1
params.MinProxBinSize = 2
base := pot.NewAddressFromString(b)
return &testKademlia{
NewKademlia(base, params),
false,
make(chan error),
}
}
func (k *testKademlia) newTestKadPeer(s string) Peer {
return &testDropPeer{&bzzPeer{BzzAddr: testKadPeerAddr(s)}, k.dropc}
}
func (k *testKademlia) On(ons ...string) *testKademlia {
for _, s := range ons {
k.Kademlia.On(k.newTestKadPeer(s).(OverlayConn))
}
return k
}
func (k *testKademlia) Off(offs ...string) *testKademlia {
for _, s := range offs {
k.Kademlia.Off(k.newTestKadPeer(s).(OverlayConn))
}
return k
}
func (k *testKademlia) Register(regs ...string) *testKademlia {
var as []OverlayAddr
for _, s := range regs {
as = append(as, testKadPeerAddr(s))
}
err := k.Kademlia.Register(as)
if err != nil {
panic(err.Error())
}
return k
}
func testSuggestPeer(t *testing.T, k *testKademlia, expAddr string, expPo int, expWant bool) error {
addr, o, want := k.SuggestPeer()
if binStr(addr) != expAddr {
return fmt.Errorf("incorrect peer address suggested. expected %v, got %v", expAddr, binStr(addr))
}
if o != expPo {
return fmt.Errorf("incorrect prox order suggested. expected %v, got %v", expPo, o)
}
if want != expWant {
return fmt.Errorf("expected SuggestPeer to want peers: %v", expWant)
}
return nil
}
func binStr(a OverlayPeer) string {
if a == nil {
return "<nil>"
}
return pot.ToBin(a.Address())[:8]
}
func TestSuggestPeerBug(t *testing.T) {
// 2 row gap, unsaturated proxbin, no callables -> want PO 0
k := newTestKademlia("00000000").On(
"10000000", "11000000",
"01000000",
"00010000", "00011000",
).Off(
"01000000",
)
err := testSuggestPeer(t, k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
}
func TestSuggestPeerFindPeers(t *testing.T) {
// 2 row gap, unsaturated proxbin, no callables -> want PO 0
k := newTestKademlia("00000000").On("00100000")
err := testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
// 2 row gap, saturated proxbin, no callables -> want PO 0
k.On("00010000")
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
// 1 row gap (1 less), saturated proxbin, no callables -> want PO 1
k.On("10000000")
err = testSuggestPeer(t, k, "<nil>", 1, false)
if err != nil {
t.Fatal(err.Error())
}
// no gap (1 less), saturated proxbin, no callables -> do not want more
k.On("01000000", "00100001")
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
// oversaturated proxbin, > do not want more
k.On("00100001")
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
// reintroduce gap, disconnected peer callable
// log.Info(k.String())
k.Off("01000000")
err = testSuggestPeer(t, k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
// second time disconnected peer not callable
// with reasonably set Interval
err = testSuggestPeer(t, k, "<nil>", 1, true)
if err != nil {
t.Fatal(err.Error())
}
// on and off again, peer callable again
k.On("01000000")
k.Off("01000000")
err = testSuggestPeer(t, k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
k.On("01000000")
// new closer peer appears, it is immediately wanted
k.Register("00010001")
err = testSuggestPeer(t, k, "00010001", 0, false)
if err != nil {
t.Fatal(err.Error())
}
// PO1 disconnects
k.On("00010001")
log.Info(k.String())
k.Off("01000000")
log.Info(k.String())
// second time, gap filling
err = testSuggestPeer(t, k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
k.On("01000000")
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
k.MinBinSize = 2
err = testSuggestPeer(t, k, "<nil>", 0, true)
if err != nil {
t.Fatal(err.Error())
}
k.Register("01000001")
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
k.On("10000001")
log.Trace("Kad:\n%v", k.String())
err = testSuggestPeer(t, k, "01000001", 0, false)
if err != nil {
t.Fatal(err.Error())
}
k.On("10000001")
k.On("01000001")
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
k.MinBinSize = 3
k.Register("10000010")
err = testSuggestPeer(t, k, "10000010", 0, false)
if err != nil {
t.Fatal(err.Error())
}
k.On("10000010")
err = testSuggestPeer(t, k, "<nil>", 1, false)
if err != nil {
t.Fatal(err.Error())
}
k.On("01000010")
err = testSuggestPeer(t, k, "<nil>", 2, false)
if err != nil {
t.Fatal(err.Error())
}
k.On("00100010")
err = testSuggestPeer(t, k, "<nil>", 3, false)
if err != nil {
t.Fatal(err.Error())
}
k.On("00010010")
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
}
func TestSuggestPeerRetries(t *testing.T) {
// 2 row gap, unsaturated proxbin, no callables -> want PO 0
k := newTestKademlia("00000000")
cycle := time.Second
k.RetryInterval = int(cycle)
k.MaxRetries = 50
k.RetryExponent = 2
sleep := func(n int) {
t := k.RetryInterval
for i := 1; i < n; i++ {
t *= k.RetryExponent
}
time.Sleep(time.Duration(t))
}
k.Register("01000000")
k.On("00000001", "00000010")
err := testSuggestPeer(t, k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
sleep(1)
err = testSuggestPeer(t, k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
sleep(1)
err = testSuggestPeer(t, k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
sleep(2)
err = testSuggestPeer(t, k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
sleep(2)
err = testSuggestPeer(t, k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
}
func TestPruning(t *testing.T) {
k := newTestKademlia("00000000")
k.On("10000000", "11000000", "10100000", "10010000", "10001000", "10000100")
k.On("01000000", "01100000", "01000100", "01000010", "01000001")
k.On("00100000", "00110000", "00100010", "00100001")
k.MaxBinSize = 4
k.MinBinSize = 3
prune := make(chan time.Time)
defer close(prune)
k.Prune((<-chan time.Time)(prune))
prune <- time.Now()
quitc := make(chan bool)
timeout := time.NewTimer(1000 * time.Millisecond)
n := 0
dropped := make(map[string]error)
expDropped := []string{
"10010000",
"10100000",
"11000000",
"01000100",
"01100000",
}
go func() {
for e := range k.dropc {
err := e.(*dropError)
dropped[err.addr] = err.error
n++
if n == len(expDropped) {
break
}
}
close(quitc)
}()
select {
case <-quitc:
case <-timeout.C:
t.Fatalf("timeout waiting for dropped peers. expected %v, got %v", len(expDropped), len(dropped))
}
for _, addr := range expDropped {
err := dropped[addr]
if err == nil {
t.Fatalf("expected peer %v to be dropped", addr)
}
if err.Error() != "bucket full" {
t.Fatalf("incorrect error. expected %v, got %v", "bucket full", err)
}
}
}
func TestKademliaHiveString(t *testing.T) {
k := newTestKademlia("00000000").On("01000000", "00100000").Register("10000000", "10000001")
h := k.String()
expH := "\n=========================================================================\nMon Feb 27 12:10:28 UTC 2017 KΛÐΞMLIΛ hive: queen's address: 000000\npopulation: 2 (4), MinProxBinSize: 2, MinBinSize: 1, MaxBinSize: 4\n000 0 | 2 8100 (0) 8000 (0)\n============ DEPTH: 1 ==========================================\n001 1 4000 | 1 4000 (0)\n002 1 2000 | 1 2000 (0)\n003 0 | 0\n004 0 | 0\n005 0 | 0\n006 0 | 0\n007 0 | 0\n========================================================================="
if expH[100:] != h[100:] {
t.Fatalf("incorrect hive output. expected %v, got %v", expH, h)
}
}

400
swarm/network/protocol.go Normal file
View file

@ -0,0 +1,400 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network
import (
"context"
"errors"
"fmt"
"net"
"sync"
"time"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/rpc"
)
const (
// NetworkID swarm network id
NetworkID = 322 // BZZ in l33t
// ProtocolMaxMsgSize maximum allowed message size
ProtocolMaxMsgSize = 10 * 1024 * 1024
// timeout for waiting
bzzHandshakeTimeout = 3000 * time.Millisecond
)
// BzzSpec is the spec of the generic swarm handshake
var BzzSpec = &protocols.Spec{
Name: "bzz",
Version: 1,
MaxMsgSize: 10 * 1024 * 1024,
Messages: []interface{}{
HandshakeMsg{},
},
}
// DiscoverySpec is the spec for the bzz discovery subprotocols
var DiscoverySpec = &protocols.Spec{
Name: "hive",
Version: 1,
MaxMsgSize: 10 * 1024 * 1024,
Messages: []interface{}{
peersMsg{},
subPeersMsg{},
},
}
// Addr interface that peerPool needs
type Addr interface {
OverlayPeer
Over() []byte
Under() []byte
String() string
Update(OverlayAddr) OverlayAddr
}
// Peer interface represents an live peer connection
type Peer interface {
Addr // the address of a peer
Conn // the live connection (protocols.Peer)
LastActive() time.Time // last time active
}
// Conn interface represents an live peer connection
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(interface{}) error // can send messages
Drop(error) // disconnect this peer
Run(func(interface{}) error) error // the run function to run a protocol
Off() OverlayAddr
}
// StateStore is a container interface to save/load peers across sessions
type StateStore interface {
Load(string) ([]byte, error) // load peer
Save(string, []byte) error // save peer
}
// BzzConfig captures the config params used by the hive
type BzzConfig struct {
OverlayAddr []byte // base address of the overlay network
UnderlayAddr []byte // node's underlay address
HiveParams *HiveParams
}
// Bzz is the swarm protocol bundle
type Bzz struct {
*Hive
localAddr *BzzAddr
mtx sync.Mutex
handshakes map[discover.NodeID]*HandshakeMsg
}
// NewBzz is the swarm protocol constructor
// arguments
// * bzz config
// * overlay driver
// * peer store
func NewBzz(config *BzzConfig, kad Overlay, store StateStore) *Bzz {
return &Bzz{
Hive: NewHive(config.HiveParams, kad, store),
localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr},
handshakes: make(map[discover.NodeID]*HandshakeMsg),
}
}
// UpdateLocalAddr updates underlayaddress of the running node
func (b *Bzz) UpdateLocalAddr(byteaddr []byte) *BzzAddr {
b.localAddr.Update(&BzzAddr{
UAddr: byteaddr,
OAddr: b.localAddr.OAddr,
})
return b.localAddr
}
// NodeInfo returns the node's overlay address
func (b *Bzz) NodeInfo() interface{} {
return b.localAddr.Address()
}
// Protocols return the protocols swarm offers
// Bzz implements the node.Service interface
// * handshake/hive
// * discovery
func (b *Bzz) Protocols() []p2p.Protocol {
return []p2p.Protocol{
{
Name: BzzSpec.Name,
Version: BzzSpec.Version,
Length: BzzSpec.Length(),
Run: b.runBzz,
NodeInfo: b.NodeInfo,
},
{
Name: DiscoverySpec.Name,
Version: DiscoverySpec.Version,
Length: DiscoverySpec.Length(),
Run: b.RunProtocol(DiscoverySpec, b.Hive.Run),
NodeInfo: b.Hive.NodeInfo,
PeerInfo: b.Hive.PeerInfo,
},
}
}
// APIs returns the APIs offered by bzz
// * hive
// Bzz implements the node.Service interface
func (b *Bzz) APIs() []rpc.API {
return []rpc.API{{
Namespace: "hive",
Version: "1.0",
Service: b.Hive,
}}
}
// 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 (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*bzzPeer) error) func(*p2p.Peer, p2p.MsgReadWriter) error {
return func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
// wait for the bzz protocol to perform the handshake
handshake, _ := b.GetHandshake(p.ID())
defer b.removeHandshake(p.ID())
select {
case <-handshake.done:
case <-time.After(bzzHandshakeTimeout):
return fmt.Errorf("%08x: %s protocol timeout waiting for handshake on %08x", b.BaseAddr()[:4], spec.Name, p.ID().Bytes()[:4])
}
if handshake.err != nil {
return fmt.Errorf("%08x: %s protocol closed: %v", b.BaseAddr()[:4], spec.Name, handshake.err)
}
// the handshake has succeeded so construct the bzzPeer and run the protocol
peer := &bzzPeer{
Peer: protocols.NewPeer(p, rw, spec),
localAddr: b.localAddr,
BzzAddr: handshake.peerAddr,
lastActive: time.Now(),
}
return run(peer)
}
}
// performHandshake implements the negotiation of the bzz handshake
// shared among swarm subprotocols
func performHandshake(p *protocols.Peer, handshake *HandshakeMsg) error {
ctx, _ := context.WithTimeout(context.Background(), bzzHandshakeTimeout)
// defer cancel()
// ctx, cancel := context.WithTimeout(context.Background(), bzzHandshakeTimeout)
defer close(handshake.done)
rsh, err := p.Handshake(ctx, handshake, checkHandshake)
if err != nil {
handshake.err = err
return err
}
handshake.peerAddr = rsh.(*HandshakeMsg).Addr
return nil
}
// runBzz is the p2p protocol run function for the bzz base protocol
// that negotiates the bzz handshake
func (b *Bzz) runBzz(p *p2p.Peer, rw p2p.MsgReadWriter) error {
handshake, _ := b.GetHandshake(p.ID())
if !<-handshake.init {
return fmt.Errorf("%08x: bzz already started on peer %08x", b.localAddr.Over()[:4], ToOverlayAddr(p.ID().Bytes())[:4])
}
close(handshake.init)
defer b.removeHandshake(p.ID())
peer := protocols.NewPeer(p, rw, BzzSpec)
err := performHandshake(peer, handshake)
if err != nil {
log.Warn(fmt.Sprintf("%08x: handshake failed with remote peer %08x: %v", b.localAddr.Over()[:4], ToOverlayAddr(p.ID().Bytes())[:4], err))
return err
}
// fail if we get another handshake
msg, err := rw.ReadMsg()
if err != nil {
return err
}
msg.Discard()
return errors.New("received multiple handshakes")
}
// 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
type bzzPeer struct {
*protocols.Peer // represents the connection for online peers
localAddr *BzzAddr // local Peers address
*BzzAddr // remote address -> implements Addr interface = protocols.Peer
lastActive time.Time // time is updated whenever mutexes are releasing
}
// Off returns the overlay peer record for offline persistance
func (p *bzzPeer) Off() OverlayAddr {
return p.BzzAddr
}
// LastActive returns the time the peer was last active
func (p *bzzPeer) LastActive() time.Time {
return p.lastActive
}
/*
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
*/
type HandshakeMsg struct {
Version uint64
NetworkID uint64
Addr *BzzAddr
// peerAddr is the address received in the peer handshake
peerAddr *BzzAddr
init chan bool
done chan struct{}
err error
}
// String pretty prints the handshake
func (bh *HandshakeMsg) String() string {
return fmt.Sprintf("Handshake: Version: %v, NetworkID: %v, Addr: %v", bh.Version, bh.NetworkID, bh.Addr)
}
// Perform initiates the handshake and validates the remote handshake message
func checkHandshake(hs interface{}) error {
rhs := hs.(*HandshakeMsg)
if rhs.NetworkID != NetworkID {
return fmt.Errorf("network id mismatch %d (!= %d)", rhs.NetworkID, NetworkID)
}
if rhs.Version != uint64(BzzSpec.Version) {
return fmt.Errorf("version mismatch %d (!= %d)", rhs.Version, BzzSpec.Version)
}
return nil
}
// removeHandshake removes handshake for peer with peerID
// from the bzz handshake store
func (b *Bzz) removeHandshake(peerID discover.NodeID) {
b.mtx.Lock()
defer b.mtx.Unlock()
delete(b.handshakes, peerID)
}
// GetHandshake returns the bzz handhake that the remote peer with peerID sent
func (b *Bzz) GetHandshake(peerID discover.NodeID) (*HandshakeMsg, bool) {
b.mtx.Lock()
defer b.mtx.Unlock()
handshake, found := b.handshakes[peerID]
if !found {
handshake = &HandshakeMsg{
Version: uint64(BzzSpec.Version),
NetworkID: uint64(NetworkID),
Addr: b.localAddr,
init: make(chan bool, 1),
done: make(chan struct{}),
}
// when handhsake is first created for a remote peer
// it is initialised with the init
handshake.init <- true
b.handshakes[peerID] = handshake
}
return handshake, found
}
// BzzAddr implements the PeerAddr interface
type BzzAddr struct {
OAddr []byte
UAddr []byte
}
// Address implements OverlayPeer interface to be used in Overlay
func (a *BzzAddr) Address() []byte {
return a.OAddr
}
// Over returns the overlay address
func (a *BzzAddr) Over() []byte {
return a.OAddr
}
// Under returns the underlay address
func (a *BzzAddr) Under() []byte {
return a.UAddr
}
// ID returns the nodeID from the underlay enode address
func (a *BzzAddr) ID() discover.NodeID {
return discover.MustParseNode(string(a.UAddr)).ID
}
// Update updates the underlay address of a peer record
func (a *BzzAddr) Update(na OverlayAddr) OverlayAddr {
return &BzzAddr{a.OAddr, na.(Addr).Under()}
}
// String pretty prints the address
func (a *BzzAddr) String() string {
return fmt.Sprintf("%x <%s>", a.OAddr, a.UAddr)
}
// RandomAddr is a utility method generating an address from a public key
func RandomAddr() *BzzAddr {
key, err := crypto.GenerateKey()
if err != nil {
panic("unable to generate key")
}
pubkey := crypto.FromECDSAPub(&key.PublicKey)
var id discover.NodeID
copy(id[:], pubkey[1:])
return NewAddrFromNodeID(id)
}
// NewNodeIDFromAddr transforms the underlay address to an adapters.NodeID
func NewNodeIDFromAddr(addr Addr) discover.NodeID {
log.Info(fmt.Sprintf("uaddr=%s", string(addr.Under())))
node := discover.MustParseNode(string(addr.Under()))
return node.ID
}
// NewAddrFromNodeID constucts a BzzAddr from a discover.NodeID
// the overlay address is derived as the hash of the nodeID
func NewAddrFromNodeID(id discover.NodeID) *BzzAddr {
return &BzzAddr{
OAddr: ToOverlayAddr(id.Bytes()),
UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()),
}
}
// ToOverlayAddr creates an overlayaddress from a byte slice
func ToOverlayAddr(id []byte) []byte {
return crypto.Keccak256(id)
}

203
swarm/network/protocol_test.go Normal file
View file

@ -0,0 +1,203 @@
// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network
import (
"fmt"
"sync"
"testing"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
)
type testStore struct {
sync.Mutex
values map[string][]byte
}
func newTestStore() *testStore {
return &testStore{values: make(map[string][]byte)}
}
func (t *testStore) Load(key string) ([]byte, error) {
t.Lock()
defer t.Unlock()
v, ok := t.values[key]
if !ok {
return nil, fmt.Errorf("key not found: %s", key)
}
return v, nil
}
func (t *testStore) Save(key string, v []byte) error {
t.Lock()
defer t.Unlock()
t.values[key] = v
return nil
}
func HandshakeMsgExchange(lhs, rhs *HandshakeMsg, id discover.NodeID) []p2ptest.Exchange {
return []p2ptest.Exchange{
p2ptest.Exchange{
Expects: []p2ptest.Expect{
p2ptest.Expect{
Code: 0,
Msg: lhs,
Peer: id,
},
},
},
p2ptest.Exchange{
Triggers: []p2ptest.Trigger{
p2ptest.Trigger{
Code: 0,
Msg: rhs,
Peer: id,
},
},
},
}
}
func newBzzBaseTester(t *testing.T, n int, addr *BzzAddr, spec *protocols.Spec, run func(*bzzPeer) error) *bzzTester {
cs := make(map[string]chan bool)
srv := func(p *bzzPeer) error {
defer close(cs[p.ID().String()])
return run(p)
}
protocall := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
return srv(&bzzPeer{
Peer: protocols.NewPeer(p, rw, spec),
localAddr: addr,
BzzAddr: NewAddrFromNodeID(p.ID()),
})
}
s := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), n, protocall)
for _, id := range s.IDs {
cs[id.String()] = make(chan bool)
}
return &bzzTester{
addr: addr,
ProtocolTester: s,
cs: cs,
}
}
type bzzTester struct {
*p2ptest.ProtocolTester
addr *BzzAddr
cs map[string]chan bool
}
func newBzzTester(t *testing.T, n int, addr *BzzAddr, pp *p2ptest.TestPeerPool, spec *protocols.Spec, services func(Peer) error) *bzzTester {
extraservices := func(p *bzzPeer) error {
pp.Add(p)
defer pp.Remove(p)
if services == nil {
return nil
}
return services(p)
}
return newBzzBaseTester(t, n, addr, spec, extraservices)
}
// should test handshakes in one exchange? parallelisation
func (s *bzzTester) testHandshake(lhs, rhs *HandshakeMsg, disconnects ...*p2ptest.Disconnect) {
var peers []discover.NodeID
id := NewNodeIDFromAddr(rhs.Addr)
if len(disconnects) > 0 {
for _, d := range disconnects {
peers = append(peers, d.Peer)
}
} else {
peers = []discover.NodeID{id}
}
s.TestExchanges(HandshakeMsgExchange(lhs, rhs, id)...)
s.TestDisconnected(disconnects...)
}
func (s *bzzTester) runHandshakes(ids ...discover.NodeID) {
if len(ids) == 0 {
ids = s.IDs
}
for _, id := range ids {
s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewAddrFromNodeID(id)))
<-s.cs[id.String()]
}
}
func correctBzzHandshake(addr *BzzAddr) *HandshakeMsg {
return &HandshakeMsg{
Version: 0,
NetworkID: 322,
Addr: addr,
}
}
func TestBzzHandshakeNetworkIDMismatch(t *testing.T) {
pp := p2ptest.NewTestPeerPool()
addr := RandomAddr()
s := newBzzTester(t, 1, addr, pp, nil, nil)
defer s.Stop()
id := s.IDs[0]
s.testHandshake(
correctBzzHandshake(addr),
&HandshakeMsg{Version: 0, NetworkID: 321, Addr: NewAddrFromNodeID(id)},
&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")},
)
}
func TestBzzHandshakeVersionMismatch(t *testing.T) {
pp := p2ptest.NewTestPeerPool()
addr := RandomAddr()
s := newBzzTester(t, 1, addr, pp, nil, nil)
defer s.Stop()
id := s.IDs[0]
s.testHandshake(
correctBzzHandshake(addr),
&HandshakeMsg{Version: 1, NetworkID: 322, Addr: NewAddrFromNodeID(id)},
&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("version mismatch 1 (!= 0)")},
)
}
func TestBzzHandshakeSuccess(t *testing.T) {
pp := p2ptest.NewTestPeerPool()
addr := RandomAddr()
s := newBzzTester(t, 1, addr, pp, nil, nil)
defer s.Stop()
id := s.IDs[0]
s.testHandshake(
correctBzzHandshake(addr),
&HandshakeMsg{Version: 0, NetworkID: 322, Addr: NewAddrFromNodeID(id)},
)
}

322
swarm/network/simulations/discovery/discovery_test.go Normal file
View file

@ -0,0 +1,322 @@
package discovery_test
import (
"context"
"encoding/json"
"errors"
"flag"
"fmt"
"io/ioutil"
"math/rand"
"os"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
)
// serviceName is used with the exec adapter so the exec'd binary knows which
// service to execute
const serviceName = "discovery"
const testMinProxBinSize = 2
var services = adapters.Services{
serviceName: newService,
}
var (
nodeCount = flag.Int("nodes", 16, "number of nodes to create (default 10)")
initCount = flag.Int("conns", 1, "number of originally connected peers (default 1)")
snapshotFile = flag.String("snapshot", "", "create snapshot")
loglevel = flag.Int("loglevel", 3, "verbosity of logs")
)
func init() {
flag.Parse()
// register the discovery service which will run as a devp2p
// protocol when using the exec adapter
adapters.RegisterServices(services)
log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
}
// Benchmarks to test the average time it takes for an N-node ring
// to full a healthy kademlia topology
func BenchmarkDiscovery_8_1(b *testing.B) { benchmarkDiscovery(b, 8, 1) }
func BenchmarkDiscovery_16_1(b *testing.B) { benchmarkDiscovery(b, 16, 1) }
func BenchmarkDiscovery_32_1(b *testing.B) { benchmarkDiscovery(b, 32, 1) }
func BenchmarkDiscovery_64_1(b *testing.B) { benchmarkDiscovery(b, 64, 1) }
func BenchmarkDiscovery_128_1(b *testing.B) { benchmarkDiscovery(b, 128, 1) }
func BenchmarkDiscovery_256_1(b *testing.B) { benchmarkDiscovery(b, 256, 1) }
func BenchmarkDiscovery_8_2(b *testing.B) { benchmarkDiscovery(b, 8, 2) }
func BenchmarkDiscovery_16_2(b *testing.B) { benchmarkDiscovery(b, 16, 2) }
func BenchmarkDiscovery_32_2(b *testing.B) { benchmarkDiscovery(b, 32, 2) }
func BenchmarkDiscovery_64_2(b *testing.B) { benchmarkDiscovery(b, 64, 2) }
func BenchmarkDiscovery_128_2(b *testing.B) { benchmarkDiscovery(b, 128, 2) }
func BenchmarkDiscovery_256_2(b *testing.B) { benchmarkDiscovery(b, 256, 2) }
func BenchmarkDiscovery_8_4(b *testing.B) { benchmarkDiscovery(b, 8, 4) }
func BenchmarkDiscovery_16_4(b *testing.B) { benchmarkDiscovery(b, 16, 4) }
func BenchmarkDiscovery_32_4(b *testing.B) { benchmarkDiscovery(b, 32, 4) }
func BenchmarkDiscovery_64_4(b *testing.B) { benchmarkDiscovery(b, 64, 4) }
func BenchmarkDiscovery_128_4(b *testing.B) { benchmarkDiscovery(b, 128, 4) }
func BenchmarkDiscovery_256_4(b *testing.B) { benchmarkDiscovery(b, 256, 4) }
func TestDiscoverySimulationDockerAdapter(t *testing.T) {
testDiscoverySimulationDockerAdapter(t, *nodeCount, *initCount)
}
func testDiscoverySimulationDockerAdapter(t *testing.T, nodes, conns int) {
adapter, err := adapters.NewDockerAdapter()
if err != nil {
t.Fatal(err)
}
testDiscoverySimulation(t, nodes, conns, adapter)
}
func TestDiscoverySimulationExecAdapter(t *testing.T) {
testDiscoverySimulationExecAdapter(t, *nodeCount, *initCount)
}
func testDiscoverySimulationExecAdapter(t *testing.T, nodes, conns int) {
baseDir, err := ioutil.TempDir("", "swarm-test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(baseDir)
testDiscoverySimulation(t, nodes, conns, adapters.NewExecAdapter(baseDir))
}
func TestDiscoverySimulationSimAdapter(t *testing.T) {
testDiscoverySimulationSimAdapter(t, *nodeCount, *initCount)
}
func testDiscoverySimulationSimAdapter(t *testing.T, nodes, conns int) {
testDiscoverySimulation(t, nodes, conns, adapters.NewSimAdapter(services))
}
func testDiscoverySimulation(t *testing.T, nodes, conns int, adapter adapters.NodeAdapter) {
startedAt := time.Now()
result, err := discoverySimulation(nodes, conns, adapter)
if err != nil {
t.Fatalf("Setting up simulation failed: %v", err)
}
if result.Error != nil {
t.Fatalf("Simulation failed: %s", result.Error)
}
t.Logf("Simulation with %d nodes passed in %s", nodes, result.FinishedAt.Sub(result.StartedAt))
var min, max time.Duration
var sum int
for _, pass := range result.Passes {
duration := pass.Sub(result.StartedAt)
if sum == 0 || duration < min {
min = duration
}
if duration > max {
max = duration
}
sum += int(duration.Nanoseconds())
}
t.Logf("Min: %s, Max: %s, Average: %s", min, max, time.Duration(sum/len(result.Passes))*time.Nanosecond)
finishedAt := time.Now()
t.Logf("Setup: %s, shutdown: %s", result.StartedAt.Sub(startedAt), finishedAt.Sub(result.FinishedAt))
}
func benchmarkDiscovery(b *testing.B, nodes, conns int) {
for i := 0; i < b.N; i++ {
result, err := discoverySimulation(nodes, conns, adapters.NewSimAdapter(services))
if err != nil {
b.Fatalf("setting up simulation failed", result)
}
if result.Error != nil {
b.Logf("simulation failed: %s", result.Error)
}
}
}
func discoverySimulation(nodes, conns int, adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
// create network
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
ID: "0",
DefaultService: serviceName,
})
defer net.Shutdown()
trigger := make(chan discover.NodeID)
ids := make([]discover.NodeID, nodes)
for i := 0; i < nodes; i++ {
node, err := net.NewNode()
if err != nil {
return nil, fmt.Errorf("error starting node: %s", err)
}
if err := net.Start(node.ID()); err != nil {
return nil, fmt.Errorf("error starting node %s: %s", node.ID().TerminalString(), err)
}
if err := triggerChecks(trigger, net, node.ID()); err != nil {
return nil, fmt.Errorf("error triggering checks for node %s: %s", node.ID().TerminalString(), err)
}
ids[i] = node.ID()
}
// run a simulation which connects the 10 nodes in a ring and waits
// for full peer discovery
var addrs [][]byte
action := func(ctx context.Context) error {
return nil
}
wg := sync.WaitGroup{}
for i := range ids {
// collect the overlay addresses, to
addrs = append(addrs, network.ToOverlayAddr(ids[i].Bytes()))
for j := 0; j < conns; j++ {
var k int
if j == 0 {
k = (i + 1) % len(ids)
} else {
k = rand.Intn(len(ids))
}
wg.Add(1)
go func(i, k int) {
defer wg.Done()
net.Connect(ids[i], ids[k])
}(i, k)
}
}
wg.Wait()
log.Debug(fmt.Sprintf("nodes: %v", len(addrs)))
// construct the peer pot, so that kademlia health can be checked
ppmap := network.NewPeerPot(testMinProxBinSize, ids, addrs)
check := func(ctx context.Context, id discover.NodeID) (bool, error) {
select {
case <-ctx.Done():
return false, ctx.Err()
default:
}
node := net.GetNode(id)
if node == nil {
return false, fmt.Errorf("unknown node: %s", id)
}
client, err := node.Client()
if err != nil {
return false, fmt.Errorf("error getting node client: %s", err)
}
healthy := &network.Health{}
if err := client.Call(&healthy, "hive_healthy", ppmap[id]); err != nil {
return false, fmt.Errorf("error getting node health: %s", err)
}
log.Debug(fmt.Sprintf("node %4s healthy: got nearest neighbours: %v, know nearest neighbours: %v, saturated: %v\n%v", id, healthy.GotNN, healthy.KnowNN, healthy.Full, healthy.Hive))
return healthy.KnowNN && healthy.GotNN && healthy.Full, nil
}
// 64 nodes ~ 1min
// 128 nodes ~
timeout := 600 * time.Second
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{
Action: action,
Trigger: trigger,
Expect: &simulations.Expectation{
Nodes: ids,
Check: check,
},
})
if result.Error != nil {
return result, nil
}
if *snapshotFile != "" {
snap, err := net.Snapshot()
if err != nil {
return nil, errors.New("no shapshot dude")
}
jsonsnapshot, err := json.Marshal(snap)
if err != nil {
return nil, fmt.Errorf("corrupt json snapshot: %v", err)
}
log.Info("writing snapshot", "file", *snapshotFile)
err = ioutil.WriteFile(*snapshotFile, jsonsnapshot, 0755)
if err != nil {
return nil, err
}
}
return result, nil
}
// triggerChecks triggers a simulation step check whenever a peer is added or
// removed from the given node, and also every second to avoid a race between
// peer events and kademlia becoming healthy
func triggerChecks(trigger chan discover.NodeID, net *simulations.Network, id discover.NodeID) error {
node := net.GetNode(id)
if node == nil {
return fmt.Errorf("unknown node: %s", id)
}
client, err := node.Client()
if err != nil {
return err
}
events := make(chan *p2p.PeerEvent)
sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents")
if err != nil {
return fmt.Errorf("error getting peer events for node %v: %s", id, err)
}
go func() {
defer sub.Unsubscribe()
tick := time.NewTicker(time.Second)
defer tick.Stop()
for {
select {
case <-events:
trigger <- id
case <-tick.C:
trigger <- id
case err := <-sub.Err():
if err != nil {
log.Error(fmt.Sprintf("error getting peer events for node %v", id), "err", err)
}
return
}
}
}()
return nil
}
func newService(ctx *adapters.ServiceContext) (node.Service, error) {
addr := network.NewAddrFromNodeID(ctx.Config.ID)
kp := network.NewKadParams()
kp.MinProxBinSize = testMinProxBinSize
kp.MaxBinSize = 3
kp.MinBinSize = 1
kp.MaxRetries = 1000
kp.RetryExponent = 2
kp.RetryInterval = 50000000
if ctx.Config.Reachable != nil {
kp.Reachable = func(o network.OverlayAddr) bool {
return ctx.Config.Reachable(o.(*network.BzzAddr).ID())
}
}
kad := network.NewKademlia(addr.Over(), kp)
hp := network.NewHiveParams()
hp.KeepAliveInterval = 500 * time.Millisecond
config := &network.BzzConfig{
OverlayAddr: addr.Over(),
UnderlayAddr: addr.Under(),
HiveParams: hp,
}
return network.NewBzz(config, kad, nil), nil
}

1
swarm/network/simulations/discovery/jsonsnapshot.txt Executable file
View file

File diff suppressed because one or more lines are too long

233
swarm/network/simulations/overlay.go Normal file
View file

@ -0,0 +1,233 @@
// +build none
// You can run this simulation using
//
// go run ./swarm/network/simulations/overlay.go
package main
import (
"flag"
"fmt"
"math/rand"
"net/http"
"os"
"runtime"
"sync"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
)
var noDiscovery = flag.Bool("no-discovery", false, "disable discovery (useful if you want to load a snapshot)")
type Simulation struct {
mtx sync.Mutex
stores map[discover.NodeID]*adapters.SimStateStore
}
func NewSimulation() *Simulation {
return &Simulation{
stores: make(map[discover.NodeID]*adapters.SimStateStore),
}
}
func (s *Simulation) NewService(ctx *adapters.ServiceContext) (node.Service, error) {
id := ctx.Config.ID
s.mtx.Lock()
store, ok := s.stores[id]
if !ok {
store = adapters.NewSimStateStore()
s.stores[id] = store
}
s.mtx.Unlock()
addr := network.NewAddrFromNodeID(id)
kp := network.NewKadParams()
kp.MinProxBinSize = 2
kp.MaxBinSize = 4
kp.MinBinSize = 1
kp.MaxRetries = 1000
kp.RetryExponent = 2
kp.RetryInterval = 1000000
kp.PruneInterval = 2000
kad := network.NewKademlia(addr.Over(), kp)
ticker := time.NewTicker(time.Duration(kad.PruneInterval) * time.Millisecond)
kad.Prune(ticker.C)
hp := network.NewHiveParams()
hp.Discovery = !*noDiscovery
hp.KeepAliveInterval = 300 * time.Millisecond
config := &network.BzzConfig{
OverlayAddr: addr.Over(),
UnderlayAddr: addr.Under(),
HiveParams: hp,
}
return network.NewBzz(config, kad, store), nil
}
func createMockers() map[string]*simulations.MockerConfig {
configs := make(map[string]*simulations.MockerConfig)
defaultCfg := simulations.DefaultMockerConfig()
defaultCfg.ID = "start-stop"
defaultCfg.Description = "Starts and Stops nodes in go routines"
defaultCfg.Mocker = startStopMocker
bootNetworkCfg := simulations.DefaultMockerConfig()
bootNetworkCfg.ID = "bootNet"
bootNetworkCfg.Description = "Only boots up all nodes in the config"
bootNetworkCfg.Mocker = bootMocker
randomNodesCfg := simulations.DefaultMockerConfig()
randomNodesCfg.ID = "randomNodes"
randomNodesCfg.Description = "Boots nodes and then starts and stops some picking randomly"
randomNodesCfg.Mocker = randomMocker
configs[defaultCfg.ID] = defaultCfg
configs[bootNetworkCfg.ID] = bootNetworkCfg
configs[randomNodesCfg.ID] = randomNodesCfg
return configs
}
func setupMocker(net *simulations.Network) []discover.NodeID {
nodeCount := 30
ids := make([]discover.NodeID, nodeCount)
for i := 0; i < nodeCount; i++ {
node, err := net.NewNode()
if err != nil {
panic(err.Error())
}
ids[i] = node.ID()
}
for _, id := range ids {
if err := net.Start(id); err != nil {
panic(err.Error())
}
}
for i, id := range ids {
log.Trace(fmt.Sprintf("setup mocker: register a peer on node %x", id[:4]))
var peerID discover.NodeID
if i == 0 {
peerID = ids[len(ids)-1]
} else {
peerID = ids[i-1]
}
ch := make(chan network.OverlayAddr)
go func() {
defer close(ch)
ch <- network.NewAddrFromNodeID(peerID)
}()
log.Trace(fmt.Sprintf("%x registers peer %x", id[:4], peerID[:4]))
if err := net.GetNode(id).Node.(*adapters.SimNode).Services()[0].(*network.Bzz).Hive.Register(ch); err != nil {
panic(err.Error())
}
}
return ids
}
func bootMocker(net *simulations.Network) {
setupMocker(net)
}
func randomMocker(net *simulations.Network) {
ids := setupMocker(net)
for {
var lowid, highid int
var wg sync.WaitGroup
randWait := rand.Intn(5000) + 1000
rand1 := rand.Intn(9)
rand2 := rand.Intn(9)
if rand1 < rand2 {
lowid = rand1
highid = rand2
} else if rand1 > rand2 {
highid = rand1
lowid = rand2
} else {
if rand1 == 0 {
rand2 = 9
} else if rand1 == 9 {
rand1 = 0
}
lowid = rand1
highid = rand2
}
var steps = highid - lowid
wg.Add(steps)
for i := lowid; i < highid; i++ {
log.Info(fmt.Sprintf("node %v shutting down", ids[i]))
net.Stop(ids[i])
go func(id discover.NodeID) {
time.Sleep(time.Duration(randWait) * time.Millisecond)
net.Start(id)
wg.Done()
}(ids[i])
time.Sleep(time.Duration(randWait) * time.Millisecond)
}
wg.Wait()
}
}
func startStopMocker(net *simulations.Network) {
ids := setupMocker(net)
for range time.Tick(10 * time.Second) {
id := ids[rand.Intn(len(ids))]
go func() {
log.Error("stopping node", "id", id)
if err := net.Stop(id); err != nil {
log.Error("error stopping node", "id", id, "err", err)
return
}
time.Sleep(3 * time.Second)
log.Error("starting node", "id", id)
if err := net.Start(id); err != nil {
log.Error("error starting node", "id", id, "err", err)
return
}
}()
}
}
// var server
func main() {
flag.Parse()
runtime.GOMAXPROCS(runtime.NumCPU())
log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
s := NewSimulation()
services := adapters.Services{
"overlay": s.NewService,
}
adapter := adapters.NewSimAdapter(services)
network := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
DefaultService: "overlay",
})
mockers := createMockers()
config := simulations.ServerConfig{
DefaultMockerID: "randomNodes",
// DefaultMockerID: "bootNet",
Mockers: mockers,
}
log.Info("starting simulation server on 0.0.0.0:8888...")
http.ListenAndServe(":8888", simulations.NewServer(network, config))
}