1 /*
2  * AioContext multithreading tests
3  *
4  * Copyright Red Hat, Inc. 2016
5  *
6  * Authors:
7  *  Paolo Bonzini    <pbonzini@redhat.com>
8  *
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.
11  */
12 
13 #include "qemu/osdep.h"
14 #include "block/aio.h"
15 #include "qemu/coroutine.h"
16 #include "qemu/thread.h"
17 #include "qemu/error-report.h"
18 #include "iothread.h"
19 
20 /* AioContext management */
21 
22 #define NUM_CONTEXTS 5
23 
24 static IOThread *threads[NUM_CONTEXTS];
25 static AioContext *ctx[NUM_CONTEXTS];
26 static __thread int id = -1;
27 
28 static QemuEvent done_event;
29 
30 /* Run a function synchronously on a remote iothread. */
31 
32 typedef struct CtxRunData {
33     QEMUBHFunc *cb;
34     void *arg;
35 } CtxRunData;
36 
37 static void ctx_run_bh_cb(void *opaque)
38 {
39     CtxRunData *data = opaque;
40 
41     data->cb(data->arg);
42     qemu_event_set(&done_event);
43 }
44 
45 static void ctx_run(int i, QEMUBHFunc *cb, void *opaque)
46 {
47     CtxRunData data = {
48         .cb = cb,
49         .arg = opaque
50     };
51 
52     qemu_event_reset(&done_event);
53     aio_bh_schedule_oneshot(ctx[i], ctx_run_bh_cb, &data);
54     qemu_event_wait(&done_event);
55 }
56 
57 /* Starting the iothreads. */
58 
59 static void set_id_cb(void *opaque)
60 {
61     int *i = opaque;
62 
63     id = *i;
64 }
65 
66 static void create_aio_contexts(void)
67 {
68     int i;
69 
70     for (i = 0; i < NUM_CONTEXTS; i++) {
71         threads[i] = iothread_new();
72         ctx[i] = iothread_get_aio_context(threads[i]);
73     }
74 
75     qemu_event_init(&done_event, false);
76     for (i = 0; i < NUM_CONTEXTS; i++) {
77         ctx_run(i, set_id_cb, &i);
78     }
79 }
80 
81 /* Stopping the iothreads. */
82 
83 static void join_aio_contexts(void)
84 {
85     int i;
86 
87     for (i = 0; i < NUM_CONTEXTS; i++) {
88         aio_context_ref(ctx[i]);
89     }
90     for (i = 0; i < NUM_CONTEXTS; i++) {
91         iothread_join(threads[i]);
92     }
93     for (i = 0; i < NUM_CONTEXTS; i++) {
94         aio_context_unref(ctx[i]);
95     }
96     qemu_event_destroy(&done_event);
97 }
98 
99 /* Basic test for the stuff above. */
100 
101 static void test_lifecycle(void)
102 {
103     create_aio_contexts();
104     join_aio_contexts();
105 }
106 
107 /* aio_co_schedule test.  */
108 
109 static Coroutine *to_schedule[NUM_CONTEXTS];
110 static bool stop[NUM_CONTEXTS];
111 
112 static int count_retry;
113 static int count_here;
114 static int count_other;
115 
116 static bool schedule_next(int n)
117 {
118     Coroutine *co;
119 
120     co = qatomic_xchg(&to_schedule[n], NULL);
121     if (!co) {
122         qatomic_inc(&count_retry);
123         return false;
124     }
125 
126     if (n == id) {
127         qatomic_inc(&count_here);
128     } else {
129         qatomic_inc(&count_other);
130     }
131 
132     aio_co_schedule(ctx[n], co);
133     return true;
134 }
135 
136 static void finish_cb(void *opaque)
137 {
138     stop[id] = true;
139     schedule_next(id);
140 }
141 
142 static coroutine_fn void test_multi_co_schedule_entry(void *opaque)
143 {
144     g_assert(to_schedule[id] == NULL);
145 
146     /*
147      * The next iteration will set to_schedule[id] again, but once finish_cb
148      * is scheduled there is no guarantee that it will actually be woken up,
149      * so at that point it must not go to sleep.
150      */
151     while (!stop[id]) {
152         int n;
153 
154         n = g_test_rand_int_range(0, NUM_CONTEXTS);
155         schedule_next(n);
156 
157         qatomic_set_mb(&to_schedule[id], qemu_coroutine_self());
158         /* finish_cb can run here.  */
159         qemu_coroutine_yield();
160         g_assert(to_schedule[id] == NULL);
161     }
162 }
163 
164 
165 static void test_multi_co_schedule(int seconds)
166 {
167     int i;
168 
169     count_here = count_other = count_retry = 0;
170 
171     create_aio_contexts();
172     for (i = 0; i < NUM_CONTEXTS; i++) {
173         Coroutine *co1 = qemu_coroutine_create(test_multi_co_schedule_entry, NULL);
174         aio_co_schedule(ctx[i], co1);
175     }
176 
177     g_usleep(seconds * 1000000);
178 
179     /* Guarantee that each AioContext is woken up from its last wait.  */
180     for (i = 0; i < NUM_CONTEXTS; i++) {
181         ctx_run(i, finish_cb, NULL);
182         g_assert(to_schedule[i] == NULL);
183     }
184 
185     join_aio_contexts();
186     g_test_message("scheduled %d, queued %d, retry %d, total %d",
187                   count_other, count_here, count_retry,
188                   count_here + count_other + count_retry);
189 }
190 
191 static void test_multi_co_schedule_1(void)
192 {
193     test_multi_co_schedule(1);
194 }
195 
196 static void test_multi_co_schedule_10(void)
197 {
198     test_multi_co_schedule(10);
199 }
200 
201 /* CoMutex thread-safety.  */
202 
203 static uint32_t atomic_counter;
204 static uint32_t running;
205 static uint32_t counter;
206 static CoMutex comutex;
207 static bool now_stopping;
208 
209 static void coroutine_fn test_multi_co_mutex_entry(void *opaque)
210 {
211     while (!qatomic_read(&now_stopping)) {
212         qemu_co_mutex_lock(&comutex);
213         counter++;
214         qemu_co_mutex_unlock(&comutex);
215 
216         /* Increase atomic_counter *after* releasing the mutex.  Otherwise
217          * there is a chance (it happens about 1 in 3 runs) that the iothread
218          * exits before the coroutine is woken up, causing a spurious
219          * assertion failure.
220          */
221         qatomic_inc(&atomic_counter);
222     }
223     qatomic_dec(&running);
224 }
225 
226 static void test_multi_co_mutex(int threads, int seconds)
227 {
228     int i;
229 
230     qemu_co_mutex_init(&comutex);
231     counter = 0;
232     atomic_counter = 0;
233     now_stopping = false;
234 
235     create_aio_contexts();
236     assert(threads <= NUM_CONTEXTS);
237     running = threads;
238     for (i = 0; i < threads; i++) {
239         Coroutine *co1 = qemu_coroutine_create(test_multi_co_mutex_entry, NULL);
240         aio_co_schedule(ctx[i], co1);
241     }
242 
243     g_usleep(seconds * 1000000);
244 
245     qatomic_set(&now_stopping, true);
246     while (running > 0) {
247         g_usleep(100000);
248     }
249 
250     join_aio_contexts();
251     g_test_message("%d iterations/second", counter / seconds);
252     g_assert_cmpint(counter, ==, atomic_counter);
253 }
254 
255 /* Testing with NUM_CONTEXTS threads focuses on the queue.  The mutex however
256  * is too contended (and the threads spend too much time in aio_poll)
257  * to actually stress the handoff protocol.
258  */
259 static void test_multi_co_mutex_1(void)
260 {
261     test_multi_co_mutex(NUM_CONTEXTS, 1);
262 }
263 
264 static void test_multi_co_mutex_10(void)
265 {
266     test_multi_co_mutex(NUM_CONTEXTS, 10);
267 }
268 
269 /* Testing with fewer threads stresses the handoff protocol too.  Still, the
270  * case where the locker _can_ pick up a handoff is very rare, happening
271  * about 10 times in 1 million, so increase the runtime a bit compared to
272  * other "quick" testcases that only run for 1 second.
273  */
274 static void test_multi_co_mutex_2_3(void)
275 {
276     test_multi_co_mutex(2, 3);
277 }
278 
279 static void test_multi_co_mutex_2_30(void)
280 {
281     test_multi_co_mutex(2, 30);
282 }
283 
284 /* Same test with fair mutexes, for performance comparison.  */
285 
286 #ifdef CONFIG_LINUX
287 #include "qemu/futex.h"
288 
289 /* The nodes for the mutex reside in this structure (on which we try to avoid
290  * false sharing).  The head of the mutex is in the "mutex_head" variable.
291  */
292 static struct {
293     int next, locked;
294     int padding[14];
295 } nodes[NUM_CONTEXTS] __attribute__((__aligned__(64)));
296 
297 static int mutex_head = -1;
298 
299 static void mcs_mutex_lock(void)
300 {
301     int prev;
302 
303     nodes[id].next = -1;
304     nodes[id].locked = 1;
305     prev = qatomic_xchg(&mutex_head, id);
306     if (prev != -1) {
307         qatomic_set(&nodes[prev].next, id);
308         qemu_futex_wait(&nodes[id].locked, 1);
309     }
310 }
311 
312 static void mcs_mutex_unlock(void)
313 {
314     int next;
315     if (qatomic_read(&nodes[id].next) == -1) {
316         if (qatomic_read(&mutex_head) == id &&
317             qatomic_cmpxchg(&mutex_head, id, -1) == id) {
318             /* Last item in the list, exit.  */
319             return;
320         }
321         while (qatomic_read(&nodes[id].next) == -1) {
322             /* mcs_mutex_lock did the xchg, but has not updated
323              * nodes[prev].next yet.
324              */
325         }
326     }
327 
328     /* Wake up the next in line.  */
329     next = qatomic_read(&nodes[id].next);
330     nodes[next].locked = 0;
331     qemu_futex_wake(&nodes[next].locked, 1);
332 }
333 
334 static void test_multi_fair_mutex_entry(void *opaque)
335 {
336     while (!qatomic_read(&now_stopping)) {
337         mcs_mutex_lock();
338         counter++;
339         mcs_mutex_unlock();
340         qatomic_inc(&atomic_counter);
341     }
342     qatomic_dec(&running);
343 }
344 
345 static void test_multi_fair_mutex(int threads, int seconds)
346 {
347     int i;
348 
349     assert(mutex_head == -1);
350     counter = 0;
351     atomic_counter = 0;
352     now_stopping = false;
353 
354     create_aio_contexts();
355     assert(threads <= NUM_CONTEXTS);
356     running = threads;
357     for (i = 0; i < threads; i++) {
358         Coroutine *co1 = qemu_coroutine_create(test_multi_fair_mutex_entry, NULL);
359         aio_co_schedule(ctx[i], co1);
360     }
361 
362     g_usleep(seconds * 1000000);
363 
364     qatomic_set(&now_stopping, true);
365     while (running > 0) {
366         g_usleep(100000);
367     }
368 
369     join_aio_contexts();
370     g_test_message("%d iterations/second", counter / seconds);
371     g_assert_cmpint(counter, ==, atomic_counter);
372 }
373 
374 static void test_multi_fair_mutex_1(void)
375 {
376     test_multi_fair_mutex(NUM_CONTEXTS, 1);
377 }
378 
379 static void test_multi_fair_mutex_10(void)
380 {
381     test_multi_fair_mutex(NUM_CONTEXTS, 10);
382 }
383 #endif
384 
385 /* Same test with pthread mutexes, for performance comparison and
386  * portability.  */
387 
388 static QemuMutex mutex;
389 
390 static void test_multi_mutex_entry(void *opaque)
391 {
392     while (!qatomic_read(&now_stopping)) {
393         qemu_mutex_lock(&mutex);
394         counter++;
395         qemu_mutex_unlock(&mutex);
396         qatomic_inc(&atomic_counter);
397     }
398     qatomic_dec(&running);
399 }
400 
401 static void test_multi_mutex(int threads, int seconds)
402 {
403     int i;
404 
405     qemu_mutex_init(&mutex);
406     counter = 0;
407     atomic_counter = 0;
408     now_stopping = false;
409 
410     create_aio_contexts();
411     assert(threads <= NUM_CONTEXTS);
412     running = threads;
413     for (i = 0; i < threads; i++) {
414         Coroutine *co1 = qemu_coroutine_create(test_multi_mutex_entry, NULL);
415         aio_co_schedule(ctx[i], co1);
416     }
417 
418     g_usleep(seconds * 1000000);
419 
420     qatomic_set(&now_stopping, true);
421     while (running > 0) {
422         g_usleep(100000);
423     }
424 
425     join_aio_contexts();
426     g_test_message("%d iterations/second", counter / seconds);
427     g_assert_cmpint(counter, ==, atomic_counter);
428 }
429 
430 static void test_multi_mutex_1(void)
431 {
432     test_multi_mutex(NUM_CONTEXTS, 1);
433 }
434 
435 static void test_multi_mutex_10(void)
436 {
437     test_multi_mutex(NUM_CONTEXTS, 10);
438 }
439 
440 /* End of tests.  */
441 
442 int main(int argc, char **argv)
443 {
444     init_clocks(NULL);
445 
446     g_test_init(&argc, &argv, NULL);
447     g_test_add_func("/aio/multi/lifecycle", test_lifecycle);
448     if (g_test_quick()) {
449         g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_1);
450         g_test_add_func("/aio/multi/mutex/contended", test_multi_co_mutex_1);
451         g_test_add_func("/aio/multi/mutex/handoff", test_multi_co_mutex_2_3);
452 #ifdef CONFIG_LINUX
453         g_test_add_func("/aio/multi/mutex/mcs", test_multi_fair_mutex_1);
454 #endif
455         g_test_add_func("/aio/multi/mutex/pthread", test_multi_mutex_1);
456     } else {
457         g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_10);
458         g_test_add_func("/aio/multi/mutex/contended", test_multi_co_mutex_10);
459         g_test_add_func("/aio/multi/mutex/handoff", test_multi_co_mutex_2_30);
460 #ifdef CONFIG_LINUX
461         g_test_add_func("/aio/multi/mutex/mcs", test_multi_fair_mutex_10);
462 #endif
463         g_test_add_func("/aio/multi/mutex/pthread", test_multi_mutex_10);
464     }
465     return g_test_run();
466 }
467