diff --git a/swarm/api/config_test.go b/swarm/api/config_test.go index 9bed72d23d..136967153a 100644 --- a/swarm/api/config_test.go +++ b/swarm/api/config_test.go @@ -24,8 +24,8 @@ var ( "SplitTimeout": 120, "CallInterval": 10000000000, "KadDbPath": "` + filepath.Join("TMPDIR", "0d2f62485607cf38d9d795d93682a517661e513e", "bzz-peers.json") + `", - "MaxProx": 10, - "ProxBinSize": 8, + "MaxProx": 8, + "ProxBinSize": 4, "BucketSize": 3, "PurgeInterval": 151200000000000, "InitialRetryInterval": 4200000000, diff --git a/swarm/cmd/swarm/swarm.sh b/swarm/cmd/swarm/swarm.sh index d150e25597..8546e1b8d8 100644 --- a/swarm/cmd/swarm/swarm.sh +++ b/swarm/cmd/swarm/swarm.sh @@ -104,7 +104,6 @@ function shutdown { while true ;do ps auxwww|grep geth|grep -v grep|awk '{print $2}'|grep -ql $1 || break sleep 1 - # ps auxwww|grep geth|grep -v grep |grep $1 #|awk '{print $2}' done echo "stopped" } @@ -121,8 +120,8 @@ function init { killall geth reset all cluster $* - stop all - cluster $* + enode all + connect all } 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> $enodes + echo "," >> $enodes + fi + done + echo "\"\"]" >> $enodes + cmd=$dir/connect.js + for ((i=0;i> $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> $enodes - for ((i=0;i> $enodes - echo "," >> $enodes - fi - done - echo "\"\"]" >> $enodes - for ((i=0;i 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 // discovery table is used to look it up. connectPeer(node.Url) - } else if proxLimit > -1 { + } else if need { // a random peer is taken from the table peers := self.kad.FindClosest(kademlia.RandomAddressAt(self.addr, rand.Intn(self.kad.MaxProx)), 1) if len(peers) > 0 { @@ -138,15 +140,18 @@ func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee req := &retrieveRequestMsgData{ 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) + } else { + glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: no peer") } 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 { + glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: no need for more bees") 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 @@ -165,6 +170,7 @@ func (self *Hive) keepAlive() { if self.kad.DBCount() > 0 { select { case self.more <- true: + glog.V(logger.Debug).Infof("[BZZ] KΛÐΞMLIΛ hive: buzz wakeup") default: } } @@ -241,7 +247,7 @@ func (self *Hive) DropAll() { // contructor for kademlia.NodeRecord based on peer address alone // TODO: should go away and only addr passed to kademlia func newNodeRecord(addr *peerAddr) *kademlia.NodeRecord { - now := kademlia.Time(time.Now()) + now := time.Now() return &kademlia.NodeRecord{ Addr: addr.Addr, Url: addr.String(), @@ -336,7 +342,7 @@ func (self *Hive) peers(req *retrieveRequestMsgData) { for _, peer := range self.getPeers(key, int(req.MaxPeers)) { 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{ Peers: addrs, diff --git a/swarm/network/kademlia/kaddb.go b/swarm/network/kademlia/kaddb.go index cf2a591714..22adb2cbb8 100644 --- a/swarm/network/kademlia/kaddb.go +++ b/swarm/network/kademlia/kaddb.go @@ -12,30 +12,6 @@ import ( "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 { json.Marshaler json.Unmarshaler @@ -45,17 +21,18 @@ type NodeData interface { type NodeRecord struct { Addr Address // address of node Url string // Url, used to connect to node - After Time // next call after time - Seen Time // last connected at time + After time.Time // next call after time + Seen time.Time // last connected at time Meta *json.RawMessage // arbitrary metadata saved for a peer node Node connected bool } -// set checked to current time, func (self *NodeRecord) setSeen() { - self.Seen = Time(time.Now()) + t := time.Now() + self.Seen = t + self.After = t } 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 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. 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 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 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 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 @@ -171,103 +151,91 @@ offline past peer) || (proxBin(a) < proxBin(b) && |proxBin(a)| == |proxBin(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 */ -func (self *KadDb) findBest(bucketSize int, binsize func(int) int) (node *NodeRecord, proxLimit int) { - // return value -1 indicates that buckets are filled in all - proxLimit = -1 +func (self *KadDb) findBest(maxBinSize int, binSize func(int) int) (node *NodeRecord, need bool, proxLimit int) { + // return nil, proxLimit indicates that all buckets are filled defer self.lock.Unlock() self.lock.Lock() var interval time.Duration 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: + // iterate over rows from PO 0 upto MaxProx for po, dbrow := range self.Nodes { - if po > len(self.Nodes) { - break ROUND + // if row has rounds connected peers, then take the next + if binSize(po) >= rounds { + continue ROUND } - size := binsize(po) - if size < rounds { - if proxLimit < 0 { - // set the first missing slot found - proxLimit = po + if !need { + // set proxlimit to the PO where the first missing slot is found + proxLimit = po + need = true + } + 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 they are ripe for checking - ROW: - for count < len(dbrow) { - 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 - } + // if node is scheduled to connect + if time.Time(node.After).After(time.Now()) { + 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) + continue ROW } - self.cursors[po] = n - self.delete(po, purge...) - if found { - glog.V(logger.Detail).Infof("[KΛÐ]: rounds %d: prox limit: PO%03d\n%v", rounds, proxLimit, node) - node.setSeen() - return + + delta = time.Since(time.Time(node.Seen)) + if delta < self.initialRetryInterval { + delta = self.initialRetryInterval } - } // if len < rounds - } // for po-s - glog.V(logger.Detail).Infof("[KΛÐ]: rounds %d: proxlimit: PO%03d", rounds, proxLimit) - if proxLimit == 0 || proxLimit < 0 && bucketSize == rounds { - return + if delta > self.purgeInterval { + // remove node + purge = append(purge, cursor) + glog.V(logger.Debug).Infof("[KΛÐ]: kaddb record %v (PO%03d:%d) unreachable since %v. Removed", node.Addr, po, cursor, node.Seen) + 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 @@ -301,8 +269,8 @@ func (self *KadDb) save(path string, cb func(*NodeRecord, Node)) error { for _, b := range self.Nodes { for _, node := range b { n++ - node.After = Time(time.Now()) - node.Seen = Time(time.Now()) + node.After = time.Now() + node.Seen = time.Now() if cb != nil { cb(node, node.node) } @@ -350,8 +318,8 @@ func (self *KadDb) load(path string, cb func(*NodeRecord, Node) error) (err erro } } n++ - if (node.After == Time(time.Time{})) { - node.After = Time(time.Now()) + if (node.After == time.Time{}) { + node.After = time.Now() } self.index[node.Addr] = node } diff --git a/swarm/network/kademlia/kademlia.go b/swarm/network/kademlia/kademlia.go index 250dbfd9a6..50fb6b6f2c 100644 --- a/swarm/network/kademlia/kademlia.go +++ b/swarm/network/kademlia/kademlia.go @@ -279,7 +279,7 @@ func (self *Kademlia) binsize(p int) int { 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) } diff --git a/swarm/network/kademlia/kademlia_test.go b/swarm/network/kademlia/kademlia_test.go index f36158c378..ca05e1ded5 100644 --- a/swarm/network/kademlia/kademlia_test.go +++ b/swarm/network/kademlia/kademlia_test.go @@ -2,6 +2,7 @@ package kademlia import ( "fmt" + "math" "math/rand" "reflect" "testing" @@ -11,8 +12,8 @@ import ( var ( quickrand = rand.New(rand.NewSource(time.Now().Unix())) - quickcfgFindClosest = &quick.Config{MaxCount: 5000, Rand: quickrand} - quickcfgBootStrap = &quick.Config{MaxCount: 1000, Rand: quickrand} + quickcfgFindClosest = &quick.Config{MaxCount: 50, Rand: quickrand} + quickcfgBootStrap = &quick.Config{MaxCount: 100, Rand: quickrand} ) type testNode struct { @@ -60,43 +61,36 @@ func TestBootstrap(t *testing.T) { kad := New(test.Self, params) var err error - addr := RandomAddress() - prox := proximity(addr, test.Self) - - for p := 0; p <= prox; p++ { + for p := 0; p < 9; p++ { 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{ - Addr: RandomAddressAt(test.Self, p), + Addr: addr, }) } kad.Add(nrs) } - node := &testNode{addr} + node := &testNode{test.Self} n := 0 for n < 100 { err = kad.On(node, nil) - if err != nil { - t.Errorf("backend not accepting node") - return false + if err != nil && err.Error() != "bucket full" { + t.Fatalf("backend not accepting node: %v", err) } - 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() - if record == nil { + record, need, _ := kad.FindBest() + if !need { break } - node = &testNode{record.Addr} n++ + if record == nil { + continue + } + node = &testNode{record.Addr} } exp := test.BucketSize * (test.MaxProx + 1) if kad.Count() != exp { @@ -116,15 +110,14 @@ func TestFindClosest(t *testing.T) { test := func(test *FindClosestTest) bool { // for any node kad.le, Target and N params := NewKadParams() - params.MaxProx = 10 + params.MaxProx = 7 kad := New(test.Self, params) var err error // t.Logf("FindClosestTest %v: %v\n", len(test.All), test) for _, node := range test.All { err = kad.On(node, nil) - if err != nil { - t.Errorf("backend not accepting node") - return false + if err != nil && err.Error() != "bucket full" { + t.Fatalf("backend not accepting node: %v", err) } } @@ -193,7 +186,7 @@ func TestProxAdjust(t *testing.T) { r := rand.New(rand.NewSource(time.Now().UnixNano())) self := gen(Address{}, r).(Address) params := NewKadParams() - params.MaxProx = 10 + params.MaxProx = 7 kad := New(self, params) var err error @@ -201,9 +194,8 @@ func TestProxAdjust(t *testing.T) { a := gen(Address{}, r).(Address) addresses = append(addresses, a) err = kad.On(&testNode{addr: a}, nil) - if err != nil { - t.Errorf("backend not accepting node") - return + if err != nil && err.Error() != "bucket full" { + t.Fatalf("backend not accepting node: %v", err) } if !kad.proxCheck(t) { return @@ -229,16 +221,15 @@ func TestSaveLoad(t *testing.T) { addresses := gen([]Address{}, r).([]Address) self := RandomAddress() params := NewKadParams() - params.MaxProx = 10 + params.MaxProx = 7 kad := New(self, params) var err error for _, a := range addresses { err = kad.On(&testNode{addr: a}, nil) - if err != nil { - t.Errorf("backend not accepting node") - return + if err != nil && err.Error() != "bucket full" { + t.Fatalf("backend not accepting node: %v", err) } } nodes := kad.FindClosest(self, 100) @@ -255,9 +246,8 @@ func TestSaveLoad(t *testing.T) { for _, b := range kad.db.Nodes { for _, node := range b { err = kad.On(&testNode{node.Addr}, nil) - if err != nil { - t.Errorf("backend not accepting node") - return + if err != nil && err.Error() != "bucket full" { + t.Fatalf("backend not accepting node: %v", err) } } } @@ -309,7 +299,7 @@ type bootstrapTest struct { func (*bootstrapTest) Generate(rand *rand.Rand, size int) reflect.Value { t := &bootstrapTest{ Self: gen(Address{}, rand).(Address), - MaxProx: 10 + rand.Intn(3), + MaxProx: 5 + rand.Intn(2), BucketSize: rand.Intn(3) + 1, } return reflect.ValueOf(t) diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index c6c51934e6..80893336b7 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -316,7 +316,7 @@ func (self *syncer) syncState(state *syncState) { // stop quits both request processor and saves the request cache to disk func (self *syncer) stop() { 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 { db.stop() }