swarm/network: Kademlia fixes

swarm/network/discovery.go

@@ -120,9 +120,8 @@ func (self *discPeer) handleSubPeersMsg(msg *subPeersMsg) error {
 			if uint8(po) < self.depth {
 				return false
 			}
-			log.Warn(fmt.Sprintf("peer %#v depth %v", p, self.depth))
 			if !self.seen(p) {
-				peers = append(peers, ToAddr(p))
+				peers = append(peers, ToAddr(p.Off()))
 			}
 			return true
 		})
@@ -147,13 +146,15 @@ func (self *discPeer) handlePeersMsg(msg *peersMsg) error {
 		return nil
 	}
 
-	var c chan OverlayAddr
+	c := make(chan OverlayAddr)
 	go func() {
+		defer close(c)
 		for _, a := range msg.Peers {
 			self.seen(a)
 			c <- a
 		}
 	}()
+	log.Info("discovery overlay register")
 	return self.overlay.Register(c)
 }
 
@@ -168,7 +169,7 @@ func (self *discPeer) handleGetPeersMsg(msg *getPeersMsg) error {
 	self.overlay.EachConn(self.Over(), int(msg.Order), func(p OverlayConn, po int, isproxbin bool) bool {
 		i++
 		// only send peers we have not sent before in this session
-		a := ToAddr(p)
+		a := ToAddr(p.Off())
 		if self.seen(a) {
 			peers = append(peers, a)
 		}

swarm/network/hive.go

@@ -126,7 +126,8 @@ func (self *Hive) Start(server *p2p.Server) error {
 				// to attempt to write to more (remove Peer when shutting down)
 				return
 			}
-			log.Trace("hive delegate to overlay driver: suggest addr to connect to")
+			log.Debug("hive delegate to overlay driver: suggest addr to connect to")
+			// log.Trace("hive delegate to overlay driver: suggest addr to connect to")
 			addr, order, want := self.SuggestPeer()
 
 			if addr != nil {
@@ -143,16 +144,17 @@ func (self *Hive) Start(server *p2p.Server) error {
 
 			want = want && self.Discovery
 			if want {
+				log.Debug(fmt.Sprintf("========> request peers nearest %v", addr))
 				RequestOrder(self.Overlay, uint8(order), self.PeersBroadcastSetSize, self.MaxPeersPerRequest)
 			}
 
+			log.Info(fmt.Sprintf("%v", self))
 			select {
 			case self.toggle <- want:
 				log.Trace(fmt.Sprintf("keep hive alive: %v", want))
 			case <-self.quit:
 				return
 			}
-			// log.Info(fmt.Sprintf("%v", self))
 		}
 	}()
 	return nil
@@ -177,13 +179,6 @@ func (self *Hive) Run(p *bzzPeer) error {
 	return p.Run(dp.HandleMsg)
 }
 
-// Remove called after peer is disconnected
-// func (self *Hive) Remove(p *bzzPeer) {
-// 	defer self.wake()
-// 	log.Debug(fmt.Sprintf("remove bee %v", 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{} {
@@ -229,6 +224,9 @@ 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
 }
 
@@ -262,7 +260,7 @@ func (self *Hive) keepAlive() {
 	for {
 		select {
 		case <-tick:
-			log.Trace("wake up: make hive alive")
+			log.Debug("wake up: make hive alive")
 			self.wake()
 		case need := <-self.toggle:
 			if ticker == nil && need {
@@ -295,9 +293,9 @@ func (self *Hive) loadPeers() error {
 		return err
 	}
 
-	var c chan OverlayAddr
-	defer close(c)
+	c := make(chan OverlayAddr)
 	go func() {
+		defer close(c)
 		for _, a := range as {
 			c <- a
 		}

swarm/network/kademlia.go

@@ -162,10 +162,11 @@ func (self *Kademlia) Register(peers chan OverlayAddr) error {
 		if bytes.Equal(p.Address(), self.base) {
 			return fmt.Errorf("add peers: %x is self", self.base)
 		}
-		np, _, _ = pot.Add(np, pot.PotVal(newEntry(p)))
+		np, _, _ = pot.Add(np, newEntry(p))
 	}
 	com := self.addrs.Merge(np)
-	log.Trace(fmt.Sprintf("merged %v peers, %v known", np.Size(), com))
+	log.Debug(fmt.Sprintf("merged %v peers, %v known, total: %v", np.Size(), com, self.addrs.Size()))
+	// log.Trace(fmt.Sprintf("merged %v peers, %v known", np.Size(), com))
 
 	// TODO: remove this check
 	m := make(map[string]bool)
@@ -189,7 +190,8 @@ func (self *Kademlia) SuggestPeer() (a OverlayAddr, o int, want bool) {
 	depth := self.Depth()
 	// if there is a callable neighbour within the current proxBin, connect
 	// this makes sure nearest neighbour set is fully connected
-	log.Trace(fmt.Sprintf("candidate prox peer checking above PO %v", depth))
+	log.Debug(fmt.Sprintf("candidate prox peer checking above PO %v", depth))
+	// log.Trace(fmt.Sprintf("candidate prox peer checking above PO %v", depth))
 	var ppo int
 	ba := pot.NewBytesVal(self.base, nil)
 	self.addrs.EachNeighbour(ba, func(val pot.PotVal, po int) bool {
@@ -266,12 +268,10 @@ func (self *Kademlia) On(p OverlayConn) {
 		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 {
@@ -284,10 +284,12 @@ func (self *Kademlia) On(p OverlayConn) {
 	f := func(val pot.PotVal, po int) {
 		dp := val.(*entry).OverlayPeer.(Notifier)
 		dp.NotifyPeer(p.Off(), uint8(po))
-		log.Trace(fmt.Sprintf("peer %v notified of %v (%v)", dp, p, po))
+		// log.Trace(fmt.Sprintf("peer %v notified of %v (%v)", dp, p, po))
+		log.Debug(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 %v", dp, depth)
+			log.Debug(fmt.Sprintf("peer %v notified of new depth %v", dp, depth))
+			// log.Trace(fmt.Sprintf("peer %v notified of new depth %v", dp, depth))
 		}
 	}
 	self.conns.EachNeighbourAsync(e, 1024, 255, f, false)

swarm/network/protocol.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net"
 	"sync"
 	"time"
 
@@ -185,6 +186,7 @@ func (b *Bzz) Stop() error {
 
 func (b *Bzz) runHandshake(p *p2p.Peer, rw p2p.MsgReadWriter) error {
 	handshake := b.getHandshake(p.ID())
+	defer b.removeHandshake(p.ID())
 
 	if err := handshake.Perform(p, rw); err != nil {
 		log.Error("handshake failed", "peer", p.ID(), "err", err)
@@ -218,6 +220,12 @@ func (b *Bzz) runProtocol(spec *protocols.Spec, run func(*bzzPeer) error) func(*
 	}
 }
 
+func (b *Bzz) removeHandshake(peerID discover.NodeID) {
+	b.mtx.Lock()
+	defer b.mtx.Unlock()
+	delete(b.handshakes, peerID)
+}
+
 func (b *Bzz) getHandshake(peerID discover.NodeID) *bzzHandshake {
 	b.mtx.Lock()
 	defer b.mtx.Unlock()
@@ -370,5 +378,8 @@ func NewNodeIdFromAddr(addr Addr) *adapters.NodeId {
 // the overlay address is derived as the hash of the nodeId
 func NewAddrFromNodeId(n *adapters.NodeId) *bzzAddr {
 	id := n.NodeID
-	return &bzzAddr{crypto.Keccak256(id[:]), id[:]}
+	return &bzzAddr{
+		OAddr: crypto.Keccak256(id[:]),
+		UAddr: []byte(discover.NewNode(id, net.IP{127, 0, 0, 1}, 30303, 30303).String()),
+	}
 }

swarm/network/simulations/overlay.go

@@ -11,66 +11,57 @@ import (
 	"net/http"
 	"os"
 	"runtime"
+	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/node"
-	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/simulations"
 	"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
-	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/ethereum/go-ethereum/swarm/network"
 )
 
-// SimNode is the adapter used by Swarm simulations.
-type SimNode struct {
-	id       *adapters.NodeId
-	rw       network.ReadWriter
-	hive     *network.Hive
-	protocol *p2p.Protocol
-}
-
-type simReadWriter struct {
+type simStore struct {
 	m map[string][]byte
 }
 
-func (self *simReadWriter) ReadAll(s string) ([]byte, error) {
+func (self *simStore) Load(s string) ([]byte, error) {
 	return self.m[s], nil
 }
 
-func (self *simReadWriter) WriteAll(s string, data []byte) error {
+func (self *simStore) Save(s string, data []byte) error {
 	self.m[s] = data
 	return nil
 }
 
-func NewSimReadWriter() *simReadWriter {
-	return &simReadWriter{
+func NewSimStore() *simStore {
+	return &simStore{
 		make(map[string][]byte),
 	}
 }
 
-func (s *SimNode) Protocols() []p2p.Protocol {
-	return []p2p.Protocol{*s.protocol}
+type Simulation struct {
+	mtx    sync.Mutex
+	stores map[discover.NodeID]*simStore
 }
 
-func (s *SimNode) APIs() []rpc.API {
-	return nil
+func NewSimulation() *Simulation {
+	return &Simulation{
+		stores: make(map[discover.NodeID]*simStore),
+	}
 }
 
-// the hive update ticker for hive
-func af() <-chan time.Time {
-	return time.NewTicker(1 * time.Second).C
-}
+func (s *Simulation) NewService(id *adapters.NodeId, snapshot []byte) node.Service {
+	s.mtx.Lock()
+	store, ok := s.stores[id.NodeID]
+	if !ok {
+		store = NewSimStore()
+		s.stores[id.NodeID] = store
+	}
+	s.mtx.Unlock()
 
-// Start() starts up the hive
-// makes SimNode implement node.Service
-func (self *SimNode) Start(server *p2p.Server) error {
-	self.init()
-	return self.hive.Start(server, af, self.rw)
-}
-
-func (self *SimNode) init() {
-	addr := network.NewPeerAddrFromNodeId(self.id)
+	addr := network.NewAddrFromNodeId(id)
 	kp := network.NewKadParams()
 
 	kp.MinProxBinSize = 2
@@ -80,42 +71,16 @@ func (self *SimNode) init() {
 	kp.RetryExponent = 2
 	kp.RetryInterval = 1000000
 
-	to := network.NewKademlia(addr.OverlayAddr(), kp) // overlay topology driver
 	hp := network.NewHiveParams()
-	hp.CallInterval = 5000
-	pp := network.NewHive(hp, to) // hive
-
-	services := func(p network.Peer) error {
-		dp := network.NewDiscovery(p, to)
-		pp.Add(dp)
-		log.Trace(fmt.Sprintf("kademlia on %v", dp))
-		p.DisconnectHook(func(err error) {
-			pp.Remove(dp)
-		})
-		return nil
+	hp.KeepAliveInterval = 5 * time.Second
+	config := &network.BzzConfig{
+		OverlayAddr:  addr.Over(),
+		UnderlayAddr: addr.Under(),
+		KadParams:    kp,
+		HiveParams:   hp,
+		Store:        store,
 	}
-
-	ct := network.BzzCodeMap(network.DiscoveryMsgs...) // bzz protocol code map
-	nodeInfo := func() interface{} { return pp.String() }
-	self.hive = pp
-	self.protocol = network.Bzz(addr.OverlayAddr(), addr.UnderlayAddr(), ct, services, nil, nodeInfo)
-}
-
-// Stop() shuts down the hive
-// makes SimNode implement node.Service
-func (self *SimNode) Stop() error {
-	self.hive.Stop()
-	return nil
-}
-
-// NewSimNode creates adapters for nodes in the simulation.
-func NewSimNode(id *adapters.NodeId, snapshot []byte) node.Service {
-	s := &SimNode{
-		id: id,
-		rw: NewSimReadWriter(),
-	}
-	s.init()
-	return s
+	return network.NewBzz(config)
 }
 
 func createMockers() map[string]*simulations.MockerConfig {
@@ -171,7 +136,12 @@ func setupMocker(net *simulations.Network) []*adapters.NodeId {
 		} else {
 			peerId = ids[i-1]
 		}
-		if err := net.Connect(id, peerId); err != nil {
+		ch := make(chan network.OverlayAddr)
+		go func() {
+			defer close(ch)
+			ch <- network.NewAddrFromNodeId(peerId)
+		}()
+		if err := net.GetNode(id).Node.(*adapters.SimNode).Service().(*network.Bzz).Hive.Register(ch); err != nil {
 			panic(err.Error())
 		}
 	}
@@ -243,10 +213,11 @@ func startStopMocker(net *simulations.Network) {
 func main() {
 	runtime.GOMAXPROCS(runtime.NumCPU())
 
-	log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
+	log.Root().SetHandler(log.LvlFilterHandler(log.LvlDebug, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
 
+	s := NewSimulation()
 	services := adapters.Services{
-		"overlay": NewSimNode,
+		"overlay": s.NewService,
 	}
 	adapters.RegisterServices(services)