Browse Source

Merge pull request #107 from tendermint/fix/cmn

Fix/cmn
pull/1842/head
Ethan Buchman 7 years ago
committed by GitHub
parent
commit
218acc2224
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 117 additions and 56 deletions
  1. +1
    -1
      common/random.go
  2. +80
    -10
      common/repeat_timer.go
  3. +36
    -45
      common/repeat_timer_test.go

+ 1
- 1
common/random.go View File

@ -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


+ 80
- 10
common/repeat_timer.go View File

@ -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)
go t.fireRoutine(t.ticker.Chan())
return t
}
func (t *RepeatTimer) fireRoutine(ticker *time.Ticker) {
func (t *RepeatTimer) fireRoutine(ch <-chan time.Time) {
for {
select {
case t_ := <-ticker.C:
case t_ := <-ch:
t.Ch <- t_
case <-t.quit:
// needed so we know when we can reset t.quit
@ -55,10 +126,10 @@ 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)
go t.fireRoutine(t.ticker.Chan())
}
// For ease of .Stop()'ing services before .Start()'ing them,
@ -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
}

+ 36
- 45
common/repeat_timer_test.go View File

@ -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)


Loading…
Cancel
Save