forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-13 07:23:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/network, p2p/protocols: fix kademlia crashes and msg asyncrony

Browse source
This commit is contained in:
zelig 2017-05-04 13:05:25 +02:00 committed by Lewis Marshall
parent ff24b603d0
commit d9af74e3fb
4 changed files with 48 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
p2p/protocols/protocol.go
View file

@ -276,8 +276,7 @@ func (self *Peer) Send(msg interface{}) error {
return errorf(ErrInvalidMsgType, "%v", code) return errorf(ErrInvalidMsgType, "%v", code)
} }
log.Trace(fmt.Sprintf("=> msg #%d TO %v : %v", code, self.ID(), msg)) log.Trace(fmt.Sprintf("=> msg #%d TO %v : %v", code, self.ID(), msg))
go p2p.Send(self.rw, uint64(code), msg) return p2p.Send(self.rw, uint64(code), msg)
return nil
} }
func (self *Peer) DisconnectHook(f func(error)) { func (self *Peer) DisconnectHook(f func(error)) {

30
swarm/network/discovery.go
View file

@ -17,9 +17,8 @@ var DiscoveryMsgs = []interface{}{
type discPeer struct { type discPeer struct {
Peer Peer
overlay Overlay overlay Overlay
peers map[string]bool peers map[string]bool
// peers map[discover.NodeID]bool
proxLimit uint8 // the proximity radius advertised by remote to subscribe to peers 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 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. // NotifyPeer notifies the receiver remote end of a peer p or PO po.
// callback for overlay driver // callback for overlay driver
func (self *discPeer) NotifyPeer(p Peer, po uint8) error { 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) { if po < self.proxLimit || self.seen(p) {
return nil return nil
} }
@ -53,9 +52,7 @@ func (self *discPeer) NotifyPeer(p Peer, po uint8) error {
resp := &peersMsg{ resp := &peersMsg{
Peers: []*peerAddr{&peerAddr{OAddr: p.OverlayAddr(), UAddr: p.UnderlayAddr()}}, // perhaps the PeerAddr interface is unnecessary generalization Peers: []*peerAddr{&peerAddr{OAddr: p.OverlayAddr(), UAddr: p.UnderlayAddr()}}, // perhaps the PeerAddr interface is unnecessary generalization
} }
self.Send(resp) return self.Send(resp)
//return err
return nil
} }
// NotifyProx sends a subPeers Msg to the receiver notifying them about // 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) // or first empty row)
// callback for overlay driver // callback for overlay driver
func (self *discPeer) NotifyProx(po uint8) error { func (self *discPeer) NotifyProx(po uint8) error {
self.Send(&subPeersMsg{ProxLimit: po}) return self.Send(&subPeersMsg{ProxLimit: po})
//return err
return nil
} }
/* /*
@ -123,13 +118,16 @@ func (self *discPeer) handleSubPeersMsg(msg interface{}) error {
if uint8(po) < self.proxLimit { if uint8(po) < self.proxLimit {
return false 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()}) peers = append(peers, &peerAddr{p.OverlayAddr(), p.UnderlayAddr()})
return true 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.proxLimit))
if len(peers) > 0 { if len(peers) > 0 {
self.Send(&peersMsg{Peers: peers}) if err := self.Send(&peersMsg{Peers: peers}); err != nil {
return err
}
} }
} }
self.sentPeers = true self.sentPeers = true
@ -181,8 +179,7 @@ func (self *discPeer) handleGetPeersMsg(msg interface{}) error {
resp := &peersMsg{ resp := &peersMsg{
Peers: peers, Peers: peers,
} }
self.Send(resp) return self.Send(resp)
return nil
} }
func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) { func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) {
@ -193,9 +190,8 @@ func RequestOrder(k Overlay, order, broadcastSize, maxPeers uint8) {
var i uint8 var i uint8
//var err error //var err error
k.EachLivePeer(nil, 255, func(n Peer, po int, isproxbin bool) bool { k.EachLivePeer(nil, 255, func(n Peer, po int, isproxbin bool) bool {
log.Trace(fmt.Sprintf("%T sent to %v", req, n.ID())) log.Trace(fmt.Sprintf("%T sent to %v", req, n))
err := n.Send(req) if err := n.Send(req); err == nil {
if err == nil {
i++ i++
if i >= broadcastSize { if i >= broadcastSize {
return false return false

87
swarm/network/kademlia.go
View file

@ -38,19 +38,14 @@ exponential scale.
It also has the property that any two item belonging to the same bin are at 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. 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 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 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 a guaranteed constant maximum limit on the number of hops needed to reach one
node from the other. 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 { type KadDiscovery interface {
NotifyPeer(Peer, uint8) error NotifyPeer(Peer, uint8) error
NotifyProx(uint8) error NotifyProx(uint8) error
@ -87,7 +82,7 @@ type Kademlia struct {
addr PeerAddr // immutable baseaddress of the table addr PeerAddr // immutable baseaddress of the table
// addr *pot.HashAddress // immutable baseaddress of the table // addr *pot.HashAddress // immutable baseaddress of the table
*KadParams // Kademlia configuration parameters *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 lastProxLimit uint8 // stores the last calculated proxlimit
} }
@ -101,7 +96,7 @@ func NewKademlia(addr []byte, params *KadParams) *Kademlia {
self := &Kademlia{ self := &Kademlia{
addr: &peerAddr{OAddr: addr}, addr: &peerAddr{OAddr: addr},
KadParams: params, KadParams: params,
conns: pot.NewPot(nil, 0), addrs: pot.NewPot(nil, 0),
peers: pot.NewPot(nil, 0), peers: pot.NewPot(nil, 0),
} }
return self return self
@ -143,9 +138,7 @@ func (self *Kademlia) Prune(c <-chan time.Time) {
// KadPeer represents a Kademlia Peer and extends // KadPeer represents a Kademlia Peer and extends
// * PeerAddr interface (overlay and underlay addresses) // * PeerAddr interface (overlay and underlay addresses)
// * Peer interface (id, last seen, drop) // * Peer interface (id, last seen, drop)
// * HashAddress as derived from PeerAddr overlay implement pot.PoVal interface
type KadPeer struct { type KadPeer struct {
// *pot.HashAddress
PeerAddr PeerAddr
Peer Peer Peer Peer
seenAt time.Time seenAt time.Time
@ -156,8 +149,6 @@ func (self *KadPeer) String() string {
if self == nil { if self == nil {
return "<nil>" return "<nil>"
} }
// return string(self.OverlayAddr())
//return self.HashAddress.Address.String()
return fmt.Sprintf("%x", self.OverlayAddr()) 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)) log.Trace(fmt.Sprintf("delta: %v", delta))
retries++ retries++
} }
// this is never called concurrently, so safe to increment // this is never called concurrently, so safe to increment
// peer can be retried again // peer can be retried again
if retries < kp.retries { if retries < kp.retries {
log.Trace(fmt.Sprintf("log time needed before retry %v, wait only warrants %v", kp.retries, retries)) log.Trace(fmt.Sprintf("log time needed before retry %v, wait only warrants %v", kp.retries, retries))
return nil return nil
@ -216,10 +207,10 @@ func (self *Kademlia) Register(nas ...PeerAddr) error {
np, _, _ = pot.Add(np, pot.PotVal(p)) np, _, _ = pot.Add(np, pot.PotVal(p))
} }
oldpeers := pot.NewPot(nil, 0) oldpeers := pot.NewPot(nil, 0)
oldpeers.Merge(self.peers) oldpeers.Merge(self.addrs)
self.peers.Merge(np) self.addrs.Merge(np)
m := make(map[string]bool) 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()] _, found := m[val.String()]
// TODO: remove this check // TODO: remove this check
// log.Debug(fmt.Sprintf("-> %v %v", val, i)) // 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 // On(p) inserts the peer as a kademlia peer into the live peers
func (self *Kademlia) On(p Peer) { func (self *Kademlia) On(p Peer) {
kp := NewKadPeer(p) 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 { if v == nil {
self.peers.Swap(kp, func(v pot.PotVal) pot.PotVal { // switch the offline peer
if v != nil { self.addrs.Swap(kp, func(v pot.PotVal) pot.PotVal {
kp = v.(*KadPeer)
}
kp.Peer = p
return pot.PotVal(kp) return pot.PotVal(kp)
}) })
// insert new peer
return pot.PotVal(kp) return pot.PotVal(kp)
} }
// found among live peers, do nothing
return v return v
}) })
kp.seenAt = time.Now()
kp.retries = 0
prox := self.proxLimit() prox := self.proxLimit()
vp, ok := kp.Peer.(KadDiscovery) vp, ok := kp.Peer.(KadDiscovery)
if !ok { if !ok {
log.Trace(fmt.Sprintf("not discovery peer %T", kp)) // log.Trace(fmt.Sprintf("not discovery peer %T", kp))
return return
} }
go vp.NotifyProx(uint8(prox)) go vp.NotifyProx(uint8(prox))
@ -268,23 +258,20 @@ func (self *Kademlia) On(p Peer) {
} }
log.Debug("peer notified") 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 // Off removes a peer from among live peers
func (self *Kademlia) Off(p Peer) { func (self *Kademlia) Off(p Peer) {
kp := NewKadPeer(p) 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 { if v != nil {
kp = v.(*KadPeer) self.peers.Swap(kp, func(v pot.PotVal) pot.PotVal {
kp.Peer = nil return nil
return nil })
} }
return nil return nil
}) })
kp.Peer = nil
kp.retries = 0
kp.seenAt = time.Now()
} }
type ByteAddr struct { type ByteAddr struct {
@ -301,7 +288,7 @@ func (self *Kademlia) EachLivePeer(base []byte, o int, f func(Peer, int, bool) b
} else { } else {
p = pot.PotVal(&peerAddr{OAddr: base}) 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 { if po > o {
return true return true
} }
@ -323,7 +310,7 @@ func (self *Kademlia) EachPeer(base []byte, o int, f func(PeerAddr, int) bool) {
} else { } else {
p = pot.NewHashAddressFromBytes(base) 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 { if po > o {
return true 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 // the nearest neighbour set with cardinality >= MinProxBinSize
// if there is altogether less than MinProxBinSize peers it returns 0 // if there is altogether less than MinProxBinSize peers it returns 0
func (self *Kademlia) proxLimit() int { func (self *Kademlia) proxLimit() int {
if self.conns.Size() < self.MinProxBinSize { if self.peers.Size() < self.MinProxBinSize {
return 0 return 0
} }
var proxLimit int var proxLimit int
@ -345,7 +332,7 @@ func (self *Kademlia) proxLimit() int {
proxLimit = i proxLimit = i
return size < self.MinProxBinSize return size < self.MinProxBinSize
} }
self.conns.EachNeighbour(pot.PotVal(self.addr), f) self.peers.EachNeighbour(pot.PotVal(self.addr), f)
return proxLimit 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 // this makes sure nearest neighbour set is fully connected
log.Trace(fmt.Sprintf("candidate prox peer checking above PO %v", proxLimit)) log.Trace(fmt.Sprintf("candidate prox peer checking above PO %v", proxLimit))
var ppo int 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) r := self.callable(val)
if r == nil { if r == nil {
return po >= proxLimit return po >= proxLimit
@ -377,7 +364,7 @@ func (self *Kademlia) SuggestPeer() (p PeerAddr, o int, want bool) {
var bpo []int var bpo []int
prev := -1 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)) log.Trace(fmt.Sprintf("check PO%02d: ", po))
prev++ prev++
if 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 // try to select a candidate peer
for i := len(bpo) - 1; i >= 0; i-- { for i := len(bpo) - 1; i >= 0; i-- {
// find the first callable peer // 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 // for each bin we find callable candidate peers
f(func(val pot.PotVal, i int) bool { f(func(val pot.PotVal, i int) bool {
r := self.callable(val) r := self.callable(val)
@ -431,15 +418,15 @@ func (self *Kademlia) String() string {
rows = append(rows, "=========================================================================") 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("%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) liverows := make([]string, self.MaxProxDisplay)
peersrows := make([]string, self.MaxProxDisplay) peersrows := make([]string, self.MaxProxDisplay)
var proxLimit int var proxLimit int
prev := -1 prev := -1
var proxLimitSet bool var proxLimitSet bool
rest := self.conns.Size() rest := self.peers.Size()
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 {
var rowlen int var rowlen int
if po >= self.MaxProxDisplay { if po >= self.MaxProxDisplay {
po = self.MaxProxDisplay - 1 po = self.MaxProxDisplay - 1
@ -463,7 +450,7 @@ func (self *Kademlia) String() string {
return true 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 var rowlen int
if po >= self.MaxProxDisplay { if po >= self.MaxProxDisplay {
po = self.MaxProxDisplay - 1 po = self.MaxProxDisplay - 1
@ -472,19 +459,13 @@ func (self *Kademlia) String() string {
panic("wtf") panic("wtf")
} }
row := []string{fmt.Sprintf("%2d", size)} row := []string{fmt.Sprintf("%2d", size)}
// we are displaying live peers too
f(func(val pot.PotVal, vpo int) bool { f(func(val pot.PotVal, vpo int) bool {
kp := val.(*KadPeer) kp := val.(*KadPeer)
// if kp.Peer != nil {
// return true
// }
row = append(row, kp.String()[:6]) row = append(row, kp.String()[:6])
rowlen++ rowlen++
return rowlen < 4 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, " ") peersrows[po] = strings.Join(row, " ")
return true return true
}) })

1
swarm/network/protocol.go
View file

@ -42,7 +42,6 @@ type bzzPeer struct {
localAddr *peerAddr localAddr *peerAddr
*peerAddr // remote address *peerAddr // remote address
lastActive time.Time lastActive time.Time
//peers map[discover.NodeID]bool
} }
func (self *bzzPeer) LastActive() time.Time { func (self *bzzPeer) LastActive() time.Time {