rate limit concurrent peer dials

whyrusleeping · whyrusleeping · commit 82084fe0286c · 2015-10-05T23:29:35.000-07:00
License: MIT
Signed-off-by: Jeromy &lt;jeromyj@gmail.com&gt;
diff --git a/p2p/net/swarm/swarm_dial.go b/p2p/net/swarm/swarm_dial.go
@@ -44,6 +44,9 @@ var (
 // add loop back in Dial(.)
 const dialAttempts = 1
 
+// dial at most 12 peers concurrently
+const concurrentPeerDials = 12
+
 // DialTimeout is the amount of time each dial attempt has. We can think about making
 // this larger down the road, or putting more granular timeouts (i.e. within each
 // subcomponent of Dial)
@@ -77,6 +80,8 @@ type dialsync struct {
 	// this way, we dont kick off N dials simultaneously.
 	ongoing map[peer.ID]chan struct{}
 	lock    sync.Mutex
+
+	ratelimit chan struct{}
 }
 
 // Lock governs the beginning of a dial attempt.
@@ -92,6 +97,7 @@ func (ds *dialsync) Lock(dst peer.ID) (bool, chan struct{}) {
 	ds.lock.Lock()
 	if ds.ongoing == nil { // init if not ready
 		ds.ongoing = make(map[peer.ID]chan struct{})
+		ds.ratelimit = make(chan struct{}, concurrentPeerDials)
 	}
 	wait, found := ds.ongoing[dst]
 	if !found {
@@ -103,6 +109,9 @@ func (ds *dialsync) Lock(dst peer.ID) (bool, chan struct{}) {
 		return false, wait
 	}
 
+	// consume rate limiting token
+	ds.ratelimit <- struct{}{}
+
 	// ok! you're signed up to dial!
 	return true, nil
 }
@@ -115,6 +124,10 @@ func (ds *dialsync) Unlock(dst peer.ID) {
 	if !found {
 		panic("called dialDone with no ongoing dials to peer: " + dst.Pretty())
 	}
+
+	// replace token for rate limiting
+	<-ds.ratelimit
+
 	delete(ds.ongoing, dst) // remove ongoing dial
 	close(wait)             // release everyone else
 	ds.lock.Unlock()
