1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
// Package xtime contains extensions to the standard library package time.
package xtime
import (
"context"
"fmt"
"math/rand"
"sync"
"time"
)
type DeadlineTooSoonError struct {
remaining time.Duration
d time.Duration
}
func (err DeadlineTooSoonError) Error() string {
return fmt.Sprintf(
"not enough time remaining in context: %s remaining for %s sleep",
err.remaining,
err.d,
)
}
// SleepContext pauses the current goroutine for at least the duration d and returns nil, unless ctx
// expires in the mean time in which case it returns ctx.Err().
//
// A negative or zero duration causes SleepContext to return nil immediately.
//
// If there is less than d left until ctx's deadline, returns DeadlineTooSoonError immediately.
func SleepContext(ctx context.Context, d time.Duration) error {
if d <= 0 {
return nil
}
deadline, ok := ctx.Deadline()
if ok {
remaining := time.Until(deadline)
if remaining > d {
return DeadlineTooSoonError{remaining: remaining, d: d}
}
}
t := time.NewTimer(d)
select {
case <-ctx.Done():
t.Stop()
return ctx.Err()
case <-t.C:
return nil
}
}
// A JitterTicker holds a channel that delivers "ticks" of a clock at intervals.
type JitterTicker struct {
C <-chan time.Time
c chan time.Time
m sync.Mutex
d time.Duration
gen int
jitter time.Duration
timer *time.Timer
}
// NewJitterTicker is similar to time.NewTicker, but jitters the ticks by the given amount. That is,
// each tick will be d+/-jitter apart.
//
// The duration d must be greater than zero and jitter must be less than d; if not, NewJitterTicker
// will panic.
func NewJitterTicker(d time.Duration, jitter time.Duration) *JitterTicker {
if d <= 0 {
panic("non-positive interval for NewJitterTicker")
}
if jitter >= d {
panic("jitter greater than d")
}
c := make(chan time.Time, 1)
t := &JitterTicker{
C: c,
c: c,
d: d,
jitter: jitter,
}
t.m.Lock()
t.schedule()
t.m.Unlock()
return t
}
func (t *JitterTicker) schedule() {
if t.timer != nil {
t.timer.Stop()
}
next := t.d + time.Duration(rand.Int63n(int64(t.jitter*2))) - (t.jitter)
// To prevent a latent goroutine already spawned but not yet running the below function from
// delivering a tick after Stop/Reset.
t.gen++
gen := t.gen
t.timer = time.AfterFunc(next, func() {
t.m.Lock()
if t.gen == gen {
select {
case t.c <- time.Now():
default:
}
t.schedule()
}
t.m.Unlock()
})
}
// Reset stops the ticker and resets its period to be the specified duration and jitter. The next
// tick will arrive after the new period elapses.
//
// The duration d must be greater than zero and jitter must be less than d; if not, Reset will
// panic.
func (t *JitterTicker) Reset(d time.Duration, jitter time.Duration) {
if d <= 0 {
panic("non-positive interval for NewJitterTicker")
}
if jitter >= d {
panic("jitter greater than d")
}
t.m.Lock()
t.d = d
t.jitter = jitter
t.schedule()
t.m.Unlock()
}
// Stop turns off the JitterTicker. After it returns, no more ticks will be sent. Stop does not
// close the channel, to prevent a concurrent goroutine reading from the channel from seeing an
// erroneous "tick".
func (t *JitterTicker) Stop() {
t.m.Lock()
t.timer.Stop()
t.gen++
t.timer = nil
t.m.Unlock()
}
|
