p2p/net/swarm: rate limit dials. max of 10 addrs at a time.
This will mitigate the fd explosion, but slow down dials majorly as any peer with more addresses than the rate limit will have to wait a whole dial timeout (~15s)
Juan Batiz-Benet committed
Feb 2, 2015 at 12:01 UTC
793048d310938f14e898bcbae4cd2d63017ad422
1 file changed
+44
-24
p2p/net/swarm/swarm_dial.go
+44
-24
@@ -15,6 +15,9 @@ import (
15
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
16
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
17
manet "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
18
+ process "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
19
+ procctx "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
20
+ ratelimit "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/ratelimit"
21
)
22
23
// Diagram of dial sync:
@@ -353,43 +356,60 @@ func (s *Swarm) dialAddrs(ctx context.Context, d *conn.Dialer, p peer.ID, remote
356
conns := make(chan conn.Conn, len(remoteAddrs))
357
errs := make(chan error, len(remoteAddrs))
358
356
- //TODO: rate limiting just in case?
357
- for _, addr := range remoteAddrs {
358
- go func(addr ma.Multiaddr) {
359
- connC, err := s.dialAddr(ctx, d, p, addr)
359
+ // dialSingleAddr is used in the rate-limited async thing below.
360
+ dialSingleAddr := func(addr ma.Multiaddr) {
361
+ connC, err := s.dialAddr(ctx, d, p, addr)
362
361
- // check parent still wants our results
362
- select {
363
- case <-foundConn:
364
- if connC != nil {
365
- connC.Close()
366
- }
367
- return
368
- default:
363
+ // check parent still wants our results
364
+ select {
365
+ case <-foundConn:
366
+ if connC != nil {
367
+ connC.Close()
368
}
369
+ return
370
+ default:
371
+ }
372
371
- if err != nil {
372
- errs <- err
373
- } else if connC == nil {
374
- errs <- fmt.Errorf("failed to dial %s %s", p, addr)
375
- } else {
376
- conns <- connC
377
- }
378
- }(addr)
373
+ if err != nil {
374
+ errs <- err
375
+ } else if connC == nil {
376
+ errs <- fmt.Errorf("failed to dial %s %s", p, addr)
377
+ } else {
378
+ conns <- connC
379
+ }
380
}
381
381
- err := fmt.Errorf("failed to dial %s", p)
382
+ // this whole thing is in a goroutine so we can use foundConn
383
+ // to end early.
384
+ go func() {
385
+ // rate limiting just in case. at most 10 addrs at once.
386
+ limiter := ratelimit.NewRateLimiter(procctx.WithContext(ctx), 10)
387
+ for _, addr := range remoteAddrs {
388
+ select {
389
+ case <-foundConn: // if one of them succeeded already
390
+ break
391
+ default:
392
+ }
393
+ workerAddr := addr // shadow variable to avoid race
394
+ limiter.Go(func(worker process.Process) {
395
+ dialSingleAddr(workerAddr)
396
+ })
397
+ }
398
+ }()
399
+
400
+ // wair fot the results.
401
+ exitErr := fmt.Errorf("failed to dial %s", p)
402
for i := 0; i < len(remoteAddrs); i++ {
403
select {
384
- case err = <-errs:
385
- log.Debug(err)
404
+ case exitErr = <-errs: //
405
+ log.Debug(exitErr)
406
case connC := <-conns:
407
// take the first + return asap
408
close(foundConn)
409
return connC, nil
410
}
411
}
392
- return nil, err
412
+ return nil, exitErr
413
}
414
415
func (s *Swarm) dialAddr(ctx context.Context, d *conn.Dialer, p peer.ID, addr ma.Multiaddr) (conn.Conn, error) {