2 * AioContext multithreading tests
4 * Copyright Red Hat, Inc. 2016
7 * Paolo Bonzini <pbonzini@redhat.com>
9 * This work is licensed under the terms of the GNU LGPL, version 2 or later.
10 * See the COPYING.LIB file in the top-level directory.
13 #include "qemu/osdep.h"
14 #include "block/aio.h"
15 #include "qapi/error.h"
16 #include "qemu/coroutine.h"
17 #include "qemu/thread.h"
18 #include "qemu/error-report.h"
21 /* AioContext management */
23 #define NUM_CONTEXTS 5
25 static IOThread
*threads
[NUM_CONTEXTS
];
26 static AioContext
*ctx
[NUM_CONTEXTS
];
27 static __thread
int id
= -1;
29 static QemuEvent done_event
;
31 /* Run a function synchronously on a remote iothread. */
33 typedef struct CtxRunData
{
38 static void ctx_run_bh_cb(void *opaque
)
40 CtxRunData
*data
= opaque
;
43 qemu_event_set(&done_event
);
46 static void ctx_run(int i
, QEMUBHFunc
*cb
, void *opaque
)
53 qemu_event_reset(&done_event
);
54 aio_bh_schedule_oneshot(ctx
[i
], ctx_run_bh_cb
, &data
);
55 qemu_event_wait(&done_event
);
58 /* Starting the iothreads. */
60 static void set_id_cb(void *opaque
)
67 static void create_aio_contexts(void)
71 for (i
= 0; i
< NUM_CONTEXTS
; i
++) {
72 threads
[i
] = iothread_new();
73 ctx
[i
] = iothread_get_aio_context(threads
[i
]);
76 qemu_event_init(&done_event
, false);
77 for (i
= 0; i
< NUM_CONTEXTS
; i
++) {
78 ctx_run(i
, set_id_cb
, &i
);
82 /* Stopping the iothreads. */
84 static void join_aio_contexts(void)
88 for (i
= 0; i
< NUM_CONTEXTS
; i
++) {
89 aio_context_ref(ctx
[i
]);
91 for (i
= 0; i
< NUM_CONTEXTS
; i
++) {
92 iothread_join(threads
[i
]);
94 for (i
= 0; i
< NUM_CONTEXTS
; i
++) {
95 aio_context_unref(ctx
[i
]);
97 qemu_event_destroy(&done_event
);
100 /* Basic test for the stuff above. */
102 static void test_lifecycle(void)
104 create_aio_contexts();
108 /* aio_co_schedule test. */
110 static Coroutine
*to_schedule
[NUM_CONTEXTS
];
112 static bool now_stopping
;
114 static int count_retry
;
115 static int count_here
;
116 static int count_other
;
118 static bool schedule_next(int n
)
122 co
= atomic_xchg(&to_schedule
[n
], NULL
);
124 atomic_inc(&count_retry
);
129 atomic_inc(&count_here
);
131 atomic_inc(&count_other
);
134 aio_co_schedule(ctx
[n
], co
);
138 static void finish_cb(void *opaque
)
143 static coroutine_fn
void test_multi_co_schedule_entry(void *opaque
)
145 g_assert(to_schedule
[id
] == NULL
);
147 while (!atomic_mb_read(&now_stopping
)) {
150 n
= g_test_rand_int_range(0, NUM_CONTEXTS
);
153 atomic_mb_set(&to_schedule
[id
], qemu_coroutine_self());
154 qemu_coroutine_yield();
155 g_assert(to_schedule
[id
] == NULL
);
160 static void test_multi_co_schedule(int seconds
)
164 count_here
= count_other
= count_retry
= 0;
165 now_stopping
= false;
167 create_aio_contexts();
168 for (i
= 0; i
< NUM_CONTEXTS
; i
++) {
169 Coroutine
*co1
= qemu_coroutine_create(test_multi_co_schedule_entry
, NULL
);
170 aio_co_schedule(ctx
[i
], co1
);
173 g_usleep(seconds
* 1000000);
175 atomic_mb_set(&now_stopping
, true);
176 for (i
= 0; i
< NUM_CONTEXTS
; i
++) {
177 ctx_run(i
, finish_cb
, NULL
);
178 to_schedule
[i
] = NULL
;
182 g_test_message("scheduled %d, queued %d, retry %d, total %d\n",
183 count_other
, count_here
, count_retry
,
184 count_here
+ count_other
+ count_retry
);
187 static void test_multi_co_schedule_1(void)
189 test_multi_co_schedule(1);
192 static void test_multi_co_schedule_10(void)
194 test_multi_co_schedule(10);
197 /* CoMutex thread-safety. */
199 static uint32_t atomic_counter
;
200 static uint32_t running
;
201 static uint32_t counter
;
202 static CoMutex comutex
;
204 static void coroutine_fn
test_multi_co_mutex_entry(void *opaque
)
206 while (!atomic_mb_read(&now_stopping
)) {
207 qemu_co_mutex_lock(&comutex
);
209 qemu_co_mutex_unlock(&comutex
);
211 /* Increase atomic_counter *after* releasing the mutex. Otherwise
212 * there is a chance (it happens about 1 in 3 runs) that the iothread
213 * exits before the coroutine is woken up, causing a spurious
216 atomic_inc(&atomic_counter
);
218 atomic_dec(&running
);
221 static void test_multi_co_mutex(int threads
, int seconds
)
225 qemu_co_mutex_init(&comutex
);
228 now_stopping
= false;
230 create_aio_contexts();
231 assert(threads
<= NUM_CONTEXTS
);
233 for (i
= 0; i
< threads
; i
++) {
234 Coroutine
*co1
= qemu_coroutine_create(test_multi_co_mutex_entry
, NULL
);
235 aio_co_schedule(ctx
[i
], co1
);
238 g_usleep(seconds
* 1000000);
240 atomic_mb_set(&now_stopping
, true);
241 while (running
> 0) {
246 g_test_message("%d iterations/second\n", counter
/ seconds
);
247 g_assert_cmpint(counter
, ==, atomic_counter
);
250 /* Testing with NUM_CONTEXTS threads focuses on the queue. The mutex however
251 * is too contended (and the threads spend too much time in aio_poll)
252 * to actually stress the handoff protocol.
254 static void test_multi_co_mutex_1(void)
256 test_multi_co_mutex(NUM_CONTEXTS
, 1);
259 static void test_multi_co_mutex_10(void)
261 test_multi_co_mutex(NUM_CONTEXTS
, 10);
264 /* Testing with fewer threads stresses the handoff protocol too. Still, the
265 * case where the locker _can_ pick up a handoff is very rare, happening
266 * about 10 times in 1 million, so increase the runtime a bit compared to
267 * other "quick" testcases that only run for 1 second.
269 static void test_multi_co_mutex_2_3(void)
271 test_multi_co_mutex(2, 3);
274 static void test_multi_co_mutex_2_30(void)
276 test_multi_co_mutex(2, 30);
279 /* Same test with fair mutexes, for performance comparison. */
282 #include "qemu/futex.h"
284 /* The nodes for the mutex reside in this structure (on which we try to avoid
285 * false sharing). The head of the mutex is in the "mutex_head" variable.
290 } nodes
[NUM_CONTEXTS
] __attribute__((__aligned__(64)));
292 static int mutex_head
= -1;
294 static void mcs_mutex_lock(void)
299 nodes
[id
].locked
= 1;
300 prev
= atomic_xchg(&mutex_head
, id
);
302 atomic_set(&nodes
[prev
].next
, id
);
303 qemu_futex_wait(&nodes
[id
].locked
, 1);
307 static void mcs_mutex_unlock(void)
310 if (atomic_read(&nodes
[id
].next
) == -1) {
311 if (atomic_read(&mutex_head
) == id
&&
312 atomic_cmpxchg(&mutex_head
, id
, -1) == id
) {
313 /* Last item in the list, exit. */
316 while (atomic_read(&nodes
[id
].next
) == -1) {
317 /* mcs_mutex_lock did the xchg, but has not updated
318 * nodes[prev].next yet.
323 /* Wake up the next in line. */
324 next
= atomic_read(&nodes
[id
].next
);
325 nodes
[next
].locked
= 0;
326 qemu_futex_wake(&nodes
[next
].locked
, 1);
329 static void test_multi_fair_mutex_entry(void *opaque
)
331 while (!atomic_mb_read(&now_stopping
)) {
335 atomic_inc(&atomic_counter
);
337 atomic_dec(&running
);
340 static void test_multi_fair_mutex(int threads
, int seconds
)
344 assert(mutex_head
== -1);
347 now_stopping
= false;
349 create_aio_contexts();
350 assert(threads
<= NUM_CONTEXTS
);
352 for (i
= 0; i
< threads
; i
++) {
353 Coroutine
*co1
= qemu_coroutine_create(test_multi_fair_mutex_entry
, NULL
);
354 aio_co_schedule(ctx
[i
], co1
);
357 g_usleep(seconds
* 1000000);
359 atomic_mb_set(&now_stopping
, true);
360 while (running
> 0) {
365 g_test_message("%d iterations/second\n", counter
/ seconds
);
366 g_assert_cmpint(counter
, ==, atomic_counter
);
369 static void test_multi_fair_mutex_1(void)
371 test_multi_fair_mutex(NUM_CONTEXTS
, 1);
374 static void test_multi_fair_mutex_10(void)
376 test_multi_fair_mutex(NUM_CONTEXTS
, 10);
380 /* Same test with pthread mutexes, for performance comparison and
383 static QemuMutex mutex
;
385 static void test_multi_mutex_entry(void *opaque
)
387 while (!atomic_mb_read(&now_stopping
)) {
388 qemu_mutex_lock(&mutex
);
390 qemu_mutex_unlock(&mutex
);
391 atomic_inc(&atomic_counter
);
393 atomic_dec(&running
);
396 static void test_multi_mutex(int threads
, int seconds
)
400 qemu_mutex_init(&mutex
);
403 now_stopping
= false;
405 create_aio_contexts();
406 assert(threads
<= NUM_CONTEXTS
);
408 for (i
= 0; i
< threads
; i
++) {
409 Coroutine
*co1
= qemu_coroutine_create(test_multi_mutex_entry
, NULL
);
410 aio_co_schedule(ctx
[i
], co1
);
413 g_usleep(seconds
* 1000000);
415 atomic_mb_set(&now_stopping
, true);
416 while (running
> 0) {
421 g_test_message("%d iterations/second\n", counter
/ seconds
);
422 g_assert_cmpint(counter
, ==, atomic_counter
);
425 static void test_multi_mutex_1(void)
427 test_multi_mutex(NUM_CONTEXTS
, 1);
430 static void test_multi_mutex_10(void)
432 test_multi_mutex(NUM_CONTEXTS
, 10);
437 int main(int argc
, char **argv
)
441 g_test_init(&argc
, &argv
, NULL
);
442 g_test_add_func("/aio/multi/lifecycle", test_lifecycle
);
443 if (g_test_quick()) {
444 g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_1
);
445 g_test_add_func("/aio/multi/mutex/contended", test_multi_co_mutex_1
);
446 g_test_add_func("/aio/multi/mutex/handoff", test_multi_co_mutex_2_3
);
448 g_test_add_func("/aio/multi/mutex/mcs", test_multi_fair_mutex_1
);
450 g_test_add_func("/aio/multi/mutex/pthread", test_multi_mutex_1
);
452 g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_10
);
453 g_test_add_func("/aio/multi/mutex/contended", test_multi_co_mutex_10
);
454 g_test_add_func("/aio/multi/mutex/handoff", test_multi_co_mutex_2_30
);
456 g_test_add_func("/aio/multi/mutex/mcs", test_multi_fair_mutex_10
);
458 g_test_add_func("/aio/multi/mutex/pthread", test_multi_mutex_10
);