1
2
3
4
5
6
7
8
9
10
11
12
13#include "qemu/osdep.h"
14#include <glib.h>
15#include "block/aio.h"
16#include "qapi/error.h"
17#include "qemu/coroutine.h"
18#include "qemu/thread.h"
19#include "qemu/error-report.h"
20#include "iothread.h"
21
22
23
24#define NUM_CONTEXTS 5
25
26static IOThread *threads[NUM_CONTEXTS];
27static AioContext *ctx[NUM_CONTEXTS];
28static __thread int id = -1;
29
30static QemuEvent done_event;
31
32
33
34typedef struct CtxRunData {
35 QEMUBHFunc *cb;
36 void *arg;
37} CtxRunData;
38
39static void ctx_run_bh_cb(void *opaque)
40{
41 CtxRunData *data = opaque;
42
43 data->cb(data->arg);
44 qemu_event_set(&done_event);
45}
46
47static void ctx_run(int i, QEMUBHFunc *cb, void *opaque)
48{
49 CtxRunData data = {
50 .cb = cb,
51 .arg = opaque
52 };
53
54 qemu_event_reset(&done_event);
55 aio_bh_schedule_oneshot(ctx[i], ctx_run_bh_cb, &data);
56 qemu_event_wait(&done_event);
57}
58
59
60
61static void set_id_cb(void *opaque)
62{
63 int *i = opaque;
64
65 id = *i;
66}
67
68static void create_aio_contexts(void)
69{
70 int i;
71
72 for (i = 0; i < NUM_CONTEXTS; i++) {
73 threads[i] = iothread_new();
74 ctx[i] = iothread_get_aio_context(threads[i]);
75 }
76
77 qemu_event_init(&done_event, false);
78 for (i = 0; i < NUM_CONTEXTS; i++) {
79 ctx_run(i, set_id_cb, &i);
80 }
81}
82
83
84
85static void join_aio_contexts(void)
86{
87 int i;
88
89 for (i = 0; i < NUM_CONTEXTS; i++) {
90 aio_context_ref(ctx[i]);
91 }
92 for (i = 0; i < NUM_CONTEXTS; i++) {
93 iothread_join(threads[i]);
94 }
95 for (i = 0; i < NUM_CONTEXTS; i++) {
96 aio_context_unref(ctx[i]);
97 }
98 qemu_event_destroy(&done_event);
99}
100
101
102
103static void test_lifecycle(void)
104{
105 create_aio_contexts();
106 join_aio_contexts();
107}
108
109
110
111static Coroutine *to_schedule[NUM_CONTEXTS];
112
113static bool now_stopping;
114
115static int count_retry;
116static int count_here;
117static int count_other;
118
119static bool schedule_next(int n)
120{
121 Coroutine *co;
122
123 co = atomic_xchg(&to_schedule[n], NULL);
124 if (!co) {
125 atomic_inc(&count_retry);
126 return false;
127 }
128
129 if (n == id) {
130 atomic_inc(&count_here);
131 } else {
132 atomic_inc(&count_other);
133 }
134
135 aio_co_schedule(ctx[n], co);
136 return true;
137}
138
139static void finish_cb(void *opaque)
140{
141 schedule_next(id);
142}
143
144static coroutine_fn void test_multi_co_schedule_entry(void *opaque)
145{
146 g_assert(to_schedule[id] == NULL);
147
148 while (!atomic_mb_read(&now_stopping)) {
149 int n;
150
151 n = g_test_rand_int_range(0, NUM_CONTEXTS);
152 schedule_next(n);
153
154 atomic_mb_set(&to_schedule[id], qemu_coroutine_self());
155 qemu_coroutine_yield();
156 g_assert(to_schedule[id] == NULL);
157 }
158}
159
160
161static void test_multi_co_schedule(int seconds)
162{
163 int i;
164
165 count_here = count_other = count_retry = 0;
166 now_stopping = false;
167
168 create_aio_contexts();
169 for (i = 0; i < NUM_CONTEXTS; i++) {
170 Coroutine *co1 = qemu_coroutine_create(test_multi_co_schedule_entry, NULL);
171 aio_co_schedule(ctx[i], co1);
172 }
173
174 g_usleep(seconds * 1000000);
175
176 atomic_mb_set(&now_stopping, true);
177 for (i = 0; i < NUM_CONTEXTS; i++) {
178 ctx_run(i, finish_cb, NULL);
179 to_schedule[i] = NULL;
180 }
181
182 join_aio_contexts();
183 g_test_message("scheduled %d, queued %d, retry %d, total %d\n",
184 count_other, count_here, count_retry,
185 count_here + count_other + count_retry);
186}
187
188static void test_multi_co_schedule_1(void)
189{
190 test_multi_co_schedule(1);
191}
192
193static void test_multi_co_schedule_10(void)
194{
195 test_multi_co_schedule(10);
196}
197
198
199
200static uint32_t atomic_counter;
201static uint32_t running;
202static uint32_t counter;
203static CoMutex comutex;
204
205static void coroutine_fn test_multi_co_mutex_entry(void *opaque)
206{
207 while (!atomic_mb_read(&now_stopping)) {
208 qemu_co_mutex_lock(&comutex);
209 counter++;
210 qemu_co_mutex_unlock(&comutex);
211
212
213
214
215
216
217 atomic_inc(&atomic_counter);
218 }
219 atomic_dec(&running);
220}
221
222static void test_multi_co_mutex(int threads, int seconds)
223{
224 int i;
225
226 qemu_co_mutex_init(&comutex);
227 counter = 0;
228 atomic_counter = 0;
229 now_stopping = false;
230
231 create_aio_contexts();
232 assert(threads <= NUM_CONTEXTS);
233 running = threads;
234 for (i = 0; i < threads; i++) {
235 Coroutine *co1 = qemu_coroutine_create(test_multi_co_mutex_entry, NULL);
236 aio_co_schedule(ctx[i], co1);
237 }
238
239 g_usleep(seconds * 1000000);
240
241 atomic_mb_set(&now_stopping, true);
242 while (running > 0) {
243 g_usleep(100000);
244 }
245
246 join_aio_contexts();
247 g_test_message("%d iterations/second\n", counter / seconds);
248 g_assert_cmpint(counter, ==, atomic_counter);
249}
250
251
252
253
254
255static void test_multi_co_mutex_1(void)
256{
257 test_multi_co_mutex(NUM_CONTEXTS, 1);
258}
259
260static void test_multi_co_mutex_10(void)
261{
262 test_multi_co_mutex(NUM_CONTEXTS, 10);
263}
264
265
266
267
268
269
270static void test_multi_co_mutex_2_3(void)
271{
272 test_multi_co_mutex(2, 3);
273}
274
275static void test_multi_co_mutex_2_30(void)
276{
277 test_multi_co_mutex(2, 30);
278}
279
280
281
282#ifdef CONFIG_LINUX
283#include "qemu/futex.h"
284
285
286
287
288static struct {
289 int next, locked;
290 int padding[14];
291} nodes[NUM_CONTEXTS] __attribute__((__aligned__(64)));
292
293static int mutex_head = -1;
294
295static void mcs_mutex_lock(void)
296{
297 int prev;
298
299 nodes[id].next = -1;
300 nodes[id].locked = 1;
301 prev = atomic_xchg(&mutex_head, id);
302 if (prev != -1) {
303 atomic_set(&nodes[prev].next, id);
304 qemu_futex_wait(&nodes[id].locked, 1);
305 }
306}
307
308static void mcs_mutex_unlock(void)
309{
310 int next;
311 if (atomic_read(&nodes[id].next) == -1) {
312 if (atomic_read(&mutex_head) == id &&
313 atomic_cmpxchg(&mutex_head, id, -1) == id) {
314
315 return;
316 }
317 while (atomic_read(&nodes[id].next) == -1) {
318
319
320
321 }
322 }
323
324
325 next = atomic_read(&nodes[id].next);
326 nodes[next].locked = 0;
327 qemu_futex_wake(&nodes[next].locked, 1);
328}
329
330static void test_multi_fair_mutex_entry(void *opaque)
331{
332 while (!atomic_mb_read(&now_stopping)) {
333 mcs_mutex_lock();
334 counter++;
335 mcs_mutex_unlock();
336 atomic_inc(&atomic_counter);
337 }
338 atomic_dec(&running);
339}
340
341static void test_multi_fair_mutex(int threads, int seconds)
342{
343 int i;
344
345 assert(mutex_head == -1);
346 counter = 0;
347 atomic_counter = 0;
348 now_stopping = false;
349
350 create_aio_contexts();
351 assert(threads <= NUM_CONTEXTS);
352 running = threads;
353 for (i = 0; i < threads; i++) {
354 Coroutine *co1 = qemu_coroutine_create(test_multi_fair_mutex_entry, NULL);
355 aio_co_schedule(ctx[i], co1);
356 }
357
358 g_usleep(seconds * 1000000);
359
360 atomic_mb_set(&now_stopping, true);
361 while (running > 0) {
362 g_usleep(100000);
363 }
364
365 join_aio_contexts();
366 g_test_message("%d iterations/second\n", counter / seconds);
367 g_assert_cmpint(counter, ==, atomic_counter);
368}
369
370static void test_multi_fair_mutex_1(void)
371{
372 test_multi_fair_mutex(NUM_CONTEXTS, 1);
373}
374
375static void test_multi_fair_mutex_10(void)
376{
377 test_multi_fair_mutex(NUM_CONTEXTS, 10);
378}
379#endif
380
381
382
383
384static QemuMutex mutex;
385
386static void test_multi_mutex_entry(void *opaque)
387{
388 while (!atomic_mb_read(&now_stopping)) {
389 qemu_mutex_lock(&mutex);
390 counter++;
391 qemu_mutex_unlock(&mutex);
392 atomic_inc(&atomic_counter);
393 }
394 atomic_dec(&running);
395}
396
397static void test_multi_mutex(int threads, int seconds)
398{
399 int i;
400
401 qemu_mutex_init(&mutex);
402 counter = 0;
403 atomic_counter = 0;
404 now_stopping = false;
405
406 create_aio_contexts();
407 assert(threads <= NUM_CONTEXTS);
408 running = threads;
409 for (i = 0; i < threads; i++) {
410 Coroutine *co1 = qemu_coroutine_create(test_multi_mutex_entry, NULL);
411 aio_co_schedule(ctx[i], co1);
412 }
413
414 g_usleep(seconds * 1000000);
415
416 atomic_mb_set(&now_stopping, true);
417 while (running > 0) {
418 g_usleep(100000);
419 }
420
421 join_aio_contexts();
422 g_test_message("%d iterations/second\n", counter / seconds);
423 g_assert_cmpint(counter, ==, atomic_counter);
424}
425
426static void test_multi_mutex_1(void)
427{
428 test_multi_mutex(NUM_CONTEXTS, 1);
429}
430
431static void test_multi_mutex_10(void)
432{
433 test_multi_mutex(NUM_CONTEXTS, 10);
434}
435
436
437
438int main(int argc, char **argv)
439{
440 init_clocks(NULL);
441
442 g_test_init(&argc, &argv, NULL);
443 g_test_add_func("/aio/multi/lifecycle", test_lifecycle);
444 if (g_test_quick()) {
445 g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_1);
446 g_test_add_func("/aio/multi/mutex/contended", test_multi_co_mutex_1);
447 g_test_add_func("/aio/multi/mutex/handoff", test_multi_co_mutex_2_3);
448#ifdef CONFIG_LINUX
449 g_test_add_func("/aio/multi/mutex/mcs", test_multi_fair_mutex_1);
450#endif
451 g_test_add_func("/aio/multi/mutex/pthread", test_multi_mutex_1);
452 } else {
453 g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_10);
454 g_test_add_func("/aio/multi/mutex/contended", test_multi_co_mutex_10);
455 g_test_add_func("/aio/multi/mutex/handoff", test_multi_co_mutex_2_30);
456#ifdef CONFIG_LINUX
457 g_test_add_func("/aio/multi/mutex/mcs", test_multi_fair_mutex_10);
458#endif
459 g_test_add_func("/aio/multi/mutex/pthread", test_multi_mutex_10);
460 }
461 return g_test_run();
462}
463