Vihren Atlas

Atlas › Test

TestConcurrentAccessAndCleanup

go.temporal.io/server/common/quotas/TestMapRequestRateLimiterSuite/TestConcurrentAccessAndCleanup

Package
go.temporal.io/server/common/quotas
Suite / test hierarchy
TestMapRequestRateLimiterSuite/TestConcurrentAccessAndCleanup
Test
TestConcurrentAccessAndCleanup
Introduced at
TestConcurrentAccessAndCleanup Test frontier
Covered ranges
12
Covered lines
65
Covered files
2

Covered source

Expand a file to inspect source; the > gutter marks covered lines.

go.temporal.io/server/common/quotas/map_request_rate_limiter_impl.go 62 covered LOC · 11 ranges

Open complete file

40 rateLimiterGenFn RequestRateLimiterFn,
41 rateLimiterKeyFn RequestRateLimiterKeyFn[K],
42 > ) *MapRequestRateLimiterImpl[K] { map_request_rate_limiter_impl.go
43 > return &MapRequestRateLimiterImpl[K]{
44 > rateLimiterGenFn: rateLimiterGenFn,
45 > rateLimiterKeyFn: rateLimiterKeyFn,
46 > rateLimiters: make(map[K]*rateLimiterEntry),
47 > ttlNano: int64(rateLimiterTTL),
48 > cleanupTicker: time.NewTicker(rateLimiterCleanupInterval),
49 > }
50 > }
51
52 func namespaceRequestRateLimiterKeyFn(req Request) string {
66 now time.Time,
67 request Request,
68 > ) bool { map_request_rate_limiter_impl.go
69 > rateLimiter := r.getOrInitRateLimiter(now, request)
70 > return rateLimiter.Allow(now, request)
71 > }
72
73 // Reserve returns a Reservation that indicates how long the caller
94 now time.Time,
95 req Request,
96 > ) RequestRateLimiter { map_request_rate_limiter_impl.go
97 > r.maybeCleanup(now)
98 >
99 > key := r.rateLimiterKeyFn(req)
100 > nowNano := now.UnixNano()
101 >
102 > r.RLock()
103 > entry, ok := r.rateLimiters[key]
104 > r.RUnlock()
105 >
106 > if ok {
107 > entry.lastAccess.Store(nowNano) map_request_rate_limiter_impl.go
108 > return entry.rateLimiter
109 > }
110
111 > newRateLimiter := r.rateLimiterGenFn(req) map_request_rate_limiter_impl.go
112 > r.Lock()
113 > defer r.Unlock()
114 >
115 > if entry, ok := r.rateLimiters[key]; ok {
116 > entry.lastAccess.Store(nowNano) map_request_rate_limiter_impl.go
117 > return entry.rateLimiter
118 > }
119
120 > entry = &rateLimiterEntry{rateLimiter: newRateLimiter} map_request_rate_limiter_impl.go
121 > entry.lastAccess.Store(nowNano)
122 > r.rateLimiters[key] = entry
123 > return newRateLimiter
124 }
125
127 // receive drains at most one ticker tick, so only one sweeper starts even if many
128 // callers reach here at once.
129 > func (r *MapRequestRateLimiterImpl[K]) maybeCleanup(now time.Time) { map_request_rate_limiter_impl.go
130 > select {
131 case <-r.cleanupTicker.C:
132 go r.cleanup(now)
133 > default: map_request_rate_limiter_impl.go
134 }
135 }
139 // 2. For each candidate: Lock, re-check, delete, Unlock
140 // This ensures read operations are only briefly blocked during actual deletions.
141 > func (r *MapRequestRateLimiterImpl[K]) cleanup(now time.Time) { map_request_rate_limiter_impl.go
142 > nowNano := now.UnixNano()
143 > ttlNano := r.ttlNano
144 >
145 > // Phase 1: collect expired keys under read lock
146 > var expiredKeys []K
147 > r.RLock()
148 > for k, e := range r.rateLimiters {
149 > if nowNano-e.lastAccess.Load() > ttlNano {
150 > expiredKeys = append(expiredKeys, k)
151 > }
152 }
153 > r.RUnlock() map_request_rate_limiter_impl.go
154 >
155 > // Phase 2: delete each expired key individually
156 > for _, k := range expiredKeys {
157 > r.Lock()
158 > if e, ok := r.rateLimiters[k]; ok && nowNano-e.lastAccess.Load() > ttlNano {
159 > delete(r.rateLimiters, k)
160 > }
161 > r.Unlock()
162 }
163 }
go.temporal.io/server/common/quotas/noop_request_rate_limiter_impl.go 3 covered LOC · 1 range

Open complete file