More efficient synchronization
Use atomic ops on the state variable. This lets new requests go ahead even when
something has the main lock, and lets us simplify that lock to a normal Mutex
instead of an RWMutex.
Evan Huus
9 years ago
| 3 | 3 | import ( |
| 4 | 4 | "errors" |
| 5 | 5 | "sync" |
| 6 | "sync/atomic" | |
| 6 | 7 | "time" |
| 7 | 8 | ) |
| 8 | 9 | |
| 10 | 11 | // because the breaker is currently open. |
| 11 | 12 | var ErrBreakerOpen = errors.New("circuit breaker is open") |
| 12 | 13 | |
| 13 | type state int | |
| 14 | ||
| 15 | 14 | const ( |
| 16 | closed state = iota | |
| 15 | closed uint32 = iota | |
| 17 | 16 | open |
| 18 | 17 | halfOpen |
| 19 | 18 | ) |
| 23 | 22 | errorThreshold, successThreshold int |
| 24 | 23 | timeout time.Duration |
| 25 | 24 | |
| 26 | lock sync.RWMutex | |
| 27 | state state | |
| 25 | lock sync.Mutex | |
| 26 | state uint32 | |
| 28 | 27 | errors, successes int |
| 29 | 28 | lastError time.Time |
| 30 | 29 | } |
| 46 | 45 | // already open, or it will run the given function and pass along its return |
| 47 | 46 | // value. It is safe to call Run concurrently on the same Breaker. |
| 48 | 47 | func (b *Breaker) Run(work func() error) error { |
| 49 | b.lock.RLock() | |
| 50 | state := b.state | |
| 51 | b.lock.RUnlock() | |
| 48 | state := atomic.LoadUint32(&b.state) | |
| 52 | 49 | |
| 53 | 50 | if state == open { |
| 54 | 51 | return ErrBreakerOpen |
| 63 | 60 | // the return value of the function. It is safe to call Go concurrently on the |
| 64 | 61 | // same Breaker. |
| 65 | 62 | func (b *Breaker) Go(work func() error) error { |
| 66 | b.lock.RLock() | |
| 67 | state := b.state | |
| 68 | b.lock.RUnlock() | |
| 63 | state := atomic.LoadUint32(&b.state) | |
| 69 | 64 | |
| 70 | 65 | if state == open { |
| 71 | 66 | return ErrBreakerOpen |
| 79 | 74 | return nil |
| 80 | 75 | } |
| 81 | 76 | |
| 82 | func (b *Breaker) doWork(state state, work func() error) error { | |
| 77 | func (b *Breaker) doWork(state uint32, work func() error) error { | |
| 83 | 78 | var panicValue interface{} |
| 84 | 79 | |
| 85 | 80 | result := func() error { |
| 158 | 153 | b.changeState(halfOpen) |
| 159 | 154 | } |
| 160 | 155 | |
| 161 | func (b *Breaker) changeState(newState state) { | |
| 156 | func (b *Breaker) changeState(newState uint32) { | |
| 162 | 157 | b.errors = 0 |
| 163 | 158 | b.successes = 0 |
| 164 | b.state = newState | |
| 159 | atomic.StoreUint32(&b.state, newState) | |
| 165 | 160 | } |