1package syncdispatch
2
3import (
4 "sync/atomic"
5 "testing"
6 "time"
7)
8
9func TestRunSynchronousDispatchRunsInlineWhenOwner(t *testing.T) {
10 var dispatched atomic.Bool
11 var ran atomic.Bool
12
13 result := RunSynchronousDispatch(SyncDispatchOptions{
14 Label: "test.inline",
15 Timeout: time.Second,
16 IsOwner: func() bool { return true },
17 Dispatch: func(func()) {
18 dispatched.Store(true)
19 },
20 }, func() {
21 ran.Store(true)
22 })
23
24 if result.Status != SyncDispatchInline {
25 t.Fatalf("status = %s, want %s", result.Status, SyncDispatchInline)
26 }
27 if !ran.Load() {
28 t.Fatal("callback did not run inline")
29 }
30 if dispatched.Load() {
31 t.Fatal("callback was dispatched despite owner context")
32 }
33}
34
35func TestRunSynchronousDispatchCompletesWhenDispatchedCallbackRuns(t *testing.T) {
36 var ran atomic.Bool
37
38 result := RunSynchronousDispatch(SyncDispatchOptions{
39 Label: "test.dispatch",
40 Timeout: time.Second,
41 IsOwner: func() bool { return false },
42 Dispatch: func(fn func()) {
43 go fn()
44 },
45 }, func() {
46 ran.Store(true)
47 })
48
49 if result.Status != SyncDispatchCompleted {
50 t.Fatalf("status = %s, want %s", result.Status, SyncDispatchCompleted)
51 }
52 if !result.Completed() {
53 t.Fatal("result.Completed() = false, want true")
54 }
55 if !ran.Load() {
56 t.Fatal("callback did not run")
57 }
58}
59
60func TestRunSynchronousDispatchTimesOutWhenDispatchBlocks(t *testing.T) {
61 releaseDispatch := make(chan struct{})
62 var ran atomic.Bool
63
64 result := RunSynchronousDispatch(SyncDispatchOptions{
65 Label: "test.blocking_dispatch",
66 Timeout: 5 * time.Millisecond,
67 IsOwner: func() bool { return false },
68 Dispatch: func(fn func()) {
69 <-releaseDispatch
70 fn()
71 },
72 }, func() {
73 ran.Store(true)
74 })
75
76 if result.Status != SyncDispatchTimedOut {
77 t.Fatalf("status = %s, want %s", result.Status, SyncDispatchTimedOut)
78 }
79 if ran.Load() {
80 t.Fatal("callback ran while dispatch was blocked")
81 }
82 close(releaseDispatch)
83 time.Sleep(20 * time.Millisecond)
84 if ran.Load() {
85 t.Fatal("callback ran after a blocking dispatch timed out")
86 }
87}
88
89func TestRunSynchronousDispatchTimesOutWhenCallbackNeverStarts(t *testing.T) {
90 var ran atomic.Bool
91
92 result := RunSynchronousDispatch(SyncDispatchOptions{
93 Label: "test.timeout",
94 Timeout: 5 * time.Millisecond,
95 IsOwner: func() bool { return false },
96 Dispatch: func(func()) {
97 // Simulate an event loop that never services the queued callback.
98 },
99 }, func() {
100 ran.Store(true)
101 })
102
103 if result.Status != SyncDispatchTimedOut {
104 t.Fatalf("status = %s, want %s", result.Status, SyncDispatchTimedOut)
105 }
106 if result.Completed() {
107 t.Fatal("result.Completed() = true, want false")
108 }
109 if ran.Load() {
110 t.Fatal("callback ran despite timeout before dispatch start")
111 }
112}
113
114func TestRunSynchronousDispatchWaitsForStartedCallbackPastTimeout(t *testing.T) {
115 started := make(chan struct{})
116 release := make(chan struct{})
117 resultCh := make(chan SyncDispatchResult, 1)
118 var ran atomic.Bool
119
120 go func() {
121 resultCh <- RunSynchronousDispatch(SyncDispatchOptions{
122 Label: "test.started_late_finish",
123 Timeout: 50 * time.Millisecond,
124 IsOwner: func() bool { return false },
125 Dispatch: func(fn func()) {
126 go fn()
127 },
128 }, func() {
129 close(started)
130 <-release
131 ran.Store(true)
132 })
133 }()
134
135 select {
136 case <-started:
137 case <-time.After(time.Second):
138 t.Fatal("callback did not start")
139 }
140 time.Sleep(100 * time.Millisecond)
141 if ran.Load() {
142 t.Fatal("callback ran before release")
143 }
144 close(release)
145
146 select {
147 case result := <-resultCh:
148 if result.Status != SyncDispatchCompletedAfterTimeout {
149 t.Fatalf("status = %s, want %s", result.Status, SyncDispatchCompletedAfterTimeout)
150 }
151 if !result.Completed() {
152 t.Fatal("result.Completed() = false, want true")
153 }
154 case <-time.After(time.Second):
155 t.Fatal("RunSynchronousDispatch did not return after started callback completed")
156 }
157 if !ran.Load() {
158 t.Fatal("callback did not run")
159 }
160}
161
162func TestRunSynchronousDispatchCancelsCallbackThatAttemptsToStartAfterDeadline(t *testing.T) {
163 callbackAttempted := make(chan struct{}, 1)
164 var ran atomic.Bool
165
166 result := RunSynchronousDispatch(SyncDispatchOptions{
167 Label: "test.after_deadline",
168 Timeout: 5 * time.Millisecond,
169 IsOwner: func() bool { return false },
170 Dispatch: func(fn func()) {
171 go func() {
172 time.Sleep(20 * time.Millisecond)
173 fn()
174 callbackAttempted <- struct{}{}
175 }()
176 },
177 }, func() {
178 ran.Store(true)
179 })
180
181 if result.Status != SyncDispatchTimedOut {
182 t.Fatalf("status = %s, want %s", result.Status, SyncDispatchTimedOut)
183 }
184 select {
185 case <-callbackAttempted:
186 case <-time.After(time.Second):
187 t.Fatal("callback did not attempt late start")
188 }
189 if ran.Load() {
190 t.Fatal("callback ran after deadline")
191 }
192}
193
194func TestRunSynchronousDispatchLeavesCallbackQueuedWhenBlockingDispatchTimesOutAndLateStartAllowed(t *testing.T) {
195 releaseDispatch := make(chan struct{})
196 var ran atomic.Bool
197
198 result := RunSynchronousDispatch(SyncDispatchOptions{
199 Label: "test.blocking_cleanup_dispatch",
200 Timeout: 5 * time.Millisecond,
201 IsOwner: func() bool { return false },
202 AllowLateStartAfterTimeout: true,
203 Dispatch: func(fn func()) {
204 <-releaseDispatch
205 fn()
206 },
207 }, func() {
208 ran.Store(true)
209 })
210
211 if result.Status != SyncDispatchQueuedAfterTimeout {
212 t.Fatalf("status = %s, want %s", result.Status, SyncDispatchQueuedAfterTimeout)
213 }
214 if ran.Load() {
215 t.Fatal("callback ran before dispatch was released")
216 }
217 close(releaseDispatch)
218 deadline := time.Now().Add(time.Second)
219 for !ran.Load() && time.Now().Before(deadline) {
220 time.Sleep(time.Millisecond)
221 }
222 if !ran.Load() {
223 t.Fatal("queued cleanup callback did not run")
224 }
225}
226
227func TestRunSynchronousDispatchLeavesCallbackQueuedAfterTimeoutWhenConfigured(t *testing.T) {
228 delayed := make(chan func(), 1)
229 var ran atomic.Bool
230
231 result := RunSynchronousDispatch(SyncDispatchOptions{
232 Label: "test.late_cleanup",
233 Timeout: 5 * time.Millisecond,
234 IsOwner: func() bool { return false },
235 AllowLateStartAfterTimeout: true,
236 Dispatch: func(fn func()) {
237 delayed <- fn
238 },
239 }, func() {
240 ran.Store(true)
241 })
242
243 if result.Status != SyncDispatchQueuedAfterTimeout {
244 t.Fatalf("status = %s, want %s", result.Status, SyncDispatchQueuedAfterTimeout)
245 }
246 if result.Completed() {
247 t.Fatal("result.Completed() = true, want false")
248 }
249 if ran.Load() {
250 t.Fatal("callback ran before late dispatch")
251 }
252
253 select {
254 case fn := <-delayed:
255 fn()
256 case <-time.After(time.Second):
257 t.Fatal("dispatch callback was not captured")
258 }
259
260 if !ran.Load() {
261 t.Fatal("queued callback did not run after timeout")
262 }
263}
264
265func TestRunSynchronousDispatchCancelsCallbackThatStartsAfterTimeout(t *testing.T) {
266 delayed := make(chan func(), 1)
267 var ran atomic.Bool
268
269 result := RunSynchronousDispatch(SyncDispatchOptions{
270 Label: "test.late",
271 Timeout: 5 * time.Millisecond,
272 IsOwner: func() bool { return false },
273 Dispatch: func(fn func()) {
274 delayed <- fn
275 },
276 }, func() {
277 ran.Store(true)
278 })
279
280 if result.Status != SyncDispatchTimedOut {
281 t.Fatalf("status = %s, want %s", result.Status, SyncDispatchTimedOut)
282 }
283
284 select {
285 case fn := <-delayed:
286 fn()
287 case <-time.After(time.Second):
288 t.Fatal("dispatch callback was not captured")
289 }
290
291 if ran.Load() {
292 t.Fatal("late callback ran after timeout")
293 }
294}