tests/test-aio-multithread.c

   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 <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 /* AioContext management */
  23
  24 #define NUM_CONTEXTS 5
  25
  26 static IOThread *threads[NUM_CONTEXTS];
  27 static AioContext *ctx[NUM_CONTEXTS];
  28 static __thread int id = -1;
  29
  30 static QemuEvent done_event;
  31
  32 /* Run a function synchronously on a remote iothread. */
  33
  34 typedef struct CtxRunData {
  35     QEMUBHFunc *cb;
  36     void *arg;
  37 } CtxRunData;
  38
  39 static 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
  47 static 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 /* Starting the iothreads. */
  60
  61 static void set_id_cb(void *opaque)
  62 {
  63     int *i = opaque;
  64
  65     id = *i;
  66 }
  67
  68 static 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 /* Stopping the iothreads. */
  84
  85 static 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 /* Basic test for the stuff above. */
 102
 103 static void test_lifecycle(void)
 104 {
 105     create_aio_contexts();
 106     join_aio_contexts();
 107 }
 108
 109 /* aio_co_schedule test.  */
 110
 111 static Coroutine *to_schedule[NUM_CONTEXTS];
 112
 113 static bool now_stopping;
 114
 115 static int count_retry;
 116 static int count_here;
 117 static int count_other;
 118
 119 static 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
 139 static void finish_cb(void *opaque)
 140 {
 141     schedule_next(id);
 142 }
 143
 144 static 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
 161 static 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
 188 static void test_multi_co_schedule_1(void)
 189 {
 190     test_multi_co_schedule(1);
 191 }
 192
 193 static void test_multi_co_schedule_10(void)
 194 {
 195     test_multi_co_schedule(10);
 196 }
 197
 198 /* CoMutex thread-safety.  */
 199
 200 static uint32_t atomic_counter;
 201 static uint32_t running;
 202 static uint32_t counter;
 203 static CoMutex comutex;
 204
 205 static 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         /* Increase atomic_counter *after* releasing the mutex.  Otherwise
 213          * there is a chance (it happens about 1 in 3 runs) that the iothread
 214          * exits before the coroutine is woken up, causing a spurious
 215          * assertion failure.
 216          */
 217         atomic_inc(&atomic_counter);
 218     }
 219     atomic_dec(&running);
 220 }
 221
 222 static 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 /* Testing with NUM_CONTEXTS threads focuses on the queue.  The mutex however
 252  * is too contended (and the threads spend too much time in aio_poll)
 253  * to actually stress the handoff protocol.
 254  */
 255 static void test_multi_co_mutex_1(void)
 256 {
 257     test_multi_co_mutex(NUM_CONTEXTS, 1);
 258 }
 259
 260 static void test_multi_co_mutex_10(void)
 261 {
 262     test_multi_co_mutex(NUM_CONTEXTS, 10);
 263 }
 264
 265 /* Testing with fewer threads stresses the handoff protocol too.  Still, the
 266  * case where the locker _can_ pick up a handoff is very rare, happening
 267  * about 10 times in 1 million, so increase the runtime a bit compared to
 268  * other "quick" testcases that only run for 1 second.
 269  */
 270 static void test_multi_co_mutex_2_3(void)
 271 {
 272     test_multi_co_mutex(2, 3);
 273 }
 274
 275 static void test_multi_co_mutex_2_30(void)
 276 {
 277     test_multi_co_mutex(2, 30);
 278 }
 279
 280 /* Same test with fair mutexes, for performance comparison.  */
 281
 282 #ifdef CONFIG_LINUX
 283 #include "qemu/futex.h"
 284
 285 /* The nodes for the mutex reside in this structure (on which we try to avoid
 286  * false sharing).  The head of the mutex is in the "mutex_head" variable.
 287  */
 288 static struct {
 289     int next, locked;
 290     int padding[14];
 291 } nodes[NUM_CONTEXTS] __attribute__((__aligned__(64)));
 292
 293 static int mutex_head = -1;
 294
 295 static 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
 308 static 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             /* Last item in the list, exit.  */
 315             return;
 316         }
 317         while (atomic_read(&nodes[id].next) == -1) {
 318             /* mcs_mutex_lock did the xchg, but has not updated
 319              * nodes[prev].next yet.
 320              */
 321         }
 322     }
 323
 324     /* Wake up the next in line.  */
 325     next = atomic_read(&nodes[id].next);
 326     nodes[next].locked = 0;
 327     qemu_futex_wake(&nodes[next].locked, 1);
 328 }
 329
 330 static 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
 341 static 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
 370 static void test_multi_fair_mutex_1(void)
 371 {
 372     test_multi_fair_mutex(NUM_CONTEXTS, 1);
 373 }
 374
 375 static void test_multi_fair_mutex_10(void)
 376 {
 377     test_multi_fair_mutex(NUM_CONTEXTS, 10);
 378 }
 379 #endif
 380
 381 /* Same test with pthread mutexes, for performance comparison and
 382  * portability.  */
 383
 384 static QemuMutex mutex;
 385
 386 static 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
 397 static 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
 426 static void test_multi_mutex_1(void)
 427 {
 428     test_multi_mutex(NUM_CONTEXTS, 1);
 429 }
 430
 431 static void test_multi_mutex_10(void)
 432 {
 433     test_multi_mutex(NUM_CONTEXTS, 10);
 434 }
 435
 436 /* End of tests.  */
 437
 438 int 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 }