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