diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md index 37ae3a510..3cbc63b7b 100644 --- a/CHANGELOG_PENDING.md +++ b/CHANGELOG_PENDING.md @@ -9,6 +9,7 @@ * Apps * Go API +- [libs/common] Remove RepeatTimer (also TimerMaker and Ticker interface) * Blockchain Protocol diff --git a/libs/common/repeat_timer.go b/libs/common/repeat_timer.go deleted file mode 100644 index 5d049738d..000000000 --- a/libs/common/repeat_timer.go +++ /dev/null @@ -1,232 +0,0 @@ -package common - -import ( - "sync" - "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() -} - -//---------------------------------------- -// defaultTicker - -var _ Ticker = (*defaultTicker)(nil) - -type defaultTicker time.Ticker - -func defaultTickerMaker(dur time.Duration) Ticker { - ticker := time.NewTicker(dur) - return (*defaultTicker)(ticker) -} - -// Implements Ticker -func (t *defaultTicker) Chan() <-chan time.Time { - return t.C -} - -// Implements Ticker -func (t *defaultTicker) Stop() { - ((*time.Ticker)(t)).Stop() -} - -//---------------------------------------- -// LogicalTickerMaker - -// Construct a TickerMaker that always uses `source`. -// It's useful for simulating a deterministic clock. -func NewLogicalTickerMaker(source chan time.Time) TickerMaker { - return func(dur time.Duration) Ticker { - return newLogicalTicker(source, dur) - } -} - -type logicalTicker struct { - source <-chan time.Time - ch chan time.Time - quit chan struct{} -} - -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 -} - -// 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 - - // Init `lasttime` - lasttime := time.Time{} - select { - case lasttime = <-source: - case <-t.quit: - return - } - // Init `lasttime` end - - for { - select { - case newtime := <-source: - elapsed := newtime.Sub(lasttime) - if interval <= elapsed { - // 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 - lasttime = newtime - } - case <-t.quit: - return // done - } - } -} - -// Implements Ticker -func (t *logicalTicker) Chan() <-chan time.Time { - return t.ch // immutable -} - -// Implements Ticker -func (t *logicalTicker) Stop() { - close(t.quit) // it *should* panic when stopped twice. -} - -//--------------------------------------------------------------------- - -/* - 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 { - name string - ch chan time.Time - tm TickerMaker - - mtx sync.Mutex - dur time.Duration - ticker Ticker - quit chan struct{} -} - -// NewRepeatTimer returns a RepeatTimer with a defaultTicker. -func NewRepeatTimer(name string, dur time.Duration) *RepeatTimer { - return NewRepeatTimerWithTickerMaker(name, dur, defaultTickerMaker) -} - -// NewRepeatTimerWithTicker returns a RepeatTimer with the given ticker -// maker. -func NewRepeatTimerWithTickerMaker(name string, dur time.Duration, tm TickerMaker) *RepeatTimer { - var t = &RepeatTimer{ - name: name, - ch: make(chan time.Time), - tm: tm, - dur: dur, - ticker: nil, - quit: nil, - } - t.reset() - return t -} - -// receive ticks on ch, send out on t.ch -func (t *RepeatTimer) fireRoutine(ch <-chan time.Time, quit <-chan struct{}) { - for { - select { - case tick := <-ch: - select { - case t.ch <- tick: - case <-quit: - return - } - 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.mtx.Lock() - defer t.mtx.Unlock() - - t.reset() -} - -//---------------------------------------- -// Misc. - -// CONTRACT: (non-constructor) caller should hold t.mtx. -func (t *RepeatTimer) reset() { - if t.ticker != nil { - t.stop() - } - 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 - /* - 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. - - select { - case <-t.ch: - // read off channel if there's anything there - default: - } - */ - close(t.quit) -} diff --git a/libs/common/repeat_timer_test.go b/libs/common/repeat_timer_test.go deleted file mode 100644 index f2a7b16c3..000000000 --- a/libs/common/repeat_timer_test.go +++ /dev/null @@ -1,136 +0,0 @@ -package common - -import ( - "sync" - "testing" - "time" - - "github.com/fortytw2/leaktest" - "github.com/stretchr/testify/assert" -) - -func TestDefaultTicker(t *testing.T) { - ticker := defaultTickerMaker(time.Millisecond * 10) - <-ticker.Chan() - ticker.Stop() -} - -func TestRepeatTimer(t *testing.T) { - - ch := make(chan time.Time, 100) - mtx := new(sync.Mutex) - - // tick() fires from start to end - // (exclusive) in milliseconds with incr. - // It locks on mtx, so subsequent calls - // run in series. - tick := func(startMs, endMs, incrMs time.Duration) { - mtx.Lock() - go func() { - for tMs := startMs; tMs < endMs; tMs += incrMs { - lt := time.Time{} - lt = lt.Add(tMs * time.Millisecond) - ch <- lt - } - mtx.Unlock() - }() - } - - // tock consumes Ticker.Chan() events and checks them against the ms in "timesMs". - tock := func(t *testing.T, rt *RepeatTimer, timesMs []int64) { - - // Check against timesMs. - for _, timeMs := range timesMs { - tyme := <-rt.Chan() - sinceMs := tyme.Sub(time.Time{}) / time.Millisecond - assert.Equal(t, timeMs, int64(sinceMs)) - } - - // TODO detect number of running - // goroutines to ensure that - // no other times will fire. - // See https://github.com/tendermint/tendermint/libs/issues/120. - time.Sleep(time.Millisecond * 100) - done := true - select { - case <-rt.Chan(): - done = false - default: - } - assert.True(t, done) - } - - tm := NewLogicalTickerMaker(ch) - rt := NewRepeatTimerWithTickerMaker("bar", time.Second, tm) - - /* NOTE: Useful for debugging deadlocks... - go func() { - time.Sleep(time.Second * 3) - trace := make([]byte, 102400) - count := runtime.Stack(trace, true) - fmt.Printf("Stack of %d bytes: %s\n", count, trace) - }() - */ - - tick(0, 1000, 10) - tock(t, rt, []int64{}) - tick(1000, 2000, 10) - tock(t, rt, []int64{1000}) - tick(2005, 5000, 10) - tock(t, rt, []int64{2005, 3005, 4005}) - tick(5001, 5999, 1) - // Read 5005 instead of 5001 because - // it's 1 second greater than 4005. - tock(t, rt, []int64{5005}) - tick(6000, 7005, 1) - tock(t, rt, []int64{6005}) - tick(7033, 8032, 1) - tock(t, rt, []int64{7033}) - - // After a reset, nothing happens - // until two ticks are received. - rt.Reset() - tock(t, rt, []int64{}) - tick(8040, 8041, 1) - tock(t, rt, []int64{}) - tick(9555, 9556, 1) - tock(t, rt, []int64{9555}) - - // After a stop, nothing more is sent. - rt.Stop() - tock(t, rt, []int64{}) - - // Another stop panics. - assert.Panics(t, func() { rt.Stop() }) -} - -func TestRepeatTimerReset(t *testing.T) { - // check that we are not leaking any go-routines - defer leaktest.Check(t)() - - timer := NewRepeatTimer("test", 20*time.Millisecond) - defer timer.Stop() - - // test we don't receive tick before duration ms. - select { - case <-timer.Chan(): - t.Fatal("did not expect to receive tick") - default: - } - - timer.Reset() - - // test we receive tick after Reset is called - select { - case <-timer.Chan(): - // all good - case <-time.After(40 * time.Millisecond): - t.Fatal("expected to receive tick after reset") - } - - // just random calls - for i := 0; i < 100; i++ { - time.Sleep(time.Duration(RandIntn(40)) * time.Millisecond) - timer.Reset() - } -} diff --git a/p2p/conn/connection.go b/p2p/conn/connection.go index c1e90ab76..e0ce062ab 100644 --- a/p2p/conn/connection.go +++ b/p2p/conn/connection.go @@ -95,13 +95,13 @@ type MConnection struct { stopMtx sync.Mutex flushTimer *cmn.ThrottleTimer // flush writes as necessary but throttled. - pingTimer *cmn.RepeatTimer // send pings periodically + pingTimer *time.Ticker // send pings periodically // close conn if pong is not received in pongTimeout pongTimer *time.Timer pongTimeoutCh chan bool // true - timeout, false - peer sent pong - chStatsTimer *cmn.RepeatTimer // update channel stats periodically + chStatsTimer *time.Ticker // update channel stats periodically created time.Time // time of creation @@ -201,9 +201,9 @@ func (c *MConnection) OnStart() error { return err } c.flushTimer = cmn.NewThrottleTimer("flush", c.config.FlushThrottle) - c.pingTimer = cmn.NewRepeatTimer("ping", c.config.PingInterval) + c.pingTimer = time.NewTicker(c.config.PingInterval) c.pongTimeoutCh = make(chan bool, 1) - c.chStatsTimer = cmn.NewRepeatTimer("chStats", updateStats) + c.chStatsTimer = time.NewTicker(updateStats) c.quitSendRoutine = make(chan struct{}) c.doneSendRoutine = make(chan struct{}) go c.sendRoutine() @@ -401,11 +401,11 @@ FOR_LOOP: // NOTE: flushTimer.Set() must be called every time // something is written to .bufConnWriter. c.flush() - case <-c.chStatsTimer.Chan(): + case <-c.chStatsTimer.C: for _, channel := range c.channels { channel.updateStats() } - case <-c.pingTimer.Chan(): + case <-c.pingTimer.C: c.Logger.Debug("Send Ping") _n, err = cdc.MarshalBinaryLengthPrefixedWriter(c.bufConnWriter, PacketPing{}) if err != nil { diff --git a/p2p/switch.go b/p2p/switch.go index a07f70ce9..9e04fe7ce 100644 --- a/p2p/switch.go +++ b/p2p/switch.go @@ -234,21 +234,26 @@ func (sw *Switch) OnStop() { // // NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved. func (sw *Switch) Broadcast(chID byte, msgBytes []byte) chan bool { - successChan := make(chan bool, len(sw.peers.List())) sw.Logger.Debug("Broadcast", "channel", chID, "msgBytes", fmt.Sprintf("%X", msgBytes)) + + peers := sw.peers.List() var wg sync.WaitGroup - for _, peer := range sw.peers.List() { - wg.Add(1) - go func(peer Peer) { + wg.Add(len(peers)) + successChan := make(chan bool, len(peers)) + + for _, peer := range peers { + go func(p Peer) { defer wg.Done() - success := peer.Send(chID, msgBytes) + success := p.Send(chID, msgBytes) successChan <- success }(peer) } + go func() { wg.Wait() close(successChan) }() + return successChan }