swarm/network: simplified neighbourhood depth calc (#1013)

* swarm/network: simplified depth calculation

* swarm/network: changed depth calls to method not under read lock. reenabled waitTillHealty

* swarm/network: moved comment, replaced neighborhoodDepth with depthForPot

* swarm/network: revert to po==256

* swarm/network: Rename wrong var in iterator
This commit is contained in:
Elad 2018-11-25 02:14:03 +05:30 committed by lash
parent 12f1e6849f
commit f316495bb0
2 changed files with 42 additions and 61 deletions

View file

@ -175,7 +175,7 @@ func (k *Kademlia) SuggestPeer() (a *BzzAddr, o int, want bool) {
k.lock.Lock() k.lock.Lock()
defer k.lock.Unlock() defer k.lock.Unlock()
minsize := k.MinBinSize minsize := k.MinBinSize
depth := k.neighbourhoodDepth() depth := depthForPot(k.conns, k.MinProxBinSize, k.base)
// if there is a callable neighbour within the current proxBin, connect // if there is a callable neighbour within the current proxBin, connect
// this makes sure nearest neighbour set is fully connected // this makes sure nearest neighbour set is fully connected
var ppo int var ppo int
@ -305,7 +305,7 @@ func (k *Kademlia) sendNeighbourhoodDepthChange() {
// It provides signaling of neighbourhood depth change. // It provides signaling of neighbourhood depth change.
// This part of the code is sending new neighbourhood depth to nDepthC if that condition is met. // This part of the code is sending new neighbourhood depth to nDepthC if that condition is met.
if k.nDepthC != nil { if k.nDepthC != nil {
nDepth := k.neighbourhoodDepth() nDepth := depthForPot(k.conns, k.MinProxBinSize, k.base)
if nDepth != k.nDepth { if nDepth != k.nDepth {
k.nDepth = nDepth k.nDepth = nDepth
k.nDepthC <- nDepth k.nDepthC <- nDepth
@ -361,7 +361,7 @@ func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(con
var startPo int var startPo int
var endPo int var endPo int
kadDepth := k.neighbourhoodDepth() kadDepth := depthForPot(k.conns, k.MinProxBinSize, k.base)
k.conns.EachBin(base, pof, o, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool { k.conns.EachBin(base, pof, o, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
if startPo > 0 && endPo != k.MaxProxDisplay { if startPo > 0 && endPo != k.MaxProxDisplay {
@ -395,7 +395,7 @@ func (k *Kademlia) eachConn(base []byte, o int, f func(*Peer, int, bool) bool) {
if len(base) == 0 { if len(base) == 0 {
base = k.base base = k.base
} }
depth := k.neighbourhoodDepth() depth := depthForPot(k.conns, k.MinProxBinSize, k.base)
k.conns.EachNeighbour(base, pof, func(val pot.Val, po int) bool { k.conns.EachNeighbour(base, pof, func(val pot.Val, po int) bool {
if po > o { if po > o {
return true return true
@ -417,7 +417,7 @@ func (k *Kademlia) eachAddr(base []byte, o int, f func(*BzzAddr, int, bool) bool
if len(base) == 0 { if len(base) == 0 {
base = k.base base = k.base
} }
depth := k.neighbourhoodDepth() depth := depthForPot(k.conns, k.MinProxBinSize, k.base)
k.addrs.EachNeighbour(base, pof, func(val pot.Val, po int) bool { k.addrs.EachNeighbour(base, pof, func(val pot.Val, po int) bool {
if po > o { if po > o {
return true return true
@ -426,18 +426,18 @@ func (k *Kademlia) eachAddr(base []byte, o int, f func(*BzzAddr, int, bool) bool
}) })
} }
// neighbourhoodDepth returns the proximity order that defines the distance of
// the nearest neighbour set with cardinality >= MinProxBinSize
// if there is altogether less than MinProxBinSize peers it returns 0
// caller must hold the lock
func (k *Kademlia) NeighbourhoodDepth() (depth int) { func (k *Kademlia) NeighbourhoodDepth() (depth int) {
k.lock.RLock() k.lock.RLock()
defer k.lock.RUnlock() defer k.lock.RUnlock()
return k.neighbourhoodDepth() return depthForPot(k.conns, k.MinProxBinSize, k.base)
} }
func (k *Kademlia) neighbourhoodDepth() (depth int) { // depthForPot returns the proximity order that defines the distance of
if k.conns.Size() <= k.MinProxBinSize { // the nearest neighbour set with cardinality >= MinProxBinSize
// if there is altogether less than MinProxBinSize peers it returns 0
// caller must hold the lock
func depthForPot(p *pot.Pot, minProxBinSize int, pivotAddr []byte) (depth int) {
if p.Size() <= minProxBinSize {
return 0 return 0
} }
@ -451,11 +451,15 @@ func (k *Kademlia) neighbourhoodDepth() (depth int) {
var lastPo int var lastPo int
f := func(v pot.Val, i int) bool { f := func(v pot.Val, i int) bool {
// po == 256 means that addr is the pivot address(self)
if i == 256 {
return true
}
size++ size++
// this means we have all nn-peers. // this means we have all nn-peers.
// depth is by default set to the bin of the farthest nn-peer // depth is by default set to the bin of the farthest nn-peer
if size == k.MinProxBinSize { if size == minProxBinSize {
b = true b = true
depth = i depth = i
return true return true
@ -481,14 +485,13 @@ func (k *Kademlia) neighbourhoodDepth() (depth int) {
lastPo = i lastPo = i
return true return true
} }
k.conns.EachNeighbour(k.base, pof, f) p.EachNeighbour(pivotAddr, pof, f)
// cover edge case where more than one farthest nn // cover edge case where more than one farthest nn
// AND we only have nn-peers // AND we only have nn-peers
if lastPo == depth { if lastPo == depth {
depth = 0 depth = 0
} }
return depth return depth
} }
@ -548,7 +551,7 @@ func (k *Kademlia) string() string {
liverows := make([]string, k.MaxProxDisplay) liverows := make([]string, k.MaxProxDisplay)
peersrows := make([]string, k.MaxProxDisplay) peersrows := make([]string, k.MaxProxDisplay)
depth := k.neighbourhoodDepth() depth := depthForPot(k.conns, k.MinProxBinSize, k.base)
rest := k.conns.Size() rest := k.conns.Size()
k.conns.EachBin(k.base, pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool { k.conns.EachBin(k.base, pof, 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool {
var rowlen int var rowlen int
@ -628,67 +631,46 @@ func NewPeerPotMap(kadMinProxSize int, addrs [][]byte) map[string]*PeerPot {
for i, a := range addrs { for i, a := range addrs {
// set to proxbin depth when all nn-peers are found // actual kademlia depth
pl := 256 depth := depthForPot(np, kadMinProxSize, a)
// upon entering a new iteration // upon entering a new iteration
// this will hold the value the po should be // this will hold the value the po should be
// if it's one higher than the po in the last iteration // if it's one higher than the po in the last iteration
prev := 256 prevPo := 256
// all bins outside proxbin depth with no peers // all empty bins which are outside neighbourhood depth
var emptyBins []int var emptyBins []int
// all nn-peers // all nn-peers
var nns [][]byte var nns [][]byte
// used to skip empty bins immediately after nn-peers np.EachNeighbour(a, pof, func(val pot.Val, po int) bool {
depthTraversed := false addr := val.([]byte)
// po == 256 means that addr is the pivot address(self)
np.EachNeighbour(addrs[i], pof, func(val pot.Val, po int) bool {
a := val.([]byte)
// 256 is self. We are selfless
if po == 256 { if po == 256 {
return true return true
} }
// if first nn-peer or peer in same bin as last // iterate through the neighbours, going from the closest to the farthest
if pl == 256 || pl == po { // we calculate the nearest neighbours that should be in the set
nns = append(nns, a) // depth in this case equates to:
// 1. Within all bins that are higher or equal than depth there are
// at least minProxBinSize peers connected
// 2. depth-1 bin is not empty
if po >= depth {
nns = append(nns, addr)
prevPo = depth - 1
return true
} }
for j := prevPo; j > po; j-- {
// if true then all nn-bins have been filled emptyBins = append(emptyBins, j)
// start counting pl and set prev to current po initially (which will skip next block)
if pl == 256 && len(nns) >= kadMinProxSize {
pl = po
prev = po
} }
prevPo = po - 1
// only true starting from first peer after nn-peers
if prev < pl {
if depthTraversed {
for j := prev; j > po; j-- {
emptyBins = append(emptyBins, j)
}
}
// after first peer after nn-peers, start counting emptybins
depthTraversed = true
}
// expected po in next iteration if there are no empty bins in between
prev = po - 1
return true return true
}) })
// add any remaining bins between po 0 and the last po in iteration log.Trace(fmt.Sprintf("%x NNS: %s, emptyBins: %s", addrs[i][:4], LogAddrs(nns), logEmptyBins(emptyBins)))
// to the list of empty bins
for j := prev; j >= 0; j-- {
emptyBins = append(emptyBins, j)
}
log.Trace(fmt.Sprintf("%x NNS: %s", addrs[i][:4], LogAddrs(nns)))
ppmap[common.Bytes2Hex(a)] = &PeerPot{nns, emptyBins} ppmap[common.Bytes2Hex(a)] = &PeerPot{nns, emptyBins}
} }
return ppmap return ppmap
@ -703,7 +685,7 @@ func (k *Kademlia) saturation(n int) int {
prev++ prev++
return prev == po && size >= n return prev == po && size >= n
}) })
depth := k.neighbourhoodDepth() depth := depthForPot(k.conns, k.MinProxBinSize, k.base)
if depth < prev { if depth < prev {
return depth return depth
} }
@ -716,7 +698,7 @@ func (k *Kademlia) full(emptyBins []int) (full bool) {
prev := 0 prev := 0
e := len(emptyBins) e := len(emptyBins)
ok := true ok := true
depth := k.neighbourhoodDepth() depth := depthForPot(k.conns, k.MinProxBinSize, k.base)
k.conns.EachBin(k.base, pof, 0, func(po, _ int, _ func(func(val pot.Val, i int) bool) bool) bool { k.conns.EachBin(k.base, pof, 0, func(po, _ int, _ func(func(val pot.Val, i int) bool) bool) bool {
if po >= depth { if po >= depth {
return false return false

View file

@ -29,7 +29,6 @@ import (
func TestWaitTillHealthy(t *testing.T) { func TestWaitTillHealthy(t *testing.T) {
t.Skip("temporarily disabled as simulations.WaitTillHealthy cannot be trusted")
sim := New(map[string]ServiceFunc{ sim := New(map[string]ServiceFunc{
"bzz": func(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) { "bzz": func(ctx *adapters.ServiceContext, b *sync.Map) (node.Service, func(), error) {
addr := network.NewAddr(ctx.Config.Node()) addr := network.NewAddr(ctx.Config.Node())