simplify rate limiting to remove potential panic
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Oct 14, 2015 at 12:20 UTC
663a0309e58c6a60184f8080f416c6c4dd19bd34
1 file changed
+48
-55
p2p/net/swarm/swarm_dial.go
+48
-55
@@ -16,9 +16,6 @@ import (
16
17
ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
18
manet "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
19
- process "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
20
- processctx "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
21
- ratelimit "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/ratelimit"
19
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
20
)
21
@@ -371,87 +368,83 @@ func (s *Swarm) dialAddrs(ctx context.Context, d *conn.Dialer, p peer.ID, remote
368
ctx, cancel := context.WithCancel(ctx)
369
defer cancel() // cancel work when we exit func
370
374
- foundConn := make(chan struct{})
375
- conns := make(chan conn.Conn, len(remoteAddrs))
371
+ conns := make(chan conn.Conn)
372
errs := make(chan error, len(remoteAddrs))
373
374
// dialSingleAddr is used in the rate-limited async thing below.
375
dialSingleAddr := func(addr ma.Multiaddr) {
376
+ // rebind chans in scope so we can nil them out easily
377
+ connsout := conns
378
+ errsout := errs
379
+
380
connC, err := s.dialAddr(ctx, d, p, addr)
381
+ if err != nil {
382
+ connsout = nil
383
+ } else if connC == nil {
384
+ // NOTE: this really should never happen
385
+ log.Errorf("failed to dial %s %s and got no error!", p, addr)
386
+ err = fmt.Errorf("failed to dial %s %s", p, addr)
387
+ connsout = nil
388
+ } else {
389
+ errsout = nil
390
+ }
391
392
// check parent still wants our results
393
select {
384
- case <-foundConn:
394
+ case <-ctx.Done():
395
if connC != nil {
396
connC.Close()
397
}
388
- return
389
- default:
390
- }
391
-
392
- if err != nil {
393
- errs <- err
394
- } else if connC == nil {
395
- errs <- fmt.Errorf("failed to dial %s %s", p, addr)
396
- } else {
397
- conns <- connC
398
+ case errsout <- err:
399
+ case connsout <- connC:
400
}
401
}
402
403
// this whole thing is in a goroutine so we can use foundConn
404
// to end early.
405
go func() {
404
- // rate limiting just in case. at most 10 addrs at once.
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)) {
409
- select {
410
- case <-foundConn: // if one of them succeeded already
411
- break
412
- case <-worker.Closing(): // our context was cancelled
413
- break
414
- default:
415
- }
416
-
417
- workerAddr := remoteAddrs[i] // shadow variable to avoid race
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
- })
406
+ limiter := make(chan struct{}, 8)
407
+ // permute addrs so we try different sets first each time.
408
+ for _, i := range rand.Perm(len(remoteAddrs)) {
409
+
410
+ addr := remoteAddrs[i]
411
+ // returns whatever ratelimiting is acceptable for workerAddr.
412
+ // may not rate limit at all.
413
+ rl := s.addrDialRateLimit(addr)
414
+ select {
415
+ case <-ctx.Done(): // our context was cancelled
416
+ return
417
+ case rl <- struct{}{}:
418
+ // take the token, move on
419
+ }
420
421
+ select {
422
+ case <-ctx.Done(): // our context was cancelled
423
+ return
424
+ case limiter <- struct{}{}:
425
+ // take the token, move on
426
}
440
- })
427
442
- processctx.CloseAfterContext(limiter, ctx)
428
+ go func(rlc <-chan struct{}, a ma.Multiaddr) {
429
+ dialSingleAddr(a)
430
+ <-limiter
431
+ <-rlc
432
+ }(rl, addr)
433
+ }
434
}()
435
445
- // wair fot the results.
436
+ // wair for the results.
437
exitErr := fmt.Errorf("failed to dial %s", p)
447
- for i := 0; i < len(remoteAddrs); i++ {
438
+ for range remoteAddrs {
439
select {
440
case exitErr = <-errs: //
441
log.Debug("dial error: ", exitErr)
442
case connC := <-conns:
443
// take the first + return asap
453
- close(foundConn)
444
return connC, nil
445
+ case <-ctx.Done():
446
+ // break out and return error
447
+ break
448
}
449
}
450
return nil, exitErr