From 8228242eedda03c548bc5e0432253dbc12ae0988 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Tue, 15 May 2018 23:57:00 -0500 Subject: [PATCH] misc dht and tree changes to stabilize coords and bootstrap the dht faster --- misc/sim/treesim.go | 13 +++++----- src/yggdrasil/dht.go | 57 +++++++++++++++++++++++++++++++---------- src/yggdrasil/router.go | 2 +- src/yggdrasil/switch.go | 12 ++++----- 4 files changed, 57 insertions(+), 27 deletions(-) diff --git a/misc/sim/treesim.go b/misc/sim/treesim.go index f2b2d0e1..91aa2234 100644 --- a/misc/sim/treesim.go +++ b/misc/sim/treesim.go @@ -6,6 +6,7 @@ import "os" import "strings" import "strconv" import "time" +import "log" import "runtime/pprof" import "flag" @@ -418,12 +419,12 @@ func main() { idxstore := loadGraph("misc/sim/fc00-2017-08-12.txt") //idxstore := loadGraph("skitter") kstore := getKeyedStore(idxstore) - /* - for _, n := range kstore { - log := n.core.DEBUG_getLogger() - log.SetOutput(os.Stderr) - } - */ + //* + logger := log.New(os.Stderr, "", log.Flags()) + for _, n := range kstore { + n.core.DEBUG_setLogger(logger) + } + //*/ startNetwork(kstore) //time.Sleep(10*time.Second) // Note that testPaths only works if pressure is turend off diff --git a/src/yggdrasil/dht.go b/src/yggdrasil/dht.go index ec4e33f3..2f196380 100644 --- a/src/yggdrasil/dht.go +++ b/src/yggdrasil/dht.go @@ -25,9 +25,9 @@ import "time" // Maximum size for buckets and lookups // Exception for buckets if the next one is non-full -const dht_bucket_size = 2 // This should be at least 2 -const dht_lookup_size = 2 // This should be at least 1, below 2 is impractical const dht_bucket_number = 8 * NodeIDLen // This shouldn't be changed +const dht_bucket_size = 2 // This should be at least 2 +const dht_lookup_size = 16 // This should be at least 1, below 2 is impractical type dhtInfo struct { nodeID_hidden *NodeID @@ -141,21 +141,14 @@ func (t *dht) handleRes(res *dhtRes) { if res.dest == *rinfo.getNodeID() { return } // No infinite recursions - // ping the nodes we were told about if len(res.infos) > dht_lookup_size { // Ignore any "extra" lookup results res.infos = res.infos[:dht_lookup_size] } for _, info := range res.infos { - bidx, isOK := t.getBucketIndex(info.getNodeID()) - if !isOK { - continue + if dht_firstCloserThanThird(info.getNodeID(), &res.dest, rinfo.getNodeID()) { + t.addToMill(info, info.getNodeID()) } - b := t.getBucket(bidx) - if b.contains(info) { - continue - } // wait for maintenance cycle to get them - t.addToMill(info, info.getNodeID()) } } @@ -167,7 +160,7 @@ func (t *dht) lookup(nodeID *NodeID) []*dhtInfo { if info == nil { panic("Should never happen!") } - if true || dht_firstCloserThanThird(info.getNodeID(), nodeID, &t.nodeID) { + if dht_firstCloserThanThird(info.getNodeID(), nodeID, &t.nodeID) { res = append(res, info) } } @@ -467,10 +460,43 @@ func (t *dht) doMaintenance() { } t.offset++ } - if len(t.rumorMill) > 0 { + for len(t.rumorMill) > 0 { var rumor dht_rumor rumor, t.rumorMill = t.rumorMill[0], t.rumorMill[1:] + if rumor.target == rumor.info.getNodeID() { + // Note that the above is a pointer comparison, and target can be nil + // This is only for adding new nodes (learned from other lookups) + // It only makes sense to ping if the node isn't already in the table + bidx, isOK := t.getBucketIndex(rumor.info.getNodeID()) + if !isOK { + continue + } + b := t.getBucket(bidx) + if b.contains(rumor.info) { + // Already know about this node + continue + } + // This is a good spot to check if a node is worth pinging + add := len(b.other) < dht_bucket_size + if bidx+1 == t.nBuckets() { + add = true + } + bnext := t.getBucket(bidx + 1) + if len(bnext.other) < dht_bucket_size { + add = true + } + for _, info := range b.other { + if dht_firstCloserThanThird(rumor.info.getNodeID(), &t.nodeID, info.getNodeID()) { + // Add the node if they are closer to us than someone in the same bucket + add = true + } + } + if !add { + continue + } + } t.ping(rumor.info, rumor.target) + break } } @@ -488,8 +514,11 @@ func dht_firstCloserThanThird(first *NodeID, return false } -func (t *dht) resetPeers() { +func (t *dht) reset() { + // This is mostly so bootstrapping will reset to resend coords into the network for _, b := range t.buckets_hidden { b.peers = b.peers[:0] + b.other = b.other[:0] } + t.offset = 0 } diff --git a/src/yggdrasil/router.go b/src/yggdrasil/router.go index a8721f22..c77c14d3 100644 --- a/src/yggdrasil/router.go +++ b/src/yggdrasil/router.go @@ -80,7 +80,7 @@ func (r *router) mainLoop() { r.core.dht.insertIfNew(info, true) case <-r.reset: r.core.sessions.resetInits() - r.core.dht.resetPeers() + r.core.dht.reset() case <-ticker.C: { // Any periodic maintenance stuff goes here diff --git a/src/yggdrasil/switch.go b/src/yggdrasil/switch.go index a297e893..d1750f00 100644 --- a/src/yggdrasil/switch.go +++ b/src/yggdrasil/switch.go @@ -127,9 +127,9 @@ type switchMessage struct { type switchPort uint64 type tableElem struct { - port switchPort - firstSeen time.Time - locator switchLocator + port switchPort + //firstSeen time.Time + locator switchLocator } type lookupTable struct { @@ -304,7 +304,7 @@ func (t *switchTable) handleMessage(msg *switchMessage, fromPort switchPort, sig doUpdate := false if !equiv(&msg.locator, &oldSender.locator) { doUpdate = true - sender.firstSeen = now + //sender.firstSeen = now } t.data.peers[fromPort] = sender updateRoot := false @@ -355,7 +355,7 @@ func (t *switchTable) handleMessage(msg *switchMessage, fromPort switchPort, sig case t.core.router.reset <- struct{}{}: default: } - //t.core.log.Println("Switch update:", msg.Locator.Root, msg.Locator.Tstamp, msg.Locator.Coords) + //t.core.log.Println("Switch update:", msg.locator.root, msg.locator.tstamp, msg.locator.coords) //fmt.Println("Switch update:", msg.Locator.Root, msg.Locator.Tstamp, msg.Locator.Coords) } if t.data.locator.tstamp != msg.locator.tstamp { @@ -397,7 +397,7 @@ func (t *switchTable) updateTable() { newTable.elems = append(newTable.elems, tableElem{ locator: loc, //degree: pinfo.degree, - firstSeen: pinfo.firstSeen, + //firstSeen: pinfo.firstSeen, //forward: pinfo.forward, port: pinfo.port, })