Merge pull request #18 from Arceliar/rumormill

Add a rumor mill to throttle dht maintenance traffic

misc/sim/treesim.go

@@ -26,6 +26,7 @@ func (n *Node) init(index int) {
 	n.core.Init()
 	n.send = n.core.DEBUG_getSend()
 	n.recv = n.core.DEBUG_getRecv()
+	n.core.DEBUG_simFixMTU()
 }
 
 func (n *Node) printTraffic() {
@@ -424,7 +425,19 @@ func main() {
 	  }
 	*/
 	startNetwork(kstore)
-	if true {
+	//time.Sleep(10*time.Second)
+	// Note that testPaths only works if pressure is turend off
+	//  Otherwise congestion can lead to routing loops?
+	for finished := false; !finished; {
+		finished = testPaths(kstore)
+	}
+	pingNodes(kstore)
+	//pingBench(kstore) // Only after disabling debug output
+	//stressTest(kstore)
+	//time.Sleep(120*time.Second)
+	dumpDHTSize(kstore) // note that this uses racey functions to read things...
+	if false {
+		// This connects the sim to the local network
 		for _, node := range kstore {
 			node.startUDP("localhost:0")
 			node.connectUDP("localhost:12345")
@@ -440,15 +453,4 @@ func main() {
 		var block chan struct{}
 		<-block
 	}
-	//time.Sleep(10*time.Second)
-	// Note that testPaths only works if pressure is turend off
-	//  Otherwise congestion can lead to routing loops?
-	for finished := false; !finished; {
-		finished = testPaths(kstore)
-	}
-	pingNodes(kstore)
-	//pingBench(kstore) // Only after disabling debug output
-	//stressTest(kstore)
-	//time.Sleep(120*time.Second)
-	dumpDHTSize(kstore) // note that this uses racey functions to read things...
 }

src/yggdrasil/debug.go

@@ -369,3 +369,7 @@ func DEBUG_simLinkPeers(p, q *peer) {
 	go p.linkLoop(plinkIn)
 	go q.linkLoop(qlinkIn)
 }
+
+func (c *Core) DEBUG_simFixMTU() {
+	c.tun.mtu = 65535
+}

src/yggdrasil/dht.go

@@ -62,6 +62,11 @@ type dhtRes struct {
 	infos  []*dhtInfo // response
 }
 
+type dht_rumor struct {
+	info   *dhtInfo
+	target *NodeID
+}
+
 type dht struct {
 	core           *Core
 	nodeID         NodeID
@@ -69,6 +74,7 @@ type dht struct {
 	peers          chan *dhtInfo             // other goroutines put incoming dht updates here
 	reqs           map[boxPubKey]map[NodeID]time.Time
 	offset         int
+	rumorMill      []dht_rumor
 }
 
 func (t *dht) init(c *Core) {
@@ -143,7 +149,7 @@ func (t *dht) handleRes(res *dhtRes) {
 		if b.contains(info) {
 			continue
 		} // wait for maintenance cycle to get them
-		t.ping(info, info.getNodeID())
+		t.addToMill(info, info.getNodeID())
 	}
 }
 
@@ -379,6 +385,14 @@ func (t *dht) ping(info *dhtInfo, target *NodeID) {
 	t.sendReq(&req, info)
 }
 
+func (t *dht) addToMill(info *dhtInfo, target *NodeID) {
+	rumor := dht_rumor{
+		info:   info,
+		target: target,
+	}
+	t.rumorMill = append(t.rumorMill, rumor)
+}
+
 func (t *dht) doMaintenance() {
 	// First clean up reqs
 	for key, reqs := range t.reqs {
@@ -391,6 +405,7 @@ func (t *dht) doMaintenance() {
 			delete(t.reqs, key)
 		}
 	}
+	if len(t.rumorMill) == 0 {
 		// Ping the least recently contacted node
 		//  This is to make sure we eventually notice when someone times out
 		var oldest *dhtInfo
@@ -409,7 +424,7 @@ func (t *dht) doMaintenance() {
 			}
 		}
 		if oldest != nil {
-		t.ping(oldest, nil)
+			t.addToMill(oldest, nil)
 		} // if the DHT isn't empty
 		// Refresh buckets
 		if t.offset > last {
@@ -417,11 +432,17 @@ func (t *dht) doMaintenance() {
 		}
 		target := t.getTarget(t.offset)
 		for _, info := range t.lookup(target) {
-		t.ping(info, target)
+			t.addToMill(info, target)
 			break
 		}
 		t.offset++
 	}
+	if len(t.rumorMill) > 0 {
+		var rumor dht_rumor
+		rumor, t.rumorMill = t.rumorMill[0], t.rumorMill[1:]
+		t.ping(rumor.info, rumor.target)
+	}
+}
 
 func dht_firstCloserThanThird(first *NodeID,
 	second *NodeID,