rate limit concurrent peer dials
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Oct 5, 2015 at 23:29 UTC
1c2223dece434650695c155c97ef211776b65af8
2 files changed
+57
-11
p2p/net/swarm/swarm.go
+12
-8
@@ -64,6 +64,9 @@ type Swarm struct {
64
// filters for addresses that shouldnt be dialed
65
Filters *filter.Filters
66
67
+ // file descriptor rate limited
68
+ fdRateLimit chan struct{}
69
+
70
proc goprocess.Process
71
ctx context.Context
72
bwc metrics.Reporter
@@ -79,14 +82,15 @@ func NewSwarm(ctx context.Context, listenAddrs []ma.Multiaddr,
82
}
83
84
s := &Swarm{
82
- swarm: ps.NewSwarm(PSTransport),
83
- local: local,
84
- peers: peers,
85
- ctx: ctx,
86
- dialT: DialTimeout,
87
- notifs: make(map[inet.Notifiee]ps.Notifiee),
88
- bwc: bwc,
89
- Filters: filter.NewFilters(),
85
+ swarm: ps.NewSwarm(PSTransport),
86
+ local: local,
87
+ peers: peers,
88
+ ctx: ctx,
89
+ dialT: DialTimeout,
90
+ notifs: make(map[inet.Notifiee]ps.Notifiee),
91
+ bwc: bwc,
92
+ fdRateLimit: make(chan struct{}, concurrentFdDials),
93
+ Filters: filter.NewFilters(),
94
}
95
96
// configure Swarm
p2p/net/swarm/swarm_dial.go
+45
-3
@@ -44,6 +44,9 @@ var (
44
// add loop back in Dial(.)
45
const dialAttempts = 1
46
47
+// number of concurrent outbound dials over transports that consume file descriptors
48
+const concurrentFdDials = 160
49
+
50
// DialTimeout is the amount of time each dial attempt has. We can think about making
51
// this larger down the road, or putting more granular timeouts (i.e. within each
52
// subcomponent of Dial)
@@ -115,6 +118,7 @@ func (ds *dialsync) Unlock(dst peer.ID) {
118
if !found {
119
panic("called dialDone with no ongoing dials to peer: " + dst.Pretty())
120
}
121
+
122
delete(ds.ongoing, dst) // remove ongoing dial
123
close(wait) // release everyone else
124
ds.lock.Unlock()
@@ -398,7 +402,7 @@ func (s *Swarm) dialAddrs(ctx context.Context, d *conn.Dialer, p peer.ID, remote
402
// to end early.
403
go func() {
404
// rate limiting just in case. at most 10 addrs at once.
401
- limiter := ratelimit.NewRateLimiter(process.Background(), 10)
405
+ limiter := ratelimit.NewRateLimiter(process.Background(), 8)
406
limiter.Go(func(worker process.Process) {
407
// permute addrs so we try different sets first each time.
408
for _, i := range rand.Perm(len(remoteAddrs)) {
@@ -411,9 +415,27 @@ func (s *Swarm) dialAddrs(ctx context.Context, d *conn.Dialer, p peer.ID, remote
415
}
416
417
workerAddr := remoteAddrs[i] // shadow variable to avoid race
414
- limiter.LimitedGo(func(worker process.Process) {
415
- dialSingleAddr(workerAddr)
418
+
419
+ // we have to do the waiting concurrently because there are addrs
420
+ // that SHOULD NOT be rate limited (utp), nor blocked by other
421
+ // rate limited addrs (tcp).
422
+ //
423
+ // (and we need to call `limiter.Go`, instead of `go` as required
424
+ // by goproc/limiter semantics. note: limiter.Go is not LimitedGo.)
425
+ limiter.Go(func(p process.Process) {
426
+
427
+ // returns whatever ratelimiting is acceptable for workerAddr.
428
+ // may not rate limit at all.
429
+ rl := s.addrDialRateLimit(workerAddr)
430
+ rl <- struct{}{}
431
+
432
+ limiter.LimitedGo(func(worker process.Process) {
433
+ dialSingleAddr(workerAddr)
434
+ })
435
+
436
+ <-rl
437
})
438
+
439
}
440
})
441
@@ -491,3 +513,23 @@ func dialConnSetup(ctx context.Context, s *Swarm, connC conn.Conn) (*Conn, error
513
514
return swarmC, err
515
}
516
+
517
+// addrDialRateLimit returns a ratelimiting channel for dialing transport
518
+// addrs like a. for example, tcp is fd-ratelimited. utp is not ratelimited.
519
+func (s *Swarm) addrDialRateLimit(a ma.Multiaddr) chan struct{} {
520
+ if isFDCostlyTransport(a) {
521
+ return s.fdRateLimit
522
+ }
523
+
524
+ // do not rate limit it at all
525
+ return make(chan struct{}, 1)
526
+}
527
+
528
+func isFDCostlyTransport(a ma.Multiaddr) bool {
529
+ return isTCPMultiaddr(a)
530
+}
531
+
532
+func isTCPMultiaddr(a ma.Multiaddr) bool {
533
+ p := a.Protocols()
534
+ return len(p) == 2 && (p[0].Name == "ip4" || p[0].Name == "ip6") && p[1].Name == "tcp"
535
+}