A better throttler?

In a previous post I wrote a simple throttler using channels. On karlseguin’s suggestion, I reimplemented it using sync.Cond. You can check it out at https://bitbucket.org/utils/sync2.

Throttler allows you to restrict a goroutine to a max number of concurrent executions. Think of it as a mutex which allows ‘n’ number of locks to be acquired.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
package sync2

import (
    "sync"
)

//Throttler allows you to restrict a goroutine to a max
//number of concurrent executions. Think of it as a mutex
//which allows 'n' number of locks to be acquired.
type Throttler interface {
    //Lock stops execution and waits if the number of goroutines
    //running has reached the max level of the throttler.
    //It returns the throttler on which it is being called to allow chaining
    //    e.g.
    //    var t= sync2.NewThrottler(10)
    //    func throttledFunc(){
    //      defer t.Lock().Unlock()
    //      ..
    //     }
    Lock() Throttler
    //Unlock releases the throttle on the function/goroutine
    //so other goroutines can continue execution if they were waiting.
    Unlock()
}

type throttler struct {
    l       sync.Locker
    c       *sync.Cond
    running int
    max     int
}

//make sure throttler implements the interface
var _ Throttler = &throttler{}

//NewThrottler creates a throttler which allows you to limit
//the number of concurrent executions of a goroutine or function
func NewThrottler(max int) Throttler {
    l := &sync.Mutex{}
    c := sync.NewCond(l)
    return &throttler{c: c, max: max, l: l}
}

//Unlock releases the throttle on the function/goroutine
//so other goroutines can continue execution if they were waiting.
func (t *throttler) Unlock() {
    t.l.Lock()
    defer t.l.Unlock()
    t.running--
    //signal one of the waiting goroutines to start executing
    t.c.Signal()
}

//Lock stops execution and waits if the number of goroutines
//running has reached the max level of the throttler.
func (t *throttler) Lock() Throttler {
    t.l.Lock()
    defer t.l.Unlock()

    //wait till we have an empty slot
    for (t.max - t.running) < 1 {
        t.c.Wait() //suspends execution of the calling goroutine
    }

    //now we are running
    t.running++
    return t
}