swarm/network: refactor, fix kademlia

pot/address.go

@@ -289,7 +289,7 @@ func (self *BoolAddress) PO(val PotVal, pos int) (po int, eq bool) {
 }
 
 type BytesAddress interface {
-	Bytes() []byte
+	Address() []byte
 }
 
 type bytesAddress struct {
@@ -312,7 +312,7 @@ func ToBytes(v AnyVal) []byte {
 		if !ok {
 			panic(fmt.Sprintf("unsupported value type %T", v))
 		}
-		b = ba.Bytes()
+		b = ba.Address()
 	}
 	return b
 }
@@ -320,7 +320,7 @@ func ToBytes(v AnyVal) []byte {
 func (a *bytesAddress) String() string {
 	return fmt.Sprintf("%08b", a.bytes)
 }
-func (a *bytesAddress) Bytes() []byte {
+func (a *bytesAddress) Address() []byte {
 	return a.bytes
 }
 

swarm/network/discovery.go

@@ -12,12 +12,11 @@ type discPeer struct {
 	*bzzPeer
 	overlay   Overlay
 	peers     map[string]bool
-	proxLimit uint8 // the proximity radius advertised by remote to subscribe to peers
+	depth     uint8 // the proximity radius advertised by remote to subscribe to peers
 	sentPeers bool  // set to true  when the peer is first notifed of peers close to them
 }
 
-// discovery peer contructor
+// NewDiscovery discovery peer contructor
-// registers the handlers for discovery messages
 func NewDiscovery(p *bzzPeer, o Overlay) *discPeer {
 	self := &discPeer{
 		overlay: o,
@@ -47,25 +46,24 @@ func (self *discPeer) HandleMsg(msg interface{}) error {
 
 // NotifyPeer notifies the receiver remote end of a peer p or PO po.
 // callback for overlay driver
-func (self *discPeer) NotifyPeer(p OverlayPeer, po uint8) error {
+func (self *discPeer) NotifyPeer(a OverlayAddr, po uint8) error {
-	log.Warn(fmt.Sprintf("peer %#v peers %v", p, self.peers))
+	if po < self.depth || self.seen(a) {
-	if po < self.proxLimit || self.seen(p) {
 		return nil
 	}
-	log.Warn(fmt.Sprintf("notification about %x", p.Address()))
+	log.Warn(fmt.Sprintf("notification about %x", a.Address()))
 
 	resp := &peersMsg{
-		Peers: []*bzzAddr{ToAddr(p)}, // perhaps the PeerAddr interface is unnecessary generalization
+		Peers: []*bzzAddr{ToAddr(a)}, // perhaps the PeerAddr interface is unnecessary generalization
 	}
 	return self.Send(resp)
 }
 
-// NotifyProx 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 prox limit (radius of the set including the nearest X peers
 // or first empty row)
 // callback for overlay driver
-func (self *discPeer) NotifyProx(po uint8) error {
+func (self *discPeer) NotifyDepth(po uint8) error {
-	return self.Send(&subPeersMsg{ProxLimit: po})
+	return self.Send(&subPeersMsg{Depth: po})
 }
 
 /*
@@ -107,28 +105,28 @@ func (self getPeersMsg) String() string {
 
 // subPeers msg is communicating the depth/sharpness/focus  of the overlay table of a peer
 type subPeersMsg struct {
-	ProxLimit uint8
+	Depth uint8
 }
 
 func (self subPeersMsg) String() string {
-	return fmt.Sprintf("%T: request peers > PO%02d. ", self, self.ProxLimit)
+	return fmt.Sprintf("%T: request peers > PO%02d. ", self, self.Depth)
 }
 
 func (self *discPeer) handleSubPeersMsg(msg *subPeersMsg) error {
-	self.proxLimit = msg.ProxLimit
+	self.depth = msg.Depth
 	if !self.sentPeers {
 		var peers []*bzzAddr
 		self.overlay.EachConn(self.Over(), 255, func(p OverlayConn, po int, isproxbin bool) bool {
-			if uint8(po) < self.proxLimit {
+			if uint8(po) < self.depth {
 				return false
 			}
-			log.Warn(fmt.Sprintf("peer %#v proxlimit %v", p, self.proxLimit))
+			log.Warn(fmt.Sprintf("peer %#v depth %v", p, self.depth))
 			if !self.seen(p) {
 				peers = append(peers, ToAddr(p))
 			}
 			return true
 		})
-		log.Warn(fmt.Sprintf("found initial %v peers not farther than %v", len(peers), self.proxLimit))
+		log.Warn(fmt.Sprintf("found initial %v peers not farther than %v", len(peers), self.depth))
 		if len(peers) > 0 {
 			if err := self.Send(&peersMsg{Peers: peers}); err != nil {
 				return err

swarm/network/discovery_test.go

@@ -33,7 +33,7 @@ func TestDiscovery(t *testing.T) {
 		Expects: []p2ptest.Expect{
 			p2ptest.Expect{
 				Code: 3,
-				Msg:  &subPeersMsg{ProxLimit: 0},
+				Msg:  &subPeersMsg{Depth: 0},
 				Peer: s.ProtocolTester.Ids[0],
 			},
 		},

swarm/network/hive.go

@@ -45,7 +45,7 @@ to keep the nodetable uptodate
 type Overlay interface {
 	Register(chan OverlayAddr) error
 
-	On(OverlayPeer)
+	On(OverlayConn)
 	Off(OverlayConn)
 
 	EachConn([]byte, int, func(OverlayConn, int, bool) bool)
@@ -57,21 +57,6 @@ type Overlay interface {
 	BaseAddr() []byte
 }
 
-// Hive implements the PeerPool interface
-type Hive struct {
-	*HiveParams // settings
-	Overlay     // the overlay topology driver
-	store       Store
-
-	// bookkeeping
-	lock   sync.Mutex
-	quit   chan bool
-	toggle chan bool
-	more   chan bool
-
-	newTicker func() hiveTicker
-}
-
 // HiveParams holds the config options to hive
 type HiveParams struct {
 	Discovery             bool  // if want discovery of not
@@ -90,6 +75,21 @@ func NewHiveParams() *HiveParams {
 	}
 }
 
+// Hive implements the PeerPool interface
+type Hive struct {
+	*HiveParams // settings
+	Overlay     // the overlay topology driver
+	store       Store
+
+	// bookkeeping
+	lock   sync.Mutex
+	quit   chan bool
+	toggle chan bool
+	more   chan bool
+
+	newTicker func() hiveTicker
+}
+
 // Hive constructor embeds both arguments
 // HiveParams: config parameters
 // Overlay: Topology Driver Interface
@@ -167,36 +167,27 @@ func (self *Hive) Stop() {
 	close(self.quit)
 }
 
-func (self *Hive) Run(peer *bzzPeer) error {
+func (self *Hive) Run(p *bzzPeer) error {
-	discPeer := NewDiscovery(peer, self)
-	self.On(discPeer)
-	defer self.Off(discPeer)
-	return peer.Run(discPeer.HandleMsg)
-}
-
-// Add is called at the end of a successful protocol handshake
-// to register a connected (live) peer
-func (self *Hive) Add(p *bzzPeer) error {
-	defer self.wake()
 	dp := NewDiscovery(p, self.Overlay)
 	log.Debug(fmt.Sprintf("to add new bee %v", p))
 	self.On(dp)
-	self.String()
+	self.wake()
-	log.Debug(fmt.Sprintf("%v", self))
+	defer self.wake()
-	return nil
+	defer self.Off(dp)
+	return p.Run(dp.HandleMsg)
 }
 
 // Remove called after peer is disconnected
-func (self *Hive) Remove(p *bzzPeer) {
+// func (self *Hive) Remove(p *bzzPeer) {
-	defer self.wake()
+// 	defer self.wake()
-	log.Debug(fmt.Sprintf("remove bee %v", p))
+// 	log.Debug(fmt.Sprintf("remove bee %v", p))
-	self.Off(p)
+// 	self.Off(p)
-}
+// }
 
 // NodeInfo function is used by the p2p.server RPC interface to display
 // protocol specific node information
 func (self *Hive) NodeInfo() interface{} {
-	return interface{}(self.String())
+	return self.String()
 }
 
 // PeerInfo function is used by the p2p.server RPC interface to display

swarm/network/hive_test.go

@@ -67,11 +67,6 @@ func TestRegisterAndConnect(t *testing.T) {
 	pp.Start(s.Server)
 	defer pp.Stop()
 	tc.ticker <- time.Now()
-
-	// if pp.Overlay.(*testOverlay).posMap[string(raddr.Over())] == nil {
-	// 	t.Fatalf("Overlay#On not called on new peer")
-	// }
-
 	// retrieve and broadcast
 	ord := raddr.Over()[0] / 32
 	o := 0

swarm/network/kademlia.go

@@ -98,7 +98,7 @@ func NewKademlia(addr []byte, params *KadParams) *Kademlia {
 }
 
 type Notifier interface {
-	NotifyPeer(OverlayConn, uint8) error
+	NotifyPeer(OverlayAddr, uint8) error
 	NotifyDepth(uint8) error
 }
 
@@ -117,7 +117,6 @@ type OverlayConn interface {
 
 type OverlayAddr interface {
 	OverlayPeer
-	On(OverlayConn) OverlayConn     // call to return the connected peer
 	Update(OverlayAddr) OverlayAddr // returns the updated version of the original
 }
 
@@ -140,18 +139,24 @@ func newEntry(p OverlayPeer) *entry {
 	}
 }
 
+func (self *entry) addr() OverlayAddr {
+	a, _ := self.OverlayPeer.(OverlayAddr)
+	return a
+}
+
+func (self *entry) conn() OverlayConn {
+	c, _ := self.OverlayPeer.(OverlayConn)
+	return c
+}
+
 func (self *entry) String() string {
-	return fmt.Sprintf("%x", self.Address())
+	return fmt.Sprintf("%x", self.OverlayPeer.Address())
 }
 
 // Register enters each OverlayAddr as kademlia peer record into the
 // database of known peer addresses
 func (self *Kademlia) Register(peers chan OverlayAddr) error {
-	if len(peers) == 0 {
-		return fmt.Errorf("empty peers list")
-	}
 	np := pot.NewPot(nil, 0)
-	defer func() { self.addrs.Merge(np) }()
 	for p := range peers {
 		// error if self received, peer should know better
 		if bytes.Equal(p.Address(), self.base) {
@@ -159,6 +164,9 @@ func (self *Kademlia) Register(peers chan OverlayAddr) error {
 		}
 		np, _, _ = pot.Add(np, pot.PotVal(newEntry(p)))
 	}
+	com := self.addrs.Merge(np)
+	log.Trace(fmt.Sprintf("merged %v peers, %v known", np.Size(), com))
+
 	// TODO: remove this check
 	m := make(map[string]bool)
 	self.addrs.Each(func(val pot.PotVal, i int) bool {
@@ -186,14 +194,15 @@ func (self *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
 	ba := pot.NewBytesVal(self.base, nil)
 	self.addrs.EachNeighbour(ba, func(val pot.PotVal, po int) bool {
 		a = self.callable(val)
+		log.Trace(fmt.Sprintf("candidate prox peer at %x: %v (%v). a == nil is %v", val.(*entry).Address(), a, po, a == nil))
 		ppo = po
-		return a != nil && po >= depth
+		return a == nil && po >= depth
 	})
 	if a != nil {
 		log.Trace(fmt.Sprintf("candidate prox peer found: %v (%v)", a, ppo))
 		return a, 0, false
 	}
-	log.Trace(fmt.Sprintf("no candidate prox peers to connect to (Depth: %v, minProxSize: %v)", depth, self.MinProxBinSize))
+	log.Trace(fmt.Sprintf("no candidate prox peers to connect to (Depth: %v, minProxSize: %v) %#v", depth, self.MinProxBinSize, a))
 
 	var bpo []int
 	prev := -1
@@ -225,7 +234,7 @@ func (self *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
 			log.Trace(fmt.Sprintf("check PO%02d: ", po))
 			f(func(val pot.PotVal, j int) bool {
 				a = self.callable(val)
-				return a != nil && po < depth
+				return a == nil && po < depth
 			})
 			return false
 		})
@@ -241,7 +250,7 @@ func (self *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
 }
 
 // On inserts the peer as a kademlia peer into the live peers
-func (self *Kademlia) On(p OverlayPeer) {
+func (self *Kademlia) On(p OverlayConn) {
 	e := newEntry(p)
 	self.conns.Swap(p, func(v pot.PotVal) pot.PotVal {
 		// if not found live
@@ -257,10 +266,12 @@ func (self *Kademlia) On(p OverlayPeer) {
 		return v
 	})
 
+	log.Trace(fmt.Sprintf("Notifier:%#v", p))
 	np, ok := p.(Notifier)
 	if !ok {
 		return
 	}
+	log.Trace(fmt.Sprintf("notify:%v", p))
 
 	depth := uint8(self.Depth())
 	if depth != self.depth {
@@ -271,12 +282,12 @@ func (self *Kademlia) On(p OverlayPeer) {
 
 	go np.NotifyDepth(depth)
 	f := func(val pot.PotVal, po int) {
-		dp := val.(Notifier)
+		dp := val.(*entry).OverlayPeer.(Notifier)
-		dp.NotifyPeer(p.(OverlayConn), uint8(po))
+		dp.NotifyPeer(p.Off(), uint8(po))
 		log.Trace(fmt.Sprintf("peer %v notified of %v (%v)", dp, p, po))
 		if depth > 0 {
 			dp.NotifyDepth(depth)
-			log.Trace("peer %v notified of new depth limit %v", dp, depth)
+			log.Trace("peer %v notified of new depth %v", dp, depth)
 		}
 	}
 	self.conns.EachNeighbourAsync(e, 1024, 255, f, false)
@@ -290,10 +301,10 @@ func (self *Kademlia) Off(p OverlayConn) {
 			panic(fmt.Sprintf("connected peer not found %v", p))
 		}
 		self.conns.Swap(p, func(v pot.PotVal) pot.PotVal {
-			// v cannot nil, but no need to check
+			// v cannot be nil, but no need to check
 			return nil
 		})
-		return newEntry(p)
+		return newEntry(p.Off())
 	})
 }
 
@@ -313,7 +324,7 @@ func (self *Kademlia) EachConn(base []byte, o int, f func(OverlayConn, int, bool
 		if l, _ := p.PO(val, 0); l >= self.Depth() {
 			isproxbin = true
 		}
-		return f(val.(OverlayConn), po, isproxbin)
+		return f(val.(*entry).conn(), po, isproxbin)
 	})
 }
 
@@ -329,7 +340,7 @@ func (self *Kademlia) EachAddr(base []byte, o int, f func(OverlayAddr, int) bool
 		if po > o {
 			return true
 		}
-		return f(val.(OverlayAddr), po)
+		return f(val.(*entry).addr(), po)
 	})
 }
 
@@ -353,7 +364,7 @@ func (self *Kademlia) Depth() (depth int) {
 func (self *Kademlia) callable(val pot.PotVal) OverlayAddr {
 	e := val.(*entry)
 	// not callable if peer is live or exceeded maxRetries
-	if _, live := val.(OverlayConn); live || e.retries > self.MaxRetries {
+	if e.conn() != nil || e.retries > self.MaxRetries {
 		log.Trace(fmt.Sprintf("peer %v (%T) not callable", e, e.OverlayPeer))
 		return nil
 	}
@@ -368,26 +379,27 @@ func (self *Kademlia) callable(val pot.PotVal) OverlayAddr {
 	// this is never called concurrently, so safe to increment
 	// peer can be retried again
 	if retries < e.retries {
-		log.Trace(fmt.Sprintf("log time needed before retry %v, wait only warrants %v", e.retries, retries))
+		log.Trace(fmt.Sprintf("long time since last try (at %v) needed before retry %v, wait only warrants %v", timeAgo, e.retries, retries))
 		return nil
 	}
 	e.retries++
 	log.Trace(fmt.Sprintf("peer %v is callable", e))
 
-	return val.(OverlayAddr)
+	return e.addr()
 }
 
+// BaseAddr return the kademlia base addres
 func (self *Kademlia) BaseAddr() []byte {
 	return self.base
 }
 
-// kademlia table + kaddb table displayed with ascii
+// String returns kademlia table + kaddb table displayed with ascii
 func (self *Kademlia) String() string {
 
 	var rows []string
 
 	rows = append(rows, "=========================================================================")
-	rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), self.BaseAddr()))
+	rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %x", time.Now().UTC().Format(time.UnixDate), self.BaseAddr()[:3]))
 	rows = append(rows, fmt.Sprintf("population: %d (%d), MinProxBinSize: %d, MinBinSize: %d, MaxBinSize: %d", self.conns.Size(), self.addrs.Size(), self.MinProxBinSize, self.MinBinSize, self.MaxBinSize))
 
 	liverows := make([]string, self.MaxProxDisplay)
@@ -468,12 +480,12 @@ func (self *Kademlia) Prune(c <-chan time.Time) {
 	go func() {
 		for range c {
 			total := 0
-			self.conns.EachBin(nil, 0, func(po, size int, f func(func(pot.PotVal, int) bool) bool) bool {
+			self.conns.EachBin(pot.NewBytesVal(self.base, nil), 0, func(po, size int, f func(func(pot.PotVal, int) bool) bool) bool {
 				extra := size - self.MinBinSize
 				if size > self.MaxBinSize {
 					n := 0
 					f(func(v pot.PotVal, po int) bool {
-						v.(OverlayConn).Drop(fmt.Errorf("bucket full"))
+						v.(*entry).conn().Drop(fmt.Errorf("bucket full"))
 						n++
 						return n < extra
 					})

swarm/network/kademlia_test.go

@@ -17,13 +17,21 @@ package network
 
 import (
 	"fmt"
+	"os"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/pot"
 )
 
+func init() {
+	h := log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
+	// h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
+	log.Root().SetHandler(h)
+}
+
 func testKadPeerAddr(s string) *bzzAddr {
 	a := pot.NewHashAddress(s).Bytes()
 	return &bzzAddr{OAddr: a, UAddr: a}
@@ -57,23 +65,24 @@ type dropError struct {
 }
 
 func (self *testDropPeer) Drop(err error) {
-	err2 := &dropError{err, overlayStr(self)}
+	err2 := &dropError{err, binStr(self)}
 	self.dropc <- err2
 }
 
-func (self *testDiscPeer) NotifyProx(po uint8) error {
+func (self *testDiscPeer) NotifyDepth(po uint8) error {
-	key := overlayStr(self)
+	key := binStr(self)
 	self.lock.Lock()
 	defer self.lock.Unlock()
 	self.notifications[key] = po
 	return nil
 }
 
-func (self *testDiscPeer) NotifyPeer(p OverlayPeer, po uint8) error {
+func (self *testDiscPeer) NotifyPeer(p OverlayAddr, po uint8) error {
-	key := overlayStr(self)
+	key := binStr(self)
-	key += overlayStr(p)
+	key += binStr(p)
 	self.lock.Lock()
 	defer self.lock.Unlock()
+	log.Trace(fmt.Sprintf("key %v=>%v", key, po))
 	self.notifications[key] = po
 	return nil
 }
@@ -108,34 +117,6 @@ func (k *testKademlia) newTestKadPeer(s string) Peer {
 	return Peer(dp)
 }
 
-func overlayStr(a OverlayPeer) string {
-	// log.Error(fmt.Sprintf("PeerAddr: %v (%T)", a, a))
-	// if a == (*KadPeer)(nil) || a == (*testDiscPeer)(nil) || a == (*bzzPeer)(nil) || a == nil {
-	// 	return "<nil>"
-	// }
-	// var p Peer
-	// s, ok := a.(*KadPeer)
-	// if ok {
-	// 	p = s.Peer
-	// } else {
-	// 	p = a.(*testDiscPeer).Peer
-	// }
-	// log.Error(fmt.Sprintf("PeerAddr: %v (%T)", p, p))
-	// if p == (Peer)(nil) || p == (*testDiscPeer)(nil) || p == (*bzzPeer)(nil) {
-	// 	return "<nil>"
-	// }
-	// return pot.NewHashAddressFromBytes(p.OverlayAddr()).Bin()[:6]
-	// if a == nil {
-	// 	return "<nil>"
-	// }
-	// k, ok := a.(*KadPeer)
-	// if ok && k.Peer != nil {
-	// 	return pot.ToBin(a.(*KadPeer).Peer.Over())[:6]
-	// }
-	// return pot.ToBin(a.Over())[:6]
-	return pot.ToBin(a.Address())
-}
-
 func (k *testKademlia) On(ons ...string) *testKademlia {
 	for _, s := range ons {
 		p := k.newTestKadPeer(s)
@@ -160,14 +141,16 @@ func (k *testKademlia) Register(regs ...string) *testKademlia {
 			ch <- testKadPeerAddr(s)
 		}
 	}()
-	k.Kademlia.Register(ch)
+	err := k.Kademlia.Register(ch)
+	log.Trace(fmt.Sprintf("register %v addresses: %v", len(regs), err))
+
 	return k
 }
 
 func testSuggestPeer(t *testing.T, k *testKademlia, expAddr string, expPo int, expWant bool) error {
 	addr, o, want := k.SuggestPeer()
-	if overlayStr(addr) != expAddr {
+	if binStr(addr) != expAddr {
-		return fmt.Errorf("incorrect peer address suggested. expected %v, got %v", expAddr, overlayStr(addr))
+		return fmt.Errorf("incorrect peer address suggested. expected %v, got %v", expAddr, binStr(addr))
 	}
 	if o != expPo {
 		return fmt.Errorf("incorrect prox order suggested. expected %v, got %v", expPo, o)
@@ -178,6 +161,13 @@ func testSuggestPeer(t *testing.T, k *testKademlia, expAddr string, expPo int, e
 	return nil
 }
 
+func binStr(a OverlayPeer) string {
+	if a == nil {
+		return "<nil>"
+	}
+	return pot.ToBin(a.Address())[:6]
+}
+
 func TestSuggestPeerFindPeers(t *testing.T) {
 	// 2 row gap, unsaturated proxbin, no callables -> want PO 0
 	k := newTestKademlia("000000").On("001000")
@@ -225,7 +215,6 @@ func TestSuggestPeerFindPeers(t *testing.T) {
 
 	// second time disconnected peer not callable
 	// with reasonably set Interval
-	// err = testSuggestPeer(t, k, "010000", 2, true)
 	err = testSuggestPeer(t, k, "<nil>", 1, true)
 	if err != nil {
 		t.Fatal(err.Error())
@@ -240,16 +229,16 @@ func TestSuggestPeerFindPeers(t *testing.T) {
 	}
 
 	k.On("010000")
-	k.Off("010000")
-	// PO1 disconnects
 	// new closer peer appears, it is immediately wanted
-	// k.Off("010000")
 	k.Register("000101")
 	err = testSuggestPeer(t, k, "000101", 0, false)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
 
+	// PO1 disconnects
+	k.On("000101")
+	k.Off("010000")
 	// second time, gap filling
 	err = testSuggestPeer(t, k, "010000", 0, false)
 	if err != nil {
@@ -268,6 +257,19 @@ func TestSuggestPeerFindPeers(t *testing.T) {
 		t.Fatal(err.Error())
 	}
 
+	k.Register("010001")
+	err = testSuggestPeer(t, k, "<nil>", 0, true)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
+	k.On("100001")
+	log.Trace("Kad:\n%v", k.String())
+	err = testSuggestPeer(t, k, "010001", 0, false)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
 	k.On("100001")
 	k.On("010001")
 	err = testSuggestPeer(t, k, "<nil>", 0, false)
@@ -276,7 +278,18 @@ func TestSuggestPeerFindPeers(t *testing.T) {
 	}
 
 	k.MinBinSize = 3
+	k.Register("100010")
+	err = testSuggestPeer(t, k, "100010", 0, false)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
 	k.On("100010")
+	err = testSuggestPeer(t, k, "<nil>", 1, true)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+
 	k.On("010010")
 	err = testSuggestPeer(t, k, "<nil>", 2, true)
 	if err != nil {
@@ -284,8 +297,16 @@ func TestSuggestPeerFindPeers(t *testing.T) {
 	}
 
 	k.On("001010")
+	err = testSuggestPeer(t, k, "<nil>", 3, true)
+	if err != nil {
+		log.Trace("Kad:\n%v", k.String())
+		t.Fatal(err.Error())
+	}
+
+	k.On("000110")
 	err = testSuggestPeer(t, k, "<nil>", 0, false)
 	if err != nil {
+		log.Trace("Kad:\n%v", k.String())
 		t.Fatal(err.Error())
 	}
 
@@ -443,7 +464,7 @@ func TestNotifications(t *testing.T) {
 	k.Discovery = true
 	k.MinProxBinSize = 3
 	k.On("010000", "001000")
-	time.Sleep(100 * time.Millisecond)
+	time.Sleep(1000 * time.Millisecond)
 	err := k.checkNotifications(
 		[]*testPeerNotification{
 			&testPeerNotification{"010000", "001000", 1},

swarm/network/protocol.go

@@ -88,6 +88,7 @@ type Conn interface {
 	Send(interface{}) error                                      // can send messages
 	Drop(error)                                                  // disconnect this peer
 	Run(func(interface{}) error) error                           // the run function to run a protocol
+	Off() OverlayAddr
 }
 
 // TODO: implement store for exec nodes
@@ -107,6 +108,18 @@ type BzzConfig struct {
 	Store Store
 }
 
+// Bzz is the swarm protocol bundle
+type Bzz struct {
+	Kademlia *Kademlia
+	Hive     *Hive
+	Pss      *Pss
+
+	localAddr  *bzzAddr
+	mtx        sync.Mutex
+	handshakes map[discover.NodeID]*bzzHandshake
+}
+
+// NewBzz is the swarm protocol constructor
 func NewBzz(config *BzzConfig) *Bzz {
 	kademlia := NewKademlia(config.OverlayAddr, config.KadParams)
 	bzz := &Bzz{
@@ -121,16 +134,10 @@ func NewBzz(config *BzzConfig) *Bzz {
 	return bzz
 }
 
-type Bzz struct {
+// Bzz implements the node.Service interface, offers Protocols
-	Kademlia *Kademlia
+// * handshake/hive
-	Hive     *Hive
+// * discovery
-	Pss      *Pss
+// * pss
-
-	localAddr  *bzzAddr
-	mtx        sync.Mutex
-	handshakes map[discover.NodeID]*bzzHandshake
-}
-
 func (b *Bzz) Protocols() []p2p.Protocol {
 	return []p2p.Protocol{
 		{
@@ -156,6 +163,9 @@ func (b *Bzz) Protocols() []p2p.Protocol {
 	}
 }
 
+// Bzz implements the node.Service interface, offers APIs:
+// * hive
+// * pss
 func (b *Bzz) APIs() []rpc.API {
 	return []rpc.API{{
 		Namespace: "hive",
@@ -200,7 +210,7 @@ func (b *Bzz) runProtocol(spec *protocols.Spec, run func(*bzzPeer) error) func(*
 
 		// the handshake has succeeded so run the service
 		peer := &bzzPeer{
-			Conn:      protocols.NewPeer(p, rw, spec),
+			Peer:      protocols.NewPeer(p, rw, spec),
 			localAddr: b.localAddr,
 			bzzAddr:   handshake.peerAddr,
 		}
@@ -227,15 +237,15 @@ func (b *Bzz) getHandshake(peerID discover.NodeID) *bzzHandshake {
 // bzzPeer is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer)
 // implements the Peer interface and all interfaces Peer implements: Addr, OverlayPeer
 type bzzPeer struct {
-	Conn                 // represents the connection for online peers
+	*protocols.Peer           // represents the connection for online peers
-	localAddr  *bzzAddr  // local Peers address
+	localAddr       *bzzAddr  // local Peers address
-	*bzzAddr             // remote address -> implements Addr interface = protocols.Peer
+	*bzzAddr                  // remote address -> implements Addr interface = protocols.Peer
-	lastActive time.Time // time is updated whenever mutexes are releasing
+	lastActive      time.Time // time is updated whenever mutexes are releasing
 }
 
-func newBzzPeer(conn Conn, over, under []byte) *bzzPeer {
+func newBzzPeer(p *protocols.Peer, over, under []byte) *bzzPeer {
 	return &bzzPeer{
-		Conn:      conn,
+		Peer:      p,
 		localAddr: &bzzAddr{over, under},
 	}
 }
@@ -318,23 +328,16 @@ func (self *bzzAddr) Address() []byte {
 	return self.OAddr
 }
 
-func (self *bzzAddr) Bytes() []byte {
+// Over returns the overlay address
-	return self.OAddr
-}
 func (self *bzzAddr) Over() []byte {
 	return self.OAddr
 }
 
+// Under retrun the underlay address
 func (self *bzzAddr) Under() []byte {
 	return self.UAddr
 }
 
-func (self *bzzAddr) On(p OverlayConn) OverlayConn {
-	bp := p.(*bzzPeer)
-	bp.bzzAddr = self
-	return bp
-}
-
 func (self *bzzAddr) Update(a OverlayAddr) OverlayAddr {
 	return &bzzAddr{self.OAddr, a.(Addr).Under()}
 }

swarm/network/protocol_test.go

@@ -5,7 +5,6 @@ import (
 	"sync"
 	"testing"
 
-	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
@@ -73,7 +72,7 @@ func newBzzBaseTester(t *testing.T, n int, addr *bzzAddr, spec *protocols.Spec,
 
 	protocall := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 		return srv(&bzzPeer{
-			Conn:      protocols.NewPeer(p, rw, spec),
+			Peer:      protocols.NewPeer(p, rw, spec),
 			localAddr: addr,
 			bzzAddr:   NewAddrFromNodeId(&adapters.NodeId{NodeID: p.ID()}),
 		})
@@ -103,6 +102,9 @@ func newBzzTester(t *testing.T, n int, addr *bzzAddr, pp *p2ptest.TestPeerPool,
 	extraservices := func(p *bzzPeer) error {
 		pp.Add(p)
 		defer pp.Remove(p)
+		if services == nil {
+			return nil
+		}
 		return services(p)
 	}
 	return newBzzBaseTester(t, n, addr, spec, extraservices)
@@ -183,67 +185,3 @@ func TestBzzHandshakeSuccess(t *testing.T) {
 		&bzzHandshake{Version: 0, NetworkId: 322, Addr: NewAddrFromNodeId(id)},
 	)
 }
-
-func TestBzzPeerPoolAdd(t *testing.T) {
-	pp := p2ptest.NewTestPeerPool()
-	addr := RandomAddr()
-	s := newBzzTester(t, 1, addr, pp, nil, nil)
-	defer s.Stop()
-
-	id := s.Ids[0]
-	log.Trace(fmt.Sprintf("handshake with %v", id))
-	s.runHandshakes()
-
-	if !pp.Has(id) {
-		t.Fatalf("peer '%v' not added: %v", id, pp)
-	}
-}
-
-func TestBzzPeerPoolRemove(t *testing.T) {
-	addr := RandomAddr()
-	pp := p2ptest.NewTestPeerPool()
-	s := newBzzTester(t, 1, addr, pp, nil, nil)
-	defer s.Stop()
-
-	s.runHandshakes()
-
-	id := s.Ids[0]
-	pp.Get(id).Drop(fmt.Errorf("p2p: read or write on closed message pipe"))
-	s.TestDisconnected(&p2ptest.Disconnect{id, fmt.Errorf("p2p: read or write on closed message pipe")})
-	if pp.Has(id) {
-		t.Fatalf("peer '%v' not removed: %v", id, pp)
-	}
-}
-
-func TestBzzPeerPoolBothAddRemove(t *testing.T) {
-	addr := RandomAddr()
-	pp := p2ptest.NewTestPeerPool()
-	s := newBzzTester(t, 1, addr, pp, nil, nil)
-	defer s.Stop()
-
-	s.runHandshakes()
-
-	id := s.Ids[0]
-	if !pp.Has(id) {
-		t.Fatalf("peer '%v' not added: %v", id, pp)
-	}
-
-	pp.Get(id).Drop(fmt.Errorf("p2p: read or write on closed message pipe"))
-	s.TestDisconnected(&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("p2p: read or write on closed message pipe")})
-	if pp.Has(id) {
-		t.Fatalf("peer '%v' not removed: %v", id, pp)
-	}
-}
-
-func TestBzzPeerPoolNotAdd(t *testing.T) {
-	addr := RandomAddr()
-	pp := p2ptest.NewTestPeerPool()
-	s := newBzzTester(t, 1, addr, pp, nil, nil)
-	defer s.Stop()
-
-	id := s.Ids[0]
-	s.testHandshake(correctBzzHandshake(addr), &bzzHandshake{Version: 0, NetworkId: 321, Addr: NewAddrFromNodeId(id)}, &p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")})
-	if pp.Has(id) {
-		t.Fatalf("peer %v incorrectly added: %v", id, pp)
-	}
-}

swarm/network/pss_test.go

@@ -12,7 +12,9 @@ const (
 )
 
 func init() {
-	h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
+	h := log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
+	//
+	// h := log.CallerFileHandler(log.StreamHandler(os.Stderr, log.TerminalFormat(true)))
 	log.Root().SetHandler(h)
 }