Atlas › Test
TestClockedRateLimiter_WaitN_NoRecycle
Exact test identity: go.temporal.io/server/common/quotas/TestClockedRateLimiter_WaitN_NoRecycle
- Package
go.temporal.io/server/common/quotas- Suite / test hierarchy
TestClockedRateLimiter_WaitN_NoRecycle- Test
TestClockedRateLimiter_WaitN_NoRecycle- Introduced at
- clocked_rate_limiter.go ×1 Frontier kind: Joint frontier
- Covered ranges
- 30
- Covered lines
- 98
- Covered files
- 2
Covered source
Expand a file to inspect source; the > gutter marks covered lines.
go.temporal.io/server/common/quotas/clocked_rate_limiter.go 50 covered LOC · 18 ranges
Open complete file
25
)
26
27
>
func NewClockedRateLimiter(rateLimiter *rate.Limiter, timeSource clock.TimeSource) ClockedRateLimiter {
clocked_rate_limiter.go
28
>
return ClockedRateLimiter{
29
>
rateLimiter: rateLimiter,
30
>
timeSource: timeSource,
31
>
recycleCh: make(chan struct{}),
32
>
}
33
>
}
34
35
func (l ClockedRateLimiter) Allow() bool {
48
}
49
51
>
return r.reservation.OK()
52
>
}
53
55
>
return r.DelayFrom(r.timeSource.Now())
56
>
}
57
59
>
return r.reservation.DelayFrom(t)
60
>
}
61
63
>
r.CancelAt(r.timeSource.Now())
64
>
}
65
67
>
r.reservation.CancelAt(t)
68
>
}
69
70
func (l ClockedRateLimiter) Reserve() ClockedReservation {
77
}
78
80
>
return l.WaitN(ctx, 1)
81
>
}
82
83
// WaitN is the only method that is different from rate.Limiter. We need to fully reimplement this method because
84
// the original method uses time.Now(), and does not allow us to pass in a time.Time. Fortunately, it can be built on
85
// top of ReserveN. However, there are some optimizations that we can make.
87
>
reservation := ClockedReservation{l.rateLimiter.ReserveN(l.timeSource.Now(), token), l.timeSource}
88
>
if !reservation.OK() {
89
return fmt.Errorf("%w: WaitN(n=%d)", ErrRateLimiterReservationCannotBeMade, token)
90
}
91
93
>
94
>
// Optimization: if the waitDuration is 0, we don't need to start a timer.
95
>
if waitDuration <= 0 {
97
>
}
98
99
// Optimization: if the waitDuration is longer than the context deadline, we can immediately return an error.
101
if l.timeSource.Now().Add(waitDuration).After(deadline) {
102
reservation.Cancel()
105
}
106
108
>
timer := l.timeSource.AfterFunc(waitDuration, func() {
109
close(waitExpired)
110
})
112
>
113
>
for {
114
>
select {
116
>
reservation.Cancel()
117
>
return fmt.Errorf("%w: %v", ErrRateLimiterWaitInterrupted, ctx.Err())
118
case <-waitExpired:
119
return nil
121
>
if token > 1 {
122
>
break // recycling 1 token to a process requesting >1 tokens is a no-op, because we only know that at least one token was not used
clocked_rate_limiter.go
123
}
124
163
// so that the actual rate of completed actions is as close to the intended rate limit as possible.
164
// If no process is waiting for a token when RecycleToken is called, this is a no-op.
166
>
select {
168
default:
169
}
go.temporal.io/server/common/clock/event_time_source.go 48 covered LOC · 12 ranges
Open complete file
39
40
// NewEventTimeSource returns a EventTimeSource with the current time set to Unix zero: 1970-01-01 00:00:00 +0000 UTC.
42
>
return &EventTimeSource{
43
>
now: time.Unix(0, 0),
44
>
}
45
>
}
46
47
// Some clients depend on the fact that the runtime's timers do _not_ run synchronously.
55
56
// Now return the current time.
58
>
ts.mu.RLock()
59
>
defer ts.mu.RUnlock()
60
>
61
>
return ts.now
62
>
}
63
64
func (ts *EventTimeSource) Since(t time.Time) time.Duration {
71
// wrap all such calls in a goroutine. If the duration is non-positive, the callback will fire immediately before
72
// AfterFunc returns.
74
>
if d < 0 {
75
d = 0
76
}
77
>
timer := &fakeTimer{timeSource: ts, deadline: ts.Now().Add(d), callback: f}
event_time_source.go
78
>
ts.addTimer(timer)
79
>
return timer
80
}
81
96
}
97
99
>
ts.mu.Lock()
100
>
defer ts.mu.Unlock()
101
>
t.index = len(ts.timers)
102
>
ts.timers = append(ts.timers, t)
103
>
ts.fireTimers()
104
>
}
105
106
// Update the fake current time. It returns the timeSource so that you can chain calls like this:
156
157
// fireTimers fires all timers that are ready.
159
>
n := 0
160
>
for _, t := range ts.timers {
163
>
t.index = n
164
>
n++
166
if ts.async {
167
go t.callback()
200
201
// Stop the timer. Returns true if the timer was active.
203
>
t.timeSource.mu.Lock()
204
>
defer t.timeSource.mu.Unlock()
205
>
206
>
if t.done {
207
return false
208
}
209
211
>
timers := t.timeSource.timers
212
>
213
>
timers[i] = timers[len(timers)-1] // swap with last timer
214
>
timers[i].index = i // update index of swapped timer
215
>
timers = timers[:len(timers)-1] // shrink list
216
>
217
>
t.timeSource.timers = timers
218
>
t.done = true // ensure that the timer is not reused
219
>
220
>
return true
221
}