From bfce00385f1c8dab222b7ddab6c336177a5ae731 Mon Sep 17 00:00:00 2001
From: Viktor Trón <viktor.tron@gmail.com>
Date: Wed, 12 Sep 2018 11:24:56 +0200
Subject: Kademlia refactor (#17641)
* swarm/network: simplify kademlia/hive; rid interfaces
* swarm, swarm/network/stream, swarm/netork/simulations,, swarm/pss: adapt to new Kad API
* swarm/network: minor changes re review; add missing lock to NeighbourhoodDepthC
---
swarm/network/discovery.go | 76 ++++-----
swarm/network/discovery_test.go | 2 +-
swarm/network/hive.go | 71 ++------
swarm/network/hive_test.go | 8 +-
swarm/network/kademlia.go | 154 ++++++++---------
swarm/network/kademlia_test.go | 183 +++++++++------------
swarm/network/networkid_test.go | 2 +-
swarm/network/protocol.go | 47 +-----
.../simulations/discovery/discovery_test.go | 4 +-
swarm/network/stream/delivery.go | 12 +-
swarm/network/stream/snapshot_sync_test.go | 9 +-
swarm/network/stream/stream.go | 18 +-
swarm/network_test.go | 3 +-
swarm/pss/pss.go | 43 ++---
swarm/pss/pss_test.go | 47 ++----
swarm/swarm.go | 2 +-
16 files changed, 260 insertions(+), 421 deletions(-)
(limited to 'swarm')
diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go
index 55bf7c033..301959480 100644
--- a/swarm/network/discovery.go
+++ b/swarm/network/discovery.go
@@ -26,30 +26,30 @@ import (
// discovery bzz extension for requesting and relaying node address records
-// discPeer wraps BzzPeer and embeds an Overlay connectivity driver
-type discPeer struct {
+// Peer wraps BzzPeer and embeds Kademlia overlay connectivity driver
+type Peer struct {
*BzzPeer
- overlay Overlay
- sentPeers bool // whether we already sent peer closer to this address
- mtx sync.RWMutex
+ kad *Kademlia
+ sentPeers bool // whether we already sent peer closer to this address
+ mtx sync.RWMutex //
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,
+// NewPeer constructs a discovery peer
+func NewPeer(p *BzzPeer, kad *Kademlia) *Peer {
+ d := &Peer{
+ kad: kad,
BzzPeer: p,
peers: make(map[string]bool),
}
// record remote as seen so we never send a peer its own record
- d.seen(d)
+ d.seen(p.BzzAddr)
return d
}
// HandleMsg is the message handler that delegates incoming messages
-func (d *discPeer) HandleMsg(ctx context.Context, msg interface{}) error {
+func (d *Peer) HandleMsg(ctx context.Context, msg interface{}) error {
switch msg := msg.(type) {
case *peersMsg:
@@ -64,24 +64,18 @@ func (d *discPeer) HandleMsg(ctx context.Context, msg interface{}) error {
}
// 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 {
- dp.NotifyDepth(depth)
- }
+func NotifyDepth(depth uint8, kad *Kademlia) {
+ f := func(val *Peer, po int, _ bool) bool {
+ val.NotifyDepth(depth)
return true
}
- h.EachConn(nil, 255, f)
+ kad.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 {
- dp.NotifyPeer(p, uint8(po))
- }
+func NotifyPeer(p *BzzAddr, k *Kademlia) {
+ f := func(val *Peer, po int, _ bool) bool {
+ val.NotifyPeer(p, uint8(po))
return true
}
k.EachConn(p.Address(), 255, f)
@@ -91,22 +85,20 @@ func NotifyPeer(p OverlayAddr, k Overlay) {
// the peer's PO is within the recipients advertised depth
// OR the peer is closer to the recipient than self
// unless already notified during the connection session
-func (d *discPeer) NotifyPeer(a OverlayAddr, po uint8) {
+func (d *Peer) NotifyPeer(a *BzzAddr, po uint8) {
// immediately return
if (po < d.getDepth() && pot.ProxCmp(d.localAddr, d, a) != 1) || d.seen(a) {
return
}
- // 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)},
+ Peers: []*BzzAddr{a},
}
go d.Send(context.TODO(), 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) {
- // log.Trace(fmt.Sprintf("%08x peer %08x notified of new depth %v", d.localAddr.Over()[:4], d.Address()[:4], po))
+func (d *Peer) NotifyDepth(po uint8) {
go d.Send(context.TODO(), &subPeersMsg{Depth: po})
}
@@ -141,7 +133,7 @@ func (msg peersMsg) String() string {
// 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 {
+func (d *Peer) handlePeersMsg(msg *peersMsg) error {
// register all addresses
if len(msg.Peers) == 0 {
return nil
@@ -149,12 +141,12 @@ func (d *discPeer) handlePeersMsg(msg *peersMsg) error {
for _, a := range msg.Peers {
d.seen(a)
- NotifyPeer(a, d.overlay)
+ NotifyPeer(a, d.kad)
}
- return d.overlay.Register(toOverlayAddrs(msg.Peers...))
+ return d.kad.Register(msg.Peers...)
}
-// subPeers msg is communicating the depth/sharpness/focus of the overlay table of a peer
+// subPeers msg is communicating the depth of the overlay table of a peer
type subPeersMsg struct {
Depth uint8
}
@@ -164,21 +156,20 @@ func (msg subPeersMsg) String() string {
return fmt.Sprintf("%T: request peers > PO%02d. ", msg, msg.Depth)
}
-func (d *discPeer) handleSubPeersMsg(msg *subPeersMsg) error {
+func (d *Peer) handleSubPeersMsg(msg *subPeersMsg) error {
if !d.sentPeers {
d.setDepth(msg.Depth)
var peers []*BzzAddr
- d.overlay.EachConn(d.Over(), 255, func(p OverlayConn, po int, isproxbin bool) bool {
+ d.kad.EachConn(d.Over(), 255, func(p *Peer, 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()))
+ if !d.seen(p.BzzAddr) {
+ peers = append(peers, p.BzzAddr)
}
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(context.TODO(), &peersMsg{Peers: peers})
}
}
@@ -186,9 +177,9 @@ func (d *discPeer) handleSubPeersMsg(msg *subPeersMsg) error {
return nil
}
-// seen takes an Overlay peer and checks if it was sent to a peer already
+// seen takes an peer address and checks if it was sent to a peer already
// if not, marks the peer as sent
-func (d *discPeer) seen(p OverlayPeer) bool {
+func (d *Peer) seen(p *BzzAddr) bool {
d.mtx.Lock()
defer d.mtx.Unlock()
k := string(p.Address())
@@ -199,12 +190,13 @@ func (d *discPeer) seen(p OverlayPeer) bool {
return false
}
-func (d *discPeer) getDepth() uint8 {
+func (d *Peer) getDepth() uint8 {
d.mtx.RLock()
defer d.mtx.RUnlock()
return d.depth
}
-func (d *discPeer) setDepth(depth uint8) {
+
+func (d *Peer) setDepth(depth uint8) {
d.mtx.Lock()
defer d.mtx.Unlock()
d.depth = depth
diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go
index 0427d81ca..494bc8196 100644
--- a/swarm/network/discovery_test.go
+++ b/swarm/network/discovery_test.go
@@ -33,7 +33,7 @@ func TestDiscovery(t *testing.T) {
id := s.IDs[0]
raddr := NewAddrFromNodeID(id)
- pp.Register([]OverlayAddr{OverlayAddr(raddr)})
+ pp.Register(raddr)
// start the hive and wait for the connection
pp.Start(s.Server)
diff --git a/swarm/network/hive.go b/swarm/network/hive.go
index 366021088..425c1d5a1 100644
--- a/swarm/network/hive.go
+++ b/swarm/network/hive.go
@@ -32,31 +32,10 @@ import (
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)
+asks the 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
- // 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
@@ -78,7 +57,7 @@ func NewHiveParams() *HiveParams {
// Hive manages network connections of the swarm node
type Hive struct {
*HiveParams // settings
- Overlay // the overlay connectiviy driver
+ *Kademlia // the overlay connectiviy driver
Store state.Store // storage interface to save peers across sessions
addPeer func(*discover.Node) // server callback to connect to a peer
// bookkeeping
@@ -88,12 +67,12 @@ type Hive struct {
// NewHive constructs a new hive
// HiveParams: config parameters
-// Overlay: connectivity driver using a network topology
+// Kademlia: connectivity driver using a network topology
// StateStore: to save peers across sessions
-func NewHive(params *HiveParams, overlay Overlay, store state.Store) *Hive {
+func NewHive(params *HiveParams, kad *Kademlia, store state.Store) *Hive {
return &Hive{
HiveParams: params,
- Overlay: overlay,
+ Kademlia: kad,
Store: store,
}
}
@@ -133,7 +112,7 @@ func (h *Hive) Stop() error {
}
}
log.Info(fmt.Sprintf("%08x hive stopped, dropping peers", h.BaseAddr()[:4]))
- h.EachConn(nil, 255, func(p OverlayConn, _ int, _ bool) bool {
+ h.EachConn(nil, 255, func(p *Peer, _ int, _ bool) bool {
log.Info(fmt.Sprintf("%08x dropping peer %08x", h.BaseAddr()[:4], p.Address()[:4]))
p.Drop(nil)
return true
@@ -151,14 +130,14 @@ func (h *Hive) connect() {
addr, depth, changed := h.SuggestPeer()
if h.Discovery && changed {
- NotifyDepth(uint8(depth), h)
+ NotifyDepth(uint8(depth), h.Kademlia)
}
if addr == nil {
continue
}
log.Trace(fmt.Sprintf("%08x hive connect() suggested %08x", h.BaseAddr()[:4], addr.Address()[:4]))
- under, err := discover.ParseNode(string(addr.(Addr).Under()))
+ under, err := discover.ParseNode(string(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
@@ -170,19 +149,19 @@ func (h *Hive) connect() {
// Run protocol run function
func (h *Hive) Run(p *BzzPeer) error {
- dp := newDiscovery(p, h)
+ dp := NewPeer(p, h.Kademlia)
depth, changed := h.On(dp)
// if we want discovery, advertise change of depth
if h.Discovery {
if changed {
// if depth changed, send to all peers
- NotifyDepth(depth, h)
+ NotifyDepth(depth, h.Kademlia)
} else {
// otherwise just send depth to new peer
dp.NotifyDepth(depth)
}
}
- NotifyPeer(p.Off(), h)
+ NotifyPeer(p.BzzAddr, h.Kademlia)
defer h.Off(dp)
return dp.Run(dp.HandleMsg)
}
@@ -206,17 +185,6 @@ func (h *Hive) PeerInfo(id discover.NodeID) interface{} {
}
}
-// 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 {
var as []*BzzAddr
@@ -230,28 +198,19 @@ func (h *Hive) loadPeers() error {
}
log.Info(fmt.Sprintf("hive %08x: peers loaded", h.BaseAddr()[:4]))
- 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
+ return h.Register(as...)
}
// 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 {
+ h.Kademlia.EachAddr(nil, 256, func(pa *BzzAddr, i int, _ bool) bool {
if pa == nil {
log.Warn(fmt.Sprintf("empty addr: %v", i))
return true
}
- apa := ToAddr(pa)
- log.Trace("saving peer", "peer", apa)
- peers = append(peers, apa)
+ log.Trace("saving peer", "peer", pa)
+ peers = append(peers, pa)
return true
})
if err := h.Store.Put("peers", peers); err != nil {
diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go
index c2abfb2aa..7ea000c1a 100644
--- a/swarm/network/hive_test.go
+++ b/swarm/network/hive_test.go
@@ -41,7 +41,7 @@ func TestRegisterAndConnect(t *testing.T) {
id := s.IDs[0]
raddr := NewAddrFromNodeID(id)
- pp.Register([]OverlayAddr{OverlayAddr(raddr)})
+ pp.Register(raddr)
// start the hive and wait for the connection
err := pp.Start(s.Server)
@@ -77,7 +77,7 @@ func TestHiveStatePersistance(t *testing.T) {
peers := make(map[string]bool)
for _, id := range s.IDs {
raddr := NewAddrFromNodeID(id)
- pp.Register([]OverlayAddr{OverlayAddr(raddr)})
+ pp.Register(raddr)
peers[raddr.String()] = true
}
@@ -97,8 +97,8 @@ func TestHiveStatePersistance(t *testing.T) {
pp.Start(s1.Server)
i := 0
- pp.Overlay.EachAddr(nil, 256, func(addr OverlayAddr, po int, nn bool) bool {
- delete(peers, addr.(*BzzAddr).String())
+ pp.Kademlia.EachAddr(nil, 256, func(addr *BzzAddr, po int, nn bool) bool {
+ delete(peers, addr.String())
i++
return true
})
diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go
index 0177d449c..55a0c6f13 100644
--- a/swarm/network/kademlia.go
+++ b/swarm/network/kademlia.go
@@ -62,7 +62,7 @@ type KadParams struct {
RetryExponent int // exponent to multiply retry intervals with
MaxRetries int // maximum number of redial attempts
// function to sanction or prevent suggesting a peer
- Reachable func(OverlayAddr) bool
+ Reachable func(*BzzAddr) bool
}
// NewKadParams returns a params struct with default values
@@ -106,45 +106,22 @@ func NewKademlia(addr []byte, params *KadParams) *Kademlia {
}
}
-// 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)
+// entry represents a Kademlia table entry (an extension of BzzAddr)
type entry struct {
- OverlayPeer
+ *BzzAddr
+ conn *Peer
seenAt time.Time
retries int
}
-// newEntry creates a kademlia peer from an OverlayPeer interface
-func newEntry(p OverlayPeer) *entry {
+// newEntry creates a kademlia peer from a *Peer
+func newEntry(p *BzzAddr) *entry {
return &entry{
- OverlayPeer: p,
- seenAt: time.Now(),
+ BzzAddr: 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)
@@ -152,29 +129,12 @@ func Label(e *entry) string {
// Hex is the hexadecimal serialisation of the entry address
func (e *entry) Hex() string {
- return fmt.Sprintf("%x", e.addr().Address())
+ return fmt.Sprintf("%x", e.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
+// Register enters each address as kademlia peer record into the
// database of known peer addresses
-func (k *Kademlia) Register(peers []OverlayAddr) error {
+func (k *Kademlia) Register(peers ...*BzzAddr) error {
k.lock.Lock()
defer k.lock.Unlock()
var known, size int
@@ -203,7 +163,6 @@ func (k *Kademlia) Register(peers []OverlayAddr) error {
if k.addrCountC != nil && size-known > 0 {
k.addrCountC <- k.addrs.Size()
}
- // log.Trace(fmt.Sprintf("%x registered %v peers, %v known, total: %v", k.BaseAddr()[:4], size, known, k.addrs.Size()))
k.sendNeighbourhoodDepthChange()
return nil
@@ -212,7 +171,7 @@ func (k *Kademlia) Register(peers []OverlayAddr) error {
// 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) {
+func (k *Kademlia) SuggestPeer() (a *BzzAddr, o int, want bool) {
k.lock.Lock()
defer k.lock.Unlock()
minsize := k.MinBinSize
@@ -224,15 +183,18 @@ func (k *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
if po < depth {
return false
}
- a = k.callable(val)
+ e := val.(*entry)
+ c := k.callable(e)
+ if c {
+ a = e.BzzAddr
+ }
ppo = po
- return a == nil
+ return !c
})
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
@@ -250,7 +212,6 @@ func (k *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
})
// 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
@@ -264,8 +225,12 @@ func (k *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
return false
}
return f(func(val pot.Val, _ int) bool {
- a = k.callable(val)
- return a == nil
+ e := val.(*entry)
+ c := k.callable(e)
+ if c {
+ a = e.BzzAddr
+ }
+ return !c
})
})
// found a candidate
@@ -282,25 +247,26 @@ func (k *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
}
// On inserts the peer as a kademlia peer into the live peers
-func (k *Kademlia) On(p OverlayConn) (uint8, bool) {
+func (k *Kademlia) On(p *Peer) (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
+ return p
}
// found among live peers, do nothing
return v
})
if ins {
+ a := newEntry(p.BzzAddr)
+ a.conn = p
// insert new online peer into addrs
k.addrs, _, _, _ = pot.Swap(k.addrs, p, pof, func(v pot.Val) pot.Val {
- return e
+ return a
})
// send new address count value only if the peer is inserted
if k.addrCountC != nil {
@@ -324,6 +290,8 @@ func (k *Kademlia) On(p OverlayConn) (uint8, bool) {
// Not receiving from the returned channel will block On function
// when the neighbourhood depth is changed.
func (k *Kademlia) NeighbourhoodDepthC() <-chan int {
+ k.lock.Lock()
+ defer k.lock.Unlock()
if k.nDepthC == nil {
k.nDepthC = make(chan int)
}
@@ -357,7 +325,7 @@ func (k *Kademlia) AddrCountC() <-chan int {
}
// Off removes a peer from among live peers
-func (k *Kademlia) Off(p OverlayConn) {
+func (k *Kademlia) Off(p *Peer) {
k.lock.Lock()
defer k.lock.Unlock()
var del bool
@@ -367,7 +335,7 @@ func (k *Kademlia) Off(p OverlayConn) {
panic(fmt.Sprintf("connected peer not found %v", p))
}
del = true
- return newEntry(p.Off())
+ return newEntry(p.BzzAddr)
})
if del {
@@ -383,7 +351,7 @@ func (k *Kademlia) Off(p OverlayConn) {
}
}
-func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(conn OverlayConn, po int) bool) {
+func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(conn *Peer, po int) bool) {
k.lock.RLock()
defer k.lock.RUnlock()
@@ -403,7 +371,7 @@ func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(con
for bin := startPo; bin <= endPo; bin++ {
f(func(val pot.Val, _ int) bool {
- return eachBinFunc(val.(*entry).conn(), bin)
+ return eachBinFunc(val.(*Peer), bin)
})
}
return true
@@ -413,13 +381,13 @@ func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(con
// 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) {
+func (k *Kademlia) EachConn(base []byte, o int, f func(*Peer, 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) {
+func (k *Kademlia) eachConn(base []byte, o int, f func(*Peer, int, bool) bool) {
if len(base) == 0 {
base = k.base
}
@@ -428,20 +396,20 @@ func (k *Kademlia) eachConn(base []byte, o int, f func(OverlayConn, int, bool) b
if po > o {
return true
}
- return f(val.(*entry).conn(), po, po >= depth)
+ return f(val.(*Peer), 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) {
+func (k *Kademlia) EachAddr(base []byte, o int, f func(*BzzAddr, 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) {
+func (k *Kademlia) eachAddr(base []byte, o int, f func(*BzzAddr, int, bool) bool) {
if len(base) == 0 {
base = k.base
}
@@ -450,7 +418,7 @@ func (k *Kademlia) eachAddr(base []byte, o int, f func(OverlayAddr, int, bool) b
if po > o {
return true
}
- return f(val.(*entry).addr(), po, po >= depth)
+ return f(val.(*entry).BzzAddr, po, po >= depth)
})
}
@@ -472,12 +440,11 @@ func (k *Kademlia) neighbourhoodDepth() (depth int) {
return depth
}
-// callable when called with val,
-func (k *Kademlia) callable(val pot.Val) OverlayAddr {
- e := val.(*entry)
+// callable decides if an address entry represents a callable peer
+func (k *Kademlia) callable(e *entry) bool {
// not callable if peer is live or exceeded maxRetries
- if e.conn() != nil || e.retries > k.MaxRetries {
- return nil
+ if e.conn != nil || e.retries > k.MaxRetries {
+ return false
}
// calculate the allowed number of retries based on time lapsed since last seen
timeAgo := int64(time.Since(e.seenAt))
@@ -491,17 +458,17 @@ func (k *Kademlia) callable(val pot.Val) OverlayAddr {
// 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
+ return false
}
// function to sanction or prevent suggesting a peer
- if k.Reachable != nil && !k.Reachable(e.addr()) {
+ if k.Reachable != nil && !k.Reachable(e.BzzAddr) {
log.Trace(fmt.Sprintf("%08x: peer %v is temporarily not callable", k.BaseAddr()[:4], e))
- return nil
+ return false
}
e.retries++
log.Trace(fmt.Sprintf("%08x: peer %v is callable", k.BaseAddr()[:4], e))
- return e.addr()
+ return true
}
// BaseAddr return the kademlia base address
@@ -516,7 +483,8 @@ func (k *Kademlia) String() string {
return k.string()
}
-// String returns kademlia table + kaddb table displayed with ascii
+// string returns kademlia table + kaddb table displayed with ascii
+// caller must hold the lock
func (k *Kademlia) string() string {
wsrow := " "
var rows []string
@@ -538,7 +506,7 @@ func (k *Kademlia) string() string {
row := []string{fmt.Sprintf("%2d", size)}
rest -= size
f(func(val pot.Val, vpo int) bool {
- e := val.(*entry)
+ e := val.(*Peer)
row = append(row, fmt.Sprintf("%x", e.Address()[:2]))
rowlen++
return rowlen < 4
@@ -594,8 +562,9 @@ type PeerPot struct {
EmptyBins []int
}
-// NewPeerPotMap creates a map of pot record of OverlayAddr with keys
+// NewPeerPotMap creates a map of pot record of *BzzAddr with keys
// as hexadecimal representations of the address.
+// used for testing only
func NewPeerPotMap(kadMinProxSize int, addrs [][]byte) map[string]*PeerPot {
// create a table of all nodes for health check
np := pot.NewPot(nil, 0)
@@ -640,6 +609,7 @@ func NewPeerPotMap(kadMinProxSize int, addrs [][]byte) map[string]*PeerPot {
// saturation returns the lowest proximity order that the bin for that order
// has less than n peers
+// It is used in Healthy function for testing only
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 {
@@ -654,7 +624,7 @@ func (k *Kademlia) saturation(n int) int {
}
// full returns true if all required bins have connected peers.
-// It is used in Healthy function.
+// It is used in Healthy function for testing only
func (k *Kademlia) full(emptyBins []int) (full bool) {
prev := 0
e := len(emptyBins)
@@ -688,10 +658,13 @@ func (k *Kademlia) full(emptyBins []int) (full bool) {
return e == 0
}
+// knowNearestNeighbours tests if all known nearest neighbours given as arguments
+// are found in the addressbook
+// It is used in Healthy function for testing only
func (k *Kademlia) knowNearestNeighbours(peers [][]byte) bool {
pm := make(map[string]bool)
- k.eachAddr(nil, 255, func(p OverlayAddr, po int, nn bool) bool {
+ k.eachAddr(nil, 255, func(p *BzzAddr, po int, nn bool) bool {
if !nn {
return false
}
@@ -709,10 +682,13 @@ func (k *Kademlia) knowNearestNeighbours(peers [][]byte) bool {
return true
}
+// gotNearestNeighbours tests if all known nearest neighbours given as arguments
+// are connected peers
+// It is used in Healthy function for testing only
func (k *Kademlia) gotNearestNeighbours(peers [][]byte) (got bool, n int, missing [][]byte) {
pm := make(map[string]bool)
- k.eachConn(nil, 255, func(p OverlayConn, po int, nn bool) bool {
+ k.eachConn(nil, 255, func(p *Peer, po int, nn bool) bool {
if !nn {
return false
}
@@ -735,6 +711,7 @@ func (k *Kademlia) gotNearestNeighbours(peers [][]byte) (got bool, n int, missin
}
// Health state of the Kademlia
+// used for testing only
type Health struct {
KnowNN bool // whether node knows all its nearest neighbours
GotNN bool // whether node is connected to all its nearest neighbours
@@ -746,6 +723,7 @@ type Health struct {
// Healthy reports the health state of the kademlia connectivity
// returns a Health struct
+// used for testing only
func (k *Kademlia) Healthy(pp *PeerPot) *Health {
k.lock.RLock()
defer k.lock.RUnlock()
diff --git a/swarm/network/kademlia_test.go b/swarm/network/kademlia_test.go
index b60e1e9a3..903c8dbda 100644
--- a/swarm/network/kademlia_test.go
+++ b/swarm/network/kademlia_test.go
@@ -38,71 +38,42 @@ func testKadPeerAddr(s string) *BzzAddr {
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 {
+func newTestKademlia(b string) *Kademlia {
params := NewKadParams()
params.MinBinSize = 1
params.MinProxBinSize = 2
base := pot.NewAddressFromString(b)
- return &testKademlia{
- NewKademlia(base, params),
- false,
- make(chan error),
- }
+ return NewKademlia(base, params)
}
-func (k *testKademlia) newTestKadPeer(s string) Peer {
- return &testDropPeer{&BzzPeer{BzzAddr: testKadPeerAddr(s)}, k.dropc}
+func newTestKadPeer(k *Kademlia, s string) *Peer {
+ return NewPeer(&BzzPeer{BzzAddr: testKadPeerAddr(s)}, k)
}
-func (k *testKademlia) On(ons ...string) *testKademlia {
+func On(k *Kademlia, ons ...string) {
for _, s := range ons {
- k.Kademlia.On(k.newTestKadPeer(s).(OverlayConn))
+ k.On(newTestKadPeer(k, s))
}
- return k
}
-func (k *testKademlia) Off(offs ...string) *testKademlia {
+func Off(k *Kademlia, offs ...string) {
for _, s := range offs {
- k.Kademlia.Off(k.newTestKadPeer(s).(OverlayConn))
+ k.Off(newTestKadPeer(k, s))
}
-
- return k
}
-func (k *testKademlia) Register(regs ...string) *testKademlia {
- var as []OverlayAddr
+func Register(k *Kademlia, regs ...string) {
+ var as []*BzzAddr
for _, s := range regs {
as = append(as, testKadPeerAddr(s))
}
- err := k.Kademlia.Register(as)
+ err := k.Register(as...)
if err != nil {
panic(err.Error())
}
- return k
}
-func testSuggestPeer(t *testing.T, k *testKademlia, expAddr string, expPo int, expWant bool) error {
+func testSuggestPeer(k *Kademlia, 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))
@@ -116,7 +87,7 @@ func testSuggestPeer(t *testing.T, k *testKademlia, expAddr string, expPo int, e
return nil
}
-func binStr(a OverlayPeer) string {
+func binStr(a *BzzAddr) string {
if a == nil {
return "<nil>"
}
@@ -125,15 +96,17 @@ func binStr(a OverlayPeer) string {
func TestSuggestPeerBug(t *testing.T) {
// 2 row gap, unsaturated proxbin, no callables -> want PO 0
- k := newTestKademlia("00000000").On(
+ k := newTestKademlia("00000000")
+ On(k,
"10000000", "11000000",
"01000000",
"00010000", "00011000",
- ).Off(
+ )
+ Off(k,
"01000000",
)
- err := testSuggestPeer(t, k, "01000000", 0, false)
+ err := testSuggestPeer(k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
@@ -141,140 +114,140 @@ func TestSuggestPeerBug(t *testing.T) {
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)
+ k := newTestKademlia("00000000")
+ On(k, "00100000")
+ err := testSuggestPeer(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)
+ On(k, "00010000")
+ err = testSuggestPeer(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)
+ On(k, "10000000")
+ err = testSuggestPeer(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)
+ On(k, "01000000", "00100001")
+ err = testSuggestPeer(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)
+ On(k, "00100001")
+ err = testSuggestPeer(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)
+ Off(k, "01000000")
+ err = testSuggestPeer(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)
+ err = testSuggestPeer(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)
+ On(k, "01000000")
+ Off(k, "01000000")
+ err = testSuggestPeer(k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
- k.On("01000000")
+ On(k, "01000000")
// new closer peer appears, it is immediately wanted
- k.Register("00010001")
- err = testSuggestPeer(t, k, "00010001", 0, false)
+ Register(k, "00010001")
+ err = testSuggestPeer(k, "00010001", 0, false)
if err != nil {
t.Fatal(err.Error())
}
// PO1 disconnects
- k.On("00010001")
+ On(k, "00010001")
log.Info(k.String())
- k.Off("01000000")
+ Off(k, "01000000")
log.Info(k.String())
// second time, gap filling
- err = testSuggestPeer(t, k, "01000000", 0, false)
+ err = testSuggestPeer(k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
- k.On("01000000")
- err = testSuggestPeer(t, k, "<nil>", 0, false)
+ On(k, "01000000")
+ err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
k.MinBinSize = 2
- err = testSuggestPeer(t, k, "<nil>", 0, true)
+ err = testSuggestPeer(k, "<nil>", 0, true)
if err != nil {
t.Fatal(err.Error())
}
- k.Register("01000001")
- err = testSuggestPeer(t, k, "01000001", 0, false)
+ Register(k, "01000001")
+ err = testSuggestPeer(k, "01000001", 0, false)
if err != nil {
t.Fatal(err.Error())
}
- k.On("10000001")
+ On(k, "10000001")
log.Trace(fmt.Sprintf("Kad:\n%v", k.String()))
- err = testSuggestPeer(t, k, "<nil>", 1, true)
+ err = testSuggestPeer(k, "<nil>", 1, true)
if err != nil {
t.Fatal(err.Error())
}
- k.On("01000001")
- err = testSuggestPeer(t, k, "<nil>", 0, false)
+ On(k, "01000001")
+ err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
k.MinBinSize = 3
- k.Register("10000010")
- err = testSuggestPeer(t, k, "10000010", 0, false)
+ Register(k, "10000010")
+ err = testSuggestPeer(k, "10000010", 0, false)
if err != nil {
t.Fatal(err.Error())
}
- k.On("10000010")
- err = testSuggestPeer(t, k, "<nil>", 1, false)
+ On(k, "10000010")
+ err = testSuggestPeer(k, "<nil>", 1, false)
if err != nil {
t.Fatal(err.Error())
}
- k.On("01000010")
- err = testSuggestPeer(t, k, "<nil>", 2, false)
+ On(k, "01000010")
+ err = testSuggestPeer(k, "<nil>", 2, false)
if err != nil {
t.Fatal(err.Error())
}
- k.On("00100010")
- err = testSuggestPeer(t, k, "<nil>", 3, false)
+ On(k, "00100010")
+ err = testSuggestPeer(k, "<nil>", 3, false)
if err != nil {
t.Fatal(err.Error())
}
- k.On("00010010")
- err = testSuggestPeer(t, k, "<nil>", 0, false)
+ On(k, "00010010")
+ err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
@@ -282,10 +255,8 @@ func TestSuggestPeerFindPeers(t *testing.T) {
}
func TestSuggestPeerRetries(t *testing.T) {
- t.Skip("Test is disabled, because it is flaky. It fails with kademlia_test.go:346: incorrect peer address suggested. expected <nil>, got 01000000")
- // 2 row gap, unsaturated proxbin, no callables -> want PO 0
k := newTestKademlia("00000000")
- k.RetryInterval = int64(100 * time.Millisecond) // cycle
+ k.RetryInterval = int64(300 * time.Millisecond) // cycle
k.MaxRetries = 50
k.RetryExponent = 2
sleep := func(n int) {
@@ -296,53 +267,53 @@ func TestSuggestPeerRetries(t *testing.T) {
time.Sleep(time.Duration(ts))
}
- k.Register("01000000")
- k.On("00000001", "00000010")
- err := testSuggestPeer(t, k, "01000000", 0, false)
+ Register(k, "01000000")
+ On(k, "00000001", "00000010")
+ err := testSuggestPeer(k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
- err = testSuggestPeer(t, k, "<nil>", 0, false)
+ err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
sleep(1)
- err = testSuggestPeer(t, k, "01000000", 0, false)
+ err = testSuggestPeer(k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
- err = testSuggestPeer(t, k, "<nil>", 0, false)
+ err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
sleep(1)
- err = testSuggestPeer(t, k, "01000000", 0, false)
+ err = testSuggestPeer(k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
- err = testSuggestPeer(t, k, "<nil>", 0, false)
+ err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
sleep(2)
- err = testSuggestPeer(t, k, "01000000", 0, false)
+ err = testSuggestPeer(k, "01000000", 0, false)
if err != nil {
t.Fatal(err.Error())
}
- err = testSuggestPeer(t, k, "<nil>", 0, false)
+ err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
sleep(2)
- err = testSuggestPeer(t, k, "<nil>", 0, false)
+ err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil {
t.Fatal(err.Error())
}
@@ -350,7 +321,9 @@ func TestSuggestPeerRetries(t *testing.T) {
}
func TestKademliaHiveString(t *testing.T) {
- k := newTestKademlia("00000000").On("01000000", "00100000").Register("10000000", "10000001")
+ k := newTestKademlia("00000000")
+ On(k, "01000000", "00100000")
+ Register(k, "10000000", "10000001")
k.MaxProxDisplay = 8
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========================================================================="
@@ -378,7 +351,7 @@ func testKademliaCase(t *testing.T, pivotAddr string, addrs ...string) {
continue
}
p := &BzzAddr{OAddr: a, UAddr: a}
- if err := k.Register([]OverlayAddr{p}); err != nil {
+ if err := k.Register(p); err != nil {
t.Fatal(err)
}
}
@@ -392,12 +365,12 @@ func testKademliaCase(t *testing.T, pivotAddr string, addrs ...string) {
if a == nil {
break
}
- k.On(&BzzPeer{BzzAddr: a.(*BzzAddr)})
+ k.On(NewPeer(&BzzPeer{BzzAddr: a}, k))
}
h := k.Healthy(pp)
if !(h.GotNN && h.KnowNN && h.Full) {
- t.Error("not healthy")
+ t.Fatalf("not healthy: %#v\n%v", h, k.String())
}
}
diff --git a/swarm/network/networkid_test.go b/swarm/network/networkid_test.go
index 05134b083..91a1f6d7b 100644
--- a/swarm/network/networkid_test.go
+++ b/swarm/network/networkid_test.go
@@ -92,7 +92,7 @@ func TestNetworkID(t *testing.T) {
if kademlias[node].addrs.Size() != len(netIDGroup)-1 {
t.Fatalf("Kademlia size has not expected peer size. Kademlia size: %d, expected size: %d", kademlias[node].addrs.Size(), len(netIDGroup)-1)
}
- kademlias[node].EachAddr(nil, 0, func(addr OverlayAddr, _ int, _ bool) bool {
+ kademlias[node].EachAddr(nil, 0, func(addr *BzzAddr, _ int, _ bool) bool {
found := false
for _, nd := range netIDGroup {
p := ToOverlayAddr(nd.Bytes())
diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go
index 7f7ca5eed..ef0956d5f 100644
--- a/swarm/network/protocol.go
+++ b/swarm/network/protocol.go
@@ -62,32 +62,6 @@ var DiscoverySpec = &protocols.Spec{
},
}
-// 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(context.Context, interface{}) error // can send messages
- Drop(error) // disconnect this peer
- Run(func(context.Context, interface{}) error) error // the run function to run a protocol
- Off() OverlayAddr
-}
-
// BzzConfig captures the config params used by the hive
type BzzConfig struct {
OverlayAddr []byte // base address of the overlay network
@@ -114,7 +88,7 @@ type Bzz struct {
// * bzz config
// * overlay driver
// * peer store
-func NewBzz(config *BzzConfig, kad Overlay, store state.Store, streamerSpec *protocols.Spec, streamerRun func(*BzzPeer) error) *Bzz {
+func NewBzz(config *BzzConfig, kad *Kademlia, store state.Store, streamerSpec *protocols.Spec, streamerRun func(*BzzPeer) error) *Bzz {
return &Bzz{
Hive: NewHive(config.HiveParams, kad, store),
NetworkID: config.NetworkID,
@@ -131,7 +105,7 @@ func (b *Bzz) UpdateLocalAddr(byteaddr []byte) *BzzAddr {
b.localAddr = b.localAddr.Update(&BzzAddr{
UAddr: byteaddr,
OAddr: b.localAddr.OAddr,
- }).(*BzzAddr)
+ })
return b.localAddr
}
@@ -274,7 +248,7 @@ type BzzPeer struct {
LightNode bool
}
-func NewBzzTestPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer {
+func NewBzzPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer {
return &BzzPeer{
Peer: p,
localAddr: addr,
@@ -282,11 +256,6 @@ func NewBzzTestPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer {
}
}
-// Off returns the overlay peer record for offline persistence
-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
@@ -388,8 +357,8 @@ func (a *BzzAddr) ID() discover.NodeID {
}
// Update updates the underlay address of a peer record
-func (a *BzzAddr) Update(na OverlayAddr) OverlayAddr {
- return &BzzAddr{a.OAddr, na.(Addr).Under()}
+func (a *BzzAddr) Update(na *BzzAddr) *BzzAddr {
+ return &BzzAddr{a.OAddr, na.UAddr}
}
// String pretty prints the address
@@ -410,9 +379,9 @@ func RandomAddr() *BzzAddr {
}
// 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()))
+func NewNodeIDFromAddr(addr *BzzAddr) discover.NodeID {
+ log.Info(fmt.Sprintf("uaddr=%s", string(addr.UAddr)))
+ node := discover.MustParseNode(string(addr.UAddr))
return node.ID
}
diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go
index acf3479e5..913d6d837 100644
--- a/swarm/network/simulations/discovery/discovery_test.go
+++ b/swarm/network/simulations/discovery/discovery_test.go
@@ -556,8 +556,8 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) {
kp.MinProxBinSize = testMinProxBinSize
if ctx.Config.Reachable != nil {
- kp.Reachable = func(o network.OverlayAddr) bool {
- return ctx.Config.Reachable(o.(*network.BzzAddr).ID())
+ kp.Reachable = func(o *network.BzzAddr) bool {
+ return ctx.Config.Reachable(o.ID())
}
}
kad := network.NewKademlia(addr.Over(), kp)
diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go
index 36040339d..627352535 100644
--- a/swarm/network/stream/delivery.go
+++ b/swarm/network/stream/delivery.go
@@ -47,15 +47,15 @@ var (
type Delivery struct {
db *storage.DBAPI
- overlay network.Overlay
+ kad *network.Kademlia
receiveC chan *ChunkDeliveryMsg
getPeer func(discover.NodeID) *Peer
}
-func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery {
+func NewDelivery(kad *network.Kademlia, db *storage.DBAPI) *Delivery {
d := &Delivery{
db: db,
- overlay: overlay,
+ kad: kad,
receiveC: make(chan *ChunkDeliveryMsg, deliveryCap),
}
@@ -172,7 +172,7 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *
t := time.NewTimer(10 * time.Minute)
defer t.Stop()
- log.Debug("waiting delivery", "peer", sp.ID(), "hash", req.Addr, "node", common.Bytes2Hex(d.overlay.BaseAddr()), "created", created)
+ log.Debug("waiting delivery", "peer", sp.ID(), "hash", req.Addr, "node", common.Bytes2Hex(d.kad.BaseAddr()), "created", created)
start := time.Now()
select {
case <-chunk.ReqC:
@@ -269,8 +269,8 @@ func (d *Delivery) RequestFromPeers(ctx context.Context, hash []byte, skipCheck
var err error
requestFromPeersCount.Inc(1)
- d.overlay.EachConn(hash, 255, func(p network.OverlayConn, po int, nn bool) bool {
- spId := p.(network.Peer).ID()
+ d.kad.EachConn(hash, 255, func(p *network.Peer, po int, nn bool) bool {
+ spId := p.ID()
for _, p := range peersToSkip {
if p == spId {
log.Trace("Delivery.RequestFromPeers: skip peer", "peer", spId)
diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go
index 4e1ab09fc..313019d6a 100644
--- a/swarm/network/stream/snapshot_sync_test.go
+++ b/swarm/network/stream/snapshot_sync_test.go
@@ -457,15 +457,10 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error {
//returns the number of subscriptions requested
func startSyncing(r *Registry, conf *synctestConfig) (int, error) {
var err error
-
- kad, ok := r.delivery.overlay.(*network.Kademlia)
- if !ok {
- return 0, fmt.Errorf("Not a Kademlia!")
- }
-
+ kad := r.delivery.kad
subCnt := 0
//iterate over each bin and solicit needed subscription to bins
- kad.EachBin(r.addr.Over(), pof, 0, func(conn network.OverlayConn, po int) bool {
+ kad.EachBin(r.addr.Over(), pof, 0, func(conn *network.Peer, po int) bool {
//identify begin and start index of the bin(s) we want to subscribe to
histRange := &Range{}
diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go
index cd0580a0c..deffdfc3f 100644
--- a/swarm/network/stream/stream.go
+++ b/swarm/network/stream/stream.go
@@ -130,7 +130,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i
// wait for kademlia table to be healthy
time.Sleep(options.SyncUpdateDelay)
- kad := streamer.delivery.overlay.(*network.Kademlia)
+ kad := streamer.delivery.kad
depthC := latestIntC(kad.NeighbourhoodDepthC())
addressBookSizeC := latestIntC(kad.AddrCountC())
@@ -398,9 +398,7 @@ func (r *Registry) Run(p *network.BzzPeer) error {
// and they are no longer required after iteration, request to Quit
// them will be send to appropriate peers.
func (r *Registry) updateSyncing() {
- // if overlay in not Kademlia, panic
- kad := r.delivery.overlay.(*network.Kademlia)
-
+ kad := r.delivery.kad
// map of all SYNC streams for all peers
// used at the and of the function to remove servers
// that are not needed anymore
@@ -421,8 +419,7 @@ func (r *Registry) updateSyncing() {
r.peersMu.RUnlock()
// request subscriptions for all nodes and bins
- kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(conn network.OverlayConn, bin int) bool {
- p := conn.(network.Peer)
+ kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(p *network.Peer, bin int) bool {
log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin))
// bin is always less then 256 and it is safe to convert it to type uint8
@@ -461,10 +458,11 @@ func (r *Registry) updateSyncing() {
func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
peer := protocols.NewPeer(p, rw, Spec)
- bzzPeer := network.NewBzzTestPeer(peer, r.addr)
- r.delivery.overlay.On(bzzPeer)
- defer r.delivery.overlay.Off(bzzPeer)
- return r.Run(bzzPeer)
+ bp := network.NewBzzPeer(peer, r.addr)
+ np := network.NewPeer(bp, r.delivery.kad)
+ r.delivery.kad.On(np)
+ defer r.delivery.kad.Off(np)
+ return r.Run(bp)
}
// HandleMsg is the message handler that delegates incoming messages
diff --git a/swarm/network_test.go b/swarm/network_test.go
index 176c635d8..9bc6143d8 100644
--- a/swarm/network_test.go
+++ b/swarm/network_test.go
@@ -34,7 +34,6 @@ import (
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/api"
- "github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/storage"
colorable "github.com/mattn/go-colorable"
@@ -293,7 +292,7 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa
if err != nil {
return nil, cleanup, err
}
- bucket.Store(simulation.BucketKeyKademlia, swarm.bzz.Hive.Overlay.(*network.Kademlia))
+ bucket.Store(simulation.BucketKeyKademlia, swarm.bzz.Hive.Kademlia)
log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()))
return swarm, cleanup, nil
},
diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go
index 8459211dd..b55c97fdd 100644
--- a/swarm/pss/pss.go
+++ b/swarm/pss/pss.go
@@ -110,10 +110,10 @@ func (params *PssParams) WithPrivateKey(privatekey *ecdsa.PrivateKey) *PssParams
//
// Implements node.Service
type Pss struct {
- network.Overlay // we can get the overlayaddress from this
- privateKey *ecdsa.PrivateKey // pss can have it's own independent key
- w *whisper.Whisper // key and encryption backend
- auxAPIs []rpc.API // builtins (handshake, test) can add APIs
+ *network.Kademlia // we can get the Kademlia address from this
+ privateKey *ecdsa.PrivateKey // pss can have it's own independent key
+ w *whisper.Whisper // key and encryption backend
+ auxAPIs []rpc.API // builtins (handshake, test) can add APIs
// sending and forwarding
fwdPool map[string]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer
@@ -151,9 +151,9 @@ func (p *Pss) String() string {
// Creates a new Pss instance.
//
-// In addition to params, it takes a swarm network overlay
+// In addition to params, it takes a swarm network Kademlia
// and a FileStore storage for message cache storage.
-func NewPss(k network.Overlay, params *PssParams) (*Pss, error) {
+func NewPss(k *network.Kademlia, params *PssParams) (*Pss, error) {
if params.privateKey == nil {
return nil, errors.New("missing private key for pss")
}
@@ -162,7 +162,7 @@ func NewPss(k network.Overlay, params *PssParams) (*Pss, error) {
Version: pssVersion,
}
ps := &Pss{
- Overlay: k,
+ Kademlia: k,
privateKey: params.privateKey,
w: whisper.New(&whisper.DefaultConfig),
quitC: make(chan struct{}),
@@ -290,9 +290,9 @@ func (p *Pss) addAPI(api rpc.API) {
p.auxAPIs = append(p.auxAPIs, api)
}
-// Returns the swarm overlay address of the pss node
+// Returns the swarm Kademlia address of the pss node
func (p *Pss) BaseAddr() []byte {
- return p.Overlay.BaseAddr()
+ return p.Kademlia.BaseAddr()
}
// Returns the pss node's public key
@@ -356,11 +356,11 @@ func (p *Pss) handlePssMsg(ctx context.Context, msg interface{}) error {
}
if int64(pssmsg.Expire) < time.Now().Unix() {
metrics.GetOrRegisterCounter("pss.expire", nil).Inc(1)
- log.Warn("pss filtered expired message", "from", common.ToHex(p.Overlay.BaseAddr()), "to", common.ToHex(pssmsg.To))
+ log.Warn("pss filtered expired message", "from", common.ToHex(p.Kademlia.BaseAddr()), "to", common.ToHex(pssmsg.To))
return nil
}
if p.checkFwdCache(pssmsg) {
- log.Trace("pss relay block-cache match (process)", "from", common.ToHex(p.Overlay.BaseAddr()), "to", (common.ToHex(pssmsg.To)))
+ log.Trace("pss relay block-cache match (process)", "from", common.ToHex(p.Kademlia.BaseAddr()), "to", (common.ToHex(pssmsg.To)))
return nil
}
p.addFwdCache(pssmsg)
@@ -442,12 +442,12 @@ func (p *Pss) executeHandlers(topic Topic, payload []byte, from *PssAddress, asy
// will return false if using partial address
func (p *Pss) isSelfRecipient(msg *PssMsg) bool {
- return bytes.Equal(msg.To, p.Overlay.BaseAddr())
+ return bytes.Equal(msg.To, p.Kademlia.BaseAddr())
}
-// test match of leftmost bytes in given message to node's overlay address
+// test match of leftmost bytes in given message to node's Kademlia address
func (p *Pss) isSelfPossibleRecipient(msg *PssMsg) bool {
- local := p.Overlay.BaseAddr()
+ local := p.Kademlia.BaseAddr()
return bytes.Equal(msg.To[:], local[:len(msg.To)])
}
@@ -816,14 +816,7 @@ func (p *Pss) forward(msg *PssMsg) error {
// send with kademlia
// find the closest peer to the recipient and attempt to send
sent := 0
- p.Overlay.EachConn(to, 256, func(op network.OverlayConn, po int, isproxbin bool) bool {
- // we need p2p.protocols.Peer.Send
- // cast and resolve
- sp, ok := op.(senderPeer)
- if !ok {
- log.Crit("Pss cannot use kademlia peer type")
- return false
- }
+ p.Kademlia.EachConn(to, 256, func(sp *network.Peer, po int, isproxbin bool) bool {
info := sp.Info()
// check if the peer is running pss
@@ -840,7 +833,7 @@ func (p *Pss) forward(msg *PssMsg) error {
}
// get the protocol peer from the forwarding peer cache
- sendMsg := fmt.Sprintf("MSG TO %x FROM %x VIA %x", to, p.BaseAddr(), op.Address())
+ sendMsg := fmt.Sprintf("MSG TO %x FROM %x VIA %x", to, p.BaseAddr(), sp.Address())
p.fwdPoolMu.RLock()
pp := p.fwdPool[sp.Info().ID]
p.fwdPoolMu.RUnlock()
@@ -859,11 +852,11 @@ func (p *Pss) forward(msg *PssMsg) error {
// - if the peer is end recipient but the full address has not been disclosed
// - if the peer address matches the partial address fully
// - if the peer is in proxbin
- if len(msg.To) < addressLength && bytes.Equal(msg.To, op.Address()[:len(msg.To)]) {
+ if len(msg.To) < addressLength && bytes.Equal(msg.To, sp.Address()[:len(msg.To)]) {
log.Trace(fmt.Sprintf("Pss keep forwarding: Partial address + full partial match"))
return true
} else if isproxbin {
- log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ToHex(op.Address())))
+ log.Trace(fmt.Sprintf("%x is in proxbin, keep forwarding", common.ToHex(sp.Address())))
return true
}
// at this point we stop forwarding, and the state is as follows:
diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go
index 41b03db28..6ba04cb5d 100644
--- a/swarm/pss/pss_test.go
+++ b/swarm/pss/pss_test.go
@@ -556,23 +556,6 @@ OUTER:
}
}
-type pssTestPeer struct {
- *protocols.Peer
- addr []byte
-}
-
-func (t *pssTestPeer) Address() []byte {
- return t.addr
-}
-
-func (t *pssTestPeer) Update(addr network.OverlayAddr) network.OverlayAddr {
- return addr
-}
-
-func (t *pssTestPeer) Off() network.OverlayAddr {
- return &pssTestPeer{}
-}
-
// forwarding should skip peers that do not have matching pss capabilities
func TestMismatch(t *testing.T) {
@@ -582,7 +565,7 @@ func TestMismatch(t *testing.T) {
t.Fatal(err)
}
- // initialize overlay
+ // initialize kad
baseaddr := network.RandomAddr()
kad := network.NewKademlia((baseaddr).Over(), network.NewKadParams())
rw := &p2p.MsgPipeRW{}
@@ -594,10 +577,10 @@ func TestMismatch(t *testing.T) {
Version: 0,
}
nid, _ := discover.HexID("0x01")
- wrongpsspeer := &pssTestPeer{
- Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(wrongpssaddr.Over()), []p2p.Cap{wrongpsscap}), rw, nil),
- addr: wrongpssaddr.Over(),
- }
+ wrongpsspeer := network.NewPeer(&network.BzzPeer{
+ Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(wrongpssaddr.Over()), []p2p.Cap{wrongpsscap}), rw, nil),
+ BzzAddr: &network.BzzAddr{OAddr: wrongpssaddr.Over(), UAddr: nil},
+ }, kad)
// one peer doesn't even have pss (boo!)
nopssaddr := network.RandomAddr()
@@ -606,16 +589,16 @@ func TestMismatch(t *testing.T) {
Version: 1,
}
nid, _ = discover.HexID("0x02")
- nopsspeer := &pssTestPeer{
- Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(nopssaddr.Over()), []p2p.Cap{nopsscap}), rw, nil),
- addr: nopssaddr.Over(),
- }
+ nopsspeer := network.NewPeer(&network.BzzPeer{
+ Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(nopssaddr.Over()), []p2p.Cap{nopsscap}), rw, nil),
+ BzzAddr: &network.BzzAddr{OAddr: nopssaddr.Over(), UAddr: nil},
+ }, kad)
// add peers to kademlia and activate them
// it's safe so don't check errors
- kad.Register([]network.OverlayAddr{wrongpsspeer})
+ kad.Register(wrongpsspeer.BzzAddr)
kad.On(wrongpsspeer)
- kad.Register([]network.OverlayAddr{nopsspeer})
+ kad.Register(nopsspeer.BzzAddr)
kad.On(nopsspeer)
// create pss
@@ -1636,17 +1619,17 @@ func newServices(allowRaw bool) adapters.Services {
}
}
-func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *PssParams) *Pss {
+func newTestPss(privkey *ecdsa.PrivateKey, kad *network.Kademlia, ppextra *PssParams) *Pss {
var nid discover.NodeID
copy(nid[:], crypto.FromECDSAPub(&privkey.PublicKey))
addr := network.NewAddrFromNodeID(nid)
// set up routing if kademlia is not passed to us
- if overlay == nil {
+ if kad == nil {
kp := network.NewKadParams()
kp.MinProxBinSize = 3
- overlay = network.NewKademlia(addr.Over(), kp)
+ kad = network.NewKademlia(addr.Over(), kp)
}
// create pss
@@ -1654,7 +1637,7 @@ func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *Pss
if ppextra != nil {
pp.SymKeyCacheCapacity = ppextra.SymKeyCacheCapacity
}
- ps, err := NewPss(overlay, pp)
+ ps, err := NewPss(kad, pp)
if err != nil {
return nil
}
diff --git a/swarm/swarm.go b/swarm/swarm.go
index 736cd37de..baf71b962 100644
--- a/swarm/swarm.go
+++ b/swarm/swarm.go
@@ -356,7 +356,7 @@ func (self *Swarm) Start(srv *p2p.Server) error {
log.Error("bzz failed", "err", err)
return err
}
- log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.Overlay.BaseAddr()))
+ log.Info("Swarm network started", "bzzaddr", fmt.Sprintf("%x", self.bzz.Hive.BaseAddr()))
if self.ps != nil {
self.ps.Start(srv)
--
cgit