cerc-io/plugeth
18
0
Fork 1
Code Issues 4 Pull Requests Packages Projects Releases Wiki Activity

Kademlia refactor (#17641)

Browse Source 
* 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
This commit is contained in:
Viktor Trón 2018-09-12 11:24:56 +02:00 committed by Balint Gabor
parent b06ff563a1
commit bfce00385f
16 changed files with 260 additions and 421 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
swarm/network/discovery.go
View File

@ -26,30 +26,30 @@ import (
// discovery bzz extension for requesting and relaying node address records // discovery bzz extension for requesting and relaying node address records
// discPeer wraps BzzPeer and embeds an Overlay connectivity driver // Peer wraps BzzPeer and embeds Kademlia overlay connectivity driver
type discPeer struct { type Peer struct {
*BzzPeer *BzzPeer
overlay Overlay kad *Kademlia
sentPeers bool // whether we already sent peer closer to this address sentPeers bool // whether we already sent peer closer to this address
mtx sync.RWMutex mtx sync.RWMutex //
peers map[string]bool // tracks node records sent to the peer peers map[string]bool // tracks node records sent to the peer
depth uint8 // the proximity order advertised by remote as depth of saturation depth uint8 // the proximity order advertised by remote as depth of saturation
} }
// NewDiscovery constructs a discovery peer // NewPeer constructs a discovery peer
func newDiscovery(p *BzzPeer, o Overlay) *discPeer { func NewPeer(p *BzzPeer, kad *Kademlia) *Peer {
d := &discPeer{ d := &Peer{
overlay: o, kad: kad,
BzzPeer: p, BzzPeer: p,
peers: make(map[string]bool), peers: make(map[string]bool),
} }
// record remote as seen so we never send a peer its own record // record remote as seen so we never send a peer its own record
d.seen(d) d.seen(p.BzzAddr)
return d return d
} }
// HandleMsg is the message handler that delegates incoming messages // 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) { switch msg := msg.(type) {
case *peersMsg: 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 // NotifyDepth sends a message to all connections if depth of saturation is changed
func NotifyDepth(depth uint8, h Overlay) { func NotifyDepth(depth uint8, kad *Kademlia) {
f := func(val OverlayConn, po int, _ bool) bool { f := func(val *Peer, po int, _ bool) bool {
dp, ok := val.(*discPeer) val.NotifyDepth(depth)
if ok {
dp.NotifyDepth(depth)
}
return true return true
} }
h.EachConn(nil, 255, f) kad.EachConn(nil, 255, f)
} }
// NotifyPeer informs all peers about a newly added node // NotifyPeer informs all peers about a newly added node
func NotifyPeer(p OverlayAddr, k Overlay) { func NotifyPeer(p *BzzAddr, k *Kademlia) {
f := func(val OverlayConn, po int, _ bool) bool { f := func(val *Peer, po int, _ bool) bool {
dp, ok := val.(*discPeer) val.NotifyPeer(p, uint8(po))
if ok {
dp.NotifyPeer(p, uint8(po))
}
return true return true
} }
k.EachConn(p.Address(), 255, f) 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 // the peer's PO is within the recipients advertised depth
// OR the peer is closer to the recipient than self // OR the peer is closer to the recipient than self
// unless already notified during the connection session // 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 // immediately return
if (po < d.getDepth() && pot.ProxCmp(d.localAddr, d, a) != 1) || d.seen(a) { if (po < d.getDepth() && pot.ProxCmp(d.localAddr, d, a) != 1) || d.seen(a) {
return return
} }
// log.Trace(fmt.Sprintf("%08x peer %08x notified of peer %08x", d.localAddr.Over()[:4], d.Address()[:4], a.Address()[:4]))
resp := &peersMsg{ resp := &peersMsg{
Peers: []*BzzAddr{ToAddr(a)}, Peers: []*BzzAddr{a},
} }
go d.Send(context.TODO(), resp) go d.Send(context.TODO(), resp)
} }
// NotifyDepth sends a subPeers Msg to the receiver notifying them about // NotifyDepth sends a subPeers Msg to the receiver notifying them about
// a change in the depth of saturation // a change in the depth of saturation
func (d *discPeer) NotifyDepth(po uint8) { func (d *Peer) NotifyDepth(po uint8) {
// log.Trace(fmt.Sprintf("%08x peer %08x notified of new depth %v", d.localAddr.Over()[:4], d.Address()[:4], po))
go d.Send(context.TODO(), &subPeersMsg{Depth: po}) 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) // 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 // list of nodes ([]PeerAddr in peersMsg) is added to the overlay db using the
// Register interface method // Register interface method
func (d *discPeer) handlePeersMsg(msg *peersMsg) error { func (d *Peer) handlePeersMsg(msg *peersMsg) error {
// register all addresses // register all addresses
if len(msg.Peers) == 0 { if len(msg.Peers) == 0 {
return nil return nil
@ -149,12 +141,12 @@ func (d *discPeer) handlePeersMsg(msg *peersMsg) error {
for _, a := range msg.Peers { for _, a := range msg.Peers {
d.seen(a) 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 { type subPeersMsg struct {
Depth uint8 Depth uint8
} }
@ -164,21 +156,20 @@ func (msg subPeersMsg) String() string {
return fmt.Sprintf("%T: request peers > PO%02d. ", msg, msg.Depth) 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 { if !d.sentPeers {
d.setDepth(msg.Depth) d.setDepth(msg.Depth)
var peers []*BzzAddr 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 { if pob, _ := pof(d, d.localAddr, 0); pob > po {
return false return false
} }
if !d.seen(p) { if !d.seen(p.BzzAddr) {
peers = append(peers, ToAddr(p.Off())) peers = append(peers, p.BzzAddr)
} }
return true return true
}) })
if len(peers) > 0 { 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}) go d.Send(context.TODO(), &peersMsg{Peers: peers})
} }
} }
@ -186,9 +177,9 @@ func (d *discPeer) handleSubPeersMsg(msg *subPeersMsg) error {
return nil 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 // if not, marks the peer as sent
func (d *discPeer) seen(p OverlayPeer) bool { func (d *Peer) seen(p *BzzAddr) bool {
d.mtx.Lock() d.mtx.Lock()
defer d.mtx.Unlock() defer d.mtx.Unlock()
k := string(p.Address()) k := string(p.Address())
@ -199,12 +190,13 @@ func (d *discPeer) seen(p OverlayPeer) bool {
return false return false
} }
func (d *discPeer) getDepth() uint8 { func (d *Peer) getDepth() uint8 {
d.mtx.RLock() d.mtx.RLock()
defer d.mtx.RUnlock() defer d.mtx.RUnlock()
return d.depth return d.depth
} }
func (d *discPeer) setDepth(depth uint8) {
func (d *Peer) setDepth(depth uint8) {
d.mtx.Lock() d.mtx.Lock()
defer d.mtx.Unlock() defer d.mtx.Unlock()
d.depth = depth d.depth = depth

2
swarm/network/discovery_test.go
View File

@ -33,7 +33,7 @@ func TestDiscovery(t *testing.T) {
id := s.IDs[0] id := s.IDs[0]
raddr := NewAddrFromNodeID(id) raddr := NewAddrFromNodeID(id)
pp.Register([]OverlayAddr{OverlayAddr(raddr)}) pp.Register(raddr)
// start the hive and wait for the connection // start the hive and wait for the connection
pp.Start(s.Server) pp.Start(s.Server)

71
swarm/network/hive.go
View File

@ -32,31 +32,10 @@ import (
Hive is the logistic manager of the swarm Hive is the logistic manager of the swarm
When the hive is started, a forever loop is launched that 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 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 // HiveParams holds the config options to hive
type HiveParams struct { type HiveParams struct {
Discovery bool // if want discovery of not Discovery bool // if want discovery of not
@ -78,7 +57,7 @@ func NewHiveParams() *HiveParams {
// Hive manages network connections of the swarm node // Hive manages network connections of the swarm node
type Hive struct { type Hive struct {
*HiveParams // settings *HiveParams // settings
Overlay // the overlay connectiviy driver *Kademlia // the overlay connectiviy driver
Store state.Store // storage interface to save peers across sessions Store state.Store // storage interface to save peers across sessions
addPeer func(*discover.Node) // server callback to connect to a peer addPeer func(*discover.Node) // server callback to connect to a peer
// bookkeeping // bookkeeping
@ -88,12 +67,12 @@ type Hive struct {
// NewHive constructs a new hive // NewHive constructs a new hive
// HiveParams: config parameters // HiveParams: config parameters
// Overlay: connectivity driver using a network topology // Kademlia: connectivity driver using a network topology
// StateStore: to save peers across sessions // 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{ return &Hive{
HiveParams: params, HiveParams: params,
Overlay: overlay, Kademlia: kad,
Store: store, Store: store,
} }
} }
@ -133,7 +112,7 @@ func (h *Hive) Stop() error {
} }
} }
log.Info(fmt.Sprintf("%08x hive stopped, dropping peers", h.BaseAddr()[:4])) 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])) log.Info(fmt.Sprintf("%08x dropping peer %08x", h.BaseAddr()[:4], p.Address()[:4]))
p.Drop(nil) p.Drop(nil)
return true return true
@ -151,14 +130,14 @@ func (h *Hive) connect() {
addr, depth, changed := h.SuggestPeer() addr, depth, changed := h.SuggestPeer()
if h.Discovery && changed { if h.Discovery && changed {
NotifyDepth(uint8(depth), h) NotifyDepth(uint8(depth), h.Kademlia)
} }
if addr == nil { if addr == nil {
continue continue
} }
log.Trace(fmt.Sprintf("%08x hive connect() suggested %08x", h.BaseAddr()[:4], addr.Address()[:4])) 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 { 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)) log.Warn(fmt.Sprintf("%08x unable to connect to bee %08x: invalid node URL: %v", h.BaseAddr()[:4], addr.Address()[:4], err))
continue continue
@ -170,19 +149,19 @@ func (h *Hive) connect() {
// Run protocol run function // Run protocol run function
func (h *Hive) Run(p *BzzPeer) error { func (h *Hive) Run(p *BzzPeer) error {
dp := newDiscovery(p, h) dp := NewPeer(p, h.Kademlia)
depth, changed := h.On(dp) depth, changed := h.On(dp)
// if we want discovery, advertise change of depth // if we want discovery, advertise change of depth
if h.Discovery { if h.Discovery {
if changed { if changed {
// if depth changed, send to all peers // if depth changed, send to all peers
NotifyDepth(depth, h) NotifyDepth(depth, h.Kademlia)
} else { } else {
// otherwise just send depth to new peer // otherwise just send depth to new peer
dp.NotifyDepth(depth) dp.NotifyDepth(depth)
} }
} }
NotifyPeer(p.Off(), h) NotifyPeer(p.BzzAddr, h.Kademlia)
defer h.Off(dp) defer h.Off(dp)
return dp.Run(dp.HandleMsg) 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/ // loadPeers, savePeer implement persistence callback/
func (h *Hive) loadPeers() error { func (h *Hive) loadPeers() error {
var as []*BzzAddr var as []*BzzAddr
@ -230,28 +198,19 @@ func (h *Hive) loadPeers() error {
} }
log.Info(fmt.Sprintf("hive %08x: peers loaded", h.BaseAddr()[:4])) log.Info(fmt.Sprintf("hive %08x: peers loaded", h.BaseAddr()[:4]))
return h.Register(toOverlayAddrs(as...)) return h.Register(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/ // savePeers, savePeer implement persistence callback/
func (h *Hive) savePeers() error { func (h *Hive) savePeers() error {
var peers []*BzzAddr 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 { if pa == nil {
log.Warn(fmt.Sprintf("empty addr: %v", i)) log.Warn(fmt.Sprintf("empty addr: %v", i))
return true return true
} }
apa := ToAddr(pa) log.Trace("saving peer", "peer", pa)
log.Trace("saving peer", "peer", apa) peers = append(peers, pa)
peers = append(peers, apa)
return true return true
}) })
if err := h.Store.Put("peers", peers); err != nil { if err := h.Store.Put("peers", peers); err != nil {

8
swarm/network/hive_test.go
View File

@ -41,7 +41,7 @@ func TestRegisterAndConnect(t *testing.T) {
id := s.IDs[0] id := s.IDs[0]
raddr := NewAddrFromNodeID(id) raddr := NewAddrFromNodeID(id)
pp.Register([]OverlayAddr{OverlayAddr(raddr)}) pp.Register(raddr)
// start the hive and wait for the connection // start the hive and wait for the connection
err := pp.Start(s.Server) err := pp.Start(s.Server)
@ -77,7 +77,7 @@ func TestHiveStatePersistance(t *testing.T) {
peers := make(map[string]bool) peers := make(map[string]bool)
for _, id := range s.IDs { for _, id := range s.IDs {
raddr := NewAddrFromNodeID(id) raddr := NewAddrFromNodeID(id)
pp.Register([]OverlayAddr{OverlayAddr(raddr)}) pp.Register(raddr)
peers[raddr.String()] = true peers[raddr.String()] = true
} }
@ -97,8 +97,8 @@ func TestHiveStatePersistance(t *testing.T) {
pp.Start(s1.Server) pp.Start(s1.Server)
i := 0 i := 0
pp.Overlay.EachAddr(nil, 256, func(addr OverlayAddr, po int, nn bool) bool { pp.Kademlia.EachAddr(nil, 256, func(addr *BzzAddr, po int, nn bool) bool {
delete(peers, addr.(*BzzAddr).String()) delete(peers, addr.String())
i++ i++
return true return true
}) })

154
swarm/network/kademlia.go
View File

@ -62,7 +62,7 @@ type KadParams struct {
RetryExponent int // exponent to multiply retry intervals with RetryExponent int // exponent to multiply retry intervals with
MaxRetries int // maximum number of redial attempts MaxRetries int // maximum number of redial attempts
// function to sanction or prevent suggesting a peer // function to sanction or prevent suggesting a peer
Reachable func(OverlayAddr) bool Reachable func(*BzzAddr) bool
} }
// NewKadParams returns a params struct with default values // 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 // entry represents a Kademlia table entry (an extension of BzzAddr)
// 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 { type entry struct {
OverlayPeer *BzzAddr
conn *Peer
seenAt time.Time seenAt time.Time
retries int retries int
} }
// newEntry creates a kademlia peer from an OverlayPeer interface // newEntry creates a kademlia peer from a *Peer
func newEntry(p OverlayPeer) *entry { func newEntry(p *BzzAddr) *entry {
return &entry{ return &entry{
OverlayPeer: p, BzzAddr: p,
seenAt: time.Now(), 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 // Label is a short tag for the entry for debug
func Label(e *entry) string { func Label(e *entry) string {
return fmt.Sprintf("%s (%d)", e.Hex()[:4], e.retries) 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 // Hex is the hexadecimal serialisation of the entry address
func (e *entry) Hex() string { 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 // Register enters each address as kademlia peer record into the
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 // database of known peer addresses
func (k *Kademlia) Register(peers []OverlayAddr) error { func (k *Kademlia) Register(peers ...*BzzAddr) error {
k.lock.Lock() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
var known, size int var known, size int
@ -203,7 +163,6 @@ func (k *Kademlia) Register(peers []OverlayAddr) error {
if k.addrCountC != nil && size-known > 0 { if k.addrCountC != nil && size-known > 0 {
k.addrCountC <- k.addrs.Size() 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() k.sendNeighbourhoodDepthChange()
return nil 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 // SuggestPeer returns a known peer for the lowest proximity bin for the
// lowest bincount below depth // lowest bincount below depth
// naturally if there is an empty row it returns a peer for that // 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() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
minsize := k.MinBinSize minsize := k.MinBinSize
@ -224,15 +183,18 @@ func (k *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
if po < depth { if po < depth {
return false return false
} }
a = k.callable(val) e := val.(*entry)
c := k.callable(e)
if c {
a = e.BzzAddr
}
ppo = po ppo = po
return a == nil return !c
}) })
if a != nil { if a != nil {
log.Trace(fmt.Sprintf("%08x candidate nearest neighbour found: %v (%v)", k.BaseAddr()[:4], a, ppo)) log.Trace(fmt.Sprintf("%08x candidate nearest neighbour found: %v (%v)", k.BaseAddr()[:4], a, ppo))
return a, 0, false 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 var bpo []int
prev := -1 prev := -1
@ -250,7 +212,6 @@ func (k *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
}) })
// all buckets are full, ie., minsize == k.MinBinSize // all buckets are full, ie., minsize == k.MinBinSize
if len(bpo) == 0 { if len(bpo) == 0 {
// log.Debug(fmt.Sprintf("%08x: all bins saturated", k.BaseAddr()[:4]))
return nil, 0, false return nil, 0, false
} }
// as long as we got candidate peers to connect to // 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 false
} }
return f(func(val pot.Val, _ int) bool { return f(func(val pot.Val, _ int) bool {
a = k.callable(val) e := val.(*entry)
return a == nil c := k.callable(e)
if c {
a = e.BzzAddr
}
return !c
}) })
}) })
// found a candidate // 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 // 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() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
e := newEntry(p)
var ins bool var ins bool
k.conns, _, _, _ = pot.Swap(k.conns, p, pof, func(v pot.Val) pot.Val { k.conns, _, _, _ = pot.Swap(k.conns, p, pof, func(v pot.Val) pot.Val {
// if not found live // if not found live
if v == nil { if v == nil {
ins = true ins = true
// insert new online peer into conns // insert new online peer into conns
return e return p
} }
// found among live peers, do nothing // found among live peers, do nothing
return v return v
}) })
if ins { if ins {
a := newEntry(p.BzzAddr)
a.conn = p
// insert new online peer into addrs // insert new online peer into addrs
k.addrs, _, _, _ = pot.Swap(k.addrs, p, pof, func(v pot.Val) pot.Val { 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 // send new address count value only if the peer is inserted
if k.addrCountC != nil { 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 // Not receiving from the returned channel will block On function
// when the neighbourhood depth is changed. // when the neighbourhood depth is changed.
func (k *Kademlia) NeighbourhoodDepthC() <-chan int { func (k *Kademlia) NeighbourhoodDepthC() <-chan int {
k.lock.Lock()
defer k.lock.Unlock()
if k.nDepthC == nil { if k.nDepthC == nil {
k.nDepthC = make(chan int) k.nDepthC = make(chan int)
} }
@ -357,7 +325,7 @@ func (k *Kademlia) AddrCountC() <-chan int {
} }
// Off removes a peer from among live peers // Off removes a peer from among live peers
func (k *Kademlia) Off(p OverlayConn) { func (k *Kademlia) Off(p *Peer) {
k.lock.Lock() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
var del bool var del bool
@ -367,7 +335,7 @@ func (k *Kademlia) Off(p OverlayConn) {
panic(fmt.Sprintf("connected peer not found %v", p)) panic(fmt.Sprintf("connected peer not found %v", p))
} }
del = true del = true
return newEntry(p.Off()) return newEntry(p.BzzAddr)
}) })
if del { 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() k.lock.RLock()
defer k.lock.RUnlock() 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++ { for bin := startPo; bin <= endPo; bin++ {
f(func(val pot.Val, _ int) bool { f(func(val pot.Val, _ int) bool {
return eachBinFunc(val.(*entry).conn(), bin) return eachBinFunc(val.(*Peer), bin)
}) })
} }
return true 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 // 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 // that has proximity order po or less as measured from the base
// if base is nil, kademlia base address is used // 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() k.lock.RLock()
defer k.lock.RUnlock() defer k.lock.RUnlock()
k.eachConn(base, o, f) 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 { if len(base) == 0 {
base = k.base base = k.base
} }
@ -428,20 +396,20 @@ func (k *Kademlia) eachConn(base []byte, o int, f func(OverlayConn, int, bool) b
if po > o { if po > o {
return true 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 // 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 // that has proximity order po or less as measured from the base
// if base is nil, kademlia base address is used // 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() k.lock.RLock()
defer k.lock.RUnlock() defer k.lock.RUnlock()
k.eachAddr(base, o, f) 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 { if len(base) == 0 {
base = k.base base = k.base
} }
@ -450,7 +418,7 @@ func (k *Kademlia) eachAddr(base []byte, o int, f func(OverlayAddr, int, bool) b
if po > o { if po > o {
return true 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 return depth
} }
// callable when called with val, // callable decides if an address entry represents a callable peer
func (k *Kademlia) callable(val pot.Val) OverlayAddr { func (k *Kademlia) callable(e *entry) bool {
e := val.(*entry)
// not callable if peer is live or exceeded maxRetries // not callable if peer is live or exceeded maxRetries
if e.conn() != nil || e.retries > k.MaxRetries { if e.conn != nil || e.retries > k.MaxRetries {
return nil return false
} }
// calculate the allowed number of retries based on time lapsed since last seen // calculate the allowed number of retries based on time lapsed since last seen
timeAgo := int64(time.Since(e.seenAt)) timeAgo := int64(time.Since(e.seenAt))
@ -491,17 +458,17 @@ func (k *Kademlia) callable(val pot.Val) OverlayAddr {
// peer can be retried again // peer can be retried again
if retries < e.retries { 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)) 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 // 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)) log.Trace(fmt.Sprintf("%08x: peer %v is temporarily not callable", k.BaseAddr()[:4], e))
return nil return false
} }
e.retries++ e.retries++
log.Trace(fmt.Sprintf("%08x: peer %v is callable", k.BaseAddr()[:4], e)) log.Trace(fmt.Sprintf("%08x: peer %v is callable", k.BaseAddr()[:4], e))
return e.addr() return true
} }
// BaseAddr return the kademlia base address // BaseAddr return the kademlia base address
@ -516,7 +483,8 @@ func (k *Kademlia) String() string {
return k.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 { func (k *Kademlia) string() string {
wsrow := " " wsrow := " "
var rows []string var rows []string
@ -538,7 +506,7 @@ func (k *Kademlia) string() string {
row := []string{fmt.Sprintf("%2d", size)} row := []string{fmt.Sprintf("%2d", size)}
rest -= size rest -= size
f(func(val pot.Val, vpo int) bool { f(func(val pot.Val, vpo int) bool {
e := val.(*entry) e := val.(*Peer)
row = append(row, fmt.Sprintf("%x", e.Address()[:2])) row = append(row, fmt.Sprintf("%x", e.Address()[:2]))
rowlen++ rowlen++
return rowlen < 4 return rowlen < 4
@ -594,8 +562,9 @@ type PeerPot struct {
EmptyBins []int 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. // as hexadecimal representations of the address.
// used for testing only
func NewPeerPotMap(kadMinProxSize int, addrs [][]byte) map[string]*PeerPot { func NewPeerPotMap(kadMinProxSize int, addrs [][]byte) map[string]*PeerPot {
// create a table of all nodes for health check // create a table of all nodes for health check
np := pot.NewPot(nil, 0) 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 // saturation returns the lowest proximity order that the bin for that order
// has less than n peers // has less than n peers
// It is used in Healthy function for testing only
func (k *Kademlia) saturation(n int) int { func (k *Kademlia) saturation(n int) int {
prev := -1 prev := -1
k.addrs.EachBin(k.base, pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool { 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. // 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) { func (k *Kademlia) full(emptyBins []int) (full bool) {
prev := 0 prev := 0
e := len(emptyBins) e := len(emptyBins)
@ -688,10 +658,13 @@ func (k *Kademlia) full(emptyBins []int) (full bool) {
return e == 0 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 { func (k *Kademlia) knowNearestNeighbours(peers [][]byte) bool {
pm := make(map[string]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 { if !nn {
return false return false
} }
@ -709,10 +682,13 @@ func (k *Kademlia) knowNearestNeighbours(peers [][]byte) bool {
return true 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) { func (k *Kademlia) gotNearestNeighbours(peers [][]byte) (got bool, n int, missing [][]byte) {
pm := make(map[string]bool) 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 { if !nn {
return false return false
} }
@ -735,6 +711,7 @@ func (k *Kademlia) gotNearestNeighbours(peers [][]byte) (got bool, n int, missin
} }
// Health state of the Kademlia // Health state of the Kademlia
// used for testing only
type Health struct { type Health struct {
KnowNN bool // whether node knows all its nearest neighbours KnowNN bool // whether node knows all its nearest neighbours
GotNN bool // whether node is connected to 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 // Healthy reports the health state of the kademlia connectivity
// returns a Health struct // returns a Health struct
// used for testing only
func (k *Kademlia) Healthy(pp *PeerPot) *Health { func (k *Kademlia) Healthy(pp *PeerPot) *Health {
k.lock.RLock() k.lock.RLock()
defer k.lock.RUnlock() defer k.lock.RUnlock()

183
swarm/network/kademlia_test.go
View File

@ -38,71 +38,42 @@ func testKadPeerAddr(s string) *BzzAddr {
return &BzzAddr{OAddr: a, UAddr: a} return &BzzAddr{OAddr: a, UAddr: a}
} }
type testDropPeer struct { func newTestKademlia(b string) *Kademlia {
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 := NewKadParams()
params.MinBinSize = 1 params.MinBinSize = 1
params.MinProxBinSize = 2 params.MinProxBinSize = 2
base := pot.NewAddressFromString(b) base := pot.NewAddressFromString(b)
return &testKademlia{ return NewKademlia(base, params)
NewKademlia(base, params),
false,
make(chan error),
}
} }
func (k *testKademlia) newTestKadPeer(s string) Peer { func newTestKadPeer(k *Kademlia, s string) *Peer {
return &testDropPeer{&BzzPeer{BzzAddr: testKadPeerAddr(s)}, k.dropc} return NewPeer(&BzzPeer{BzzAddr: testKadPeerAddr(s)}, k)
} }
func (k *testKademlia) On(ons ...string) *testKademlia { func On(k *Kademlia, ons ...string) {
for _, s := range ons { 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 { 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 { func Register(k *Kademlia, regs ...string) {
var as []OverlayAddr var as []*BzzAddr
for _, s := range regs { for _, s := range regs {
as = append(as, testKadPeerAddr(s)) as = append(as, testKadPeerAddr(s))
} }
err := k.Kademlia.Register(as) err := k.Register(as...)
if err != nil { if err != nil {
panic(err.Error()) 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() addr, o, want := k.SuggestPeer()
if binStr(addr) != expAddr { if binStr(addr) != expAddr {
return fmt.Errorf("incorrect peer address suggested. expected %v, got %v", expAddr, binStr(addr)) 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 return nil
} }
func binStr(a OverlayPeer) string { func binStr(a *BzzAddr) string {
if a == nil { if a == nil {
return "<nil>" return "<nil>"
} }
@ -125,15 +96,17 @@ func binStr(a OverlayPeer) string {
func TestSuggestPeerBug(t *testing.T) { func TestSuggestPeerBug(t *testing.T) {
// 2 row gap, unsaturated proxbin, no callables -> want PO 0 // 2 row gap, unsaturated proxbin, no callables -> want PO 0
k := newTestKademlia("00000000").On( k := newTestKademlia("00000000")
On(k,
"10000000", "11000000", "10000000", "11000000",
"01000000", "01000000",
"00010000", "00011000", "00010000", "00011000",
).Off( )
Off(k,
"01000000", "01000000",
) )
err := testSuggestPeer(t, k, "01000000", 0, false) err := testSuggestPeer(k, "01000000", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
@ -141,140 +114,140 @@ func TestSuggestPeerBug(t *testing.T) {
func TestSuggestPeerFindPeers(t *testing.T) { func TestSuggestPeerFindPeers(t *testing.T) {
// 2 row gap, unsaturated proxbin, no callables -> want PO 0 // 2 row gap, unsaturated proxbin, no callables -> want PO 0
k := newTestKademlia("00000000").On("00100000") k := newTestKademlia("00000000")
err := testSuggestPeer(t, k, "<nil>", 0, false) On(k, "00100000")
err := testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
// 2 row gap, saturated proxbin, no callables -> want PO 0 // 2 row gap, saturated proxbin, no callables -> want PO 0
k.On("00010000") On(k, "00010000")
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
// 1 row gap (1 less), saturated proxbin, no callables -> want PO 1 // 1 row gap (1 less), saturated proxbin, no callables -> want PO 1
k.On("10000000") On(k, "10000000")
err = testSuggestPeer(t, k, "<nil>", 1, false) err = testSuggestPeer(k, "<nil>", 1, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
// no gap (1 less), saturated proxbin, no callables -> do not want more // no gap (1 less), saturated proxbin, no callables -> do not want more
k.On("01000000", "00100001") On(k, "01000000", "00100001")
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
// oversaturated proxbin, > do not want more // oversaturated proxbin, > do not want more
k.On("00100001") On(k, "00100001")
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
// reintroduce gap, disconnected peer callable // reintroduce gap, disconnected peer callable
// log.Info(k.String()) Off(k, "01000000")
k.Off("01000000") err = testSuggestPeer(k, "01000000", 0, false)
err = testSuggestPeer(t, k, "01000000", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
// second time disconnected peer not callable // second time disconnected peer not callable
// with reasonably set Interval // with reasonably set Interval
err = testSuggestPeer(t, k, "<nil>", 1, true) err = testSuggestPeer(k, "<nil>", 1, true)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
// on and off again, peer callable again // on and off again, peer callable again
k.On("01000000") On(k, "01000000")
k.Off("01000000") Off(k, "01000000")
err = testSuggestPeer(t, k, "01000000", 0, false) err = testSuggestPeer(k, "01000000", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.On("01000000") On(k, "01000000")
// new closer peer appears, it is immediately wanted // new closer peer appears, it is immediately wanted
k.Register("00010001") Register(k, "00010001")
err = testSuggestPeer(t, k, "00010001", 0, false) err = testSuggestPeer(k, "00010001", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
// PO1 disconnects // PO1 disconnects
k.On("00010001") On(k, "00010001")
log.Info(k.String()) log.Info(k.String())
k.Off("01000000") Off(k, "01000000")
log.Info(k.String()) log.Info(k.String())
// second time, gap filling // second time, gap filling
err = testSuggestPeer(t, k, "01000000", 0, false) err = testSuggestPeer(k, "01000000", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.On("01000000") On(k, "01000000")
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.MinBinSize = 2 k.MinBinSize = 2
err = testSuggestPeer(t, k, "<nil>", 0, true) err = testSuggestPeer(k, "<nil>", 0, true)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.Register("01000001") Register(k, "01000001")
err = testSuggestPeer(t, k, "01000001", 0, false) err = testSuggestPeer(k, "01000001", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.On("10000001") On(k, "10000001")
log.Trace(fmt.Sprintf("Kad:\n%v", k.String())) 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 { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.On("01000001") On(k, "01000001")
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.MinBinSize = 3 k.MinBinSize = 3
k.Register("10000010") Register(k, "10000010")
err = testSuggestPeer(t, k, "10000010", 0, false) err = testSuggestPeer(k, "10000010", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.On("10000010") On(k, "10000010")
err = testSuggestPeer(t, k, "<nil>", 1, false) err = testSuggestPeer(k, "<nil>", 1, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.On("01000010") On(k, "01000010")
err = testSuggestPeer(t, k, "<nil>", 2, false) err = testSuggestPeer(k, "<nil>", 2, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.On("00100010") On(k, "00100010")
err = testSuggestPeer(t, k, "<nil>", 3, false) err = testSuggestPeer(k, "<nil>", 3, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
k.On("00010010") On(k, "00010010")
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
@ -282,10 +255,8 @@ func TestSuggestPeerFindPeers(t *testing.T) {
} }
func TestSuggestPeerRetries(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 := newTestKademlia("00000000")
k.RetryInterval = int64(100 * time.Millisecond) // cycle k.RetryInterval = int64(300 * time.Millisecond) // cycle
k.MaxRetries = 50 k.MaxRetries = 50
k.RetryExponent = 2 k.RetryExponent = 2
sleep := func(n int) { sleep := func(n int) {
@ -296,53 +267,53 @@ func TestSuggestPeerRetries(t *testing.T) {
time.Sleep(time.Duration(ts)) time.Sleep(time.Duration(ts))
} }
k.Register("01000000") Register(k, "01000000")
k.On("00000001", "00000010") On(k, "00000001", "00000010")
err := testSuggestPeer(t, k, "01000000", 0, false) err := testSuggestPeer(k, "01000000", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
sleep(1) sleep(1)
err = testSuggestPeer(t, k, "01000000", 0, false) err = testSuggestPeer(k, "01000000", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
sleep(1) sleep(1)
err = testSuggestPeer(t, k, "01000000", 0, false) err = testSuggestPeer(k, "01000000", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
sleep(2) sleep(2)
err = testSuggestPeer(t, k, "01000000", 0, false) err = testSuggestPeer(k, "01000000", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
sleep(2) sleep(2)
err = testSuggestPeer(t, k, "<nil>", 0, false) err = testSuggestPeer(k, "<nil>", 0, false)
if err != nil { if err != nil {
t.Fatal(err.Error()) t.Fatal(err.Error())
} }
@ -350,7 +321,9 @@ func TestSuggestPeerRetries(t *testing.T) {
} }
func TestKademliaHiveString(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 k.MaxProxDisplay = 8
h := k.String() 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=========================================================================" 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 continue
} }
p := &BzzAddr{OAddr: a, UAddr: a} p := &BzzAddr{OAddr: a, UAddr: a}
if err := k.Register([]OverlayAddr{p}); err != nil { if err := k.Register(p); err != nil {
t.Fatal(err) t.Fatal(err)
} }
} }
@ -392,12 +365,12 @@ func testKademliaCase(t *testing.T, pivotAddr string, addrs ...string) {
if a == nil { if a == nil {
break break
} }
k.On(&BzzPeer{BzzAddr: a.(*BzzAddr)}) k.On(NewPeer(&BzzPeer{BzzAddr: a}, k))
} }
h := k.Healthy(pp) h := k.Healthy(pp)
if !(h.GotNN && h.KnowNN && h.Full) { if !(h.GotNN && h.KnowNN && h.Full) {
t.Error("not healthy") t.Fatalf("not healthy: %#v\n%v", h, k.String())
} }
} }

2
swarm/network/networkid_test.go
View File

@ -92,7 +92,7 @@ func TestNetworkID(t *testing.T) {
if kademlias[node].addrs.Size() != len(netIDGroup)-1 { 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) 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 found := false
for _, nd := range netIDGroup { for _, nd := range netIDGroup {
p := ToOverlayAddr(nd.Bytes()) p := ToOverlayAddr(nd.Bytes())

47
swarm/network/protocol.go
View File

@ -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 // BzzConfig captures the config params used by the hive
type BzzConfig struct { type BzzConfig struct {
OverlayAddr []byte // base address of the overlay network OverlayAddr []byte // base address of the overlay network
@ -114,7 +88,7 @@ type Bzz struct {
// * bzz config // * bzz config
// * overlay driver // * overlay driver
// * peer store // * 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{ return &Bzz{
Hive: NewHive(config.HiveParams, kad, store), Hive: NewHive(config.HiveParams, kad, store),
NetworkID: config.NetworkID, NetworkID: config.NetworkID,
@ -131,7 +105,7 @@ func (b *Bzz) UpdateLocalAddr(byteaddr []byte) *BzzAddr {
b.localAddr = b.localAddr.Update(&BzzAddr{ b.localAddr = b.localAddr.Update(&BzzAddr{
UAddr: byteaddr, UAddr: byteaddr,
OAddr: b.localAddr.OAddr, OAddr: b.localAddr.OAddr,
}).(*BzzAddr) })
return b.localAddr return b.localAddr
} }
@ -274,7 +248,7 @@ type BzzPeer struct {
LightNode bool LightNode bool
} }
func NewBzzTestPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer { func NewBzzPeer(p *protocols.Peer, addr *BzzAddr) *BzzPeer {
return &BzzPeer{ return &BzzPeer{
Peer: p, Peer: p,
localAddr: addr, 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 // LastActive returns the time the peer was last active
func (p *BzzPeer) LastActive() time.Time { func (p *BzzPeer) LastActive() time.Time {
return p.lastActive return p.lastActive
@ -388,8 +357,8 @@ func (a *BzzAddr) ID() discover.NodeID {
} }
// Update updates the underlay address of a peer record // Update updates the underlay address of a peer record
func (a *BzzAddr) Update(na OverlayAddr) OverlayAddr { func (a *BzzAddr) Update(na *BzzAddr) *BzzAddr {
return &BzzAddr{a.OAddr, na.(Addr).Under()} return &BzzAddr{a.OAddr, na.UAddr}
} }
// String pretty prints the address // String pretty prints the address
@ -410,9 +379,9 @@ func RandomAddr() *BzzAddr {
} }
// NewNodeIDFromAddr transforms the underlay address to an adapters.NodeID // NewNodeIDFromAddr transforms the underlay address to an adapters.NodeID
func NewNodeIDFromAddr(addr Addr) discover.NodeID { func NewNodeIDFromAddr(addr *BzzAddr) discover.NodeID {
log.Info(fmt.Sprintf("uaddr=%s", string(addr.Under()))) log.Info(fmt.Sprintf("uaddr=%s", string(addr.UAddr)))
node := discover.MustParseNode(string(addr.Under())) node := discover.MustParseNode(string(addr.UAddr))
return node.ID return node.ID
} }

4
swarm/network/simulations/discovery/discovery_test.go
View File

@ -556,8 +556,8 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) {
kp.MinProxBinSize = testMinProxBinSize kp.MinProxBinSize = testMinProxBinSize
if ctx.Config.Reachable != nil { if ctx.Config.Reachable != nil {
kp.Reachable = func(o network.OverlayAddr) bool { kp.Reachable = func(o *network.BzzAddr) bool {
return ctx.Config.Reachable(o.(*network.BzzAddr).ID()) return ctx.Config.Reachable(o.ID())
} }
} }
kad := network.NewKademlia(addr.Over(), kp) kad := network.NewKademlia(addr.Over(), kp)

12
swarm/network/stream/delivery.go
View File

@ -47,15 +47,15 @@ var (
type Delivery struct { type Delivery struct {
db *storage.DBAPI db *storage.DBAPI
overlay network.Overlay kad *network.Kademlia
receiveC chan *ChunkDeliveryMsg receiveC chan *ChunkDeliveryMsg
getPeer func(discover.NodeID) *Peer 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{ d := &Delivery{
db: db, db: db,
overlay: overlay, kad: kad,
receiveC: make(chan *ChunkDeliveryMsg, deliveryCap), 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) t := time.NewTimer(10 * time.Minute)
defer t.Stop() 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() start := time.Now()
select { select {
case <-chunk.ReqC: case <-chunk.ReqC:
@ -269,8 +269,8 @@ func (d *Delivery) RequestFromPeers(ctx context.Context, hash []byte, skipCheck
var err error var err error
requestFromPeersCount.Inc(1) requestFromPeersCount.Inc(1)
d.overlay.EachConn(hash, 255, func(p network.OverlayConn, po int, nn bool) bool { d.kad.EachConn(hash, 255, func(p *network.Peer, po int, nn bool) bool {
spId := p.(network.Peer).ID() spId := p.ID()
for _, p := range peersToSkip { for _, p := range peersToSkip {
if p == spId { if p == spId {
log.Trace("Delivery.RequestFromPeers: skip peer", "peer", spId) log.Trace("Delivery.RequestFromPeers: skip peer", "peer", spId)

9
swarm/network/stream/snapshot_sync_test.go
View File

@ -457,15 +457,10 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error {
//returns the number of subscriptions requested //returns the number of subscriptions requested
func startSyncing(r *Registry, conf *synctestConfig) (int, error) { func startSyncing(r *Registry, conf *synctestConfig) (int, error) {
var err error var err error
kad := r.delivery.kad
kad, ok := r.delivery.overlay.(*network.Kademlia)
if !ok {
return 0, fmt.Errorf("Not a Kademlia!")
}
subCnt := 0 subCnt := 0
//iterate over each bin and solicit needed subscription to bins //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 //identify begin and start index of the bin(s) we want to subscribe to
histRange := &Range{} histRange := &Range{}

18
swarm/network/stream/stream.go
View File

@ -130,7 +130,7 @@ func NewRegistry(addr *network.BzzAddr, delivery *Delivery, db *storage.DBAPI, i
// wait for kademlia table to be healthy // wait for kademlia table to be healthy
time.Sleep(options.SyncUpdateDelay) time.Sleep(options.SyncUpdateDelay)
kad := streamer.delivery.overlay.(*network.Kademlia) kad := streamer.delivery.kad
depthC := latestIntC(kad.NeighbourhoodDepthC()) depthC := latestIntC(kad.NeighbourhoodDepthC())
addressBookSizeC := latestIntC(kad.AddrCountC()) 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 // and they are no longer required after iteration, request to Quit
// them will be send to appropriate peers. // them will be send to appropriate peers.
func (r *Registry) updateSyncing() { func (r *Registry) updateSyncing() {
// if overlay in not Kademlia, panic kad := r.delivery.kad
kad := r.delivery.overlay.(*network.Kademlia)
// map of all SYNC streams for all peers // map of all SYNC streams for all peers
// used at the and of the function to remove servers // used at the and of the function to remove servers
// that are not needed anymore // that are not needed anymore
@ -421,8 +419,7 @@ func (r *Registry) updateSyncing() {
r.peersMu.RUnlock() r.peersMu.RUnlock()
// request subscriptions for all nodes and bins // request subscriptions for all nodes and bins
kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(conn network.OverlayConn, bin int) bool { kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(p *network.Peer, bin int) bool {
p := conn.(network.Peer)
log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin)) 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 // 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 { func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error {
peer := protocols.NewPeer(p, rw, Spec) peer := protocols.NewPeer(p, rw, Spec)
bzzPeer := network.NewBzzTestPeer(peer, r.addr) bp := network.NewBzzPeer(peer, r.addr)
r.delivery.overlay.On(bzzPeer) np := network.NewPeer(bp, r.delivery.kad)
defer r.delivery.overlay.Off(bzzPeer) r.delivery.kad.On(np)
return r.Run(bzzPeer) defer r.delivery.kad.Off(np)
return r.Run(bp)
} }
// HandleMsg is the message handler that delegates incoming messages // HandleMsg is the message handler that delegates incoming messages

3
swarm/network_test.go
View File

@ -34,7 +34,6 @@ import (
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/api" "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/network/simulation"
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
colorable "github.com/mattn/go-colorable" colorable "github.com/mattn/go-colorable"
@ -293,7 +292,7 @@ func testSwarmNetwork(t *testing.T, o *testSwarmNetworkOptions, steps ...testSwa
if err != nil { if err != nil {
return nil, cleanup, err 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())) log.Info("new swarm", "bzzKey", config.BzzKey, "baseAddr", fmt.Sprintf("%x", swarm.bzz.BaseAddr()))
return swarm, cleanup, nil return swarm, cleanup, nil
}, },

43
swarm/pss/pss.go
View File

@ -110,10 +110,10 @@ func (params *PssParams) WithPrivateKey(privatekey *ecdsa.PrivateKey) *PssParams
// //
// Implements node.Service // Implements node.Service
type Pss struct { type Pss struct {
network.Overlay // we can get the overlayaddress from this *network.Kademlia // we can get the Kademlia address from this
privateKey *ecdsa.PrivateKey // pss can have it's own independent key privateKey *ecdsa.PrivateKey // pss can have it's own independent key
w *whisper.Whisper // key and encryption backend w *whisper.Whisper // key and encryption backend
auxAPIs []rpc.API // builtins (handshake, test) can add APIs auxAPIs []rpc.API // builtins (handshake, test) can add APIs
// sending and forwarding // sending and forwarding
fwdPool map[string]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer 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. // 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. // 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 { if params.privateKey == nil {
return nil, errors.New("missing private key for pss") return nil, errors.New("missing private key for pss")
} }
@ -162,7 +162,7 @@ func NewPss(k network.Overlay, params *PssParams) (*Pss, error) {
Version: pssVersion, Version: pssVersion,
} }
ps := &Pss{ ps := &Pss{
Overlay: k, Kademlia: k,
privateKey: params.privateKey, privateKey: params.privateKey,
w: whisper.New(&whisper.DefaultConfig), w: whisper.New(&whisper.DefaultConfig),
quitC: make(chan struct{}), quitC: make(chan struct{}),
@ -290,9 +290,9 @@ func (p *Pss) addAPI(api rpc.API) {
p.auxAPIs = append(p.auxAPIs, 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 { func (p *Pss) BaseAddr() []byte {
return p.Overlay.BaseAddr() return p.Kademlia.BaseAddr()
} }
// Returns the pss node's public key // 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() { if int64(pssmsg.Expire) < time.Now().Unix() {
metrics.GetOrRegisterCounter("pss.expire", nil).Inc(1) 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 return nil
} }
if p.checkFwdCache(pssmsg) { 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 return nil
} }
p.addFwdCache(pssmsg) 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 // will return false if using partial address
func (p *Pss) isSelfRecipient(msg *PssMsg) bool { 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 { func (p *Pss) isSelfPossibleRecipient(msg *PssMsg) bool {
local := p.Overlay.BaseAddr() local := p.Kademlia.BaseAddr()
return bytes.Equal(msg.To[:], local[:len(msg.To)]) return bytes.Equal(msg.To[:], local[:len(msg.To)])
} }
@ -816,14 +816,7 @@ func (p *Pss) forward(msg *PssMsg) error {
// send with kademlia // send with kademlia
// find the closest peer to the recipient and attempt to send // find the closest peer to the recipient and attempt to send
sent := 0 sent := 0
p.Overlay.EachConn(to, 256, func(op network.OverlayConn, po int, isproxbin bool) bool { p.Kademlia.EachConn(to, 256, func(sp *network.Peer, 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
}
info := sp.Info() info := sp.Info()
// check if the peer is running pss // 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 // 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() p.fwdPoolMu.RLock()
pp := p.fwdPool[sp.Info().ID] pp := p.fwdPool[sp.Info().ID]
p.fwdPoolMu.RUnlock() 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 is end recipient but the full address has not been disclosed
// - if the peer address matches the partial address fully // - if the peer address matches the partial address fully
// - if the peer is in proxbin // - 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")) log.Trace(fmt.Sprintf("Pss keep forwarding: Partial address + full partial match"))
return true return true
} else if isproxbin { } 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 return true
} }
// at this point we stop forwarding, and the state is as follows: // at this point we stop forwarding, and the state is as follows:

47
swarm/pss/pss_test.go
View File

@ -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 // forwarding should skip peers that do not have matching pss capabilities
func TestMismatch(t *testing.T) { func TestMismatch(t *testing.T) {
@ -582,7 +565,7 @@ func TestMismatch(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
// initialize overlay // initialize kad
baseaddr := network.RandomAddr() baseaddr := network.RandomAddr()
kad := network.NewKademlia((baseaddr).Over(), network.NewKadParams()) kad := network.NewKademlia((baseaddr).Over(), network.NewKadParams())
rw := &p2p.MsgPipeRW{} rw := &p2p.MsgPipeRW{}
@ -594,10 +577,10 @@ func TestMismatch(t *testing.T) {
Version: 0, Version: 0,
} }
nid, _ := discover.HexID("0x01") nid, _ := discover.HexID("0x01")
wrongpsspeer := &pssTestPeer{ wrongpsspeer := network.NewPeer(&network.BzzPeer{
Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(wrongpssaddr.Over()), []p2p.Cap{wrongpsscap}), rw, nil), Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(wrongpssaddr.Over()), []p2p.Cap{wrongpsscap}), rw, nil),
addr: wrongpssaddr.Over(), BzzAddr: &network.BzzAddr{OAddr: wrongpssaddr.Over(), UAddr: nil},
} }, kad)
// one peer doesn't even have pss (boo!) // one peer doesn't even have pss (boo!)
nopssaddr := network.RandomAddr() nopssaddr := network.RandomAddr()
@ -606,16 +589,16 @@ func TestMismatch(t *testing.T) {
Version: 1, Version: 1,
} }
nid, _ = discover.HexID("0x02") nid, _ = discover.HexID("0x02")
nopsspeer := &pssTestPeer{ nopsspeer := network.NewPeer(&network.BzzPeer{
Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(nopssaddr.Over()), []p2p.Cap{nopsscap}), rw, nil), Peer: protocols.NewPeer(p2p.NewPeer(nid, common.ToHex(nopssaddr.Over()), []p2p.Cap{nopsscap}), rw, nil),
addr: nopssaddr.Over(), BzzAddr: &network.BzzAddr{OAddr: nopssaddr.Over(), UAddr: nil},
} }, kad)
// add peers to kademlia and activate them // add peers to kademlia and activate them
// it's safe so don't check errors // it's safe so don't check errors
kad.Register([]network.OverlayAddr{wrongpsspeer}) kad.Register(wrongpsspeer.BzzAddr)
kad.On(wrongpsspeer) kad.On(wrongpsspeer)
kad.Register([]network.OverlayAddr{nopsspeer}) kad.Register(nopsspeer.BzzAddr)
kad.On(nopsspeer) kad.On(nopsspeer)
// create pss // 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 var nid discover.NodeID
copy(nid[:], crypto.FromECDSAPub(&privkey.PublicKey)) copy(nid[:], crypto.FromECDSAPub(&privkey.PublicKey))
addr := network.NewAddrFromNodeID(nid) addr := network.NewAddrFromNodeID(nid)
// set up routing if kademlia is not passed to us // set up routing if kademlia is not passed to us
if overlay == nil { if kad == nil {
kp := network.NewKadParams() kp := network.NewKadParams()
kp.MinProxBinSize = 3 kp.MinProxBinSize = 3
overlay = network.NewKademlia(addr.Over(), kp) kad = network.NewKademlia(addr.Over(), kp)
} }
// create pss // create pss
@ -1654,7 +1637,7 @@ func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *Pss
if ppextra != nil { if ppextra != nil {
pp.SymKeyCacheCapacity = ppextra.SymKeyCacheCapacity pp.SymKeyCacheCapacity = ppextra.SymKeyCacheCapacity
} }
ps, err := NewPss(overlay, pp) ps, err := NewPss(kad, pp)
if err != nil { if err != nil {
return nil return nil
} }

2
swarm/swarm.go
View File

@ -356,7 +356,7 @@ func (self *Swarm) Start(srv *p2p.Server) error {
log.Error("bzz failed", "err", err) log.Error("bzz failed", "err", err)
return 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 { if self.ps != nil {
self.ps.Start(srv) self.ps.Start(srv)