Atlas › Test

TestShrinksAgain

Exact test identity: go.temporal.io/server/common/goro/TestAdaptivePoolSuite/TestShrinksAgain

Package
go.temporal.io/server/common/goro
Suite / test hierarchy
TestAdaptivePoolSuite/TestShrinksAgain
Test
TestShrinksAgain
Introduced at
TestShrinksAgain Frontier kind: Test frontier
Covered ranges
143
Covered lines
479
Covered files
12

Covered source

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

go.temporal.io/server/common/testing/parallelsuite/suite.go 107 covered LOC · 30 ranges

Open complete file

74 //
75 //nolint:revive // ctx is last so callers can pass nil to mean "no override"; SA1012 forbids passing nil as the first ctx arg.
76 > func (s *Suite[T]) copySuite(t *testing.T, parallel bool, assertT require.TestingT, ctx context.Context) testingSuite { suite.go
77 > cp := reflect.New(reflect.TypeFor[T]().Elem()).Interface().(T)
78 > cp.initSuite(t, parallel, assertT, ctx)
79 > return cp
80 > }
81
82 //nolint:revive // see copySuite above.
83 > func (s *Suite[T]) initSuite(t *testing.T, parallel bool, assertT require.TestingT, ctx context.Context) { suite.go
84 > g := &s.guardT
85 > g.name = t.Name()
86 > g.T = t
87 > g.hasSubtests.Store(false)
88 > s.runParallel = parallel
89 > s.ctx = ctx
90 > s.ctxOnce = sync.Once{}
91 > if s.runParallel {
92 > t.Parallel() //nolint:testifylint // parallelsuite intentionally supports parallel tests
93 > }
94 > if assertT == nil {
95 > assertT = g
96 > }
97 > s.assertT = assertT
98 > s.Assertions = require.New(assertT)
99 > s.ProtoAssertions = protorequire.New(assertT)
100 > s.HistoryRequire = historyrequire.New(assertT)
101 }
102
103 // T returns the *testing.T, panicking if the guard has been sealed.
104 > func (s *Suite[T]) T() *testing.T { suite.go
105 > if s.guardT.hasSubtests.Load() {
106 panic("parallelsuite: do not call T() after Run(); use the subtest callback's parameter instead")
107 }
108 > return s.guardT.T suite.go
109 }
110
126 // Context returns the test-scoped context (created from [testcontext]).
127 // Inside an [Await] callback, it returns the await-scoped context.
128 > func (s *Suite[T]) Context() context.Context { suite.go
129 > s.ctxOnce.Do(func() {
130 > if s.ctx == nil {
131 > s.ctx = testcontext.For(s.T())
132 > }
133 })
134 > return s.ctx suite.go
135 }
136
146
147 // Await calls fn repeatedly until all assertions pass or timeout is reached.
148 > func (s *Suite[T]) Await(fn func(T), timeout, interval time.Duration) { suite.go
149 > s.Awaitf(fn, timeout, interval, "")
150 > }
151
152 // Awaitf is like [Await] but includes a format string appended to the failure message.
153 > func (s *Suite[T]) Awaitf(fn func(T), timeout, interval time.Duration, msg string, args ...any) { suite.go
154 > t := s.T()
155 > await.Requiref(s.Context(), t, func(at *await.T) {
156 > fn(s.copySuite(t, false, at, at.Context()).(T))
157 > }, timeout, interval, msg, args...)
158 }
159
161 //
162 // Use it for simple local predicates only. Do not use assertions or side effects; use [Await] instead.
163 > func (s *Suite[T]) AwaitTrue(fn func() bool, timeout, interval time.Duration) { suite.go
164 > s.AwaitTruef(fn, timeout, interval, "")
165 > }
166
167 // AwaitTruef is like [AwaitTrue] but includes a format string appended to the failure message.
168 > func (s *Suite[T]) AwaitTruef(fn func() bool, timeout, interval time.Duration, msg string, args ...any) { suite.go
169 > await.RequireTruef(s.T(), fn, timeout, interval, msg, args...)
170 > }
171
172 // Run discovers and runs all exported Test* methods on the given suite in parallel.
177 //
178 // The suite must embed [Suite] and have no other fields.
179 > func Run[T testingSuite](t *testing.T, s T, args ...any) { suite.go
180 > run(t, s, true, args...)
181 > }
182
183 // RunLegacySequential behaves like [Run] but does not mark test methods as parallel.
189 }
190
191 > func run[T testingSuite](t *testing.T, s T, methodsParallel bool, args ...any) { suite.go
192 > t.Helper()
193 >
194 > typ := reflect.TypeFor[T]()
195 > if typ.Kind() != reflect.Pointer || typ.Elem().Kind() != reflect.Struct {
196 panic(fmt.Sprintf("parallelsuite.Run: suite must be a pointer to a struct, got %v", typ))
197 }
198 > structType := typ.Elem() suite.go
199 >
200 > validateSuiteStruct(structType)
201 >
202 > methods := discoverTestMethods(typ, structType, args)
203 > if len(methods) == 0 {
204 panic(fmt.Sprintf("parallelsuite.Run: suite %s has no Test* methods", structType.Name()))
205 }
206
207 > methods = applyTestifyMFilter(methods) suite.go
208 > if len(methods) == 0 {
209 return // all methods filtered by -testify.m; nothing to run
210 }
211
212 > argVals := make([]reflect.Value, len(args)) suite.go
213 > for i, a := range args {
214 argVals[i] = reflect.ValueOf(a)
215 }
216
217 > s.initSuite(t, true, nil, nil) suite.go
218 >
219 > for _, method := range methods {
220 > t.Run(method.Name, func(t *testing.T) {
221 > cpS := s.copySuite(t, methodsParallel, nil, nil)
222 > callArgs := append([]reflect.Value{reflect.ValueOf(cpS)}, argVals...)
223 > method.Func.Call(callArgs)
224 > })
225 }
226 }
228 var inheritedMethods map[string]bool
229
230 > func init() { suite.go
231 > type ds struct{ Suite[*ds] }
232 > ptrType := reflect.TypeFor[*ds]()
233 > inheritedMethods = make(map[string]bool, ptrType.NumMethod())
234 > for method := range ptrType.Methods() {
235 > inheritedMethods[method.Name] = true
236 > }
237 }
238
239 > func validateSuiteStruct(structType reflect.Type) { suite.go
240 > if !strings.HasSuffix(structType.Name(), "Suite") {
241 panic(fmt.Sprintf("parallelsuite.Run: struct name %q must end with \"Suite\"", structType.Name()))
242 }
243
244 > if structType.NumField() != 1 { suite.go
245 panic(fmt.Sprintf(
246 "parallelsuite.Run: suite %s must have no fields besides the embedded parallelsuite.Suite; "+
249 ))
250 }
251 > f := structType.Field(0) suite.go
252 > if !f.Anonymous {
253 panic(fmt.Sprintf(
254 "parallelsuite.Run: suite %s must embed parallelsuite.Suite, found named field %q",
263 // The flag is registered by testify's suite package (imported above); we share
264 // that registration via flag.Lookup rather than registering it a second time.
265 > func applyTestifyMFilter(methods []reflect.Method) []reflect.Method { suite.go
266 > f := flag.Lookup("testify.m")
267 > if f == nil {
268 return methods
269 }
270 > pattern := f.Value.String() suite.go
271 > if pattern == "" {
272 > return methods suite.go
273 > }
274 re, err := regexp.Compile(pattern)
275 if err != nil {
285 }
286
287 > func discoverTestMethods(ptrType, structType reflect.Type, args []any) []reflect.Method { suite.go
288 > expectedNumIn := 1 + len(args)
289 >
290 > for method := range ptrType.Methods() {
291 > name := method.Name
292 > if !strings.HasPrefix(name, "Test") && !inheritedMethods[name] {
293 panic(fmt.Sprintf(
294 "parallelsuite.Run: suite %s has exported method %s that does not start with Test; "+
299 }
300
301 > var methods []reflect.Method suite.go
302 > for method := range ptrType.Methods() {
303 > if !strings.HasPrefix(method.Name, "Test") {
304 > continue
305 }
306
307 > mt := method.Type suite.go
308 > if mt.NumOut() != 0 {
309 panic(fmt.Sprintf(
310 "parallelsuite.Run: method %s.%s must not have return values, got %v",
312 ))
313 }
314 > if mt.NumIn() != expectedNumIn { suite.go
315 panic(fmt.Sprintf(
316 "parallelsuite.Run: method %s.%s has wrong number of parameters: expected %d, got %d (%v)",
319 }
320
321 > for j, a := range args { suite.go
322 paramType := mt.In(1 + j)
323 argType := reflect.TypeOf(a)
330 }
331
332 > methods = append(methods, method) suite.go
333 }
334 > return methods suite.go
335 }
go.temporal.io/server/common/testing/await/require_ctx.go 96 covered LOC · 35 ranges

Open complete file

20 const defaultSoftDeadlockTimeout = 30 * time.Second
21
22 > func softDeadlockTimeout() time.Duration { require_ctx.go
23 > if s := os.Getenv(softDeadlockTimeoutEnvVar); s != "" {
24 if d, err := time.ParseDuration(s); err == nil {
25 return d
26 }
27 }
28 > return defaultSoftDeadlockTimeout require_ctx.go
29 }
30
40 const defaultHardDeadlockTimeout = 10 * time.Second
41
42 > func hardDeadlockTimeout() time.Duration { require_ctx.go
43 > if s := os.Getenv(hardDeadlockTimeoutEnvVar); s != "" {
44 if d, err := time.ParseDuration(s); err == nil {
45 return d
46 }
47 }
48 > return defaultHardDeadlockTimeout require_ctx.go
49 }
50
61 // Requiref is like [Require] but adds a formatted message to the timeout
62 // failure.
63 > func Requiref(ctx context.Context, tb testing.TB, condition func(*T), timeout, pollInterval time.Duration, msg string, args ...any) { require_ctx.go
64 > tb.Helper()
65 > run(ctx, tb, condition, legacyConfig(timeout, pollInterval, fmt.Sprintf(msg, args...)), "Requiref", requireMisuseHint, true)
66 > }
67
68 func run(
74 misuseHint string,
75 cancellable bool,
76 > ) { require_ctx.go
77 > tb.Helper()
78 >
79 > // Skip if the test already failed — no point polling.
80 > if tb.Failed() {
81 tb.Logf("%s: skipping (test already failed)", funcName)
82 return
83 }
84 // Guard: context.WithDeadline panics on a nil parent.
85 > if parentCtx == nil { require_ctx.go
86 tb.Fatalf("%s: nil context", funcName)
87 return
88 }
89
90 > deadline := time.Now().Add(cfg.totalTimeout) require_ctx.go
91 >
92 > // Cap at the parent context's deadline if it's earlier than our timeout.
93 > if parentDeadline, hasDeadline := parentCtx.Deadline(); hasDeadline && parentDeadline.Before(deadline) {
94 deadline = parentDeadline
95 }
97 // Cap at the test's deadline if it's earlier than our deadline.
98 // Ideally, the parent context already accounts for the test's deadline - but we are being defensive.
99 > if d, ok := tb.(interface{ Deadline() (time.Time, bool) }); ok { require_ctx.go
100 > if testDeadline, hasDeadline := d.Deadline(); hasDeadline && testDeadline.Before(deadline) { require_ctx.go
101 deadline = testDeadline
102 }
103 }
104
105 > effectiveTimeout := max(0, time.Until(deadline)) require_ctx.go
106 > awaitCtx, awaitCancel := context.WithDeadline(parentCtx, deadline)
107 > defer awaitCancel()
108 >
109 > report := timeoutReport{effectiveTimeout: effectiveTimeout}
110 >
111 > for {
112 > // Parent context was canceled while we were sleeping (not our deadline).
113 > if err := awaitCtx.Err(); err != nil && !deadlineReached(deadline) {
114 report.reportAttemptErrors(tb)
115 tb.Fatalf("%s: context canceled before condition was satisfied: %v", funcName, err)
117 }
118
119 > report.nextPoll() require_ctx.go
120 >
121 > // Per-attempt context: bounded by the configured attempt timeout and
122 > // further capped by the overall awaitCtx.
123 > attemptCtx, attemptCancel := context.WithTimeout(awaitCtx, cfg.attemptTimeout)
124 > t := &T{tb: tb, ctx: attemptCtx}
125 >
126 > // Run attempt.
127 > res := runAttempt(t, condition, attemptCancel, funcName, cancellable)
128 > attemptCancel()
129 > if res.panicVal != nil {
130 panic(res.panicVal) // propagate to caller
131 }
132 > if res.deadlocked { require_ctx.go
133 report.reportAttemptErrors(tb)
134 if cancellable {
141 return
142 }
143 > report.recordErrors(t.errors) require_ctx.go
144 >
145 > // Attempt-timeout expiry: attemptCtx is done but awaitCtx is not.
146 > // Record nothing special - the attempt's recorded errors (if any)
147 > // already describe what went wrong; otherwise we just retry.
148 > attemptHitOwnTimeout := attemptCtx.Err() == context.DeadlineExceeded && awaitCtx.Err() == nil
149 > if attemptHitOwnTimeout {
150 report.recordAttemptTimeout()
151 }
152
153 // Check misuse where the real test failed instead of just the attempt.
154 > if tb.Failed() { require_ctx.go
155 tb.Fatalf("%s: the test was marked failed directly — %s", funcName, misuseHint)
156 return
158
159 // Parent context was canceled during the attempt (not our deadline).
160 > if err := awaitCtx.Err(); err != nil && !deadlineReached(deadline) { require_ctx.go
161 report.reportAttemptErrors(tb)
162 tb.Fatalf("%s: context canceled before condition was satisfied: %v", funcName, err)
165
166 // Our deadline expired.
167 > if deadlineReached(deadline) { require_ctx.go
168 report.reportTimeout(tb, funcName, cfg.timeoutMsg)
169 return
171
172 // Success: attempt completed without failures.
173 > if !res.stopped && !t.Failed() && !attemptHitOwnTimeout { require_ctx.go
174 > return require_ctx.go
175 > }
176
177 // Wait for pollInterval, or context is canceled or deadline is reached.
178 > sleep(awaitCtx, deadline, cfg.pollInterval) require_ctx.go
179 }
180 }
211 funcName string,
212 cancellable bool,
213 > ) attemptResult { require_ctx.go
214 > done := make(chan attemptResult, 1)
215 >
216 > go func() {
217 > completed := false
218 > defer func() {
219 > if r := recover(); r != nil { require_ctx.go
220 > if _, ok := r.(attemptFailed); ok { require_ctx.go
221 > done <- attemptResult{stopped: true} require_ctx.go
222 > return
223 > }
224 done <- attemptResult{panicVal: r}
225 return
227 // recover returned nil: either normal return (completed=true) or
228 // runtime.Goexit (completed=false; Goexit is not a panic).
229 > done <- attemptResult{stopped: !completed} require_ctx.go
230 }()
231 > condition(t) require_ctx.go
232 > completed = true
233 }()
234
235 > if cancellable { require_ctx.go
236 > // Soft phase: wait for the condition, our soft timer, or parent cancel. require_ctx.go
237 > softTimer := time.NewTimer(softDeadlockTimeout())
238 > defer softTimer.Stop()
239 >
240 > select {
241 > case r := <-done: require_ctx.go
242 > return r
243 case <-softTimer.C:
244 // Soft deadlock: log a warning.
255
256 // Hard phase: wait for the condition or the hard timer.
257 > hardTimer := time.NewTimer(hardDeadlockTimeout()) require_ctx.go
258 > defer hardTimer.Stop()
259 >
260 > select {
261 > case r := <-done: require_ctx.go
262 > return r
263 case <-hardTimer.C:
264 return attemptResult{deadlocked: true}
266 }
267
268 > func sleep(ctx context.Context, deadline time.Time, pollInterval time.Duration) { require_ctx.go
269 > remaining := time.Until(deadline)
270 > if remaining < pollInterval {
271 pollInterval = remaining
272 }
273
274 > timer := time.NewTimer(pollInterval) require_ctx.go
275 > defer timer.Stop()
276 >
277 > select {
278 case <-ctx.Done():
279 > case <-timer.C: require_ctx.go
280 }
281 }
282
283 > func deadlineReached(deadline time.Time) bool { require_ctx.go
284 > return !time.Now().Before(deadline)
285 > }
go.temporal.io/server/common/clock/event_time_source.go 80 covered LOC · 21 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.
41 > func NewEventTimeSource() *EventTimeSource { event_time_source.go
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.
57 > func (ts *EventTimeSource) Now() time.Time { event_time_source.go
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 {
82 // NewTimer creates a Timer that will send the current time on a channel after at least
83 // duration d. It returns the channel and the Timer.
84 > func (ts *EventTimeSource) NewTimer(d time.Duration) (<-chan time.Time, Timer) { event_time_source.go
85 > c := make(chan time.Time, 1)
86 > // we can't call ts.Now() from the callback so just calculate what it should be
87 > target := ts.Now().Add(d)
88 > timer := &fakeTimer{
89 > timeSource: ts,
90 > deadline: target,
91 > callback: func() { c <- target },
92 c: c,
93 }
94 > ts.addTimer(timer) event_time_source.go
95 > return c, timer
96 }
97
98 > func (ts *EventTimeSource) addTimer(t *fakeTimer) { event_time_source.go
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:
116
117 // Advance the timer by the specified duration.
118 > func (ts *EventTimeSource) Advance(d time.Duration) { event_time_source.go
119 > ts.mu.Lock()
120 > defer ts.mu.Unlock()
121 >
122 > ts.now = ts.now.Add(d)
123 > ts.fireTimers()
124 > }
125
126 // AdvanceNext advances to the next timer.
127 > func (ts *EventTimeSource) AdvanceNext() { event_time_source.go
128 > ts.mu.Lock()
129 > defer ts.mu.Unlock()
130 >
131 > if len(ts.timers) == 0 {
132 return
133 }
134 // just do linear search, this is efficient enough for now
135 > tmin := ts.timers[0].deadline event_time_source.go
136 > for _, t := range ts.timers[1:] {
137 tmin = util.MinTime(tmin, t.deadline)
138 }
139 > ts.now = tmin event_time_source.go
140 > ts.fireTimers()
141 }
142
143 // NumTimers returns the number of outstanding timers.
144 > func (ts *EventTimeSource) NumTimers() int { event_time_source.go
145 > ts.mu.Lock()
146 > defer ts.mu.Unlock()
147 >
148 > return len(ts.timers)
149 > }
150
151 // Sleep is a convenience function for waiting on a new timer.
156
157 // fireTimers fires all timers that are ready.
158 > func (ts *EventTimeSource) fireTimers() { event_time_source.go
159 > n := 0
160 > for _, t := range ts.timers {
161 > if t.deadline.After(ts.now) { event_time_source.go
162 > ts.timers[n] = t event_time_source.go
163 > t.index = n
164 > n++
165 > } else { event_time_source.go
166 > if ts.async { event_time_source.go
167 go t.callback()
168 > } else { event_time_source.go
169 > t.callback() event_time_source.go
170 > }
171 > t.done = true event_time_source.go
172 }
173 }
174 > ts.timers = ts.timers[:n] event_time_source.go
175 }
176
200
201 // Stop the timer. Returns true if the timer was active.
202 > func (t *fakeTimer) Stop() bool { event_time_source.go
203 > t.timeSource.mu.Lock()
204 > defer t.timeSource.mu.Unlock()
205 >
206 > if t.done {
207 return false
208 }
209
210 > i := t.index event_time_source.go
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 }
go.temporal.io/server/common/testing/testcontext/context.go 72 covered LOC · 16 ranges

Open complete file

39
40 // DefaultTimeout returns the effective default timeout for test-scoped contexts.
41 > func DefaultTimeout() time.Duration { context.go
42 > return effectiveTimeout(0)
43 > }
44
45 // For returns the test-scoped context for tb. The context is canceled
49 // return the same context, but an explicit different timeout fails instead of
50 // being silently ignored.
51 > func For(tb testing.TB, opts ...Option) context.Context { context.go
52 > tb.Helper()
53 >
54 > cfg := config{timeout: DefaultTimeout()}
55 > for _, opt := range opts {
56 opt(&cfg)
57 }
58
59 > st := getContextState(tb, cfg.timeout) context.go
60 > st.configure(tb, cfg)
61 > return st.context()
62 }
63
98 }
99
100 > func getContextState(tb testing.TB, timeout time.Duration) *contextState { context.go
101 > tb.Helper()
102 >
103 > testContexts.Lock()
104 > defer testContexts.Unlock()
105 >
106 > if st, ok := testContexts.byTest[tb]; ok {
107 > return st context.go
108 > }
109
110 > ctx, cancel := context.WithTimeout(tb.Context(), timeout) context.go
111 >
112 > // Annotate gRPC requests with the test name for OTEL tracing.
113 > ctx = metadata.AppendToOutgoingContext(ctx, testNameMetadataKey, tb.Name())
114 >
115 > st := &contextState{
116 > ctx: ctx,
117 > cancel: cancel,
118 > timeout: timeout,
119 > decorators: make(map[any]struct{}),
120 > }
121 > testContexts.byTest[tb] = st
122 >
123 > tb.Cleanup(func() {
124 > err := st.err()
125 > st.cancel()
126 > testContexts.Lock()
127 > delete(testContexts.byTest, tb)
128 > testContexts.Unlock()
129 > if err == context.DeadlineExceeded {
130 tb.Errorf("test exceeded timeout of %v", st.timeout)
131 }
132 > st.release() context.go
133 })
134 > return st context.go
135 }
136
137 > func (s *contextState) configure(tb testing.TB, cfg config) { context.go
138 > tb.Helper()
139 >
140 > s.mu.Lock()
141 > defer s.mu.Unlock()
142 >
143 > if cfg.timeoutSet && cfg.timeout != s.timeout {
144 tb.Fatalf("testcontext: test context already exists with timeout %v; cannot change it to %v", s.timeout, cfg.timeout)
145 }
168 }
169
170 > func (s *contextState) context() context.Context { context.go
171 > s.mu.Lock()
172 > defer s.mu.Unlock()
173 > return s.ctx
174 > }
175
176 > func (s *contextState) err() error { context.go
177 > s.mu.Lock()
178 > defer s.mu.Unlock()
179 > return s.ctx.Err()
180 > }
181
182 > func (s *contextState) release() { context.go
183 > s.mu.Lock()
184 > defer s.mu.Unlock()
185 > s.ctx = nil
186 > }
187
188 > func effectiveTimeout(customTimeout time.Duration) (timeout time.Duration) { context.go
189 > defer func() {
190 > // Build flag TEMPORAL_DEBUG applies a timeout multiplier to all test timeouts.
191 > timeout *= debug.TimeoutMultiplier
192 > }()
193
194 // 1. Custom timeout (via WithTimeout option).
195 > if customTimeout > 0 { context.go
196 return customTimeout
197 }
198
199 // 2. TEMPORAL_TEST_TIMEOUT environment variable.
200 > if envTimeout := os.Getenv("TEMPORAL_TEST_TIMEOUT"); envTimeout != "" { context.go
201 if dur, err := time.ParseDuration(envTimeout); err == nil && dur > 0 {
202 return dur
205
206 // 3. Default timeout.
207 > return defaultTimeout context.go
208 }
go.temporal.io/server/common/goro/adaptive_pool.go 62 covered LOC · 22 ranges

Open complete file

32 targetDelay time.Duration,
33 shrinkFactor float64,
34 > ) *AdaptivePool { adaptive_pool.go
35 > p := &AdaptivePool{
36 > ts: ts,
37 > minWorkers: minWorkers,
38 > maxWorkers: maxWorkers,
39 > targetDelay: targetDelay,
40 > shrinkFactor: shrinkFactor,
41 > ch: make(chan func()),
42 > stopCh: make(chan struct{}),
43 > }
44 > for range minWorkers {
45 > go p.work() adaptive_pool.go
46 > }
47 > p.workers.Store(int64(minWorkers)) adaptive_pool.go
48 > return p
49 }
50
52 // When Stop is called, concurrent calls to Do may or may not call their function, and future
53 // calls definitely won't.
54 > func (p *AdaptivePool) Stop() { adaptive_pool.go
55 > close(p.stopCh)
56 > }
57
58 // Do calls f() on a worker goroutine. If the call can't be started within targetDelay, it adds
59 // another worker. If Stop is called concurrently, Do may or may not call f. If Stop has been
60 // called already, Do does nothing.
61 > func (p *AdaptivePool) Do(f func()) { adaptive_pool.go
62 > // try send first
63 > select {
64 > case p.ch <- f: adaptive_pool.go
65 > return
66 > default: adaptive_pool.go
67 }
68
69 // we might want to add a worker, send with timeout
70 > have := p.workers.Load() adaptive_pool.go
71 > if have < int64(p.maxWorkers) {
72 > timech, timer := p.ts.NewTimer(p.targetDelay) adaptive_pool.go
73 > select {
74 case <-p.stopCh:
75 timer.Stop()
76 return
77 > case p.ch <- f: adaptive_pool.go
78 > timer.Stop()
79 > return
80 > case <-timech: adaptive_pool.go
81 }
82
83 > if p.workers.CompareAndSwap(have, have+1) { adaptive_pool.go
84 > go p.work()
85 > }
86 }
87
88 // blocking send
89 > select { adaptive_pool.go
90 > case p.ch <- f:
91 case <-p.stopCh:
92 }
93 }
94
95 > func (p *AdaptivePool) work() { adaptive_pool.go
96 > for {
97 > // try receive first
98 > select {
99 > case f := <-p.ch: adaptive_pool.go
100 > f()
101 > continue
102 > default: adaptive_pool.go
103 }
104
105 > have := p.workers.Load() adaptive_pool.go
106 > if have > int64(p.minWorkers) {
107 > // we might want to exit, receive with timeout adaptive_pool.go
108 > // jitter this so we shrink slower than we grow
109 > timech, timer := p.ts.NewTimer(time.Duration(float64(p.targetDelay) * p.shrinkFactor * rand.Float64()))
110 > select {
111 case <-p.stopCh:
112 timer.Stop()
113 return
114 > case f := <-p.ch: adaptive_pool.go
115 > timer.Stop()
116 > f()
117 > continue
118 > case <-timech: adaptive_pool.go
119 }
120 > if p.workers.CompareAndSwap(have, have-1) { adaptive_pool.go
121 > return
122 > }
123 }
124
134
135 // NumWorkers returns the current number of workers. Probably only useful for testing.
136 > func (p *AdaptivePool) NumWorkers() int { adaptive_pool.go
137 > return int(p.workers.Load())
138 > }
go.temporal.io/server/common/testing/await/t.go 21 covered LOC · 7 ranges

Open complete file

24
25 // Context returns the await-scoped context for the current attempt.
26 > func (t *T) Context() context.Context { t.go
27 > if t.ctx != nil {
28 > return t.ctx
29 > }
30 return t.tb.Context()
31 }
47
48 // Fail marks the current attempt as failed without stopping it.
49 > func (t *T) Fail() { t.go
50 > t.failed = true
51 > }
52
53 // Error records an error message for reporting on timeout.
58
59 // Errorf records an error message for reporting on timeout.
60 > func (t *T) Errorf(format string, args ...any) { t.go
61 > t.Fail()
62 > t.errors = append(t.errors, fmt.Sprintf(format, args...))
63 > }
64
65 // FailNow is called by require.* on failure. It stops the current attempt.
66 // Unlike testing.TB.FailNow(), this does NOT mark the test as failed.
67 > func (t *T) FailNow() { t.go
68 > t.Fail()
69 > panic(attemptFailed{})
70 }
71
83
84 // Failed reports whether this attempt has failed.
85 > func (t *T) Failed() bool { t.go
86 > return t.failed
87 > }
88
89 // Helper marks the calling function as a test helper.
90 > func (t *T) Helper() { t.go
91 > if t.tb != nil {
92 > t.tb.Helper() t.go
93 > }
94 }
go.temporal.io/server/common/testing/await/config.go 17 covered LOC · 3 ranges

Open complete file

17 }
18
19 > func newConfig() config { config.go
20 > return config{
21 > attemptTimeout: envDuration(attemptTimeoutEnvVar, 10*time.Second) * debug.TimeoutMultiplier,
22 > }
23 > }
24
25 > func legacyConfig(timeout, pollInterval time.Duration, timeoutMsg string) config { config.go
26 > cfg := newConfig()
27 > cfg.totalTimeout = timeout
28 > cfg.pollInterval = pollInterval
29 > cfg.timeoutMsg = timeoutMsg
30 > return cfg
31 > }
32
33 > func envDuration(name string, fallback time.Duration) time.Duration { config.go
34 > if s := os.Getenv(name); s != "" {
35 > if d, err := time.ParseDuration(s); err == nil && d > 0 {
36 > return d
37 > }
38 }
39 return fallback
go.temporal.io/server/common/testing/await/report.go 7 covered LOC · 3 ranges

Open complete file

28 }
29
30 > func (r *timeoutReport) nextPoll() { report.go
31 > r.attempts++
32 > }
33
34 > func (r *timeoutReport) recordErrors(errors []string) { report.go
35 > if len(errors) > 0 {
36 > r.failures = append(r.failures, attemptFailure{attempt: r.attempts, errors: errors}) report.go
37 > }
38 }
39
go.temporal.io/server/common/testing/await/require_true.go 6 covered LOC · 2 ranges

Open complete file

27 // RequireTruef is like [RequireTrue] but accepts a format string that is included
28 // in the failure message when the condition is not satisfied before the timeout.
29 > func RequireTruef(tb testing.TB, condition func() bool, timeout, pollInterval time.Duration, msg string, args ...any) { require_true.go
30 > tb.Helper()
31 > run(testcontext.For(tb), tb, func(t *T) {
32 > if !condition() {
33 > t.Fail() require_true.go
34 > }
35 }, legacyConfig(timeout, pollInterval, fmt.Sprintf(msg, args...)), "RequireTruef", requireTrueMisuseHint, false)
36 }
go.temporal.io/server/common/testing/historyrequire/history_require.go 5 covered LOC · 1 range

Open complete file

36 )
37
38 > func New(t require.TestingT) HistoryRequire { history_require.go
39 > return HistoryRequire{
40 > t: t,
41 > }
42 > }
43
44 // TODO (maybe):
go.temporal.io/server/common/testing/parallelsuite/guard.go 3 covered LOC · 2 ranges

Open complete file

16 }
17
18 > func (g *guardT) Helper() { guard.go
19 > if g.hasSubtests.Load() {
20 panic(fmt.Sprintf(
21 "parallelsuite: assertion called on %q after Run() was called; "+
24 ))
25 }
26 > g.T.Helper() guard.go
27 }
28
go.temporal.io/server/common/testing/protorequire/require.go 3 covered LOC · 1 range

Open complete file

38 }
39
40 > func New(t require.TestingT) ProtoAssertions { require.go
41 > return ProtoAssertions{t}
42 > }
43
44 // ProtoEqual compares two proto messages for equality using proto semantics. Options can be passed to customize