From 5307e1ea771bf8fb49d419d2589da58972180f47 Mon Sep 17 00:00:00 2001 From: zelig Date: Tue, 16 May 2017 15:49:44 -0700 Subject: [PATCH] swarm/network: Kademlia fixes --- swarm/network/discovery.go | 9 ++- swarm/network/hive.go | 22 +++--- swarm/network/kademlia.go | 16 ++-- swarm/network/protocol.go | 13 +++- swarm/network/simulations/overlay.go | 109 ++++++++++----------------- 5 files changed, 76 insertions(+), 93 deletions(-) diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go index 80e8e27c70..e6492e2b35 100644 --- a/swarm/network/discovery.go +++ b/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) } diff --git a/swarm/network/hive.go b/swarm/network/hive.go index dfa356214b..1a361cb86f 100644 --- a/swarm/network/hive.go +++ b/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 } diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index e3d4604597..80ecad5962 100644 --- a/swarm/network/kademlia.go +++ b/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) diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index fd42276589..9ef52f3dc4 100644 --- a/swarm/network/protocol.go +++ b/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()), + } } diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go index 3498be5bca..85ac83d2f5 100644 --- a/swarm/network/simulations/overlay.go +++ b/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)