ratelimiter: fixing rate limiter use
Use of the ratelimiter should be conscious of the ratelimiter's potential closing. any loops that add work to ratelimiter should (a) only do so if the rate limiter is not closed, or (b) prevent limiter while work is added (i.e. use limiter.Go(addWorkHere))
Juan Batiz-Benet committed
Feb 6, 2015 at 10:59 UTC
91a79bc203218b76a9ac8c3e4ca6d56eb39875f5
3 files changed
+39
-28
blockservice/worker/worker.go
+3
-4
@@ -117,11 +117,10 @@ func (w *Worker) start(c Config) {
117
}
118
})
119
120
- // reads from |workerChan| until process closes
121
- w.process.Go(func(proc process.Process) {
120
+ // reads from |workerChan| until w.process closes
121
+ limiter := ratelimit.NewRateLimiter(w.process, c.NumWorkers)
122
+ limiter.Go(func(proc process.Process) {
123
ctx := waitable.Context(proc) // shut down in-progress HasBlock when time to die
123
- limiter := ratelimit.NewRateLimiter(process.Background(), c.NumWorkers)
124
- defer limiter.Close()
124
for {
125
select {
126
case <-proc.Closing():
p2p/net/swarm/swarm_dial.go
+16
-13
@@ -385,20 +385,23 @@ func (s *Swarm) dialAddrs(ctx context.Context, d *conn.Dialer, p peer.ID, remote
385
go func() {
386
// rate limiting just in case. at most 10 addrs at once.
387
limiter := ratelimit.NewRateLimiter(procctx.WithContext(ctx), 10)
388
-
389
- // permute addrs so we try different sets first each time.
390
- for _, i := range rand.Perm(len(remoteAddrs)) {
391
- select {
392
- case <-foundConn: // if one of them succeeded already
393
- break
394
- default:
388
+ limiter.Go(func(worker process.Process) {
389
+ // permute addrs so we try different sets first each time.
390
+ for _, i := range rand.Perm(len(remoteAddrs)) {
391
+ select {
392
+ case <-foundConn: // if one of them succeeded already
393
+ break
394
+ case <-worker.Closing(): // our context was cancelled
395
+ break
396
+ default:
397
+ }
398
+
399
+ workerAddr := remoteAddrs[i] // shadow variable to avoid race
400
+ limiter.LimitedGo(func(worker process.Process) {
401
+ dialSingleAddr(workerAddr)
402
+ })
403
}
396
-
397
- workerAddr := remoteAddrs[i] // shadow variable to avoid race
398
- limiter.Go(func(worker process.Process) {
399
- dialSingleAddr(workerAddr)
400
- })
401
- }
404
+ })
405
}()
406
407
// wair fot the results.
thirdparty/notifier/notifier.go
+20
-11
@@ -120,18 +120,27 @@ func (n *Notifier) StopNotify(e Notifiee) {
120
// hooks into your object that block you accidentally.
121
func (n *Notifier) NotifyAll(notify func(Notifiee)) {
122
n.mu.Lock()
123
- if n.nots != nil { // so that zero-value is ready to be used.
124
- for notifiee := range n.nots {
123
+ defer n.mu.Unlock()
124
+
125
+ if n.nots == nil { // so that zero-value is ready to be used.
126
+ return
127
+ }
128
126
- if n.lim == nil { // no rate limit
127
- go notify(notifiee)
128
- } else {
129
- notifiee := notifiee // rebind for data races
130
- n.lim.LimitedGo(func(worker process.Process) {
131
- notify(notifiee)
132
- })
133
- }
129
+ // no rate limiting.
130
+ if n.lim == nil {
131
+ for notifiee := range n.nots {
132
+ go notify(notifiee)
133
}
134
+ return
135
}
136
- n.mu.Unlock()
136
+
137
+ // with rate limiting.
138
+ n.lim.Go(func(worker process.Process) {
139
+ for notifiee := range n.nots {
140
+ notifiee := notifiee // rebind for loop data races
141
+ n.lim.LimitedGo(func(worker process.Process) {
142
+ notify(notifiee)
143
+ })
144
+ }
145
+ })
146
}