From d9af74e3fb04dfcaa6a121fe627028eacd6c3f75 Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 4 May 2017 13:05:25 +0200 Subject: [PATCH] swarm/network, p2p/protocols: fix kademlia crashes and msg asyncrony --- p2p/protocols/protocol.go | 3 +- swarm/network/discovery.go | 30 ++++++------- swarm/network/kademlia.go | 87 +++++++++++++++----------------------- swarm/network/protocol.go | 1 - 4 files changed, 48 insertions(+), 73 deletions(-) diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index d9e9895f35..4318341e67 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -276,8 +276,7 @@ func (self *Peer) Send(msg interface{}) error { return errorf(ErrInvalidMsgType, "%v", code) } log.Trace(fmt.Sprintf("=> msg #%d TO %v : %v", code, self.ID(), msg)) - go p2p.Send(self.rw, uint64(code), msg) - return nil + return p2p.Send(self.rw, uint64(code), msg) } func (self *Peer) DisconnectHook(f func(error)) { diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go index cc032f60ff..91ad122c6b 100644 --- a/swarm/network/discovery.go +++ b/swarm/network/discovery.go @@ -17,9 +17,8 @@ var DiscoveryMsgs = []interface{}{ type discPeer struct { Peer - overlay Overlay - peers map[string]bool - // peers map[discover.NodeID]bool + overlay Overlay + peers map[string]bool proxLimit 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 } @@ -44,7 +43,7 @@ func NewDiscovery(p Peer, o Overlay) *discPeer { // NotifyPeer notifies the receiver remote end of a peer p or PO po. // callback for overlay driver func (self *discPeer) NotifyPeer(p Peer, po uint8) error { - log.Warn(fmt.Sprintf("peers %v", self.peers)) + log.Warn(fmt.Sprintf("peer %#v peers %v", p, self.peers)) if po < self.proxLimit || self.seen(p) { return nil } @@ -53,9 +52,7 @@ func (self *discPeer) NotifyPeer(p Peer, po uint8) error { resp := &peersMsg{ Peers: []*peerAddr{&peerAddr{OAddr: p.OverlayAddr(), UAddr: p.UnderlayAddr()}}, // perhaps the PeerAddr interface is unnecessary generalization } - self.Send(resp) - //return err - return nil + return self.Send(resp) } // NotifyProx sends a subPeers Msg to the receiver notifying them about @@ -63,9 +60,7 @@ func (self *discPeer) NotifyPeer(p Peer, po uint8) error { // or first empty row) // callback for overlay driver func (self *discPeer) NotifyProx(po uint8) error { - self.Send(&subPeersMsg{ProxLimit: po}) - //return err - return nil + return self.Send(&subPeersMsg{ProxLimit: po}) } /* @@ -123,13 +118,16 @@ func (self *discPeer) handleSubPeersMsg(msg interface{}) error { if uint8(po) < self.proxLimit { return false } - self.seen(p) + log.Warn(fmt.Sprintf("peer %#v proxlimit %v", p, self.proxLimit)) + self.seen(p.(*discPeer).Peer) peers = append(peers, &peerAddr{p.OverlayAddr(), p.UnderlayAddr()}) return true }) log.Warn(fmt.Sprintf("found initial %v peers not farther than %v", len(peers), self.proxLimit)) if len(peers) > 0 { - self.Send(&peersMsg{Peers: peers}) + if err := self.Send(&peersMsg{Peers: peers}); err != nil { + return err + } } } self.sentPeers = true @@ -181,8 +179,7 @@ func (self *discPeer) handleGetPeersMsg(msg interface{}) error { resp := &peersMsg{ Peers: peers, } - self.Send(resp) - return nil + return self.Send(resp) } func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) { @@ -193,9 +190,8 @@ func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) { var i uint8 //var err error k.EachLivePeer(nil, 255, func(n Peer, po int, isproxbin bool) bool { - log.Trace(fmt.Sprintf("%T sent to %v", req, n.ID())) - err := n.Send(req) - if err == nil { + log.Trace(fmt.Sprintf("%T sent to %v", req, n)) + if err := n.Send(req); err == nil { i++ if i >= broadcastSize { return false diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index 23f9b86560..789575e804 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -38,19 +38,14 @@ exponential scale. It also has the property that any two item belonging to the same bin are at most half as distant from each other as they are from x. -If we think of random sample of items in the bins as connections in a network of interconnected nodes than relative proximity can serve as the basis for local +If we think of random sample of items in the bins as connections in a network of +interconnected nodes than relative proximity can serve as the basis for local decisions for graph traversal where the task is to find a route between two points. Since in every hop, the finite distance halves, there is a guaranteed constant maximum limit on the number of hops needed to reach one node from the other. - -proxLimit is dynamically adjusted so that -1) there is no empty buckets in bin < proxLimit and -2) the sum of all items are the minimum possible but higher than ProxBinSize -adjust Prox (proxLimit and proxSize after an insertion/removal of nodes) -caller holds the lock - */ + type KadDiscovery interface { NotifyPeer(Peer, uint8) error NotifyProx(uint8) error @@ -87,7 +82,7 @@ type Kademlia struct { addr PeerAddr // immutable baseaddress of the table // addr *pot.HashAddress // immutable baseaddress of the table *KadParams // Kademlia configuration parameters - conns, peers *pot.Pot // pots container for peers + addrs, peers *pot.Pot // pots container for peers lastProxLimit uint8 // stores the last calculated proxlimit } @@ -101,7 +96,7 @@ func NewKademlia(addr []byte, params *KadParams) *Kademlia { self := &Kademlia{ addr: &peerAddr{OAddr: addr}, KadParams: params, - conns: pot.NewPot(nil, 0), + addrs: pot.NewPot(nil, 0), peers: pot.NewPot(nil, 0), } return self @@ -143,9 +138,7 @@ func (self *Kademlia) Prune(c <-chan time.Time) { // KadPeer represents a Kademlia Peer and extends // * PeerAddr interface (overlay and underlay addresses) // * Peer interface (id, last seen, drop) -// * HashAddress as derived from PeerAddr overlay implement pot.PoVal interface type KadPeer struct { - // *pot.HashAddress PeerAddr Peer Peer seenAt time.Time @@ -156,8 +149,6 @@ func (self *KadPeer) String() string { if self == nil { return "" } - // return string(self.OverlayAddr()) - //return self.HashAddress.Address.String() return fmt.Sprintf("%x", self.OverlayAddr()) } @@ -175,9 +166,9 @@ func (self *Kademlia) callable(val pot.PotVal) *KadPeer { log.Trace(fmt.Sprintf("delta: %v", delta)) retries++ } + // this is never called concurrently, so safe to increment // peer can be retried again - if retries < kp.retries { log.Trace(fmt.Sprintf("log time needed before retry %v, wait only warrants %v", kp.retries, retries)) return nil @@ -216,10 +207,10 @@ func (self *Kademlia) Register(nas ...PeerAddr) error { np, _, _ = pot.Add(np, pot.PotVal(p)) } oldpeers := pot.NewPot(nil, 0) - oldpeers.Merge(self.peers) - self.peers.Merge(np) + oldpeers.Merge(self.addrs) + self.addrs.Merge(np) m := make(map[string]bool) - self.peers.Each(func(val pot.PotVal, i int) bool { + self.addrs.Each(func(val pot.PotVal, i int) bool { _, found := m[val.String()] // TODO: remove this check // log.Debug(fmt.Sprintf("-> %v %v", val, i)) @@ -235,26 +226,25 @@ func (self *Kademlia) Register(nas ...PeerAddr) error { // On(p) inserts the peer as a kademlia peer into the live peers func (self *Kademlia) On(p Peer) { kp := NewKadPeer(p) - self.conns.Swap(kp, func(v pot.PotVal) pot.PotVal { + kp.Peer = p + self.peers.Swap(kp, func(v pot.PotVal) pot.PotVal { + // if not found live if v == nil { - self.peers.Swap(kp, func(v pot.PotVal) pot.PotVal { - if v != nil { - kp = v.(*KadPeer) - } - kp.Peer = p + // switch the offline peer + self.addrs.Swap(kp, func(v pot.PotVal) pot.PotVal { return pot.PotVal(kp) }) + // insert new peer return pot.PotVal(kp) } + // found among live peers, do nothing return v }) - kp.seenAt = time.Now() - kp.retries = 0 prox := self.proxLimit() vp, ok := kp.Peer.(KadDiscovery) if !ok { - log.Trace(fmt.Sprintf("not discovery peer %T", kp)) + // log.Trace(fmt.Sprintf("not discovery peer %T", kp)) return } go vp.NotifyProx(uint8(prox)) @@ -268,23 +258,20 @@ func (self *Kademlia) On(p Peer) { } log.Debug("peer notified") } - self.conns.EachNeighbourAsync(kp, 1024, 255, f, false) + self.peers.EachNeighbourAsync(kp, 1024, 255, f, false) } // Off removes a peer from among live peers func (self *Kademlia) Off(p Peer) { kp := NewKadPeer(p) - self.conns.Swap(kp, func(v pot.PotVal) pot.PotVal { + self.addrs.Swap(kp, func(v pot.PotVal) pot.PotVal { if v != nil { - kp = v.(*KadPeer) - kp.Peer = nil - return nil + self.peers.Swap(kp, func(v pot.PotVal) pot.PotVal { + return nil + }) } return nil }) - kp.Peer = nil - kp.retries = 0 - kp.seenAt = time.Now() } type ByteAddr struct { @@ -301,7 +288,7 @@ func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int, bool) b } else { p = pot.PotVal(&peerAddr{OAddr: base}) } - self.conns.EachNeighbour(p, func(val pot.PotVal, po int) bool { + self.peers.EachNeighbour(p, func(val pot.PotVal, po int) bool { if po > o { return true } @@ -323,7 +310,7 @@ func (self *Kademlia) EachPeer(base []byte, o int, f func(PeerAddr, int) bool) { } else { p = pot.NewHashAddressFromBytes(base) } - self.peers.EachNeighbour(p, func(val pot.PotVal, po int) bool { + self.addrs.EachNeighbour(p, func(val pot.PotVal, po int) bool { if po > o { return true } @@ -335,7 +322,7 @@ func (self *Kademlia) EachPeer(base []byte, o int, f func(PeerAddr, int) bool) { // the nearest neighbour set with cardinality >= MinProxBinSize // if there is altogether less than MinProxBinSize peers it returns 0 func (self *Kademlia) proxLimit() int { - if self.conns.Size() < self.MinProxBinSize { + if self.peers.Size() < self.MinProxBinSize { return 0 } var proxLimit int @@ -345,7 +332,7 @@ func (self *Kademlia) proxLimit() int { proxLimit = i return size < self.MinProxBinSize } - self.conns.EachNeighbour(pot.PotVal(self.addr), f) + self.peers.EachNeighbour(pot.PotVal(self.addr), f) return proxLimit } @@ -360,7 +347,7 @@ func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) { // this makes sure nearest neighbour set is fully connected log.Trace(fmt.Sprintf("candidate prox peer checking above PO %v", proxLimit)) var ppo int - self.peers.EachNeighbour(self.addr, func(val pot.PotVal, po int) bool { + self.addrs.EachNeighbour(self.addr, func(val pot.PotVal, po int) bool { r := self.callable(val) if r == nil { return po >= proxLimit @@ -377,7 +364,7 @@ func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) { var bpo []int prev := -1 - self.conns.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + self.peers.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { log.Trace(fmt.Sprintf("check PO%02d: ", po)) prev++ if po > prev { @@ -400,7 +387,7 @@ func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) { // try to select a candidate peer for i := len(bpo) - 1; i >= 0; i-- { // find the first callable peer - self.peers.EachBin(self.addr, bpo[i], func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + self.addrs.EachBin(self.addr, bpo[i], func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { // for each bin we find callable candidate peers f(func(val pot.PotVal, i int) bool { r := self.callable(val) @@ -431,15 +418,15 @@ func (self *Kademlia) String() string { rows = append(rows, "=========================================================================") rows = append(rows, fmt.Sprintf("%v KΛÐΞMLIΛ hive: queen's address: %v", time.Now().UTC().Format(time.UnixDate), fmt.Sprintf("%x", self.addr.OverlayAddr()[:3]))) - rows = append(rows, fmt.Sprintf("population: %d (%d), MinProxBinSize: %d, MinBinSize: %d, MaxBinSize: %d", self.conns.Size(), self.peers.Size(), self.MinProxBinSize, self.MinBinSize, self.MaxBinSize)) + rows = append(rows, fmt.Sprintf("population: %d (%d), MinProxBinSize: %d, MinBinSize: %d, MaxBinSize: %d", self.peers.Size(), self.addrs.Size(), self.MinProxBinSize, self.MinBinSize, self.MaxBinSize)) liverows := make([]string, self.MaxProxDisplay) peersrows := make([]string, self.MaxProxDisplay) var proxLimit int prev := -1 var proxLimitSet bool - rest := self.conns.Size() - self.conns.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + rest := self.peers.Size() + self.peers.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { var rowlen int if po >= self.MaxProxDisplay { po = self.MaxProxDisplay - 1 @@ -463,7 +450,7 @@ func (self *Kademlia) String() string { return true }) - self.peers.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { + self.addrs.EachBin(self.addr, 0, func(po, size int, f func(func(val pot.PotVal, i int) bool) bool) bool { var rowlen int if po >= self.MaxProxDisplay { po = self.MaxProxDisplay - 1 @@ -472,19 +459,13 @@ func (self *Kademlia) String() string { panic("wtf") } row := []string{fmt.Sprintf("%2d", size)} + // we are displaying live peers too f(func(val pot.PotVal, vpo int) bool { kp := val.(*KadPeer) - // if kp.Peer != nil { - // return true - // } row = append(row, kp.String()[:6]) rowlen++ return rowlen < 4 }) - // log.Debug(fmt.Sprintf("po: %v, peerrows length: %v, maxProxDisplay: %v", po, len(peersrows), self.MaxProxDisplay)) - // if po < self.MaxProxDisplay { - // peersrows[po] = strings.Join(row, " ") - // } peersrows[po] = strings.Join(row, " ") return true }) diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 59bd48debb..66963215e4 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -42,7 +42,6 @@ type bzzPeer struct { localAddr *peerAddr *peerAddr // remote address lastActive time.Time - //peers map[discover.NodeID]bool } func (self *bzzPeer) LastActive() time.Time {