Merge pull request #2660 from ethersphere/s/kademlia-findbest

[swarm] kademlia peer selection strategy fixes
TODO: add comments to swarm.sh functions documenting API and make them a bit more logical.
This commit is contained in:
Daniel A. Nagy 2016-06-06 19:51:29 +02:00
commit e2853948da
7 changed files with 191 additions and 183 deletions

View file

@ -24,8 +24,8 @@ var (
"SplitTimeout": 120, "SplitTimeout": 120,
"CallInterval": 10000000000, "CallInterval": 10000000000,
"KadDbPath": "` + filepath.Join("TMPDIR", "0d2f62485607cf38d9d795d93682a517661e513e", "bzz-peers.json") + `", "KadDbPath": "` + filepath.Join("TMPDIR", "0d2f62485607cf38d9d795d93682a517661e513e", "bzz-peers.json") + `",
"MaxProx": 10, "MaxProx": 8,
"ProxBinSize": 8, "ProxBinSize": 4,
"BucketSize": 3, "BucketSize": 3,
"PurgeInterval": 151200000000000, "PurgeInterval": 151200000000000,
"InitialRetryInterval": 4200000000, "InitialRetryInterval": 4200000000,

View file

@ -104,7 +104,6 @@ function shutdown {
while true ;do while true ;do
ps auxwww|grep geth|grep -v grep|awk '{print $2}'|grep -ql $1 || break ps auxwww|grep geth|grep -v grep|awk '{print $2}'|grep -ql $1 || break
sleep 1 sleep 1
# ps auxwww|grep geth|grep -v grep |grep $1 #|awk '{print $2}'
done done
echo "stopped" echo "stopped"
} }
@ -121,8 +120,8 @@ function init {
killall geth killall geth
reset all reset all
cluster $* cluster $*
stop all enode all
cluster $* connect all
} }
function reset { function reset {
@ -136,6 +135,61 @@ function reset {
} }
function enode {
dir=$root/$network_id
id=$1
shift
if [ $id = "all" ]; then
N=`ls -1 $dir/enodes/|wc -l`
enodes=$dir/enodes.all
rm -f $enodes
# build a static nodes(-like) list of all enodes of the local cluster
echo "[" >> $enodes
for ((i=0;i<N;++i)); do
id=`printf "%02d" $i`
enode=$dir/enodes/$id.enode
enode $id
if [ -f "$enode" ] && [ ! -z "$enode" ]; then
cat "$enode" >> $enodes
echo "," >> $enodes
fi
done
echo "\"\"]" >> $enodes
cmd=$dir/connect.js
for ((i=0;i<N;++i)); do
id=`printf "%02d" $i`
enode=$dir/enodes/$id.enode
if [ -f "$enode" ] && [ ! -z "$enode" ]; then
echo -n "admin.addPeer(" >> $cmd
cat "$enode" >> $cmd
echo ");" >> $cmd
fi
done
else
enode=$dir/enodes/$id.enode
attach $id --exec "'console.log(admin.nodeInfo.enode)'" |head -2 |tail -1| perl -pe 's/^/"/;s/$/"/'|perl -pe 's/\s*$//' > $enode
# cat $enode
fi
}
function connect {
dir=$root/$network_id
id=$1
shift
if [ $id = "all" ]; then
for ((i=0;i<N;++i)); do
id=`printf "%02d" $i`
connect $id
done
else
cmd="$GETH --preload $dir/connect.js --exec '\"admin.peers\"' attach ipc:$root/$network_id/data/$id/geth.ipc $dir/connect.js"
# echo $cmd
eval $cmd
cat $dir/connect.js
fi
}
function cluster { function cluster {
N=$1 N=$1
shift shift
@ -149,20 +203,6 @@ function cluster {
mkdir -p $dir/pids mkdir -p $dir/pids
mkdir -p $dir/log mkdir -p $dir/log
enodes=$dir/enodes.all
rm -f $enodes
# build a static nodes(-like) list of all enodes of the local cluster
echo "[" >> $enodes
for ((i=0;i<N;++i)); do
id=`printf "%02d" $i`
enode=$dir/enodes/$id.enode
if [ -f "$enode" ]; then
cat "$enode" >> $enodes
echo "," >> $enodes
fi
done
echo "\"\"]" >> $enodes
for ((i=0;i<N;++i)); do for ((i=0;i<N;++i)); do
id=`printf "%02d" $i` id=`printf "%02d" $i`
mkdir -p $dir/data/$id mkdir -p $dir/data/$id
@ -282,6 +322,10 @@ function netstatconf {
case $cmd in case $cmd in
"info" ) "info" )
info $*;; info $*;;
"enode" )
enode $*;;
"connect" )
connect $*;;
"status" ) "status" )
status $*;; status $*;;
"clean" ) "clean" )

View file

@ -123,13 +123,15 @@ func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee
// to attempt to write to more (remove Peer when shutting down) // to attempt to write to more (remove Peer when shutting down)
return return
} }
node, proxLimit := self.kad.FindBest() node, need, proxLimit := self.kad.FindBest()
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: select candidate peer")
if node != nil && len(node.Url) > 0 { if node != nil && len(node.Url) > 0 {
glog.V(logger.Detail).Infof("[BZZ] KΛÐΞMLIΛ hive: call for bee %v", node.Url) glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: call for bee %v", node.Url)
// enode or any lower level connection address is unnecessary in future // enode or any lower level connection address is unnecessary in future
// discovery table is used to look it up. // discovery table is used to look it up.
connectPeer(node.Url) connectPeer(node.Url)
} else if proxLimit > -1 { } else if need {
// a random peer is taken from the table // a random peer is taken from the table
peers := self.kad.FindClosest(kademlia.RandomAddressAt(self.addr, rand.Intn(self.kad.MaxProx)), 1) peers := self.kad.FindClosest(kademlia.RandomAddressAt(self.addr, rand.Intn(self.kad.MaxProx)), 1)
if len(peers) > 0 { if len(peers) > 0 {
@ -138,15 +140,18 @@ func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee
req := &retrieveRequestMsgData{ req := &retrieveRequestMsgData{
Key: storage.Key(randAddr[:]), Key: storage.Key(randAddr[:]),
} }
glog.V(logger.Detail).Infof("[BZZ] KΛÐΞMLIΛ hive: call any bee in area %v messenger bee %v", randAddr, peers[0]) glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: call any bee in area %v messenger bee %v", randAddr, peers[0])
peers[0].(*peer).retrieve(req) peers[0].(*peer).retrieve(req)
} else {
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: no peer")
} }
self.toggle <- true self.toggle <- true
glog.V(logger.Detail).Infof("[BZZ] KΛÐΞMLIΛ hive: buzz kept alive") glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: buzz kept alive")
} else { } else {
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: no need for more bees")
self.toggle <- false self.toggle <- false
} }
glog.V(logger.Detail).Infof("[BZZ] KΛÐΞMLIΛ hive: queen's address: %v, population: %d (%d)", self.addr, self.kad.Count(), self.kad.DBCount()) glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: queen's address: %v, population: %d (%d)", self.addr, self.kad.Count(), self.kad.DBCount())
} }
}() }()
return return
@ -165,6 +170,7 @@ func (self *Hive) keepAlive() {
if self.kad.DBCount() > 0 { if self.kad.DBCount() > 0 {
select { select {
case self.more <- true: case self.more <- true:
glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: buzz wakeup")
default: default:
} }
} }
@ -241,7 +247,7 @@ func (self *Hive) DropAll() {
// contructor for kademlia.NodeRecord based on peer address alone // contructor for kademlia.NodeRecord based on peer address alone
// TODO: should go away and only addr passed to kademlia // TODO: should go away and only addr passed to kademlia
func newNodeRecord(addr *peerAddr) *kademlia.NodeRecord { func newNodeRecord(addr *peerAddr) *kademlia.NodeRecord {
now := kademlia.Time(time.Now()) now := time.Now()
return &kademlia.NodeRecord{ return &kademlia.NodeRecord{
Addr: addr.Addr, Addr: addr.Addr,
Url: addr.String(), Url: addr.String(),
@ -336,7 +342,7 @@ func (self *Hive) peers(req *retrieveRequestMsgData) {
for _, peer := range self.getPeers(key, int(req.MaxPeers)) { for _, peer := range self.getPeers(key, int(req.MaxPeers)) {
addrs = append(addrs, peer.remoteAddr) addrs = append(addrs, peer.remoteAddr)
} }
glog.V(logger.Detail).Infof("[BZZ] Hive sending %d peer addresses to %v. req.Id: %v, req.Key: %x", len(addrs), req.from, req.Id, req.Key.Log()) glog.V(logger.Debug).Infof("[BZZ] Hive sending %d peer addresses to %v. req.Id: %v, req.Key: %x", len(addrs), req.from, req.Id, req.Key.Log())
peersData := &peersMsgData{ peersData := &peersMsgData{
Peers: addrs, Peers: addrs,

View file

@ -12,30 +12,6 @@ import (
"github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/logger/glog"
) )
type Time time.Time
func (t *Time) MarshalJSON() (out []byte, err error) {
return []byte(fmt.Sprintf("%d", t.Unix())), nil
}
func (t *Time) UnmarshalJSON(value []byte) error {
var i int64
_, err := fmt.Sscanf(string(value), "%d", &i)
if err != nil {
return err
}
*t = Time(time.Unix(i, 0))
return nil
}
func (t Time) Unix() int64 {
return time.Time(t).Unix()
}
func (t Time) String() string {
return time.Time(t).Format("2000-01-01 20:41:00")
}
type NodeData interface { type NodeData interface {
json.Marshaler json.Marshaler
json.Unmarshaler json.Unmarshaler
@ -45,17 +21,18 @@ type NodeData interface {
type NodeRecord struct { type NodeRecord struct {
Addr Address // address of node Addr Address // address of node
Url string // Url, used to connect to node Url string // Url, used to connect to node
After Time // next call after time After time.Time // next call after time
Seen Time // last connected at time Seen time.Time // last connected at time
Meta *json.RawMessage // arbitrary metadata saved for a peer Meta *json.RawMessage // arbitrary metadata saved for a peer
node Node node Node
connected bool connected bool
} }
// set checked to current time,
func (self *NodeRecord) setSeen() { func (self *NodeRecord) setSeen() {
self.Seen = Time(time.Now()) t := time.Now()
self.Seen = t
self.After = t
} }
func (self *NodeRecord) String() string { func (self *NodeRecord) String() string {
@ -146,16 +123,19 @@ This is used to pick candidates for live nodes that are most wanted for
a higly connected low centrality network structure for Swarm which best suits a higly connected low centrality network structure for Swarm which best suits
for a Kademlia-style routing. for a Kademlia-style routing.
The candidate is chosen using the following strategy. * Starting as naive node with empty db, this implements Kademlia bootstrapping
* As a mature node, it fills short lines. All on demand.
The candidate is chosen using the following strategy:
We check for missing online nodes in the buckets for 1 upto Max BucketSize rounds. We check for missing online nodes in the buckets for 1 upto Max BucketSize rounds.
On each round we proceed from the low to high proximity order buckets. On each round we proceed from the low to high proximity order buckets.
If the number of active nodes (=connected peers) is < rounds, then start looking If the number of active nodes (=connected peers) is < rounds, then start looking
for a known candidate. To determine if there is a candidate to recommend the for a known candidate. To determine if there is a candidate to recommend the
node record database row corresponding to the bucket is checked. kaddb node record database row corresponding to the bucket is checked.
If the row cursor is on position i, the ith element in the row is chosen. If the row cursor is on position i, the ith element in the row is chosen.
If the record is scheduled not to be retried before NOW, the next element is taken. If the record is scheduled not to be retried before NOW, the next element is taken.
If the record is scheduled can be retried, it is set as checked, scheduled for If the record is scheduled to be retried, it is set as checked, scheduled for
checking and is returned. The time of the next check is in X (duration) such that checking and is returned. The time of the next check is in X (duration) such that
X = ConnRetryExp * delta where delta is the time past since the last check and X = ConnRetryExp * delta where delta is the time past since the last check and
ConnRetryExp is constant obsoletion factor. (Note that when node records are added ConnRetryExp is constant obsoletion factor. (Note that when node records are added
@ -171,103 +151,91 @@ offline past peer)
|| (proxBin(a) < proxBin(b) && |proxBin(a)| == |proxBin(b)|) || (proxBin(a) < proxBin(b) && |proxBin(a)| == |proxBin(b)|)
|| (proxBin(a) == proxBin(b) && lastChecked(a) < lastChecked(b)) || (proxBin(a) == proxBin(b) && lastChecked(a) < lastChecked(b))
This has double role. Starting as naive node with empty db, this implements
Kademlia bootstrapping
As a mature node, it fills short lines. All on demand.
The second argument returned names the first missing slot found The second argument returned names the first missing slot found
*/ */
func (self *KadDb) findBest(bucketSize int, binsize func(int) int) (node *NodeRecord, proxLimit int) { func (self *KadDb) findBest(maxBinSize int, binSize func(int) int) (node *NodeRecord, need bool, proxLimit int) {
// return value -1 indicates that buckets are filled in all // return nil, proxLimit indicates that all buckets are filled
proxLimit = -1
defer self.lock.Unlock() defer self.lock.Unlock()
self.lock.Lock() self.lock.Lock()
var interval time.Duration var interval time.Duration
var found bool var found bool
for rounds := 1; rounds <= bucketSize; rounds++ { var count int
var purge []int
var delta time.Duration
var cursor int
var after time.Time
// iterate over columns maximum bucketsize times
for rounds := 1; rounds <= maxBinSize; rounds++ {
ROUND: ROUND:
// iterate over rows from PO 0 upto MaxProx
for po, dbrow := range self.Nodes { for po, dbrow := range self.Nodes {
if po > len(self.Nodes) { // if row has rounds connected peers, then take the next
break ROUND if binSize(po) >= rounds {
continue ROUND
} }
size := binsize(po) if !need {
if size < rounds { // set proxlimit to the PO where the first missing slot is found
if proxLimit < 0 { proxLimit = po
// set the first missing slot found need = true
proxLimit = po }
cursor = self.cursors[po]
// there is a missing slot - finding a node to connect to
// select a node record from the relavant kaddb row (of identical prox order)
ROW:
for cursor = self.cursors[po]; !found && count < len(dbrow); cursor = (cursor + 1) % len(dbrow) {
count++
node = dbrow[cursor]
// skip already connected nodes
if node.connected {
glog.V(logger.Debug).Infof("[KΛÐ]: kaddb record %v (PO%03d:%d/%d) already connected", node.Addr, po, cursor, len(dbrow))
continue ROW
} }
var count int
var purge []int
n := self.cursors[po]
// try node records in the relavant kaddb row (of identical prox order) // if node is scheduled to connect
// if they are ripe for checking if time.Time(node.After).After(time.Now()) {
ROW: glog.V(logger.Debug).Infof("[KΛÐ]: kaddb record %v (PO%03d:%d) skipped. seen at %v (%v ago), scheduled at %v", node.Addr, po, cursor, node.Seen, delta, node.After)
for count < len(dbrow) { continue ROW
node = dbrow[n]
// skip already connected nodes
if !node.connected {
glog.V(logger.Debug).Infof("[KΛÐ]: kaddb record %v (PO%03d:%d) not to be retried before %v", node.Addr, po, n, node.After)
// time since last known connection attempt
delta := node.After.Unix() - node.Seen.Unix()
// if delta < 4 {
// node.After = Time(time.Time{})
// }
// if node is scheduled to connect
if time.Time(node.After).Before(time.Now()) {
glog.V(logger.Debug).Infof("[KΛÐ]: scheduled at %v, seen at %v", node.After, node.Seen)
// if checked longer than purge interval
period := time.Since(time.Time(node.Seen))
if period > self.purgeInterval {
glog.V(logger.Debug).Infof("[KΛÐ]: expired node last seen %v ago", period)
// delete nodes
purge = append(purge, n)
glog.V(logger.Debug).Infof("[KΛÐ]: deleting inactive node record %v (PO%03d:%d) last check: %v, next check: %v", node.Addr, po, n, node.Seen, node.After)
} else {
// scheduling next check
if (node.After == Time(time.Time{})) {
node.After = Time(time.Now().Add(self.initialRetryInterval))
} else {
interval = time.Duration(delta * int64(self.connRetryExp))
node.After = Time(time.Now().Add(interval))
}
glog.V(logger.Debug).Infof("[KΛÐ]: serve node record %v (PO%03d:%d), last check: %v, next check: %v", node.Addr, po, n, node.Seen, node.After)
}
found = true
break ROW
}
glog.V(logger.Debug).Infof("[KΛÐ]: kaddb record %v (PO%03d:%d) not ready. skipped. not to be retried before: %v", node.Addr, po, n, node.After)
} // if node.node == nil
n++
count++
// cycle: n = n % len(dbrow)
if n >= len(dbrow) {
n = 0
}
} }
self.cursors[po] = n
self.delete(po, purge...) delta = time.Since(time.Time(node.Seen))
if found { if delta < self.initialRetryInterval {
glog.V(logger.Detail).Infof("[KΛÐ]: rounds %d: prox limit: PO%03d\n%v", rounds, proxLimit, node) delta = self.initialRetryInterval
node.setSeen()
return
} }
} // if len < rounds if delta > self.purgeInterval {
} // for po-s // remove node
glog.V(logger.Detail).Infof("[KΛÐ]: rounds %d: proxlimit: PO%03d", rounds, proxLimit) purge = append(purge, cursor)
if proxLimit == 0 || proxLimit < 0 && bucketSize == rounds { glog.V(logger.Debug).Infof("[KΛÐ]: kaddb record %v (PO%03d:%d) unreachable since %v. Removed", node.Addr, po, cursor, node.Seen)
return continue ROW
}
glog.V(logger.Debug).Infof("[KΛÐ]: kaddb record %v (PO%03d:%d) ready to be tried. seen at %v (%v ago), scheduled at %v", node.Addr, po, cursor, node.Seen, delta, node.After)
// scheduling next check
interval = time.Duration(delta * time.Duration(self.connRetryExp))
after = time.Now().Add(interval)
glog.V(logger.Debug).Infof("[KΛÐ]: kaddb record %v (PO%03d:%d) selected as candidate connection %v. seen at %v (%v ago), selectable since %v, retry after %v (in %v)", node.Addr, po, cursor, rounds, node.Seen, delta, node.After, after, interval)
node.After = after
found = true
break ROW
} // ROW
self.cursors[po] = cursor
self.delete(po, purge...)
if found {
return node, true, proxLimit
}
} // ROUND
if need {
return nil, true, proxLimit
} }
} // for round } // ROUNDS
return return nil, false, proxLimit
} }
// deletes the noderecords of a kaddb row corresponding to the indexes // deletes the noderecords of a kaddb row corresponding to the indexes
@ -301,8 +269,8 @@ func (self *KadDb) save(path string, cb func(*NodeRecord, Node)) error {
for _, b := range self.Nodes { for _, b := range self.Nodes {
for _, node := range b { for _, node := range b {
n++ n++
node.After = Time(time.Now()) node.After = time.Now()
node.Seen = Time(time.Now()) node.Seen = time.Now()
if cb != nil { if cb != nil {
cb(node, node.node) cb(node, node.node)
} }
@ -350,8 +318,8 @@ func (self *KadDb) load(path string, cb func(*NodeRecord, Node) error) (err erro
} }
} }
n++ n++
if (node.After == Time(time.Time{})) { if (node.After == time.Time{}) {
node.After = Time(time.Now()) node.After = time.Now()
} }
self.index[node.Addr] = node self.index[node.Addr] = node
} }

View file

@ -279,7 +279,7 @@ func (self *Kademlia) binsize(p int) int {
return len(b.nodes) return len(b.nodes)
} }
func (self *Kademlia) FindBest() (node *NodeRecord, proxLimit int) { func (self *Kademlia) FindBest() (*NodeRecord, bool, int) {
return self.db.findBest(self.BucketSize, self.binsize) return self.db.findBest(self.BucketSize, self.binsize)
} }

View file

@ -2,6 +2,7 @@ package kademlia
import ( import (
"fmt" "fmt"
"math"
"math/rand" "math/rand"
"reflect" "reflect"
"testing" "testing"
@ -11,8 +12,8 @@ import (
var ( var (
quickrand = rand.New(rand.NewSource(time.Now().Unix())) quickrand = rand.New(rand.NewSource(time.Now().Unix()))
quickcfgFindClosest = &quick.Config{MaxCount: 5000, Rand: quickrand} quickcfgFindClosest = &quick.Config{MaxCount: 50, Rand: quickrand}
quickcfgBootStrap = &quick.Config{MaxCount: 1000, Rand: quickrand} quickcfgBootStrap = &quick.Config{MaxCount: 100, Rand: quickrand}
) )
type testNode struct { type testNode struct {
@ -60,43 +61,36 @@ func TestBootstrap(t *testing.T) {
kad := New(test.Self, params) kad := New(test.Self, params)
var err error var err error
addr := RandomAddress() for p := 0; p < 9; p++ {
prox := proximity(addr, test.Self)
for p := 0; p <= prox; p++ {
var nrs []*NodeRecord var nrs []*NodeRecord
for i := 0; i < 3; i++ { n := math.Pow(float64(2), float64(7-p))
for i := 0; i < int(n); i++ {
addr := RandomAddressAt(test.Self, p)
nrs = append(nrs, &NodeRecord{ nrs = append(nrs, &NodeRecord{
Addr: RandomAddressAt(test.Self, p), Addr: addr,
}) })
} }
kad.Add(nrs) kad.Add(nrs)
} }
node := &testNode{addr} node := &testNode{test.Self}
n := 0 n := 0
for n < 100 { for n < 100 {
err = kad.On(node, nil) err = kad.On(node, nil)
if err != nil { if err != nil && err.Error() != "bucket full" {
t.Errorf("backend not accepting node") t.Fatalf("backend not accepting node: %v", err)
return false
} }
var nrs []*NodeRecord
prox := proximity(test.Self, node.addr)
for i := 0; i < 13; i++ {
nrs = append(nrs, &NodeRecord{
Addr: RandomAddressAt(test.Self, prox+1),
})
}
kad.Add(nrs)
record, _ := kad.FindBest() record, need, _ := kad.FindBest()
if record == nil { if !need {
break break
} }
node = &testNode{record.Addr}
n++ n++
if record == nil {
continue
}
node = &testNode{record.Addr}
} }
exp := test.BucketSize * (test.MaxProx + 1) exp := test.BucketSize * (test.MaxProx + 1)
if kad.Count() != exp { if kad.Count() != exp {
@ -116,15 +110,14 @@ func TestFindClosest(t *testing.T) {
test := func(test *FindClosestTest) bool { test := func(test *FindClosestTest) bool {
// for any node kad.le, Target and N // for any node kad.le, Target and N
params := NewKadParams() params := NewKadParams()
params.MaxProx = 10 params.MaxProx = 7
kad := New(test.Self, params) kad := New(test.Self, params)
var err error var err error
// t.Logf("FindClosestTest %v: %v\n", len(test.All), test) // t.Logf("FindClosestTest %v: %v\n", len(test.All), test)
for _, node := range test.All { for _, node := range test.All {
err = kad.On(node, nil) err = kad.On(node, nil)
if err != nil { if err != nil && err.Error() != "bucket full" {
t.Errorf("backend not accepting node") t.Fatalf("backend not accepting node: %v", err)
return false
} }
} }
@ -193,7 +186,7 @@ func TestProxAdjust(t *testing.T) {
r := rand.New(rand.NewSource(time.Now().UnixNano())) r := rand.New(rand.NewSource(time.Now().UnixNano()))
self := gen(Address{}, r).(Address) self := gen(Address{}, r).(Address)
params := NewKadParams() params := NewKadParams()
params.MaxProx = 10 params.MaxProx = 7
kad := New(self, params) kad := New(self, params)
var err error var err error
@ -201,9 +194,8 @@ func TestProxAdjust(t *testing.T) {
a := gen(Address{}, r).(Address) a := gen(Address{}, r).(Address)
addresses = append(addresses, a) addresses = append(addresses, a)
err = kad.On(&testNode{addr: a}, nil) err = kad.On(&testNode{addr: a}, nil)
if err != nil { if err != nil && err.Error() != "bucket full" {
t.Errorf("backend not accepting node") t.Fatalf("backend not accepting node: %v", err)
return
} }
if !kad.proxCheck(t) { if !kad.proxCheck(t) {
return return
@ -229,16 +221,15 @@ func TestSaveLoad(t *testing.T) {
addresses := gen([]Address{}, r).([]Address) addresses := gen([]Address{}, r).([]Address)
self := RandomAddress() self := RandomAddress()
params := NewKadParams() params := NewKadParams()
params.MaxProx = 10 params.MaxProx = 7
kad := New(self, params) kad := New(self, params)
var err error var err error
for _, a := range addresses { for _, a := range addresses {
err = kad.On(&testNode{addr: a}, nil) err = kad.On(&testNode{addr: a}, nil)
if err != nil { if err != nil && err.Error() != "bucket full" {
t.Errorf("backend not accepting node") t.Fatalf("backend not accepting node: %v", err)
return
} }
} }
nodes := kad.FindClosest(self, 100) nodes := kad.FindClosest(self, 100)
@ -255,9 +246,8 @@ func TestSaveLoad(t *testing.T) {
for _, b := range kad.db.Nodes { for _, b := range kad.db.Nodes {
for _, node := range b { for _, node := range b {
err = kad.On(&testNode{node.Addr}, nil) err = kad.On(&testNode{node.Addr}, nil)
if err != nil { if err != nil && err.Error() != "bucket full" {
t.Errorf("backend not accepting node") t.Fatalf("backend not accepting node: %v", err)
return
} }
} }
} }
@ -309,7 +299,7 @@ type bootstrapTest struct {
func (*bootstrapTest) Generate(rand *rand.Rand, size int) reflect.Value { func (*bootstrapTest) Generate(rand *rand.Rand, size int) reflect.Value {
t := &bootstrapTest{ t := &bootstrapTest{
Self: gen(Address{}, rand).(Address), Self: gen(Address{}, rand).(Address),
MaxProx: 10 + rand.Intn(3), MaxProx: 5 + rand.Intn(2),
BucketSize: rand.Intn(3) + 1, BucketSize: rand.Intn(3) + 1,
} }
return reflect.ValueOf(t) return reflect.ValueOf(t)

View file

@ -316,7 +316,7 @@ func (self *syncer) syncState(state *syncState) {
// stop quits both request processor and saves the request cache to disk // stop quits both request processor and saves the request cache to disk
func (self *syncer) stop() { func (self *syncer) stop() {
close(self.quit) close(self.quit)
glog.V(logger.Detail).Infof("[BZZ] syncer[%v]: stopand save sync request db backlog", self.key.Log()) glog.V(logger.Detail).Infof("[BZZ] syncer[%v]: stop and save sync request db backlog", self.key.Log())
for _, db := range self.queues { for _, db := range self.queues {
db.stop() db.stop()
} }