From 3d9113c16e08fe53f31a2403a5280202c8c9cc14 Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Wed, 6 Dec 2017 09:18:04 +0100 Subject: [PATCH 01/36] Add a bit more padding to tests so they pass on osx with -race --- common/repeat_timer_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index 9f03f41df..87f34b950 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -43,7 +43,7 @@ func TestRepeat(test *testing.T) { short := time.Duration(20) * time.Millisecond // delay waits for cnt durations, an a little extra delay := func(cnt int) time.Duration { - return time.Duration(cnt)*dur + time.Millisecond + return time.Duration(cnt)*dur + time.Duration(5)*time.Millisecond } t := NewRepeatTimer("bar", dur) From dcb43956048f0d38495f39e43fd4438ec6d47de7 Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Wed, 6 Dec 2017 11:17:50 +0100 Subject: [PATCH 02/36] Refactor throttle timer --- common/throttle_timer.go | 102 ++++++++++++++++++++++------------ common/throttle_timer_test.go | 19 ++++++- 2 files changed, 85 insertions(+), 36 deletions(-) diff --git a/common/throttle_timer.go b/common/throttle_timer.go index 38ef4e9a3..e260e01bd 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -1,7 +1,7 @@ package common import ( - "sync" + "fmt" "time" ) @@ -12,54 +12,88 @@ If a long continuous burst of .Set() calls happens, ThrottleTimer fires at most once every "dur". */ type ThrottleTimer struct { - Name string - Ch chan struct{} - quit chan struct{} - dur time.Duration + Name string + Ch chan struct{} + input chan command + dur time.Duration - mtx sync.Mutex timer *time.Timer isSet bool } +type command int32 + +const ( + Set command = iota + Unset + Quit +) + func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer { - var ch = make(chan struct{}) - var quit = make(chan struct{}) - var t = &ThrottleTimer{Name: name, Ch: ch, dur: dur, quit: quit} - t.mtx.Lock() - t.timer = time.AfterFunc(dur, t.fireRoutine) - t.mtx.Unlock() + var t = &ThrottleTimer{ + Name: name, + Ch: make(chan struct{}, 1), + dur: dur, + input: make(chan command), + timer: time.NewTimer(dur), + } t.timer.Stop() + go t.run() return t } -func (t *ThrottleTimer) fireRoutine() { - t.mtx.Lock() - defer t.mtx.Unlock() - select { - case t.Ch <- struct{}{}: - t.isSet = false - case <-t.quit: - // do nothing +func (t *ThrottleTimer) run() { + for { + select { + case cmd := <-t.input: + // stop goroutine if the input says so + if t.processInput(cmd) { + // TODO: do we want to close the channels??? + // close(t.Ch) + // close(t.input) + return + } + case <-t.timer.C: + t.isSet = false + t.Ch <- struct{}{} + } + } +} + +// all modifications of the internal state of ThrottleTimer +// happen in this method. It is only called from the run goroutine +// so we avoid any race conditions +func (t *ThrottleTimer) processInput(cmd command) (shutdown bool) { + fmt.Printf("processInput: %d\n", cmd) + switch cmd { + case Set: + if !t.isSet { + t.isSet = true + t.timer.Reset(t.dur) + } + case Quit: + shutdown = true + fallthrough + case Unset: + if t.isSet { + t.isSet = false + if !t.timer.Stop() { + <-t.timer.C + } + } default: - t.timer.Reset(t.dur) + panic("unknown command!") } + // return true + return shutdown } func (t *ThrottleTimer) Set() { - t.mtx.Lock() - defer t.mtx.Unlock() - if !t.isSet { - t.isSet = true - t.timer.Reset(t.dur) - } + t.input <- Set } func (t *ThrottleTimer) Unset() { - t.mtx.Lock() - defer t.mtx.Unlock() - t.isSet = false - t.timer.Stop() + t.input <- Unset } // For ease of .Stop()'ing services before .Start()'ing them, @@ -68,8 +102,6 @@ func (t *ThrottleTimer) Stop() bool { if t == nil { return false } - close(t.quit) - t.mtx.Lock() - defer t.mtx.Unlock() - return t.timer.Stop() + t.input <- Quit + return true } diff --git a/common/throttle_timer_test.go b/common/throttle_timer_test.go index 00f5abdec..014f9dcdc 100644 --- a/common/throttle_timer_test.go +++ b/common/throttle_timer_test.go @@ -41,6 +41,7 @@ func TestThrottle(test *testing.T) { ms := 50 delay := time.Duration(ms) * time.Millisecond + shortwait := time.Duration(ms/2) * time.Millisecond longwait := time.Duration(2) * delay t := NewThrottleTimer("foo", delay) @@ -65,6 +66,21 @@ func TestThrottle(test *testing.T) { time.Sleep(longwait) assert.Equal(2, c.Count()) + // keep cancelling before it is ready + for i := 0; i < 10; i++ { + t.Set() + time.Sleep(shortwait) + t.Unset() + } + time.Sleep(longwait) + assert.Equal(2, c.Count()) + + // a few unsets do nothing... + for i := 0; i < 5; i++ { + t.Unset() + } + assert.Equal(2, c.Count()) + // send 12, over 2 delay sections, adds 3 short := time.Duration(ms/5) * time.Millisecond for i := 0; i < 13; i++ { @@ -74,5 +90,6 @@ func TestThrottle(test *testing.T) { time.Sleep(longwait) assert.Equal(5, c.Count()) - close(t.Ch) + stopped := t.Stop() + assert.True(stopped) } From 4ec7883891fa9700ce4b122252b8fc697df0bfca Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Wed, 6 Dec 2017 11:21:01 +0100 Subject: [PATCH 03/36] Cleanup --- common/throttle_timer.go | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/common/throttle_timer.go b/common/throttle_timer.go index e260e01bd..705a12a12 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -1,7 +1,6 @@ package common import ( - "fmt" "time" ) @@ -64,7 +63,6 @@ func (t *ThrottleTimer) run() { // happen in this method. It is only called from the run goroutine // so we avoid any race conditions func (t *ThrottleTimer) processInput(cmd command) (shutdown bool) { - fmt.Printf("processInput: %d\n", cmd) switch cmd { case Set: if !t.isSet { @@ -77,9 +75,7 @@ func (t *ThrottleTimer) processInput(cmd command) (shutdown bool) { case Unset: if t.isSet { t.isSet = false - if !t.timer.Stop() { - <-t.timer.C - } + t.timer.Stop() } default: panic("unknown command!") From 0a8721113a67b3c05f58e12328a0fe0216811b0c Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Wed, 6 Dec 2017 21:08:55 +0100 Subject: [PATCH 04/36] First pass of PR updates --- common/throttle_timer.go | 28 ++++++++++++++-------------- common/throttle_timer_test.go | 2 +- 2 files changed, 15 insertions(+), 15 deletions(-) diff --git a/common/throttle_timer.go b/common/throttle_timer.go index 705a12a12..f2ce60b2a 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -11,10 +11,11 @@ If a long continuous burst of .Set() calls happens, ThrottleTimer fires at most once every "dur". */ type ThrottleTimer struct { - Name string - Ch chan struct{} - input chan command - dur time.Duration + Name string + Ch <-chan struct{} + output chan<- struct{} + input chan command + dur time.Duration timer *time.Timer isSet bool @@ -29,12 +30,14 @@ const ( ) func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer { + c := make(chan struct{}, 1) var t = &ThrottleTimer{ - Name: name, - Ch: make(chan struct{}, 1), - dur: dur, - input: make(chan command), - timer: time.NewTimer(dur), + Name: name, + Ch: c, + dur: dur, + output: c, + input: make(chan command), + timer: time.NewTimer(dur), } t.timer.Stop() go t.run() @@ -47,14 +50,12 @@ func (t *ThrottleTimer) run() { case cmd := <-t.input: // stop goroutine if the input says so if t.processInput(cmd) { - // TODO: do we want to close the channels??? - // close(t.Ch) - // close(t.input) + close(t.output) return } case <-t.timer.C: t.isSet = false - t.Ch <- struct{}{} + t.output <- struct{}{} } } } @@ -80,7 +81,6 @@ func (t *ThrottleTimer) processInput(cmd command) (shutdown bool) { default: panic("unknown command!") } - // return true return shutdown } diff --git a/common/throttle_timer_test.go b/common/throttle_timer_test.go index 014f9dcdc..81b817038 100644 --- a/common/throttle_timer_test.go +++ b/common/throttle_timer_test.go @@ -10,7 +10,7 @@ import ( ) type thCounter struct { - input chan struct{} + input <-chan struct{} mtx sync.Mutex count int } From 1ac4c5dd6d007a708337e1ad2636e456e1e4b8db Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Wed, 6 Dec 2017 21:20:30 +0100 Subject: [PATCH 05/36] Made throttle output non-blocking --- common/throttle_timer.go | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/common/throttle_timer.go b/common/throttle_timer.go index f2ce60b2a..069b6d84b 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -30,7 +30,7 @@ const ( ) func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer { - c := make(chan struct{}, 1) + c := make(chan struct{}) var t = &ThrottleTimer{ Name: name, Ch: c, @@ -54,12 +54,22 @@ func (t *ThrottleTimer) run() { return } case <-t.timer.C: - t.isSet = false - t.output <- struct{}{} + t.trySend() } } } +// trySend performs non-blocking send on t.output (t.Ch) +func (t *ThrottleTimer) trySend() { + select { + case t.output <- struct{}{}: + t.isSet = false + default: + // if we just want to drop, replace this with t.isSet = false + t.timer.Reset(t.dur) + } +} + // all modifications of the internal state of ThrottleTimer // happen in this method. It is only called from the run goroutine // so we avoid any race conditions From e430d3f8447d23b739840d5137ae75c37ff33a1d Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Wed, 6 Dec 2017 21:51:23 +0100 Subject: [PATCH 06/36] One more attempt with a read-only channel --- common/throttle_timer.go | 33 ++++++++++++++++++--------------- common/throttle_timer_test.go | 2 +- 2 files changed, 19 insertions(+), 16 deletions(-) diff --git a/common/throttle_timer.go b/common/throttle_timer.go index 069b6d84b..4a4b30033 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -11,11 +11,10 @@ If a long continuous burst of .Set() calls happens, ThrottleTimer fires at most once every "dur". */ type ThrottleTimer struct { - Name string - Ch <-chan struct{} - output chan<- struct{} - input chan command - dur time.Duration + Name string + Ch chan struct{} + input chan command + dur time.Duration timer *time.Timer isSet bool @@ -30,27 +29,31 @@ const ( ) func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer { - c := make(chan struct{}) var t = &ThrottleTimer{ - Name: name, - Ch: c, - dur: dur, - output: c, - input: make(chan command), - timer: time.NewTimer(dur), + Name: name, + Ch: make(chan struct{}), + dur: dur, + input: make(chan command), + timer: time.NewTimer(dur), } t.timer.Stop() go t.run() return t } +// C is the proper way to listen to the timer output. +// t.Ch will be made private in the (near?) future +func (t *ThrottleTimer) C() <-chan struct{} { + return t.Ch +} + func (t *ThrottleTimer) run() { for { select { case cmd := <-t.input: // stop goroutine if the input says so if t.processInput(cmd) { - close(t.output) + close(t.Ch) return } case <-t.timer.C: @@ -59,10 +62,10 @@ func (t *ThrottleTimer) run() { } } -// trySend performs non-blocking send on t.output (t.Ch) +// trySend performs non-blocking send on t.Ch func (t *ThrottleTimer) trySend() { select { - case t.output <- struct{}{}: + case t.Ch <- struct{}{}: t.isSet = false default: // if we just want to drop, replace this with t.isSet = false diff --git a/common/throttle_timer_test.go b/common/throttle_timer_test.go index 81b817038..f6b5d1df5 100644 --- a/common/throttle_timer_test.go +++ b/common/throttle_timer_test.go @@ -46,7 +46,7 @@ func TestThrottle(test *testing.T) { t := NewThrottleTimer("foo", delay) // start at 0 - c := &thCounter{input: t.Ch} + c := &thCounter{input: t.C()} assert.Equal(0, c.Count()) go c.Read() From 8b518fadb2f3eb928ce5d5a014b4087c5b31309a Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Wed, 6 Dec 2017 22:28:18 +0100 Subject: [PATCH 07/36] Don't close throttle channel, explain why --- common/throttle_timer.go | 2 +- common/throttle_timer_test.go | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/common/throttle_timer.go b/common/throttle_timer.go index 4a4b30033..051d44376 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -52,8 +52,8 @@ func (t *ThrottleTimer) run() { select { case cmd := <-t.input: // stop goroutine if the input says so + // don't close channels, as closed channels mess up select reads if t.processInput(cmd) { - close(t.Ch) return } case <-t.timer.C: diff --git a/common/throttle_timer_test.go b/common/throttle_timer_test.go index f6b5d1df5..7d96ac7c5 100644 --- a/common/throttle_timer_test.go +++ b/common/throttle_timer_test.go @@ -31,6 +31,9 @@ func (c *thCounter) Count() int { // Read should run in a go-routine and // updates count by one every time a packet comes in func (c *thCounter) Read() { + // note, since this channel never closes, this will never end + // if thCounter was used in anything beyond trivial test cases. + // it would have to be smarter. for range c.input { c.Increment() } From 3779310c72c93173b9e87561281e697e7cdf9437 Mon Sep 17 00:00:00 2001 From: Anton Kaliaev Date: Wed, 6 Dec 2017 18:48:39 -0600 Subject: [PATCH 08/36] return back output internal channel (way go does with Timer) --- common/throttle_timer.go | 47 +++++++++++++++++++++-------------- common/throttle_timer_test.go | 2 +- 2 files changed, 30 insertions(+), 19 deletions(-) diff --git a/common/throttle_timer.go b/common/throttle_timer.go index 051d44376..ab2ad2e62 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -11,10 +11,11 @@ If a long continuous burst of .Set() calls happens, ThrottleTimer fires at most once every "dur". */ type ThrottleTimer struct { - Name string - Ch chan struct{} - input chan command - dur time.Duration + Name string + Ch <-chan struct{} + input chan command + output chan<- struct{} + dur time.Duration timer *time.Timer isSet bool @@ -28,25 +29,22 @@ const ( Quit ) +// NewThrottleTimer creates a new ThrottleTimer. func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer { + c := make(chan struct{}) var t = &ThrottleTimer{ - Name: name, - Ch: make(chan struct{}), - dur: dur, - input: make(chan command), - timer: time.NewTimer(dur), + Name: name, + Ch: c, + dur: dur, + input: make(chan command), + output: c, + timer: time.NewTimer(dur), } t.timer.Stop() go t.run() return t } -// C is the proper way to listen to the timer output. -// t.Ch will be made private in the (near?) future -func (t *ThrottleTimer) C() <-chan struct{} { - return t.Ch -} - func (t *ThrottleTimer) run() { for { select { @@ -65,7 +63,7 @@ func (t *ThrottleTimer) run() { // trySend performs non-blocking send on t.Ch func (t *ThrottleTimer) trySend() { select { - case t.Ch <- struct{}{}: + case t.output <- struct{}{}: t.isSet = false default: // if we just want to drop, replace this with t.isSet = false @@ -105,8 +103,21 @@ func (t *ThrottleTimer) Unset() { t.input <- Unset } -// For ease of .Stop()'ing services before .Start()'ing them, -// we ignore .Stop()'s on nil ThrottleTimers +// Stop prevents the ThrottleTimer from firing. It always returns true. Stop does not +// close the channel, to prevent a read from the channel succeeding +// incorrectly. +// +// To prevent a timer created with NewThrottleTimer from firing after a call to +// Stop, check the return value and drain the channel. +// +// For example, assuming the program has not received from t.C already: +// +// if !t.Stop() { +// <-t.C +// } +// +// For ease of stopping services before starting them, we ignore Stop on nil +// ThrottleTimers. func (t *ThrottleTimer) Stop() bool { if t == nil { return false diff --git a/common/throttle_timer_test.go b/common/throttle_timer_test.go index 7d96ac7c5..a1b6606f5 100644 --- a/common/throttle_timer_test.go +++ b/common/throttle_timer_test.go @@ -49,7 +49,7 @@ func TestThrottle(test *testing.T) { t := NewThrottleTimer("foo", delay) // start at 0 - c := &thCounter{input: t.C()} + c := &thCounter{input: t.Ch} assert.Equal(0, c.Count()) go c.Read() From 887d766c86f1f217653915a2042374972c8f38ae Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Thu, 7 Dec 2017 10:15:38 +0100 Subject: [PATCH 09/36] Refactored RepeatTimer, tests hang --- common/repeat_timer.go | 121 +++++++++++++++++++++--------------- common/repeat_timer_test.go | 4 +- common/throttle_timer.go | 16 ++--- 3 files changed, 81 insertions(+), 60 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index d7d9154d4..0f6501131 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -1,7 +1,7 @@ package common import ( - "sync" + "fmt" "time" ) @@ -11,54 +11,40 @@ It's good for keeping connections alive. A RepeatTimer must be Stop()'d or it will keep a goroutine alive. */ type RepeatTimer struct { - Ch chan time.Time + Name string + Ch <-chan time.Time + output chan<- time.Time + input chan repeatCommand - mtx sync.Mutex - name string - ticker *time.Ticker - quit chan struct{} - wg *sync.WaitGroup - dur time.Duration + dur time.Duration + timer *time.Timer } +type repeatCommand int32 + +const ( + Reset repeatCommand = iota + RQuit +) + func NewRepeatTimer(name string, dur time.Duration) *RepeatTimer { + c := make(chan time.Time) var t = &RepeatTimer{ - Ch: make(chan time.Time), - ticker: time.NewTicker(dur), - quit: make(chan struct{}), - wg: new(sync.WaitGroup), - name: name, - dur: dur, - } - t.wg.Add(1) - go t.fireRoutine(t.ticker) - return t -} + Name: name, + Ch: c, + output: c, + input: make(chan repeatCommand), -func (t *RepeatTimer) fireRoutine(ticker *time.Ticker) { - for { - select { - case t_ := <-ticker.C: - t.Ch <- t_ - case <-t.quit: - // needed so we know when we can reset t.quit - t.wg.Done() - return - } + timer: time.NewTimer(dur), + dur: dur, } + go t.run() + return t } // Wait the duration again before firing. func (t *RepeatTimer) Reset() { - t.Stop() - - t.mtx.Lock() // Lock - defer t.mtx.Unlock() - - t.ticker = time.NewTicker(t.dur) - t.quit = make(chan struct{}) - t.wg.Add(1) - go t.fireRoutine(t.ticker) + t.input <- Reset } // For ease of .Stop()'ing services before .Start()'ing them, @@ -67,20 +53,55 @@ func (t *RepeatTimer) Stop() bool { if t == nil { return false } - t.mtx.Lock() // Lock - defer t.mtx.Unlock() + t.input <- RQuit + return true +} - exists := t.ticker != nil - if exists { - t.ticker.Stop() // does not close the channel +func (t *RepeatTimer) run() { + for { + fmt.Println("for") select { - case <-t.Ch: - // read off channel if there's anything there - default: + case cmd := <-t.input: + // stop goroutine if the input says so + // don't close channels, as closed channels mess up select reads + if t.processInput(cmd) { + t.timer.Stop() + return + } + case <-t.timer.C: + fmt.Println("tick") + // send if not blocked, then start the next tick + // for blocking send, just + // t.output <- time.Now() + t.trySend() + t.timer.Reset(t.dur) } - close(t.quit) - t.wg.Wait() // must wait for quit to close else we race Reset - t.ticker = nil } - return exists +} + +// trySend performs non-blocking send on t.Ch +func (t *RepeatTimer) trySend() { + // TODO: this was blocking in previous version (t.Ch <- t_) + // should I use that behavior unstead of unblocking as per throttle? + select { + case t.output <- time.Now(): + default: + } +} + +// all modifications of the internal state of ThrottleTimer +// happen in this method. It is only called from the run goroutine +// so we avoid any race conditions +func (t *RepeatTimer) processInput(cmd repeatCommand) (shutdown bool) { + fmt.Printf("process: %d\n", cmd) + switch cmd { + case Reset: + t.timer.Reset(t.dur) + case RQuit: + t.timer.Stop() + shutdown = true + default: + panic("unknown command!") + } + return shutdown } diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index 87f34b950..d66cd3152 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -10,7 +10,7 @@ import ( ) type rCounter struct { - input chan time.Time + input <-chan time.Time mtx sync.Mutex count int } @@ -74,5 +74,5 @@ func TestRepeat(test *testing.T) { assert.Equal(6, c.Count()) // close channel to stop counter - close(t.Ch) + t.Stop() } diff --git a/common/throttle_timer.go b/common/throttle_timer.go index ab2ad2e62..c148d9904 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -13,7 +13,7 @@ at most once every "dur". type ThrottleTimer struct { Name string Ch <-chan struct{} - input chan command + input chan throttleCommand output chan<- struct{} dur time.Duration @@ -21,12 +21,12 @@ type ThrottleTimer struct { isSet bool } -type command int32 +type throttleCommand int32 const ( - Set command = iota + Set throttleCommand = iota Unset - Quit + TQuit ) // NewThrottleTimer creates a new ThrottleTimer. @@ -36,7 +36,7 @@ func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer { Name: name, Ch: c, dur: dur, - input: make(chan command), + input: make(chan throttleCommand), output: c, timer: time.NewTimer(dur), } @@ -74,14 +74,14 @@ func (t *ThrottleTimer) trySend() { // all modifications of the internal state of ThrottleTimer // happen in this method. It is only called from the run goroutine // so we avoid any race conditions -func (t *ThrottleTimer) processInput(cmd command) (shutdown bool) { +func (t *ThrottleTimer) processInput(cmd throttleCommand) (shutdown bool) { switch cmd { case Set: if !t.isSet { t.isSet = true t.timer.Reset(t.dur) } - case Quit: + case TQuit: shutdown = true fallthrough case Unset: @@ -122,6 +122,6 @@ func (t *ThrottleTimer) Stop() bool { if t == nil { return false } - t.input <- Quit + t.input <- TQuit return true } From 8797197cdfc9920e2dbce274c8aba8c09b15f86f Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Thu, 7 Dec 2017 10:36:03 +0100 Subject: [PATCH 10/36] No more blocking on multiple Stop() --- common/repeat_timer.go | 33 +++++++++++++++++---------------- common/repeat_timer_test.go | 2 +- common/throttle_timer.go | 8 +++++--- common/throttle_timer_test.go | 5 +++++ 4 files changed, 28 insertions(+), 20 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index 0f6501131..734c2d32a 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -16,8 +16,9 @@ type RepeatTimer struct { output chan<- time.Time input chan repeatCommand - dur time.Duration - timer *time.Timer + dur time.Duration + timer *time.Timer + stopped bool } type repeatCommand int32 @@ -50,43 +51,42 @@ func (t *RepeatTimer) Reset() { // For ease of .Stop()'ing services before .Start()'ing them, // we ignore .Stop()'s on nil RepeatTimers. func (t *RepeatTimer) Stop() bool { - if t == nil { + if t == nil || t.stopped { return false } t.input <- RQuit + t.stopped = true return true } func (t *RepeatTimer) run() { - for { - fmt.Println("for") + done := false + for !done { select { case cmd := <-t.input: // stop goroutine if the input says so // don't close channels, as closed channels mess up select reads - if t.processInput(cmd) { - t.timer.Stop() - return - } + done = t.processInput(cmd) case <-t.timer.C: - fmt.Println("tick") // send if not blocked, then start the next tick - // for blocking send, just - // t.output <- time.Now() t.trySend() t.timer.Reset(t.dur) } } + fmt.Println("end run") } // trySend performs non-blocking send on t.Ch func (t *RepeatTimer) trySend() { // TODO: this was blocking in previous version (t.Ch <- t_) // should I use that behavior unstead of unblocking as per throttle? - select { - case t.output <- time.Now(): - default: - } + + // select { + // case t.output <- time.Now(): + // default: + // } + + t.output <- time.Now() } // all modifications of the internal state of ThrottleTimer @@ -98,6 +98,7 @@ func (t *RepeatTimer) processInput(cmd repeatCommand) (shutdown bool) { case Reset: t.timer.Reset(t.dur) case RQuit: + fmt.Println("got quit") t.timer.Stop() shutdown = true default: diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index d66cd3152..15ca32c31 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -73,6 +73,6 @@ func TestRepeat(test *testing.T) { time.Sleep(delay(7)) assert.Equal(6, c.Count()) - // close channel to stop counter + // extra calls to stop don't block t.Stop() } diff --git a/common/throttle_timer.go b/common/throttle_timer.go index c148d9904..0e54f1027 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -17,8 +17,9 @@ type ThrottleTimer struct { output chan<- struct{} dur time.Duration - timer *time.Timer - isSet bool + timer *time.Timer + isSet bool + stopped bool } type throttleCommand int32 @@ -82,6 +83,7 @@ func (t *ThrottleTimer) processInput(cmd throttleCommand) (shutdown bool) { t.timer.Reset(t.dur) } case TQuit: + t.stopped = true shutdown = true fallthrough case Unset: @@ -119,7 +121,7 @@ func (t *ThrottleTimer) Unset() { // For ease of stopping services before starting them, we ignore Stop on nil // ThrottleTimers. func (t *ThrottleTimer) Stop() bool { - if t == nil { + if t == nil || t.stopped { return false } t.input <- TQuit diff --git a/common/throttle_timer_test.go b/common/throttle_timer_test.go index a1b6606f5..2a81bb02e 100644 --- a/common/throttle_timer_test.go +++ b/common/throttle_timer_test.go @@ -95,4 +95,9 @@ func TestThrottle(test *testing.T) { stopped := t.Stop() assert.True(stopped) + time.Sleep(longwait) + assert.Equal(5, c.Count()) + + // extra calls to stop don't block + t.Stop() } From cc7a87e27caa55ca84e984d1d081b09eeb16ffe6 Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Thu, 7 Dec 2017 11:22:54 +0100 Subject: [PATCH 11/36] Use Ticker in Repeat again to avoid drift --- common/repeat_timer.go | 34 ++++++++++++++-------------------- common/repeat_timer_test.go | 6 +++--- 2 files changed, 17 insertions(+), 23 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index 734c2d32a..b3eb107d2 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -1,7 +1,6 @@ package common import ( - "fmt" "time" ) @@ -17,7 +16,7 @@ type RepeatTimer struct { input chan repeatCommand dur time.Duration - timer *time.Timer + ticker *time.Ticker stopped bool } @@ -36,8 +35,8 @@ func NewRepeatTimer(name string, dur time.Duration) *RepeatTimer { output: c, input: make(chan repeatCommand), - timer: time.NewTimer(dur), - dur: dur, + dur: dur, + ticker: time.NewTicker(dur), } go t.run() return t @@ -51,6 +50,7 @@ func (t *RepeatTimer) Reset() { // For ease of .Stop()'ing services before .Start()'ing them, // we ignore .Stop()'s on nil RepeatTimers. func (t *RepeatTimer) Stop() bool { + // use t.stopped to gracefully handle many Stop() without blocking if t == nil || t.stopped { return false } @@ -67,39 +67,33 @@ func (t *RepeatTimer) run() { // stop goroutine if the input says so // don't close channels, as closed channels mess up select reads done = t.processInput(cmd) - case <-t.timer.C: - // send if not blocked, then start the next tick + case <-t.ticker.C: t.trySend() - t.timer.Reset(t.dur) } } - fmt.Println("end run") } // trySend performs non-blocking send on t.Ch func (t *RepeatTimer) trySend() { - // TODO: this was blocking in previous version (t.Ch <- t_) + // NOTE: this was blocking in previous version (t.Ch <- t_) // should I use that behavior unstead of unblocking as per throttle? - - // select { - // case t.output <- time.Now(): - // default: - // } - - t.output <- time.Now() + // probably not: https://golang.org/src/time/sleep.go#L132 + select { + case t.output <- time.Now(): + default: + } } // all modifications of the internal state of ThrottleTimer // happen in this method. It is only called from the run goroutine // so we avoid any race conditions func (t *RepeatTimer) processInput(cmd repeatCommand) (shutdown bool) { - fmt.Printf("process: %d\n", cmd) switch cmd { case Reset: - t.timer.Reset(t.dur) + t.ticker.Stop() + t.ticker = time.NewTicker(t.dur) case RQuit: - fmt.Println("got quit") - t.timer.Stop() + t.ticker.Stop() shutdown = true default: panic("unknown command!") diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index 15ca32c31..db53aa614 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -39,11 +39,11 @@ func (c *rCounter) Read() { func TestRepeat(test *testing.T) { assert := asrt.New(test) - dur := time.Duration(50) * time.Millisecond + dur := time.Duration(100) * time.Millisecond short := time.Duration(20) * time.Millisecond // delay waits for cnt durations, an a little extra delay := func(cnt int) time.Duration { - return time.Duration(cnt)*dur + time.Duration(5)*time.Millisecond + return time.Duration(cnt)*dur + time.Duration(10)*time.Millisecond } t := NewRepeatTimer("bar", dur) @@ -70,7 +70,7 @@ func TestRepeat(test *testing.T) { // after a stop, nothing more is sent stopped := t.Stop() assert.True(stopped) - time.Sleep(delay(7)) + time.Sleep(delay(2)) assert.Equal(6, c.Count()) // extra calls to stop don't block From ec4adf21e0451f3fb7da33932d6cac168ddeaa93 Mon Sep 17 00:00:00 2001 From: Ethan Frey Date: Fri, 8 Dec 2017 10:07:04 +0100 Subject: [PATCH 12/36] Cleanup from PR comments --- common/repeat_timer.go | 20 ++++++++++---------- common/throttle_timer.go | 4 ++-- common/throttle_timer_test.go | 3 --- 3 files changed, 12 insertions(+), 15 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index b3eb107d2..77f736034 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -20,7 +20,7 @@ type RepeatTimer struct { stopped bool } -type repeatCommand int32 +type repeatCommand int8 const ( Reset repeatCommand = iota @@ -67,21 +67,21 @@ func (t *RepeatTimer) run() { // stop goroutine if the input says so // don't close channels, as closed channels mess up select reads done = t.processInput(cmd) - case <-t.ticker.C: - t.trySend() + case tick := <-t.ticker.C: + t.trySend(tick) } } } // trySend performs non-blocking send on t.Ch -func (t *RepeatTimer) trySend() { +func (t *RepeatTimer) trySend(tick time.Time) { // NOTE: this was blocking in previous version (t.Ch <- t_) - // should I use that behavior unstead of unblocking as per throttle? - // probably not: https://golang.org/src/time/sleep.go#L132 - select { - case t.output <- time.Now(): - default: - } + // probably better not: https://golang.org/src/time/sleep.go#L132 + t.output <- tick + // select { + // case t.output <- tick: + // default: + // } } // all modifications of the internal state of ThrottleTimer diff --git a/common/throttle_timer.go b/common/throttle_timer.go index 0e54f1027..a5bd6ded8 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -22,7 +22,7 @@ type ThrottleTimer struct { stopped bool } -type throttleCommand int32 +type throttleCommand int8 const ( Set throttleCommand = iota @@ -83,7 +83,6 @@ func (t *ThrottleTimer) processInput(cmd throttleCommand) (shutdown bool) { t.timer.Reset(t.dur) } case TQuit: - t.stopped = true shutdown = true fallthrough case Unset: @@ -125,5 +124,6 @@ func (t *ThrottleTimer) Stop() bool { return false } t.input <- TQuit + t.stopped = true return true } diff --git a/common/throttle_timer_test.go b/common/throttle_timer_test.go index 2a81bb02e..94ec1b43c 100644 --- a/common/throttle_timer_test.go +++ b/common/throttle_timer_test.go @@ -95,9 +95,6 @@ func TestThrottle(test *testing.T) { stopped := t.Stop() assert.True(stopped) - time.Sleep(longwait) - assert.Equal(5, c.Count()) - // extra calls to stop don't block t.Stop() } From ff2fd63bf7db6373e5fb0c1d311c6a139b99dfe0 Mon Sep 17 00:00:00 2001 From: Anton Kaliaev Date: Fri, 8 Dec 2017 11:17:07 -0600 Subject: [PATCH 13/36] rename trySend to send --- common/repeat_timer.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index 77f736034..23faf74ae 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -68,20 +68,20 @@ func (t *RepeatTimer) run() { // don't close channels, as closed channels mess up select reads done = t.processInput(cmd) case tick := <-t.ticker.C: - t.trySend(tick) + t.send(tick) } } } -// trySend performs non-blocking send on t.Ch -func (t *RepeatTimer) trySend(tick time.Time) { - // NOTE: this was blocking in previous version (t.Ch <- t_) - // probably better not: https://golang.org/src/time/sleep.go#L132 - t.output <- tick +// send performs blocking send on t.Ch +func (t *RepeatTimer) send(tick time.Time) { + // XXX: possibly it is better to not block: + // https://golang.org/src/time/sleep.go#L132 // select { // case t.output <- tick: // default: // } + t.output <- tick } // all modifications of the internal state of ThrottleTimer From cb4ba522ef643073c1b1ae372ef0a5e32078cb5f Mon Sep 17 00:00:00 2001 From: Anton Kaliaev Date: Sat, 9 Dec 2017 23:05:13 -0600 Subject: [PATCH 14/36] add String method to Query interface Required for https://github.com/tendermint/tendermint/issues/945 --- pubsub/pubsub.go | 1 + 1 file changed, 1 insertion(+) diff --git a/pubsub/pubsub.go b/pubsub/pubsub.go index 52b8361f8..27f15cbeb 100644 --- a/pubsub/pubsub.go +++ b/pubsub/pubsub.go @@ -38,6 +38,7 @@ type cmd struct { // Query defines an interface for a query to be used for subscribing. type Query interface { Matches(tags map[string]interface{}) bool + String() string } // Server allows clients to subscribe/unsubscribe for messages, publishing From e4ef2835f0081c2ece83b9c1f777cf071f956e81 Mon Sep 17 00:00:00 2001 From: Anton Kaliaev Date: Sat, 9 Dec 2017 23:35:14 -0600 Subject: [PATCH 15/36] return error if client already subscribed --- pubsub/pubsub.go | 68 ++++++++++++++++++++++++++++++++++--------- pubsub/pubsub_test.go | 32 +++++++++++++++----- 2 files changed, 80 insertions(+), 20 deletions(-) diff --git a/pubsub/pubsub.go b/pubsub/pubsub.go index 27f15cbeb..54a4b8aed 100644 --- a/pubsub/pubsub.go +++ b/pubsub/pubsub.go @@ -13,6 +13,8 @@ package pubsub import ( "context" + "errors" + "sync" cmn "github.com/tendermint/tmlibs/common" ) @@ -48,6 +50,9 @@ type Server struct { cmds chan cmd cmdsCap int + + mtx sync.RWMutex + subscriptions map[string]map[string]struct{} // subscriber -> query -> struct{} } // Option sets a parameter for the server. @@ -57,7 +62,9 @@ type Option func(*Server) // for a detailed description of how to configure buffering. If no options are // provided, the resulting server's queue is unbuffered. func NewServer(options ...Option) *Server { - s := &Server{} + s := &Server{ + subscriptions: make(map[string]map[string]struct{}), + } s.BaseService = *cmn.NewBaseService(nil, "PubSub", s) for _, option := range options { @@ -83,17 +90,33 @@ func BufferCapacity(cap int) Option { } // BufferCapacity returns capacity of the internal server's queue. -func (s Server) BufferCapacity() int { +func (s *Server) BufferCapacity() int { return s.cmdsCap } // Subscribe creates a subscription for the given client. It accepts a channel -// on which messages matching the given query can be received. If the -// subscription already exists, the old channel will be closed. An error will -// be returned to the caller if the context is canceled. +// on which messages matching the given query can be received. An error will be +// returned to the caller if the context is canceled or if subscription already +// exist for pair clientID and query. func (s *Server) Subscribe(ctx context.Context, clientID string, query Query, out chan<- interface{}) error { + s.mtx.RLock() + clientSubscriptions, ok := s.subscriptions[clientID] + if ok { + _, ok = clientSubscriptions[query.String()] + } + s.mtx.RUnlock() + if ok { + return errors.New("already subscribed") + } + select { case s.cmds <- cmd{op: sub, clientID: clientID, query: query, ch: out}: + s.mtx.Lock() + if _, ok = s.subscriptions[clientID]; !ok { + s.subscriptions[clientID] = make(map[string]struct{}) + } + s.subscriptions[clientID][query.String()] = struct{}{} + s.mtx.Unlock() return nil case <-ctx.Done(): return ctx.Err() @@ -101,10 +124,24 @@ func (s *Server) Subscribe(ctx context.Context, clientID string, query Query, ou } // Unsubscribe removes the subscription on the given query. An error will be -// returned to the caller if the context is canceled. +// returned to the caller if the context is canceled or if subscription does +// not exist. func (s *Server) Unsubscribe(ctx context.Context, clientID string, query Query) error { + s.mtx.RLock() + clientSubscriptions, ok := s.subscriptions[clientID] + if ok { + _, ok = clientSubscriptions[query.String()] + } + s.mtx.RUnlock() + if !ok { + return errors.New("subscription not found") + } + select { case s.cmds <- cmd{op: unsub, clientID: clientID, query: query}: + s.mtx.Lock() + delete(clientSubscriptions, query.String()) + s.mtx.Unlock() return nil case <-ctx.Done(): return ctx.Err() @@ -112,10 +149,20 @@ func (s *Server) Unsubscribe(ctx context.Context, clientID string, query Query) } // UnsubscribeAll removes all client subscriptions. An error will be returned -// to the caller if the context is canceled. +// to the caller if the context is canceled or if subscription does not exist. func (s *Server) UnsubscribeAll(ctx context.Context, clientID string) error { + s.mtx.RLock() + _, ok := s.subscriptions[clientID] + s.mtx.RUnlock() + if !ok { + return errors.New("subscription not found") + } + select { case s.cmds <- cmd{op: unsub, clientID: clientID}: + s.mtx.Lock() + delete(s.subscriptions, clientID) + s.mtx.Unlock() return nil case <-ctx.Done(): return ctx.Err() @@ -187,13 +234,8 @@ loop: func (state *state) add(clientID string, q Query, ch chan<- interface{}) { // add query if needed - if clientToChannelMap, ok := state.queries[q]; !ok { + if _, ok := state.queries[q]; !ok { state.queries[q] = make(map[string]chan<- interface{}) - } else { - // check if already subscribed - if oldCh, ok := clientToChannelMap[clientID]; ok { - close(oldCh) - } } // create subscription diff --git a/pubsub/pubsub_test.go b/pubsub/pubsub_test.go index 7bf7b41f7..84b6aa218 100644 --- a/pubsub/pubsub_test.go +++ b/pubsub/pubsub_test.go @@ -86,14 +86,11 @@ func TestClientSubscribesTwice(t *testing.T) { ch2 := make(chan interface{}, 1) err = s.Subscribe(ctx, clientID, q, ch2) - require.NoError(t, err) - - _, ok := <-ch1 - assert.False(t, ok) + require.Error(t, err) err = s.PublishWithTags(ctx, "Spider-Man", map[string]interface{}{"tm.events.type": "NewBlock"}) require.NoError(t, err) - assertReceive(t, "Spider-Man", ch2) + assertReceive(t, "Spider-Man", ch1) } func TestUnsubscribe(t *testing.T) { @@ -117,6 +114,27 @@ func TestUnsubscribe(t *testing.T) { assert.False(t, ok) } +func TestResubscribe(t *testing.T) { + s := pubsub.NewServer() + s.SetLogger(log.TestingLogger()) + s.Start() + defer s.Stop() + + ctx := context.Background() + ch := make(chan interface{}) + err := s.Subscribe(ctx, clientID, query.Empty{}, ch) + require.NoError(t, err) + err = s.Unsubscribe(ctx, clientID, query.Empty{}) + require.NoError(t, err) + ch = make(chan interface{}) + err = s.Subscribe(ctx, clientID, query.Empty{}, ch) + require.NoError(t, err) + + err = s.Publish(ctx, "Cable") + require.NoError(t, err) + assertReceive(t, "Cable", ch) +} + func TestUnsubscribeAll(t *testing.T) { s := pubsub.NewServer() s.SetLogger(log.TestingLogger()) @@ -125,9 +143,9 @@ func TestUnsubscribeAll(t *testing.T) { ctx := context.Background() ch1, ch2 := make(chan interface{}, 1), make(chan interface{}, 1) - err := s.Subscribe(ctx, clientID, query.Empty{}, ch1) + err := s.Subscribe(ctx, clientID, query.MustParse("tm.events.type='NewBlock'"), ch1) require.NoError(t, err) - err = s.Subscribe(ctx, clientID, query.Empty{}, ch2) + err = s.Subscribe(ctx, clientID, query.MustParse("tm.events.type='NewBlockHeader'"), ch2) require.NoError(t, err) err = s.UnsubscribeAll(ctx, clientID) From f39b575503b80cf22753f70ddc2956925b7b1ac4 Mon Sep 17 00:00:00 2001 From: Zach Ramsay Date: Tue, 12 Dec 2017 16:55:41 +0000 Subject: [PATCH 16/36] remove deprecated --root flag --- cli/setup.go | 27 +++++++-------------------- cli/setup_test.go | 20 ++++++++++---------- 2 files changed, 17 insertions(+), 30 deletions(-) diff --git a/cli/setup.go b/cli/setup.go index 78151015b..295477598 100644 --- a/cli/setup.go +++ b/cli/setup.go @@ -14,7 +14,6 @@ import ( ) const ( - RootFlag = "root" HomeFlag = "home" TraceFlag = "trace" OutputFlag = "output" @@ -28,14 +27,9 @@ type Executable interface { } // PrepareBaseCmd is meant for tendermint and other servers -func PrepareBaseCmd(cmd *cobra.Command, envPrefix, defautRoot string) Executor { +func PrepareBaseCmd(cmd *cobra.Command, envPrefix, defaultHome string) Executor { cobra.OnInitialize(func() { initEnv(envPrefix) }) - cmd.PersistentFlags().StringP(RootFlag, "r", defautRoot, "DEPRECATED. Use --home") - // -h is already reserved for --help as part of the cobra framework - // do you want to try something else?? - // also, default must be empty, so we can detect this unset and fall back - // to --root / TM_ROOT / TMROOT - cmd.PersistentFlags().String(HomeFlag, "", "root directory for config and data") + cmd.PersistentFlags().StringP(HomeFlag, "", defaultHome, "directory for config and data") cmd.PersistentFlags().Bool(TraceFlag, false, "print out full stack trace on errors") cmd.PersistentPreRunE = concatCobraCmdFuncs(bindFlagsLoadViper, cmd.PersistentPreRunE) return Executor{cmd, os.Exit} @@ -45,11 +39,11 @@ func PrepareBaseCmd(cmd *cobra.Command, envPrefix, defautRoot string) Executor { // // This adds --encoding (hex, btc, base64) and --output (text, json) to // the command. These only really make sense in interactive commands. -func PrepareMainCmd(cmd *cobra.Command, envPrefix, defautRoot string) Executor { +func PrepareMainCmd(cmd *cobra.Command, envPrefix, defaultHome string) Executor { cmd.PersistentFlags().StringP(EncodingFlag, "e", "hex", "Binary encoding (hex|b64|btc)") cmd.PersistentFlags().StringP(OutputFlag, "o", "text", "Output format (text|json)") cmd.PersistentPreRunE = concatCobraCmdFuncs(setEncoding, validateOutput, cmd.PersistentPreRunE) - return PrepareBaseCmd(cmd, envPrefix, defautRoot) + return PrepareBaseCmd(cmd, envPrefix, defaultHome) } // initEnv sets to use ENV variables if set. @@ -136,17 +130,10 @@ func bindFlagsLoadViper(cmd *cobra.Command, args []string) error { return err } - // rootDir is command line flag, env variable, or default $HOME/.tlc - // NOTE: we support both --root and --home for now, but eventually only --home - // Also ensure we set the correct rootDir under HomeFlag so we dont need to - // repeat this logic elsewhere. - rootDir := viper.GetString(HomeFlag) - if rootDir == "" { - rootDir = viper.GetString(RootFlag) - viper.Set(HomeFlag, rootDir) - } + homeDir := viper.GetString(HomeFlag) + viper.Set(HomeFlag, homeDir) viper.SetConfigName("config") // name of config file (without extension) - viper.AddConfigPath(rootDir) // search root directory + viper.AddConfigPath(homeDir) // search root directory // If a config file is found, read it in. if err := viper.ReadInConfig(); err == nil { diff --git a/cli/setup_test.go b/cli/setup_test.go index 692da26d3..2f085f7d5 100644 --- a/cli/setup_test.go +++ b/cli/setup_test.go @@ -74,16 +74,16 @@ func TestSetupConfig(t *testing.T) { // setting on the command line {[]string{"--boo", "haha"}, nil, "haha", ""}, {[]string{"--two-words", "rocks"}, nil, "", "rocks"}, - {[]string{"--root", conf1}, nil, cval1, ""}, + {[]string{"--home", conf1}, nil, cval1, ""}, // test both variants of the prefix - {nil, map[string]string{"RD_BOO": "bang"}, "bang", ""}, - {nil, map[string]string{"RD_TWO_WORDS": "fly"}, "", "fly"}, - {nil, map[string]string{"RDTWO_WORDS": "fly"}, "", "fly"}, - {nil, map[string]string{"RD_ROOT": conf1}, cval1, ""}, - {nil, map[string]string{"RDROOT": conf2}, cval2, "WORD"}, - {nil, map[string]string{"RDHOME": conf1}, cval1, ""}, + //{nil, map[string]string{"RD_BOO": "bang"}, "bang", ""}, + //{nil, map[string]string{"RD_TWO_WORDS": "fly"}, "", "fly"}, + //{nil, map[string]string{"RDTWO_WORDS": "fly"}, "", "fly"}, + //{nil, map[string]string{"RD_ROOT": conf1}, cval1, ""}, + //{nil, map[string]string{"RDROOT": conf2}, cval2, "WORD"}, + //{nil, map[string]string{"RDHOME": conf1}, cval1, ""}, // and when both are set??? HOME wins every time! - {[]string{"--root", conf1}, map[string]string{"RDHOME": conf2}, cval2, "WORD"}, + {[]string{"--home", conf1}, map[string]string{"RDHOME": conf2}, cval2, "WORD"}, } for idx, tc := range cases { @@ -156,10 +156,10 @@ func TestSetupUnmarshal(t *testing.T) { {nil, nil, c("", 0)}, // setting on the command line {[]string{"--name", "haha"}, nil, c("haha", 0)}, - {[]string{"--root", conf1}, nil, c(cval1, 0)}, + {[]string{"--home", conf1}, nil, c(cval1, 0)}, // test both variants of the prefix {nil, map[string]string{"MR_AGE": "56"}, c("", 56)}, - {nil, map[string]string{"MR_ROOT": conf1}, c(cval1, 0)}, + //{nil, map[string]string{"MR_ROOT": conf1}, c(cval1, 0)}, {[]string{"--age", "17"}, map[string]string{"MRHOME": conf2}, c(cval2, 17)}, } From 541780c6dff65a2d3554ac297ae2c7e61d8217f6 Mon Sep 17 00:00:00 2001 From: Anton Kaliaev Date: Tue, 12 Dec 2017 23:23:49 -0600 Subject: [PATCH 17/36] uncomment tests --- cli/setup_test.go | 20 +++++++------------- 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/cli/setup_test.go b/cli/setup_test.go index 2f085f7d5..e0fd75d8a 100644 --- a/cli/setup_test.go +++ b/cli/setup_test.go @@ -57,12 +57,9 @@ func TestSetupEnv(t *testing.T) { func TestSetupConfig(t *testing.T) { // we pre-create two config files we can refer to in the rest of // the test cases. - cval1, cval2 := "fubble", "wubble" + cval1 := "fubble" conf1, err := WriteDemoConfig(map[string]string{"boo": cval1}) require.Nil(t, err) - // make sure it handles dashed-words in the config, and ignores random info - conf2, err := WriteDemoConfig(map[string]string{"boo": cval2, "foo": "bar", "two-words": "WORD"}) - require.Nil(t, err) cases := []struct { args []string @@ -76,14 +73,11 @@ func TestSetupConfig(t *testing.T) { {[]string{"--two-words", "rocks"}, nil, "", "rocks"}, {[]string{"--home", conf1}, nil, cval1, ""}, // test both variants of the prefix - //{nil, map[string]string{"RD_BOO": "bang"}, "bang", ""}, - //{nil, map[string]string{"RD_TWO_WORDS": "fly"}, "", "fly"}, - //{nil, map[string]string{"RDTWO_WORDS": "fly"}, "", "fly"}, - //{nil, map[string]string{"RD_ROOT": conf1}, cval1, ""}, - //{nil, map[string]string{"RDROOT": conf2}, cval2, "WORD"}, - //{nil, map[string]string{"RDHOME": conf1}, cval1, ""}, - // and when both are set??? HOME wins every time! - {[]string{"--home", conf1}, map[string]string{"RDHOME": conf2}, cval2, "WORD"}, + {nil, map[string]string{"RD_BOO": "bang"}, "bang", ""}, + {nil, map[string]string{"RD_TWO_WORDS": "fly"}, "", "fly"}, + {nil, map[string]string{"RDTWO_WORDS": "fly"}, "", "fly"}, + {nil, map[string]string{"RD_HOME": conf1}, cval1, ""}, + {nil, map[string]string{"RDHOME": conf1}, cval1, ""}, } for idx, tc := range cases { @@ -159,7 +153,7 @@ func TestSetupUnmarshal(t *testing.T) { {[]string{"--home", conf1}, nil, c(cval1, 0)}, // test both variants of the prefix {nil, map[string]string{"MR_AGE": "56"}, c("", 56)}, - //{nil, map[string]string{"MR_ROOT": conf1}, c(cval1, 0)}, + {nil, map[string]string{"MR_HOME": conf1}, c(cval1, 0)}, {[]string{"--age", "17"}, map[string]string{"MRHOME": conf2}, c(cval2, 17)}, } From 29471d75cb50eb4cea5878b8bd1be25e8150564c Mon Sep 17 00:00:00 2001 From: Emmanuel Odeke Date: Wed, 13 Dec 2017 22:53:02 -0700 Subject: [PATCH 18/36] common: no more relying on math/rand.DefaultSource Fixes https://github.com/tendermint/tmlibs/issues/99 Updates https://github.com/tendermint/tendermint/issues/973 Removed usages of math/rand.DefaultSource in favour of our own source that's seeded with a completely random source and is safe for use in concurrent in multiple goroutines. Also extend some functionality that the stdlib exposes such as * RandPerm * RandIntn * RandInt31 * RandInt63 Also added an integration test whose purpose is to be run as a consistency check to ensure that our results never repeat hence that our internal PRNG is uniquely seeded each time. This integration test can be triggered by setting environment variable: `TENDERMINT_INTEGRATION_TESTS=true` for example ```shell TENDERMINT_INTEGRATION_TESTS=true go test ``` --- common/bit_array.go | 7 ++- common/random.go | 89 +++++++++++++++++++++++++--------- common/random_test.go | 108 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 178 insertions(+), 26 deletions(-) create mode 100644 common/random_test.go diff --git a/common/bit_array.go b/common/bit_array.go index 5590fe61b..848763b48 100644 --- a/common/bit_array.go +++ b/common/bit_array.go @@ -3,7 +3,6 @@ package common import ( "encoding/binary" "fmt" - "math/rand" "strings" "sync" ) @@ -212,12 +211,12 @@ func (bA *BitArray) PickRandom() (int, bool) { if length == 0 { return 0, false } - randElemStart := rand.Intn(length) + randElemStart := RandIntn(length) for i := 0; i < length; i++ { elemIdx := ((i + randElemStart) % length) if elemIdx < length-1 { if bA.Elems[elemIdx] > 0 { - randBitStart := rand.Intn(64) + randBitStart := RandIntn(64) for j := 0; j < 64; j++ { bitIdx := ((j + randBitStart) % 64) if (bA.Elems[elemIdx] & (uint64(1) << uint(bitIdx))) > 0 { @@ -232,7 +231,7 @@ func (bA *BitArray) PickRandom() (int, bool) { if elemBits == 0 { elemBits = 64 } - randBitStart := rand.Intn(elemBits) + randBitStart := RandIntn(elemBits) for j := 0; j < elemBits; j++ { bitIdx := ((j + randBitStart) % elemBits) if (bA.Elems[elemIdx] & (uint64(1) << uint(bitIdx))) > 0 { diff --git a/common/random.go b/common/random.go index 73bd16356..f0d169e09 100644 --- a/common/random.go +++ b/common/random.go @@ -3,6 +3,7 @@ package common import ( crand "crypto/rand" "math/rand" + "sync" "time" ) @@ -10,6 +11,11 @@ const ( strChars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" // 62 characters ) +var rng struct { + sync.Mutex + *rand.Rand +} + func init() { b := cRandBytes(8) var seed uint64 @@ -17,7 +23,7 @@ func init() { seed |= uint64(b[i]) seed <<= 8 } - rand.Seed(int64(seed)) + rng.Rand = rand.New(rand.NewSource(int64(seed))) } // Constructs an alphanumeric string of given length. @@ -25,7 +31,7 @@ func RandStr(length int) string { chars := []byte{} MAIN_LOOP: for { - val := rand.Int63() + val := rng.Int63() for i := 0; i < 10; i++ { v := int(val & 0x3f) // rightmost 6 bits if v >= 62 { // only 62 characters in strChars @@ -45,72 +51,98 @@ MAIN_LOOP: } func RandUint16() uint16 { - return uint16(rand.Uint32() & (1<<16 - 1)) + return uint16(RandUint32() & (1<<16 - 1)) } func RandUint32() uint32 { - return rand.Uint32() + rng.Lock() + u32 := rng.Uint32() + rng.Unlock() + return u32 } func RandUint64() uint64 { - return uint64(rand.Uint32())<<32 + uint64(rand.Uint32()) + return uint64(RandUint32())<<32 + uint64(RandUint32()) } func RandUint() uint { - return uint(rand.Int()) + rng.Lock() + i := rng.Int() + rng.Unlock() + return uint(i) } func RandInt16() int16 { - return int16(rand.Uint32() & (1<<16 - 1)) + return int16(RandUint32() & (1<<16 - 1)) } func RandInt32() int32 { - return int32(rand.Uint32()) + return int32(RandUint32()) } func RandInt64() int64 { - return int64(rand.Uint32())<<32 + int64(rand.Uint32()) + return int64(RandUint64()) } func RandInt() int { - return rand.Int() + rng.Lock() + i := rng.Int() + rng.Unlock() + return i +} + +func RandInt31() int32 { + rng.Lock() + i31 := rng.Int31() + rng.Unlock() + return i31 +} + +func RandInt63() int64 { + rng.Lock() + i63 := rng.Int63() + rng.Unlock() + return i63 } // Distributed pseudo-exponentially to test for various cases func RandUint16Exp() uint16 { - bits := rand.Uint32() % 16 + bits := RandUint32() % 16 if bits == 0 { return 0 } n := uint16(1 << (bits - 1)) - n += uint16(rand.Int31()) & ((1 << (bits - 1)) - 1) + n += uint16(RandInt31()) & ((1 << (bits - 1)) - 1) return n } // Distributed pseudo-exponentially to test for various cases func RandUint32Exp() uint32 { - bits := rand.Uint32() % 32 + bits := RandUint32() % 32 if bits == 0 { return 0 } n := uint32(1 << (bits - 1)) - n += uint32(rand.Int31()) & ((1 << (bits - 1)) - 1) + n += uint32(RandInt31()) & ((1 << (bits - 1)) - 1) return n } // Distributed pseudo-exponentially to test for various cases func RandUint64Exp() uint64 { - bits := rand.Uint32() % 64 + bits := RandUint32() % 64 if bits == 0 { return 0 } n := uint64(1 << (bits - 1)) - n += uint64(rand.Int63()) & ((1 << (bits - 1)) - 1) + n += uint64(RandInt63()) & ((1 << (bits - 1)) - 1) return n } func RandFloat32() float32 { - return rand.Float32() + rng.Lock() + f32 := rng.Float32() + rng.Unlock() + return f32 } func RandTime() time.Time { @@ -118,11 +150,24 @@ func RandTime() time.Time { } func RandBytes(n int) []byte { - bs := make([]byte, n) - for i := 0; i < n; i++ { - bs[i] = byte(rand.Intn(256)) - } - return bs + return cRandBytes(n) +} + +// RandIntn returns, as an int, a non-negative pseudo-random number in [0, n). +// It panics if n <= 0 +func RandIntn(n int) int { + rng.Lock() + i := rng.Intn(n) + rng.Unlock() + return i +} + +// RandPerm returns a pseudo-random permutation of n integers in [0, n). +func RandPerm(n int) []int { + rng.Lock() + perm := rng.Perm(n) + rng.Unlock() + return perm } // NOTE: This relies on the os's random number generator. diff --git a/common/random_test.go b/common/random_test.go new file mode 100644 index 000000000..dd803b3f6 --- /dev/null +++ b/common/random_test.go @@ -0,0 +1,108 @@ +package common_test + +import ( + "io/ioutil" + "os" + "os/exec" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/tendermint/tmlibs/common" +) + +// It is essential that these tests run and never repeat their outputs +// lest we've been pwned and the behavior of our randomness is controlled. +// See Issues: +// * https://github.com/tendermint/tmlibs/issues/99 +// * https://github.com/tendermint/tendermint/issues/973 +func TestUniqueRng(t *testing.T) { + if os.Getenv("TENDERMINT_INTEGRATION_TESTS") == "" { + t.Skipf("Can only be run as an integration test") + } + + // The goal of this test is to invoke the + // Rand* tests externally with no repeating results, booted up. + // Any repeated results indicate that the seed is the same or that + // perhaps we are using math/rand directly. + tmpDir, err := ioutil.TempDir("", "rng-tests") + if err != nil { + t.Fatalf("Creating tempDir: %v", err) + } + defer os.RemoveAll(tmpDir) + + outpath := filepath.Join(tmpDir, "main.go") + f, err := os.Create(outpath) + if err != nil { + t.Fatalf("Setting up %q err: %v", outpath, err) + } + f.Write([]byte(integrationTestProgram)) + if err := f.Close(); err != nil { + t.Fatalf("Closing: %v", err) + } + + outputs := make(map[string][]int) + for i := 0; i < 100; i++ { + cmd := exec.Command("go", "run", outpath) + bOutput, err := cmd.CombinedOutput() + if err != nil { + t.Errorf("Run #%d: err: %v output: %s", i, err, bOutput) + continue + } + output := string(bOutput) + runs, seen := outputs[output] + if seen { + t.Errorf("Run #%d's output was already seen in previous runs: %v", i, runs) + } + outputs[output] = append(outputs[output], i) + } +} + +const integrationTestProgram = ` +package main + +import ( + "encoding/json" + "fmt" + "math/rand" + + "github.com/tendermint/tmlibs/common" +) + +func main() { + // Set math/rand's Seed so that any direct invocations + // of math/rand will reveal themselves. + rand.Seed(1) + perm := common.RandPerm(10) + blob, _ := json.Marshal(perm) + fmt.Printf("perm: %s\n", blob) + + fmt.Printf("randInt: %d\n", common.RandInt()) + fmt.Printf("randUint: %d\n", common.RandUint()) + fmt.Printf("randIntn: %d\n", common.RandIntn(97)) + fmt.Printf("randInt31: %d\n", common.RandInt31()) + fmt.Printf("randInt32: %d\n", common.RandInt32()) + fmt.Printf("randInt63: %d\n", common.RandInt63()) + fmt.Printf("randInt64: %d\n", common.RandInt64()) + fmt.Printf("randUint32: %d\n", common.RandUint32()) + fmt.Printf("randUint64: %d\n", common.RandUint64()) + fmt.Printf("randUint16Exp: %d\n", common.RandUint16Exp()) + fmt.Printf("randUint32Exp: %d\n", common.RandUint32Exp()) + fmt.Printf("randUint64Exp: %d\n", common.RandUint64Exp()) +}` + +func TestRngConcurrencySafety(t *testing.T) { + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + wg.Add(1) + go func() { + defer wg.Done() + + _ = common.RandUint64() + <-time.After(time.Millisecond * time.Duration(common.RandIntn(100))) + _ = common.RandPerm(3) + }() + } + wg.Wait() +} From b5f465b4ecb6ef85a6ced14728a971570ed477e0 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Fri, 15 Dec 2017 00:23:25 -0500 Subject: [PATCH 19/36] common: use names prng and mrand --- common/random.go | 61 +++++++++++++++++++++++-------------------- common/random_test.go | 21 +++++++++++++++ 2 files changed, 53 insertions(+), 29 deletions(-) diff --git a/common/random.go b/common/random.go index f0d169e09..b8304e898 100644 --- a/common/random.go +++ b/common/random.go @@ -2,7 +2,7 @@ package common import ( crand "crypto/rand" - "math/rand" + mrand "math/rand" "sync" "time" ) @@ -11,9 +11,11 @@ const ( strChars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" // 62 characters ) -var rng struct { +// pseudo random number generator. +// seeded with OS randomness (crand) +var prng struct { sync.Mutex - *rand.Rand + *mrand.Rand } func init() { @@ -23,7 +25,7 @@ func init() { seed |= uint64(b[i]) seed <<= 8 } - rng.Rand = rand.New(rand.NewSource(int64(seed))) + prng.Rand = mrand.New(mrand.NewSource(int64(seed))) } // Constructs an alphanumeric string of given length. @@ -31,7 +33,7 @@ func RandStr(length int) string { chars := []byte{} MAIN_LOOP: for { - val := rng.Int63() + val := prng.Int63() for i := 0; i < 10; i++ { v := int(val & 0x3f) // rightmost 6 bits if v >= 62 { // only 62 characters in strChars @@ -55,9 +57,9 @@ func RandUint16() uint16 { } func RandUint32() uint32 { - rng.Lock() - u32 := rng.Uint32() - rng.Unlock() + prng.Lock() + u32 := prng.Uint32() + prng.Unlock() return u32 } @@ -66,9 +68,9 @@ func RandUint64() uint64 { } func RandUint() uint { - rng.Lock() - i := rng.Int() - rng.Unlock() + prng.Lock() + i := prng.Int() + prng.Unlock() return uint(i) } @@ -85,23 +87,23 @@ func RandInt64() int64 { } func RandInt() int { - rng.Lock() - i := rng.Int() - rng.Unlock() + prng.Lock() + i := prng.Int() + prng.Unlock() return i } func RandInt31() int32 { - rng.Lock() - i31 := rng.Int31() - rng.Unlock() + prng.Lock() + i31 := prng.Int31() + prng.Unlock() return i31 } func RandInt63() int64 { - rng.Lock() - i63 := rng.Int63() - rng.Unlock() + prng.Lock() + i63 := prng.Int63() + prng.Unlock() return i63 } @@ -139,9 +141,9 @@ func RandUint64Exp() uint64 { } func RandFloat32() float32 { - rng.Lock() - f32 := rng.Float32() - rng.Unlock() + prng.Lock() + f32 := prng.Float32() + prng.Unlock() return f32 } @@ -149,6 +151,7 @@ func RandTime() time.Time { return time.Unix(int64(RandUint64Exp()), 0) } +// RandBytes returns n random bytes from the OS's source of entropy ie. via crypto/rand. func RandBytes(n int) []byte { return cRandBytes(n) } @@ -156,17 +159,17 @@ func RandBytes(n int) []byte { // RandIntn returns, as an int, a non-negative pseudo-random number in [0, n). // It panics if n <= 0 func RandIntn(n int) int { - rng.Lock() - i := rng.Intn(n) - rng.Unlock() + prng.Lock() + i := prng.Intn(n) + prng.Unlock() return i } // RandPerm returns a pseudo-random permutation of n integers in [0, n). func RandPerm(n int) []int { - rng.Lock() - perm := rng.Perm(n) - rng.Unlock() + prng.Lock() + perm := prng.Perm(n) + prng.Unlock() return perm } diff --git a/common/random_test.go b/common/random_test.go index dd803b3f6..3fe0bbc06 100644 --- a/common/random_test.go +++ b/common/random_test.go @@ -9,9 +9,30 @@ import ( "testing" "time" + "github.com/stretchr/testify/assert" "github.com/tendermint/tmlibs/common" ) +func TestRandStr(t *testing.T) { + l := 243 + s := common.RandStr(l) + assert.Equal(t, l, len(s)) +} + +func TestRandBytes(t *testing.T) { + l := 243 + b := common.RandBytes(l) + assert.Equal(t, l, len(b)) +} + +func TestRandIntn(t *testing.T) { + n := 243 + for i := 0; i < 100; i++ { + x := common.RandIntn(n) + assert.True(t, x < n) + } +} + // It is essential that these tests run and never repeat their outputs // lest we've been pwned and the behavior of our randomness is controlled. // See Issues: From cdc798882326a722040706a87ec0397e7c91d517 Mon Sep 17 00:00:00 2001 From: Emmanuel Odeke Date: Fri, 15 Dec 2017 02:14:05 -0700 Subject: [PATCH 20/36] common: use genius simplification of tests from @ebuchman Massive test simplication for more reliable tests from @ebuchman --- common/random.go | 8 ++- common/random_test.go | 115 +++++++++++++++--------------------------- 2 files changed, 47 insertions(+), 76 deletions(-) diff --git a/common/random.go b/common/random.go index b8304e898..37b8b2773 100644 --- a/common/random.go +++ b/common/random.go @@ -18,14 +18,20 @@ var prng struct { *mrand.Rand } -func init() { +func reset() { b := cRandBytes(8) var seed uint64 for i := 0; i < 8; i++ { seed |= uint64(b[i]) seed <<= 8 } + prng.Lock() prng.Rand = mrand.New(mrand.NewSource(int64(seed))) + prng.Unlock() +} + +func init() { + reset() } // Constructs an alphanumeric string of given length. diff --git a/common/random_test.go b/common/random_test.go index 3fe0bbc06..bed8e7650 100644 --- a/common/random_test.go +++ b/common/random_test.go @@ -1,34 +1,34 @@ -package common_test +package common import ( - "io/ioutil" - "os" - "os/exec" - "path/filepath" + "bytes" + "encoding/json" + "fmt" + "io" + mrand "math/rand" "sync" "testing" "time" "github.com/stretchr/testify/assert" - "github.com/tendermint/tmlibs/common" ) func TestRandStr(t *testing.T) { l := 243 - s := common.RandStr(l) + s := RandStr(l) assert.Equal(t, l, len(s)) } func TestRandBytes(t *testing.T) { l := 243 - b := common.RandBytes(l) + b := RandBytes(l) assert.Equal(t, l, len(b)) } func TestRandIntn(t *testing.T) { n := 243 for i := 0; i < 100; i++ { - x := common.RandIntn(n) + x := RandIntn(n) assert.True(t, x < n) } } @@ -39,39 +39,12 @@ func TestRandIntn(t *testing.T) { // * https://github.com/tendermint/tmlibs/issues/99 // * https://github.com/tendermint/tendermint/issues/973 func TestUniqueRng(t *testing.T) { - if os.Getenv("TENDERMINT_INTEGRATION_TESTS") == "" { - t.Skipf("Can only be run as an integration test") - } - - // The goal of this test is to invoke the - // Rand* tests externally with no repeating results, booted up. - // Any repeated results indicate that the seed is the same or that - // perhaps we are using math/rand directly. - tmpDir, err := ioutil.TempDir("", "rng-tests") - if err != nil { - t.Fatalf("Creating tempDir: %v", err) - } - defer os.RemoveAll(tmpDir) - - outpath := filepath.Join(tmpDir, "main.go") - f, err := os.Create(outpath) - if err != nil { - t.Fatalf("Setting up %q err: %v", outpath, err) - } - f.Write([]byte(integrationTestProgram)) - if err := f.Close(); err != nil { - t.Fatalf("Closing: %v", err) - } - + buf := new(bytes.Buffer) outputs := make(map[string][]int) for i := 0; i < 100; i++ { - cmd := exec.Command("go", "run", outpath) - bOutput, err := cmd.CombinedOutput() - if err != nil { - t.Errorf("Run #%d: err: %v output: %s", i, err, bOutput) - continue - } - output := string(bOutput) + testThemAll(buf) + output := buf.String() + buf.Reset() runs, seen := outputs[output] if seen { t.Errorf("Run #%d's output was already seen in previous runs: %v", i, runs) @@ -80,38 +53,30 @@ func TestUniqueRng(t *testing.T) { } } -const integrationTestProgram = ` -package main - -import ( - "encoding/json" - "fmt" - "math/rand" - - "github.com/tendermint/tmlibs/common" -) - -func main() { - // Set math/rand's Seed so that any direct invocations - // of math/rand will reveal themselves. - rand.Seed(1) - perm := common.RandPerm(10) - blob, _ := json.Marshal(perm) - fmt.Printf("perm: %s\n", blob) - - fmt.Printf("randInt: %d\n", common.RandInt()) - fmt.Printf("randUint: %d\n", common.RandUint()) - fmt.Printf("randIntn: %d\n", common.RandIntn(97)) - fmt.Printf("randInt31: %d\n", common.RandInt31()) - fmt.Printf("randInt32: %d\n", common.RandInt32()) - fmt.Printf("randInt63: %d\n", common.RandInt63()) - fmt.Printf("randInt64: %d\n", common.RandInt64()) - fmt.Printf("randUint32: %d\n", common.RandUint32()) - fmt.Printf("randUint64: %d\n", common.RandUint64()) - fmt.Printf("randUint16Exp: %d\n", common.RandUint16Exp()) - fmt.Printf("randUint32Exp: %d\n", common.RandUint32Exp()) - fmt.Printf("randUint64Exp: %d\n", common.RandUint64Exp()) -}` +func testThemAll(out io.Writer) { + // Reset the internal PRNG + reset() + + // Set math/rand's Seed so that any direct invocations + // of math/rand will reveal themselves. + mrand.Seed(1) + perm := RandPerm(10) + blob, _ := json.Marshal(perm) + fmt.Fprintf(out, "perm: %s\n", blob) + + fmt.Fprintf(out, "randInt: %d\n", RandInt()) + fmt.Fprintf(out, "randUint: %d\n", RandUint()) + fmt.Fprintf(out, "randIntn: %d\n", RandIntn(97)) + fmt.Fprintf(out, "randInt31: %d\n", RandInt31()) + fmt.Fprintf(out, "randInt32: %d\n", RandInt32()) + fmt.Fprintf(out, "randInt63: %d\n", RandInt63()) + fmt.Fprintf(out, "randInt64: %d\n", RandInt64()) + fmt.Fprintf(out, "randUint32: %d\n", RandUint32()) + fmt.Fprintf(out, "randUint64: %d\n", RandUint64()) + fmt.Fprintf(out, "randUint16Exp: %d\n", RandUint16Exp()) + fmt.Fprintf(out, "randUint32Exp: %d\n", RandUint32Exp()) + fmt.Fprintf(out, "randUint64Exp: %d\n", RandUint64Exp()) +} func TestRngConcurrencySafety(t *testing.T) { var wg sync.WaitGroup @@ -120,9 +85,9 @@ func TestRngConcurrencySafety(t *testing.T) { go func() { defer wg.Done() - _ = common.RandUint64() - <-time.After(time.Millisecond * time.Duration(common.RandIntn(100))) - _ = common.RandPerm(3) + _ = RandUint64() + <-time.After(time.Millisecond * time.Duration(RandIntn(100))) + _ = RandPerm(3) }() } wg.Wait() From 8638961f02833def91f743cbccaa2cecdccffa74 Mon Sep 17 00:00:00 2001 From: Emmanuel Odeke Date: Fri, 15 Dec 2017 22:40:12 -0700 Subject: [PATCH 21/36] common: Rand* warnings about cryptographic unsafety Lesson articulated by @jaekwon on why we need 80 bits of entropy at least before we can think of cryptographic safety. math/rand's seed is a max of 64 bits so can never be cryptographically secure. Also added some benchmarks for RandBytes --- common/random.go | 29 +++++++++++++++++++++++++++-- common/random_test.go | 26 ++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/common/random.go b/common/random.go index 37b8b2773..9df55ff81 100644 --- a/common/random.go +++ b/common/random.go @@ -35,6 +35,7 @@ func init() { } // Constructs an alphanumeric string of given length. +// It is not safe for cryptographic usage. func RandStr(length int) string { chars := []byte{} MAIN_LOOP: @@ -58,10 +59,12 @@ MAIN_LOOP: return string(chars) } +// It is not safe for cryptographic usage. func RandUint16() uint16 { return uint16(RandUint32() & (1<<16 - 1)) } +// It is not safe for cryptographic usage. func RandUint32() uint32 { prng.Lock() u32 := prng.Uint32() @@ -69,10 +72,12 @@ func RandUint32() uint32 { return u32 } +// It is not safe for cryptographic usage. func RandUint64() uint64 { return uint64(RandUint32())<<32 + uint64(RandUint32()) } +// It is not safe for cryptographic usage. func RandUint() uint { prng.Lock() i := prng.Int() @@ -80,18 +85,22 @@ func RandUint() uint { return uint(i) } +// It is not safe for cryptographic usage. func RandInt16() int16 { return int16(RandUint32() & (1<<16 - 1)) } +// It is not safe for cryptographic usage. func RandInt32() int32 { return int32(RandUint32()) } +// It is not safe for cryptographic usage. func RandInt64() int64 { return int64(RandUint64()) } +// It is not safe for cryptographic usage. func RandInt() int { prng.Lock() i := prng.Int() @@ -99,6 +108,7 @@ func RandInt() int { return i } +// It is not safe for cryptographic usage. func RandInt31() int32 { prng.Lock() i31 := prng.Int31() @@ -106,6 +116,7 @@ func RandInt31() int32 { return i31 } +// It is not safe for cryptographic usage. func RandInt63() int64 { prng.Lock() i63 := prng.Int63() @@ -114,6 +125,7 @@ func RandInt63() int64 { } // Distributed pseudo-exponentially to test for various cases +// It is not safe for cryptographic usage. func RandUint16Exp() uint16 { bits := RandUint32() % 16 if bits == 0 { @@ -125,6 +137,7 @@ func RandUint16Exp() uint16 { } // Distributed pseudo-exponentially to test for various cases +// It is not safe for cryptographic usage. func RandUint32Exp() uint32 { bits := RandUint32() % 32 if bits == 0 { @@ -136,6 +149,7 @@ func RandUint32Exp() uint32 { } // Distributed pseudo-exponentially to test for various cases +// It is not safe for cryptographic usage. func RandUint64Exp() uint64 { bits := RandUint32() % 64 if bits == 0 { @@ -146,6 +160,7 @@ func RandUint64Exp() uint64 { return n } +// It is not safe for cryptographic usage. func RandFloat32() float32 { prng.Lock() f32 := prng.Float32() @@ -153,17 +168,26 @@ func RandFloat32() float32 { return f32 } +// It is not safe for cryptographic usage. func RandTime() time.Time { return time.Unix(int64(RandUint64Exp()), 0) } // RandBytes returns n random bytes from the OS's source of entropy ie. via crypto/rand. +// It is not safe for cryptographic usage. func RandBytes(n int) []byte { - return cRandBytes(n) + // cRandBytes isn't guaranteed to be fast so instead + // use random bytes generated from the internal PRNG + bs := make([]byte, n) + for i := 0; i < len(bs); i++ { + bs[i] = byte(RandInt() & 0xFF) + } + return bs } // RandIntn returns, as an int, a non-negative pseudo-random number in [0, n). -// It panics if n <= 0 +// It panics if n <= 0. +// It is not safe for cryptographic usage. func RandIntn(n int) int { prng.Lock() i := prng.Intn(n) @@ -172,6 +196,7 @@ func RandIntn(n int) int { } // RandPerm returns a pseudo-random permutation of n integers in [0, n). +// It is not safe for cryptographic usage. func RandPerm(n int) []int { prng.Lock() perm := prng.Perm(n) diff --git a/common/random_test.go b/common/random_test.go index bed8e7650..216f2f8bc 100644 --- a/common/random_test.go +++ b/common/random_test.go @@ -92,3 +92,29 @@ func TestRngConcurrencySafety(t *testing.T) { } wg.Wait() } + +func BenchmarkRandBytes10B(b *testing.B) { + benchmarkRandBytes(b, 10) +} +func BenchmarkRandBytes100B(b *testing.B) { + benchmarkRandBytes(b, 100) +} +func BenchmarkRandBytes1KiB(b *testing.B) { + benchmarkRandBytes(b, 1024) +} +func BenchmarkRandBytes10KiB(b *testing.B) { + benchmarkRandBytes(b, 10*1024) +} +func BenchmarkRandBytes100KiB(b *testing.B) { + benchmarkRandBytes(b, 100*1024) +} +func BenchmarkRandBytes1MiB(b *testing.B) { + benchmarkRandBytes(b, 1024*1024) +} + +func benchmarkRandBytes(b *testing.B, n int) { + for i := 0; i < b.N; i++ { + _ = RandBytes(n) + } + b.ReportAllocs() +} From 70e30f74e60b2710c3c270178b0be2c4c7319722 Mon Sep 17 00:00:00 2001 From: Anton Kaliaev Date: Tue, 19 Dec 2017 16:16:16 -0600 Subject: [PATCH 22/36] Revert "Refactor repeat timer" --- common/repeat_timer.go | 116 +++++++++++++++------------------- common/repeat_timer_test.go | 12 ++-- common/throttle_timer.go | 24 ++++--- common/throttle_timer_test.go | 2 - 4 files changed, 67 insertions(+), 87 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index 23faf74ae..d7d9154d4 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -1,6 +1,7 @@ package common import ( + "sync" "time" ) @@ -10,93 +11,76 @@ It's good for keeping connections alive. A RepeatTimer must be Stop()'d or it will keep a goroutine alive. */ type RepeatTimer struct { - Name string - Ch <-chan time.Time - output chan<- time.Time - input chan repeatCommand + Ch chan time.Time - dur time.Duration - ticker *time.Ticker - stopped bool + mtx sync.Mutex + name string + ticker *time.Ticker + quit chan struct{} + wg *sync.WaitGroup + dur time.Duration } -type repeatCommand int8 - -const ( - Reset repeatCommand = iota - RQuit -) - func NewRepeatTimer(name string, dur time.Duration) *RepeatTimer { - c := make(chan time.Time) var t = &RepeatTimer{ - Name: name, - Ch: c, - output: c, - input: make(chan repeatCommand), - - dur: dur, + Ch: make(chan time.Time), ticker: time.NewTicker(dur), + quit: make(chan struct{}), + wg: new(sync.WaitGroup), + name: name, + dur: dur, } - go t.run() + t.wg.Add(1) + go t.fireRoutine(t.ticker) return t } +func (t *RepeatTimer) fireRoutine(ticker *time.Ticker) { + for { + select { + case t_ := <-ticker.C: + t.Ch <- t_ + case <-t.quit: + // needed so we know when we can reset t.quit + t.wg.Done() + return + } + } +} + // Wait the duration again before firing. func (t *RepeatTimer) Reset() { - t.input <- Reset + t.Stop() + + t.mtx.Lock() // Lock + defer t.mtx.Unlock() + + t.ticker = time.NewTicker(t.dur) + t.quit = make(chan struct{}) + t.wg.Add(1) + go t.fireRoutine(t.ticker) } // For ease of .Stop()'ing services before .Start()'ing them, // we ignore .Stop()'s on nil RepeatTimers. func (t *RepeatTimer) Stop() bool { - // use t.stopped to gracefully handle many Stop() without blocking - if t == nil || t.stopped { + if t == nil { return false } - t.input <- RQuit - t.stopped = true - return true -} + t.mtx.Lock() // Lock + defer t.mtx.Unlock() -func (t *RepeatTimer) run() { - done := false - for !done { + exists := t.ticker != nil + if exists { + t.ticker.Stop() // does not close the channel select { - case cmd := <-t.input: - // stop goroutine if the input says so - // don't close channels, as closed channels mess up select reads - done = t.processInput(cmd) - case tick := <-t.ticker.C: - t.send(tick) + case <-t.Ch: + // read off channel if there's anything there + default: } + close(t.quit) + t.wg.Wait() // must wait for quit to close else we race Reset + t.ticker = nil } -} - -// send performs blocking send on t.Ch -func (t *RepeatTimer) send(tick time.Time) { - // XXX: possibly it is better to not block: - // https://golang.org/src/time/sleep.go#L132 - // select { - // case t.output <- tick: - // default: - // } - t.output <- tick -} - -// all modifications of the internal state of ThrottleTimer -// happen in this method. It is only called from the run goroutine -// so we avoid any race conditions -func (t *RepeatTimer) processInput(cmd repeatCommand) (shutdown bool) { - switch cmd { - case Reset: - t.ticker.Stop() - t.ticker = time.NewTicker(t.dur) - case RQuit: - t.ticker.Stop() - shutdown = true - default: - panic("unknown command!") - } - return shutdown + return exists } diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index db53aa614..87f34b950 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -10,7 +10,7 @@ import ( ) type rCounter struct { - input <-chan time.Time + input chan time.Time mtx sync.Mutex count int } @@ -39,11 +39,11 @@ func (c *rCounter) Read() { func TestRepeat(test *testing.T) { assert := asrt.New(test) - dur := time.Duration(100) * time.Millisecond + dur := time.Duration(50) * time.Millisecond short := time.Duration(20) * time.Millisecond // delay waits for cnt durations, an a little extra delay := func(cnt int) time.Duration { - return time.Duration(cnt)*dur + time.Duration(10)*time.Millisecond + return time.Duration(cnt)*dur + time.Duration(5)*time.Millisecond } t := NewRepeatTimer("bar", dur) @@ -70,9 +70,9 @@ func TestRepeat(test *testing.T) { // after a stop, nothing more is sent stopped := t.Stop() assert.True(stopped) - time.Sleep(delay(2)) + time.Sleep(delay(7)) assert.Equal(6, c.Count()) - // extra calls to stop don't block - t.Stop() + // close channel to stop counter + close(t.Ch) } diff --git a/common/throttle_timer.go b/common/throttle_timer.go index a5bd6ded8..ab2ad2e62 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -13,21 +13,20 @@ at most once every "dur". type ThrottleTimer struct { Name string Ch <-chan struct{} - input chan throttleCommand + input chan command output chan<- struct{} dur time.Duration - timer *time.Timer - isSet bool - stopped bool + timer *time.Timer + isSet bool } -type throttleCommand int8 +type command int32 const ( - Set throttleCommand = iota + Set command = iota Unset - TQuit + Quit ) // NewThrottleTimer creates a new ThrottleTimer. @@ -37,7 +36,7 @@ func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer { Name: name, Ch: c, dur: dur, - input: make(chan throttleCommand), + input: make(chan command), output: c, timer: time.NewTimer(dur), } @@ -75,14 +74,14 @@ func (t *ThrottleTimer) trySend() { // all modifications of the internal state of ThrottleTimer // happen in this method. It is only called from the run goroutine // so we avoid any race conditions -func (t *ThrottleTimer) processInput(cmd throttleCommand) (shutdown bool) { +func (t *ThrottleTimer) processInput(cmd command) (shutdown bool) { switch cmd { case Set: if !t.isSet { t.isSet = true t.timer.Reset(t.dur) } - case TQuit: + case Quit: shutdown = true fallthrough case Unset: @@ -120,10 +119,9 @@ func (t *ThrottleTimer) Unset() { // For ease of stopping services before starting them, we ignore Stop on nil // ThrottleTimers. func (t *ThrottleTimer) Stop() bool { - if t == nil || t.stopped { + if t == nil { return false } - t.input <- TQuit - t.stopped = true + t.input <- Quit return true } diff --git a/common/throttle_timer_test.go b/common/throttle_timer_test.go index 94ec1b43c..a1b6606f5 100644 --- a/common/throttle_timer_test.go +++ b/common/throttle_timer_test.go @@ -95,6 +95,4 @@ func TestThrottle(test *testing.T) { stopped := t.Stop() assert.True(stopped) - // extra calls to stop don't block - t.Stop() } From e17e8e425f43890b207e5e316f5190d278e849c3 Mon Sep 17 00:00:00 2001 From: Anton Kaliaev Date: Tue, 19 Dec 2017 16:23:20 -0600 Subject: [PATCH 23/36] Revert "Refactor throttle timer" --- common/repeat_timer_test.go | 2 +- common/throttle_timer.go | 120 ++++++++++------------------------ common/throttle_timer_test.go | 24 +------ 3 files changed, 37 insertions(+), 109 deletions(-) diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index 87f34b950..9f03f41df 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -43,7 +43,7 @@ func TestRepeat(test *testing.T) { short := time.Duration(20) * time.Millisecond // delay waits for cnt durations, an a little extra delay := func(cnt int) time.Duration { - return time.Duration(cnt)*dur + time.Duration(5)*time.Millisecond + return time.Duration(cnt)*dur + time.Millisecond } t := NewRepeatTimer("bar", dur) diff --git a/common/throttle_timer.go b/common/throttle_timer.go index ab2ad2e62..38ef4e9a3 100644 --- a/common/throttle_timer.go +++ b/common/throttle_timer.go @@ -1,6 +1,7 @@ package common import ( + "sync" "time" ) @@ -11,117 +12,64 @@ If a long continuous burst of .Set() calls happens, ThrottleTimer fires at most once every "dur". */ type ThrottleTimer struct { - Name string - Ch <-chan struct{} - input chan command - output chan<- struct{} - dur time.Duration + Name string + Ch chan struct{} + quit chan struct{} + dur time.Duration + mtx sync.Mutex timer *time.Timer isSet bool } -type command int32 - -const ( - Set command = iota - Unset - Quit -) - -// NewThrottleTimer creates a new ThrottleTimer. func NewThrottleTimer(name string, dur time.Duration) *ThrottleTimer { - c := make(chan struct{}) - var t = &ThrottleTimer{ - Name: name, - Ch: c, - dur: dur, - input: make(chan command), - output: c, - timer: time.NewTimer(dur), - } + var ch = make(chan struct{}) + var quit = make(chan struct{}) + var t = &ThrottleTimer{Name: name, Ch: ch, dur: dur, quit: quit} + t.mtx.Lock() + t.timer = time.AfterFunc(dur, t.fireRoutine) + t.mtx.Unlock() t.timer.Stop() - go t.run() return t } -func (t *ThrottleTimer) run() { - for { - select { - case cmd := <-t.input: - // stop goroutine if the input says so - // don't close channels, as closed channels mess up select reads - if t.processInput(cmd) { - return - } - case <-t.timer.C: - t.trySend() - } - } -} - -// trySend performs non-blocking send on t.Ch -func (t *ThrottleTimer) trySend() { +func (t *ThrottleTimer) fireRoutine() { + t.mtx.Lock() + defer t.mtx.Unlock() select { - case t.output <- struct{}{}: + case t.Ch <- struct{}{}: t.isSet = false + case <-t.quit: + // do nothing default: - // if we just want to drop, replace this with t.isSet = false t.timer.Reset(t.dur) } } -// all modifications of the internal state of ThrottleTimer -// happen in this method. It is only called from the run goroutine -// so we avoid any race conditions -func (t *ThrottleTimer) processInput(cmd command) (shutdown bool) { - switch cmd { - case Set: - if !t.isSet { - t.isSet = true - t.timer.Reset(t.dur) - } - case Quit: - shutdown = true - fallthrough - case Unset: - if t.isSet { - t.isSet = false - t.timer.Stop() - } - default: - panic("unknown command!") - } - return shutdown -} - func (t *ThrottleTimer) Set() { - t.input <- Set + t.mtx.Lock() + defer t.mtx.Unlock() + if !t.isSet { + t.isSet = true + t.timer.Reset(t.dur) + } } func (t *ThrottleTimer) Unset() { - t.input <- Unset + t.mtx.Lock() + defer t.mtx.Unlock() + t.isSet = false + t.timer.Stop() } -// Stop prevents the ThrottleTimer from firing. It always returns true. Stop does not -// close the channel, to prevent a read from the channel succeeding -// incorrectly. -// -// To prevent a timer created with NewThrottleTimer from firing after a call to -// Stop, check the return value and drain the channel. -// -// For example, assuming the program has not received from t.C already: -// -// if !t.Stop() { -// <-t.C -// } -// -// For ease of stopping services before starting them, we ignore Stop on nil -// ThrottleTimers. +// For ease of .Stop()'ing services before .Start()'ing them, +// we ignore .Stop()'s on nil ThrottleTimers func (t *ThrottleTimer) Stop() bool { if t == nil { return false } - t.input <- Quit - return true + close(t.quit) + t.mtx.Lock() + defer t.mtx.Unlock() + return t.timer.Stop() } diff --git a/common/throttle_timer_test.go b/common/throttle_timer_test.go index a1b6606f5..00f5abdec 100644 --- a/common/throttle_timer_test.go +++ b/common/throttle_timer_test.go @@ -10,7 +10,7 @@ import ( ) type thCounter struct { - input <-chan struct{} + input chan struct{} mtx sync.Mutex count int } @@ -31,9 +31,6 @@ func (c *thCounter) Count() int { // Read should run in a go-routine and // updates count by one every time a packet comes in func (c *thCounter) Read() { - // note, since this channel never closes, this will never end - // if thCounter was used in anything beyond trivial test cases. - // it would have to be smarter. for range c.input { c.Increment() } @@ -44,7 +41,6 @@ func TestThrottle(test *testing.T) { ms := 50 delay := time.Duration(ms) * time.Millisecond - shortwait := time.Duration(ms/2) * time.Millisecond longwait := time.Duration(2) * delay t := NewThrottleTimer("foo", delay) @@ -69,21 +65,6 @@ func TestThrottle(test *testing.T) { time.Sleep(longwait) assert.Equal(2, c.Count()) - // keep cancelling before it is ready - for i := 0; i < 10; i++ { - t.Set() - time.Sleep(shortwait) - t.Unset() - } - time.Sleep(longwait) - assert.Equal(2, c.Count()) - - // a few unsets do nothing... - for i := 0; i < 5; i++ { - t.Unset() - } - assert.Equal(2, c.Count()) - // send 12, over 2 delay sections, adds 3 short := time.Duration(ms/5) * time.Millisecond for i := 0; i < 13; i++ { @@ -93,6 +74,5 @@ func TestThrottle(test *testing.T) { time.Sleep(longwait) assert.Equal(5, c.Count()) - stopped := t.Stop() - assert.True(stopped) + close(t.Ch) } From a25ed5ba1b0124f82f77b722cf3225cf4b3f18f5 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 21 Dec 2017 10:02:25 -0500 Subject: [PATCH 24/36] cmn: fix race condition in prng --- common/random.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/random.go b/common/random.go index 9df55ff81..ca71b6143 100644 --- a/common/random.go +++ b/common/random.go @@ -40,7 +40,7 @@ func RandStr(length int) string { chars := []byte{} MAIN_LOOP: for { - val := prng.Int63() + val := RandInt63() for i := 0; i < 10; i++ { v := int(val & 0x3f) // rightmost 6 bits if v >= 62 { // only 62 characters in strChars From b0b740210c60b7fc789382ff3a709426eb71903d Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 21 Dec 2017 11:15:17 -0500 Subject: [PATCH 25/36] cmn: fix repeate timer test with manual ticker --- common/repeat_timer.go | 86 +++++++++++++++++++++++++++++++++---- common/repeat_timer_test.go | 81 ++++++++++++++++------------------ 2 files changed, 114 insertions(+), 53 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index d7d9154d4..1500e95d1 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -5,6 +5,72 @@ import ( "time" ) +// Ticker is a basic ticker interface. +type Ticker interface { + Chan() <-chan time.Time + Stop() + Reset() +} + +// DefaultTicker wraps the stdlibs Ticker implementation. +type DefaultTicker struct { + t *time.Ticker + dur time.Duration +} + +// NewDefaultTicker returns a new DefaultTicker +func NewDefaultTicker(dur time.Duration) *DefaultTicker { + return &DefaultTicker{ + time.NewTicker(dur), + dur, + } +} + +// Implements Ticker +func (t *DefaultTicker) Chan() <-chan time.Time { + return t.t.C +} + +// Implements Ticker +func (t *DefaultTicker) Stop() { + t.t.Stop() + t.t = nil +} + +// Implements Ticker +func (t *DefaultTicker) Reset() { + t.t = time.NewTicker(t.dur) +} + +// ManualTicker wraps a channel that can be manually sent on +type ManualTicker struct { + ch chan time.Time +} + +// NewManualTicker returns a new ManualTicker +func NewManualTicker(ch chan time.Time) *ManualTicker { + return &ManualTicker{ + ch: ch, + } +} + +// Implements Ticker +func (t *ManualTicker) Chan() <-chan time.Time { + return t.ch +} + +// Implements Ticker +func (t *ManualTicker) Stop() { + // noop +} + +// Implements Ticker +func (t *ManualTicker) Reset() { + // noop +} + +//--------------------------------------------------------------------- + /* RepeatTimer repeatedly sends a struct{}{} to .Ch after each "dur" period. It's good for keeping connections alive. @@ -15,30 +81,35 @@ type RepeatTimer struct { mtx sync.Mutex name string - ticker *time.Ticker + ticker Ticker quit chan struct{} wg *sync.WaitGroup - dur time.Duration } +// NewRepeatTimer returns a RepeatTimer with the DefaultTicker. func NewRepeatTimer(name string, dur time.Duration) *RepeatTimer { + ticker := NewDefaultTicker(dur) + return NewRepeatTimerWithTicker(name, ticker) +} + +// NewRepeatTimerWithTicker returns a RepeatTimer with the given ticker. +func NewRepeatTimerWithTicker(name string, ticker Ticker) *RepeatTimer { var t = &RepeatTimer{ Ch: make(chan time.Time), - ticker: time.NewTicker(dur), + ticker: ticker, quit: make(chan struct{}), wg: new(sync.WaitGroup), name: name, - dur: dur, } t.wg.Add(1) go t.fireRoutine(t.ticker) return t } -func (t *RepeatTimer) fireRoutine(ticker *time.Ticker) { +func (t *RepeatTimer) fireRoutine(ticker Ticker) { for { select { - case t_ := <-ticker.C: + case t_ := <-ticker.Chan(): t.Ch <- t_ case <-t.quit: // needed so we know when we can reset t.quit @@ -55,7 +126,7 @@ func (t *RepeatTimer) Reset() { t.mtx.Lock() // Lock defer t.mtx.Unlock() - t.ticker = time.NewTicker(t.dur) + t.ticker.Reset() t.quit = make(chan struct{}) t.wg.Add(1) go t.fireRoutine(t.ticker) @@ -80,7 +151,6 @@ func (t *RepeatTimer) Stop() bool { } close(t.quit) t.wg.Wait() // must wait for quit to close else we race Reset - t.ticker = nil } return exists } diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index 9f03f41df..98d991e9c 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -1,7 +1,6 @@ package common import ( - "sync" "testing" "time" @@ -9,69 +8,61 @@ import ( asrt "github.com/stretchr/testify/assert" ) -type rCounter struct { - input chan time.Time - mtx sync.Mutex - count int -} - -func (c *rCounter) Increment() { - c.mtx.Lock() - c.count++ - c.mtx.Unlock() -} - -func (c *rCounter) Count() int { - c.mtx.Lock() - val := c.count - c.mtx.Unlock() - return val -} - -// Read should run in a go-routine and -// updates count by one every time a packet comes in -func (c *rCounter) Read() { - for range c.input { - c.Increment() - } -} - +// NOTE: this only tests with the ManualTicker. +// How do you test a real-clock ticker properly? func TestRepeat(test *testing.T) { assert := asrt.New(test) - dur := time.Duration(50) * time.Millisecond - short := time.Duration(20) * time.Millisecond - // delay waits for cnt durations, an a little extra - delay := func(cnt int) time.Duration { - return time.Duration(cnt)*dur + time.Millisecond + ch := make(chan time.Time, 100) + // tick fires cnt times on ch + tick := func(cnt int) { + for i := 0; i < cnt; i++ { + ch <- time.Now() + } } - t := NewRepeatTimer("bar", dur) + tock := func(test *testing.T, t *RepeatTimer, cnt int) { + for i := 0; i < cnt; i++ { + after := time.After(time.Second * 2) + select { + case <-t.Ch: + case <-after: + test.Fatal("expected ticker to fire") + } + } + done := true + select { + case <-t.Ch: + done = false + default: + } + assert.True(done) + } + + ticker := NewManualTicker(ch) + t := NewRepeatTimerWithTicker("bar", ticker) // start at 0 - c := &rCounter{input: t.Ch} - go c.Read() - assert.Equal(0, c.Count()) + tock(test, t, 0) // wait for 4 periods - time.Sleep(delay(4)) - assert.Equal(4, c.Count()) + tick(4) + tock(test, t, 4) // keep reseting leads to no firing for i := 0; i < 20; i++ { - time.Sleep(short) + time.Sleep(time.Millisecond) t.Reset() } - assert.Equal(4, c.Count()) + tock(test, t, 0) // after this, it still works normal - time.Sleep(delay(2)) - assert.Equal(6, c.Count()) + tick(2) + tock(test, t, 2) // after a stop, nothing more is sent stopped := t.Stop() assert.True(stopped) - time.Sleep(delay(7)) - assert.Equal(6, c.Count()) + tock(test, t, 0) // close channel to stop counter close(t.Ch) From e2d7f1aa41dde5f29057dd08e64371a574b84c86 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 21 Dec 2017 14:21:15 -0500 Subject: [PATCH 26/36] cmn: fix race --- common/repeat_timer.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index 1500e95d1..0bc4d87b4 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -102,14 +102,14 @@ func NewRepeatTimerWithTicker(name string, ticker Ticker) *RepeatTimer { name: name, } t.wg.Add(1) - go t.fireRoutine(t.ticker) + go t.fireRoutine(t.ticker.Chan()) return t } -func (t *RepeatTimer) fireRoutine(ticker Ticker) { +func (t *RepeatTimer) fireRoutine(ch <-chan time.Time) { for { select { - case t_ := <-ticker.Chan(): + case t_ := <-ch: t.Ch <- t_ case <-t.quit: // needed so we know when we can reset t.quit @@ -129,7 +129,7 @@ func (t *RepeatTimer) Reset() { t.ticker.Reset() t.quit = make(chan struct{}) t.wg.Add(1) - go t.fireRoutine(t.ticker) + go t.fireRoutine(t.ticker.Chan()) } // For ease of .Stop()'ing services before .Start()'ing them, From 2fd8f35b74e80382e276393b6edaa4464642a9df Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Mon, 25 Dec 2017 21:12:14 -0800 Subject: [PATCH 27/36] Fix #112 by using RWMutex per element --- clist/clist.go | 222 +++++++++++++++++++++++++++++-------------------- 1 file changed, 130 insertions(+), 92 deletions(-) diff --git a/clist/clist.go b/clist/clist.go index 5295dd995..e8cf6b93c 100644 --- a/clist/clist.go +++ b/clist/clist.go @@ -11,36 +11,38 @@ to ensure garbage collection of removed elements. import ( "sync" - "sync/atomic" - "unsafe" ) // CElement is an element of a linked-list // Traversal from a CElement are goroutine-safe. type CElement struct { - prev unsafe.Pointer + mtx sync.RWMutex + prev *CElement prevWg *sync.WaitGroup - next unsafe.Pointer + next *CElement nextWg *sync.WaitGroup - removed uint32 - Value interface{} + removed bool + + Value interface{} // immutable } // Blocking implementation of Next(). // May return nil iff CElement was tail and got removed. func (e *CElement) NextWait() *CElement { for { - e.nextWg.Wait() - next := e.Next() - if next == nil { - if e.Removed() { - return nil - } else { - continue - } - } else { + e.mtx.RLock() + next := e.next + nextWg := e.nextWg + removed := e.removed + e.mtx.RUnlock() + + if next != nil || removed { return next } + + nextWg.Wait() + // e.next doesn't necessarily exist here. + // That's why we need to continue a for-loop. } } @@ -48,82 +50,113 @@ func (e *CElement) NextWait() *CElement { // May return nil iff CElement was head and got removed. func (e *CElement) PrevWait() *CElement { for { - e.prevWg.Wait() - prev := e.Prev() - if prev == nil { - if e.Removed() { - return nil - } else { - continue - } - } else { + e.mtx.RLock() + prev := e.prev + prevWg := e.prevWg + removed := e.removed + e.mtx.RUnlock() + + if prev != nil || removed { return prev } + + prevWg.Wait() } } // Nonblocking, may return nil if at the end. func (e *CElement) Next() *CElement { - return (*CElement)(atomic.LoadPointer(&e.next)) + e.mtx.RLock() + defer e.mtx.RUnlock() + + return e.next } // Nonblocking, may return nil if at the end. func (e *CElement) Prev() *CElement { - return (*CElement)(atomic.LoadPointer(&e.prev)) + e.mtx.RLock() + defer e.mtx.RUnlock() + + return e.prev } func (e *CElement) Removed() bool { - return atomic.LoadUint32(&(e.removed)) > 0 + e.mtx.RLock() + defer e.mtx.RUnlock() + + return e.removed } func (e *CElement) DetachNext() { if !e.Removed() { panic("DetachNext() must be called after Remove(e)") } - atomic.StorePointer(&e.next, nil) + e.mtx.Lock() + defer e.mtx.Unlock() + + e.next = nil } func (e *CElement) DetachPrev() { if !e.Removed() { panic("DetachPrev() must be called after Remove(e)") } - atomic.StorePointer(&e.prev, nil) + e.mtx.Lock() + defer e.mtx.Unlock() + + e.prev = nil } -func (e *CElement) setNextAtomic(next *CElement) { - for { - oldNext := atomic.LoadPointer(&e.next) - if !atomic.CompareAndSwapPointer(&(e.next), oldNext, unsafe.Pointer(next)) { - continue - } - if next == nil && oldNext != nil { // We for-loop in NextWait() so race is ok - e.nextWg.Add(1) - } - if next != nil && oldNext == nil { - e.nextWg.Done() - } - return +// NOTE: This function needs to be safe for +// concurrent goroutines waiting on nextWg. +func (e *CElement) SetNext(newNext *CElement) { + e.mtx.Lock() + defer e.mtx.Unlock() + + oldNext := e.next + e.next = newNext + if oldNext != nil && newNext == nil { + // See https://golang.org/pkg/sync/: + // + // If a WaitGroup is reused to wait for several independent sets of + // events, new Add calls must happen after all previous Wait calls have + // returned. + e.nextWg = waitGroup1() // WaitGroups are difficult to re-use. + } + if oldNext == nil && newNext != nil { + e.nextWg.Done() } } -func (e *CElement) setPrevAtomic(prev *CElement) { - for { - oldPrev := atomic.LoadPointer(&e.prev) - if !atomic.CompareAndSwapPointer(&(e.prev), oldPrev, unsafe.Pointer(prev)) { - continue - } - if prev == nil && oldPrev != nil { // We for-loop in PrevWait() so race is ok - e.prevWg.Add(1) - } - if prev != nil && oldPrev == nil { - e.prevWg.Done() - } - return +// NOTE: This function needs to be safe for +// concurrent goroutines waiting on prevWg +func (e *CElement) SetPrev(newPrev *CElement) { + e.mtx.Lock() + defer e.mtx.Unlock() + + oldPrev := e.prev + e.prev = newPrev + if oldPrev != nil && newPrev == nil { + e.prevWg = waitGroup1() // WaitGroups are difficult to re-use. + } + if oldPrev == nil && newPrev != nil { + e.prevWg.Done() } } -func (e *CElement) setRemovedAtomic() { - atomic.StoreUint32(&(e.removed), 1) +func (e *CElement) SetRemoved() { + e.mtx.Lock() + defer e.mtx.Unlock() + + e.removed = true + + // This wakes up anyone waiting in either direction. + if e.prev == nil { + e.prevWg.Done() + } + if e.next == nil { + e.nextWg.Done() + } } //-------------------------------------------------------------------------------- @@ -132,7 +165,7 @@ func (e *CElement) setRemovedAtomic() { // The zero value for CList is an empty list ready to use. // Operations are goroutine-safe. type CList struct { - mtx sync.Mutex + mtx sync.RWMutex wg *sync.WaitGroup head *CElement // first element tail *CElement // last element @@ -142,6 +175,7 @@ type CList struct { func (l *CList) Init() *CList { l.mtx.Lock() defer l.mtx.Unlock() + l.wg = waitGroup1() l.head = nil l.tail = nil @@ -152,48 +186,55 @@ func (l *CList) Init() *CList { func New() *CList { return new(CList).Init() } func (l *CList) Len() int { - l.mtx.Lock() - defer l.mtx.Unlock() + l.mtx.RLock() + defer l.mtx.RUnlock() + return l.len } func (l *CList) Front() *CElement { - l.mtx.Lock() - defer l.mtx.Unlock() + l.mtx.RLock() + defer l.mtx.RUnlock() + return l.head } func (l *CList) FrontWait() *CElement { for { - l.mtx.Lock() + l.mtx.RLock() head := l.head wg := l.wg - l.mtx.Unlock() - if head == nil { - wg.Wait() - } else { + l.mtx.RUnlock() + + if head != nil { return head } + wg.Wait() + // l.head doesn't necessarily exist here. + // That's why we need to continue a for-loop. } } func (l *CList) Back() *CElement { - l.mtx.Lock() - defer l.mtx.Unlock() + l.mtx.RLock() + defer l.mtx.RUnlock() + return l.tail } func (l *CList) BackWait() *CElement { for { - l.mtx.Lock() + l.mtx.RLock() tail := l.tail wg := l.wg - l.mtx.Unlock() - if tail == nil { - wg.Wait() - } else { + l.mtx.RUnlock() + + if tail != nil { return tail } + wg.Wait() + // l.tail doesn't necessarily exist here. + // That's why we need to continue a for-loop. } } @@ -203,11 +244,12 @@ func (l *CList) PushBack(v interface{}) *CElement { // Construct a new element e := &CElement{ - prev: nil, - prevWg: waitGroup1(), - next: nil, - nextWg: waitGroup1(), - Value: v, + prev: nil, + prevWg: waitGroup1(), + next: nil, + nextWg: waitGroup1(), + removed: false, + Value: v, } // Release waiters on FrontWait/BackWait maybe @@ -221,9 +263,9 @@ func (l *CList) PushBack(v interface{}) *CElement { l.head = e l.tail = e } else { - l.tail.setNextAtomic(e) - e.setPrevAtomic(l.tail) - l.tail = e + e.SetPrev(l.tail) // We must init e first. + l.tail.SetNext(e) // This will make e accessible. + l.tail = e // Update the list. } return e @@ -250,30 +292,26 @@ func (l *CList) Remove(e *CElement) interface{} { // If we're removing the only item, make CList FrontWait/BackWait wait. if l.len == 1 { - l.wg.Add(1) + l.wg = waitGroup1() // WaitGroups are difficult to re-use. } + + // Update l.len l.len -= 1 // Connect next/prev and set head/tail if prev == nil { l.head = next } else { - prev.setNextAtomic(next) + prev.SetNext(next) } if next == nil { l.tail = prev } else { - next.setPrevAtomic(prev) + next.SetPrev(prev) } // Set .Done() on e, otherwise waiters will wait forever. - e.setRemovedAtomic() - if prev == nil { - e.prevWg.Done() - } - if next == nil { - e.nextWg.Done() - } + e.SetRemoved() return e.Value } From 0f8ebd024db7f32ca0d94e7f3d13049ffcb70c09 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Mon, 25 Dec 2017 22:28:15 -0800 Subject: [PATCH 28/36] Update clist.go Add more justification of synchrony primitives in documentation. --- clist/clist.go | 24 ++++++++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/clist/clist.go b/clist/clist.go index e8cf6b93c..02e31a509 100644 --- a/clist/clist.go +++ b/clist/clist.go @@ -1,20 +1,40 @@ package clist /* + The purpose of CList is to provide a goroutine-safe linked-list. This list can be traversed concurrently by any number of goroutines. However, removed CElements cannot be added back. NOTE: Not all methods of container/list are (yet) implemented. NOTE: Removed elements need to DetachPrev or DetachNext consistently to ensure garbage collection of removed elements. + */ import ( "sync" ) -// CElement is an element of a linked-list -// Traversal from a CElement are goroutine-safe. +/* + +CElement is an element of a linked-list +Traversal from a CElement are goroutine-safe. + +We can't avoid using WaitGroups or for-loops given the documentation +spec without re-implementing the primitives that already exist in +golang/sync. Notice that WaitGroup allows many go-routines to be +simultaneously released, which is what we want. Mutex doesn't do +this. RWMutex does this, but it's clumsy to use in the way that a +WaitGroup would be used -- and we'd end up having two RWMutex's for +prev/next each, which is doubly confusing. + +sync.Cond would be sort-of useful, but we don't need a write-lock in +the for-loop. Use sync.Cond when you need serial access to the +"condition". In our case our condition is if `next != nil || removed`, +and there's no reason to serialize that condition for goroutines +waiting on NextWait() (since it's just a read operation). + +*/ type CElement struct { mtx sync.RWMutex prev *CElement From e47ce81422e436459dabf803676d3a3d6924699b Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Thu, 28 Dec 2017 03:02:23 -0800 Subject: [PATCH 29/36] Comment fixes from Emmanuel --- clist/clist.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/clist/clist.go b/clist/clist.go index 02e31a509..a52920f8c 100644 --- a/clist/clist.go +++ b/clist/clist.go @@ -18,7 +18,7 @@ import ( /* CElement is an element of a linked-list -Traversal from a CElement are goroutine-safe. +Traversal from a CElement is goroutine-safe. We can't avoid using WaitGroups or for-loops given the documentation spec without re-implementing the primitives that already exist in @@ -220,6 +220,7 @@ func (l *CList) Front() *CElement { } func (l *CList) FrontWait() *CElement { + // Loop until the head is non-nil else wait and try again for { l.mtx.RLock() head := l.head @@ -230,8 +231,7 @@ func (l *CList) FrontWait() *CElement { return head } wg.Wait() - // l.head doesn't necessarily exist here. - // That's why we need to continue a for-loop. + // NOTE: If you think l.head exists here, think harder. } } From 6b5d08f7daf180036d338d7d7d729861bb58eae5 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Sat, 23 Dec 2017 04:18:50 -0800 Subject: [PATCH 30/36] RepeatTimer fix --- common/repeat_timer.go | 228 ++++++++++++++++++++++++------------ common/repeat_timer_test.go | 86 ++++++++------ 2 files changed, 203 insertions(+), 111 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index 0bc4d87b4..2947a9166 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -5,152 +5,224 @@ import ( "time" ) +// Used by RepeatTimer the first time, +// and every time it's Reset() after Stop(). +type TickerMaker func(dur time.Duration) Ticker + // Ticker is a basic ticker interface. type Ticker interface { + + // Never changes, never closes. Chan() <-chan time.Time + + // Stopping a stopped Ticker will panic. Stop() - Reset() } -// DefaultTicker wraps the stdlibs Ticker implementation. -type DefaultTicker struct { - t *time.Ticker - dur time.Duration -} +//---------------------------------------- +// defaultTickerMaker -// NewDefaultTicker returns a new DefaultTicker -func NewDefaultTicker(dur time.Duration) *DefaultTicker { - return &DefaultTicker{ - time.NewTicker(dur), - dur, - } +func defaultTickerMaker(dur time.Duration) Ticker { + ticker := time.NewTicker(dur) + return (*defaultTicker)(ticker) } +type defaultTicker time.Ticker + // Implements Ticker -func (t *DefaultTicker) Chan() <-chan time.Time { - return t.t.C +func (t *defaultTicker) Chan() <-chan time.Time { + return t.C } // Implements Ticker -func (t *DefaultTicker) Stop() { - t.t.Stop() - t.t = nil +func (t *defaultTicker) Stop() { + t.Stop() } -// Implements Ticker -func (t *DefaultTicker) Reset() { - t.t = time.NewTicker(t.dur) +//---------------------------------------- +// LogicalTickerMaker + +// Construct a TickerMaker that always uses `ch`. +// It's useful for simulating a deterministic clock. +func NewLogicalTickerMaker(ch chan time.Time) TickerMaker { + return func(dur time.Duration) Ticker { + return newLogicalTicker(ch, dur) + } } -// ManualTicker wraps a channel that can be manually sent on -type ManualTicker struct { - ch chan time.Time +type logicalTicker struct { + source <-chan time.Time + ch chan time.Time + quit chan struct{} } -// NewManualTicker returns a new ManualTicker -func NewManualTicker(ch chan time.Time) *ManualTicker { - return &ManualTicker{ - ch: ch, +func newLogicalTicker(source <-chan time.Time, interval time.Duration) Ticker { + lt := &logicalTicker{ + source: source, + ch: make(chan time.Time), + quit: make(chan struct{}), } + go lt.fireRoutine(interval) + return lt } -// Implements Ticker -func (t *ManualTicker) Chan() <-chan time.Time { - return t.ch +// We clearly need a new goroutine, for logicalTicker may have been created +// from a goroutine separate from the source. +func (t *logicalTicker) fireRoutine(interval time.Duration) { + source := t.source + + // Init `lasttime` + lasttime := time.Time{} + select { + case lasttime = <-source: + case <-t.quit: + return + } + // Init `lasttime` end + + timeleft := interval + for { + select { + case newtime := <-source: + elapsed := newtime.Sub(lasttime) + timeleft -= elapsed + if timeleft <= 0 { + // Block for determinism until the ticker is stopped. + select { + case t.ch <- newtime: + case <-t.quit: + return + } + // Reset timeleft. + // Don't try to "catch up" by sending more. + // "Ticker adjusts the intervals or drops ticks to make up for + // slow receivers" - https://golang.org/pkg/time/#Ticker + timeleft = interval + } + case <-t.quit: + return // done + } + } } // Implements Ticker -func (t *ManualTicker) Stop() { - // noop +func (t *logicalTicker) Chan() <-chan time.Time { + return t.ch // immutable } // Implements Ticker -func (t *ManualTicker) Reset() { - // noop +func (t *logicalTicker) Stop() { + close(t.quit) // it *should* panic when stopped twice. } //--------------------------------------------------------------------- /* -RepeatTimer repeatedly sends a struct{}{} to .Ch after each "dur" period. -It's good for keeping connections alive. -A RepeatTimer must be Stop()'d or it will keep a goroutine alive. + RepeatTimer repeatedly sends a struct{}{} to `.Chan()` after each `dur` + period. (It's good for keeping connections alive.) + A RepeatTimer must be stopped, or it will keep a goroutine alive. */ type RepeatTimer struct { - Ch chan time.Time + name string + ch chan time.Time + tm TickerMaker mtx sync.Mutex - name string + dur time.Duration ticker Ticker quit chan struct{} - wg *sync.WaitGroup } -// NewRepeatTimer returns a RepeatTimer with the DefaultTicker. +// NewRepeatTimer returns a RepeatTimer with a defaultTicker. func NewRepeatTimer(name string, dur time.Duration) *RepeatTimer { - ticker := NewDefaultTicker(dur) - return NewRepeatTimerWithTicker(name, ticker) + return NewRepeatTimerWithTickerMaker(name, dur, defaultTickerMaker) } -// NewRepeatTimerWithTicker returns a RepeatTimer with the given ticker. -func NewRepeatTimerWithTicker(name string, ticker Ticker) *RepeatTimer { +// NewRepeatTimerWithTicker returns a RepeatTimer with the given ticker +// maker. +func NewRepeatTimerWithTickerMaker(name string, dur time.Duration, tm TickerMaker) *RepeatTimer { var t = &RepeatTimer{ - Ch: make(chan time.Time), - ticker: ticker, - quit: make(chan struct{}), - wg: new(sync.WaitGroup), name: name, + ch: make(chan time.Time), + tm: tm, + dur: dur, + ticker: nil, + quit: nil, } - t.wg.Add(1) - go t.fireRoutine(t.ticker.Chan()) + t.reset() return t } -func (t *RepeatTimer) fireRoutine(ch <-chan time.Time) { +func (t *RepeatTimer) fireRoutine(ch <-chan time.Time, quit <-chan struct{}) { for { select { case t_ := <-ch: - t.Ch <- t_ - case <-t.quit: - // needed so we know when we can reset t.quit - t.wg.Done() + t.ch <- t_ + case <-quit: // NOTE: `t.quit` races. return } } } +func (t *RepeatTimer) Chan() <-chan time.Time { + return t.ch +} + +func (t *RepeatTimer) Stop() { + t.mtx.Lock() + defer t.mtx.Unlock() + + t.stop() +} + // Wait the duration again before firing. func (t *RepeatTimer) Reset() { - t.Stop() - - t.mtx.Lock() // Lock + t.mtx.Lock() defer t.mtx.Unlock() - t.ticker.Reset() - t.quit = make(chan struct{}) - t.wg.Add(1) - go t.fireRoutine(t.ticker.Chan()) + t.reset() } -// For ease of .Stop()'ing services before .Start()'ing them, -// we ignore .Stop()'s on nil RepeatTimers. -func (t *RepeatTimer) Stop() bool { - if t == nil { - return false +//---------------------------------------- +// Misc. + +// CONTRACT: (non-constructor) caller should hold t.mtx. +func (t *RepeatTimer) reset() { + if t.ticker != nil { + t.stop() } - t.mtx.Lock() // Lock - defer t.mtx.Unlock() + t.ticker = t.tm(t.dur) + t.quit = make(chan struct{}) + go t.fireRoutine(t.ticker.Chan(), t.quit) +} + +// CONTRACT: caller should hold t.mtx. +func (t *RepeatTimer) stop() { + if t.ticker == nil { + /* + Similar to the case of closing channels twice: + https://groups.google.com/forum/#!topic/golang-nuts/rhxMiNmRAPk + Stopping a RepeatTimer twice implies that you do + not know whether you are done or not. + If you're calling stop on a stopped RepeatTimer, + you probably have race conditions. + */ + panic("Tried to stop a stopped RepeatTimer") + } + t.ticker.Stop() + t.ticker = nil + /* + XXX + From https://golang.org/pkg/time/#Ticker: + "Stop the ticker to release associated resources" + "After Stop, no more ticks will be sent" + So we shouldn't have to do the below. - exists := t.ticker != nil - if exists { - t.ticker.Stop() // does not close the channel select { - case <-t.Ch: + case <-t.ch: // read off channel if there's anything there default: } - close(t.quit) - t.wg.Wait() // must wait for quit to close else we race Reset - } - return exists + */ + close(t.quit) } diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index 98d991e9c..f43cc7514 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -4,66 +4,86 @@ import ( "testing" "time" - // make govet noshadow happy... - asrt "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/assert" ) -// NOTE: this only tests with the ManualTicker. +// NOTE: this only tests with the LogicalTicker. // How do you test a real-clock ticker properly? -func TestRepeat(test *testing.T) { - assert := asrt.New(test) +func TestRepeat(t *testing.T) { ch := make(chan time.Time, 100) - // tick fires cnt times on ch + lt := time.Time{} // zero time is year 1 + + // tick fires `cnt` times for each second. tick := func(cnt int) { for i := 0; i < cnt; i++ { - ch <- time.Now() + lt = lt.Add(time.Second) + ch <- lt } } - tock := func(test *testing.T, t *RepeatTimer, cnt int) { + + // tock consumes Ticker.Chan() events `cnt` times. + tock := func(t *testing.T, rt *RepeatTimer, cnt int) { for i := 0; i < cnt; i++ { - after := time.After(time.Second * 2) + timeout := time.After(time.Second * 2) select { - case <-t.Ch: - case <-after: - test.Fatal("expected ticker to fire") + case _ = <-rt.Chan(): + case <-timeout: + panic("QWE") + t.Fatal("expected RepeatTimer to fire") } } done := true select { - case <-t.Ch: + case <-rt.Chan(): done = false default: } - assert.True(done) + assert.True(t, done) } - ticker := NewManualTicker(ch) - t := NewRepeatTimerWithTicker("bar", ticker) + tm := NewLogicalTickerMaker(ch) + dur := time.Duration(0) // dontcare + rt := NewRepeatTimerWithTickerMaker("bar", dur, tm) - // start at 0 - tock(test, t, 0) + // Start at 0. + tock(t, rt, 0) + tick(1) // init time - // wait for 4 periods - tick(4) - tock(test, t, 4) + tock(t, rt, 0) + tick(1) // wait 1 periods + tock(t, rt, 1) + tick(2) // wait 2 periods + tock(t, rt, 2) + tick(3) // wait 3 periods + tock(t, rt, 3) + tick(4) // wait 4 periods + tock(t, rt, 4) - // keep reseting leads to no firing + // Multiple resets leads to no firing. for i := 0; i < 20; i++ { time.Sleep(time.Millisecond) - t.Reset() + rt.Reset() } - tock(test, t, 0) - // after this, it still works normal - tick(2) - tock(test, t, 2) + // After this, it works as new. + tock(t, rt, 0) + tick(1) // init time + + tock(t, rt, 0) + tick(1) // wait 1 periods + tock(t, rt, 1) + tick(2) // wait 2 periods + tock(t, rt, 2) + tick(3) // wait 3 periods + tock(t, rt, 3) + tick(4) // wait 4 periods + tock(t, rt, 4) - // after a stop, nothing more is sent - stopped := t.Stop() - assert.True(stopped) - tock(test, t, 0) + // After a stop, nothing more is sent. + rt.Stop() + tock(t, rt, 0) - // close channel to stop counter - close(t.Ch) + // Another stop panics. + assert.Panics(t, func() { rt.Stop() }) } From 76433d904059009050393ae31c569b7f2df72350 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Mon, 25 Dec 2017 10:13:37 -0500 Subject: [PATCH 31/36] little things --- common/repeat_timer.go | 10 +++++----- common/repeat_timer_test.go | 3 +-- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index 2947a9166..7c5291844 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -42,11 +42,11 @@ func (t *defaultTicker) Stop() { //---------------------------------------- // LogicalTickerMaker -// Construct a TickerMaker that always uses `ch`. +// Construct a TickerMaker that always uses `source`. // It's useful for simulating a deterministic clock. -func NewLogicalTickerMaker(ch chan time.Time) TickerMaker { +func NewLogicalTickerMaker(source chan time.Time) TickerMaker { return func(dur time.Duration) Ticker { - return newLogicalTicker(ch, dur) + return newLogicalTicker(source, dur) } } @@ -66,8 +66,8 @@ func newLogicalTicker(source <-chan time.Time, interval time.Duration) Ticker { return lt } -// We clearly need a new goroutine, for logicalTicker may have been created -// from a goroutine separate from the source. +// We need a goroutine to read times from t.source +// and fire on t.Chan() when `interval` has passed. func (t *logicalTicker) fireRoutine(interval time.Duration) { source := t.source diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index f43cc7514..44a1a0679 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -29,7 +29,6 @@ func TestRepeat(t *testing.T) { select { case _ = <-rt.Chan(): case <-timeout: - panic("QWE") t.Fatal("expected RepeatTimer to fire") } } @@ -43,7 +42,7 @@ func TestRepeat(t *testing.T) { } tm := NewLogicalTickerMaker(ch) - dur := time.Duration(0) // dontcare + dur := time.Duration(10 * time.Millisecond) // less than a second rt := NewRepeatTimerWithTickerMaker("bar", dur, tm) // Start at 0. From 558f8e77699286ffca1f59842f54160dd30d4794 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Mon, 25 Dec 2017 11:10:48 -0500 Subject: [PATCH 32/36] fix recursion --- common/repeat_timer.go | 3 ++- common/repeat_timer_test.go | 8 ++++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index 7c5291844..96348bd19 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -36,7 +36,8 @@ func (t *defaultTicker) Chan() <-chan time.Time { // Implements Ticker func (t *defaultTicker) Stop() { - t.Stop() + tt := time.Ticker(*t) + tt.Stop() } //---------------------------------------- diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index 44a1a0679..269316bd2 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -7,8 +7,12 @@ import ( "github.com/stretchr/testify/assert" ) -// NOTE: this only tests with the LogicalTicker. -// How do you test a real-clock ticker properly? +func TestDefaultTicker(t *testing.T) { + ticker := defaultTickerMaker(time.Millisecond * 10) + <-ticker.Chan() + ticker.Stop() +} + func TestRepeat(t *testing.T) { ch := make(chan time.Time, 100) From a171d906110ea86c3e9e79f3e0bd6c7c7640abc2 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Thu, 28 Dec 2017 17:37:21 -0800 Subject: [PATCH 33/36] Fix possibly incorrect usage of conversion --- common/repeat_timer.go | 3 +-- common/repeat_timer_test.go | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/common/repeat_timer.go b/common/repeat_timer.go index 96348bd19..2e6cb81c8 100644 --- a/common/repeat_timer.go +++ b/common/repeat_timer.go @@ -36,8 +36,7 @@ func (t *defaultTicker) Chan() <-chan time.Time { // Implements Ticker func (t *defaultTicker) Stop() { - tt := time.Ticker(*t) - tt.Stop() + ((*time.Ticker)(t)).Stop() } //---------------------------------------- diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index 269316bd2..da1687073 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -31,7 +31,7 @@ func TestRepeat(t *testing.T) { for i := 0; i < cnt; i++ { timeout := time.After(time.Second * 2) select { - case _ = <-rt.Chan(): + case <-rt.Chan(): case <-timeout: t.Fatal("expected RepeatTimer to fire") } From 71f13cc071258fbcfe3fb3a3438d1a9f0ee0f4e0 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Fri, 29 Dec 2017 10:42:02 -0500 Subject: [PATCH 34/36] drop metalinter --- test.sh | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/test.sh b/test.sh index 02bdaae86..b3978d3fe 100755 --- a/test.sh +++ b/test.sh @@ -2,14 +2,14 @@ set -e # run the linter -make metalinter_test +# make metalinter_test # run the unit tests with coverage echo "" > coverage.txt for d in $(go list ./... | grep -v vendor); do - go test -race -coverprofile=profile.out -covermode=atomic "$d" - if [ -f profile.out ]; then - cat profile.out >> coverage.txt - rm profile.out - fi + go test -race -coverprofile=profile.out -covermode=atomic "$d" + if [ -f profile.out ]; then + cat profile.out >> coverage.txt + rm profile.out + fi done From 92c17f3f251d51878dc866a42dc57dc09df88ac8 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Fri, 29 Dec 2017 10:49:49 -0500 Subject: [PATCH 35/36] give test more time --- common/repeat_timer_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/common/repeat_timer_test.go b/common/repeat_timer_test.go index da1687073..5a3a4c0a6 100644 --- a/common/repeat_timer_test.go +++ b/common/repeat_timer_test.go @@ -29,11 +29,11 @@ func TestRepeat(t *testing.T) { // tock consumes Ticker.Chan() events `cnt` times. tock := func(t *testing.T, rt *RepeatTimer, cnt int) { for i := 0; i < cnt; i++ { - timeout := time.After(time.Second * 2) + timeout := time.After(time.Second * 10) select { case <-rt.Chan(): case <-timeout: - t.Fatal("expected RepeatTimer to fire") + panic("expected RepeatTimer to fire") } } done := true From 35e6f11ad445cf4cb19fefadba0517a86f00b1fc Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Fri, 29 Dec 2017 11:01:37 -0500 Subject: [PATCH 36/36] changelog and version --- CHANGELOG.md | 13 +++++++++++++ version/version.go | 2 +- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b679b839d..fe2c2fe94 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## 0.6.0 (December 29, 2017) + +BREAKING: + - [cli] remove --root + - [pubsub] add String() method to Query interface + +IMPROVEMENTS: + - [common] use a thread-safe and well seeded non-crypto rng + +BUG FIXES + - [clist] fix misuse of wait group + - [common] introduce Ticker interface and logicalTicker for better testing of timers + ## 0.5.0 (December 5, 2017) BREAKING: diff --git a/version/version.go b/version/version.go index 45222da79..6cc887286 100644 --- a/version/version.go +++ b/version/version.go @@ -1,3 +1,3 @@ package version -const Version = "0.5.0" +const Version = "0.6.0"