-
Notifications
You must be signed in to change notification settings - Fork 250
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
remotecfg: add jitter to polling frequency (#961)
Signed-off-by: Paschalis Tsilias <[email protected]>
- Loading branch information
1 parent
a31c1f5
commit 67a6239
Showing
4 changed files
with
197 additions
and
7 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| package jitter | ||
|
|
||
| import ( | ||
| "math/rand" | ||
| "sync" | ||
| "time" | ||
| ) | ||
|
|
||
| type Ticker struct { | ||
| C <-chan time.Time | ||
| stop chan struct{} | ||
| reset chan struct{} | ||
|
|
||
| mut sync.RWMutex | ||
| d time.Duration | ||
| j time.Duration | ||
| } | ||
|
|
||
| // NewTicker creates a Ticker that works similar to time.Ticker, but sends the | ||
| // time with a period specified by `duration` adjusted by a pseudorandom jitter | ||
| // in the range of [duration-jitter, duration+jitter). | ||
| // Following the behavior of time.Ticker, we use a 1-buffer channel, so if the | ||
| // client falls behind while reading, we'll drop ticks on the floor until the | ||
| // client catches up. | ||
| // Callers have to make sure that both duration and the [d-j, d+j) intervals | ||
| // are valid positive int64 values (non-negative and non-overflowing). | ||
| // Use Stop to release associated resources and the Reset methods to modify the | ||
| // duration and jitter. | ||
| func NewTicker(duration time.Duration, jitter time.Duration) *Ticker { | ||
| ticker := time.NewTicker(duration) | ||
| c := make(chan time.Time, 1) | ||
| t := &Ticker{ | ||
| C: c, | ||
|
|
||
| stop: make(chan struct{}), | ||
| reset: make(chan struct{}), | ||
| d: duration, | ||
| j: jitter, | ||
| } | ||
|
|
||
| go func() { | ||
| for { | ||
| select { | ||
| case tc := <-ticker.C: | ||
| ticker.Reset(t.getNextPeriod()) | ||
| select { | ||
| case c <- tc: | ||
| default: | ||
| } | ||
| case <-t.stop: | ||
| ticker.Stop() | ||
| return | ||
| case <-t.reset: | ||
| ticker.Reset(t.getNextPeriod()) | ||
| } | ||
| } | ||
| }() | ||
| return t | ||
| } | ||
|
|
||
| // Stop turns off the Ticker; no more ticks will be sent. Stop does not close | ||
| // Ticker's channel, to prevent a concurrent goroutine from seeing an erroneous | ||
| // "tick". | ||
| func (t *Ticker) Stop() { | ||
| close(t.reset) | ||
| close(t.stop) | ||
| } | ||
|
|
||
| // Reset stops the Ticker, resets its base duration to the specified argument | ||
| // and re-calculates the period with a jitter. | ||
| // The next tick will arrive after the new period elapses. | ||
| func (t *Ticker) Reset(d time.Duration) { | ||
| t.mut.Lock() | ||
| t.d = d | ||
| t.mut.Unlock() | ||
| t.reset <- struct{}{} | ||
| } | ||
|
|
||
| // Reset stops the Ticker, resets its jitter to the specified argument and | ||
| // re-calculates the period with the new jitter. | ||
| // The next tick will arrive after the new period elapses. | ||
| func (t *Ticker) ResetJitter(d time.Duration) { | ||
| t.mut.Lock() | ||
| t.j = d | ||
| t.mut.Unlock() | ||
| t.reset <- struct{}{} | ||
| } | ||
|
|
||
| // getNextPeriod is used to calculate the period for the Ticker. | ||
| func (t *Ticker) getNextPeriod() time.Duration { | ||
| // jitter is a random value between [0, 2j) | ||
| // the returned period is then d-j + jitter | ||
| // which results in [d-j, d+j). | ||
| t.mut.RLock() | ||
| jitter := rand.Int63n(2 * int64(t.j)) | ||
| period := t.d - t.j + time.Duration(jitter) | ||
| t.mut.RUnlock() | ||
|
|
||
| return period | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| package jitter | ||
|
|
||
| import ( | ||
| "math" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| // Inspired by a test on github.com/mroth/jitter | ||
| func TestTicker(t *testing.T) { | ||
| var ( | ||
| d = 10 * time.Millisecond | ||
| j = 3 * time.Millisecond | ||
| n = 10 | ||
| delta = 1 * time.Millisecond | ||
| min = time.Duration(math.Floor(float64(d)-float64(j)))*time.Duration(n) - delta | ||
| max = time.Duration(math.Ceil(float64(d)+float64(j)))*time.Duration(n) + delta | ||
| ) | ||
|
|
||
| // Check that the time required for N ticks is within expected range. | ||
| ticker := NewTicker(d, j) | ||
| start := time.Now() | ||
| for i := 0; i < n; i++ { | ||
| <-ticker.C | ||
| } | ||
|
|
||
| elapsed := time.Since(start) | ||
| if elapsed < min || elapsed > max { | ||
| require.Fail(t, "ticker didn't meet timing criteria", "time needed for %d ticks %v outside of expected range [%v - %v]", n, elapsed, min, max) | ||
| } | ||
| } | ||
|
|
||
| func TestTickerStop(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| var ( | ||
| d = 5 * time.Millisecond | ||
| j = 1 * time.Millisecond | ||
| before = 3 // ticks before stop | ||
| wait = d * 10 // monitor after stop | ||
| ) | ||
|
|
||
| ticker := NewTicker(d, j) | ||
| for i := 0; i < before; i++ { | ||
| <-ticker.C | ||
| } | ||
|
|
||
| ticker.Stop() | ||
| select { | ||
| case <-ticker.C: | ||
| require.Fail(t, "Got tick after Stop()") | ||
| case <-time.After(wait): | ||
| } | ||
| } | ||
|
|
||
| func TestTickerReset(t *testing.T) { | ||
| var ( | ||
| d1 = 10 * time.Millisecond | ||
| d2 = 20 * time.Millisecond | ||
| j1 = 3 * time.Millisecond | ||
| j2 = 9 * time.Millisecond | ||
| n = 10 | ||
| delta = 1 * time.Millisecond | ||
| min1 = time.Duration(math.Floor(float64(d1)-float64(j1)))*time.Duration(n) - delta | ||
| max1 = time.Duration(math.Ceil(float64(d1)+float64(j1)))*time.Duration(n) + delta | ||
| min2 = time.Duration(math.Floor(float64(d2)-float64(j2)))*time.Duration(n) - delta | ||
| max2 = time.Duration(math.Ceil(float64(d2)+float64(j2)))*time.Duration(n) + delta | ||
| ) | ||
|
|
||
| // Check that the time required for N ticks is within expected range. | ||
| ticker := NewTicker(d1, j1) | ||
| start := time.Now() | ||
| for i := 0; i < n; i++ { | ||
| <-ticker.C | ||
| } | ||
| ticker.Reset(d2) | ||
| ticker.ResetJitter(j2) | ||
| for i := 0; i < n; i++ { | ||
| <-ticker.C | ||
| } | ||
|
|
||
| elapsed := time.Since(start) | ||
| if elapsed < (min1+min2) || elapsed > (max1+max2) { | ||
| require.Fail(t, "ticker didn't meet timing criteria", "time needed for %d ticks %v outside of expected range [%v - %v]", n, elapsed, (min1 + min2), (max1 + max2)) | ||
| } | ||
| } |