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     atomic_mb_set(&to_schedule[id], qemu_coroutine_self());
 148
 149     while (!atomic_mb_read(&now_stopping)) {
 150         int n;
 151
 152         n = g_test_rand_int_range(0, NUM_CONTEXTS);
 153         schedule_next(n);
 154         qemu_coroutine_yield();
 155
 156         g_assert(to_schedule[id] == NULL);
 157         atomic_mb_set(&to_schedule[id], qemu_coroutine_self());
 158     }
 159 }
 160
 161
 162 static void test_multi_co_schedule(int seconds)
 163 {
 164     int i;
 165
 166     count_here = count_other = count_retry = 0;
 167     now_stopping = false;
 168
 169     create_aio_contexts();
 170     for (i = 0; i < NUM_CONTEXTS; i++) {
 171         Coroutine *co1 = qemu_coroutine_create(test_multi_co_schedule_entry, NULL);
 172         aio_co_schedule(ctx[i], co1);
 173     }
 174
 175     g_usleep(seconds * 1000000);
 176
 177     atomic_mb_set(&now_stopping, true);
 178     for (i = 0; i < NUM_CONTEXTS; i++) {
 179         ctx_run(i, finish_cb, NULL);
 180         to_schedule[i] = NULL;
 181     }
 182
 183     join_aio_contexts();
 184     g_test_message("scheduled %d, queued %d, retry %d, total %d\n",
 185                   count_other, count_here, count_retry,
 186                   count_here + count_other + count_retry);
 187 }
 188
 189 static void test_multi_co_schedule_1(void)
 190 {
 191     test_multi_co_schedule(1);
 192 }
 193
 194 static void test_multi_co_schedule_10(void)
 195 {
 196     test_multi_co_schedule(10);
 197 }
 198
 199 /* CoMutex thread-safety.  */
 200
 201 static uint32_t atomic_counter;
 202 static uint32_t running;
 203 static uint32_t counter;
 204 static CoMutex comutex;
 205
 206 static void coroutine_fn test_multi_co_mutex_entry(void *opaque)
 207 {
 208     while (!atomic_mb_read(&now_stopping)) {
 209         qemu_co_mutex_lock(&comutex);
 210         counter++;
 211         qemu_co_mutex_unlock(&comutex);
 212
 213         /* Increase atomic_counter *after* releasing the mutex.  Otherwise
 214          * there is a chance (it happens about 1 in 3 runs) that the iothread
 215          * exits before the coroutine is woken up, causing a spurious
 216          * assertion failure.
 217          */
 218         atomic_inc(&atomic_counter);
 219     }
 220     atomic_dec(&running);
 221 }
 222
 223 static void test_multi_co_mutex(int threads, int seconds)
 224 {
 225     int i;
 226
 227     qemu_co_mutex_init(&comutex);
 228     counter = 0;
 229     atomic_counter = 0;
 230     now_stopping = false;
 231
 232     create_aio_contexts();
 233     assert(threads <= NUM_CONTEXTS);
 234     running = threads;
 235     for (i = 0; i < threads; i++) {
 236         Coroutine *co1 = qemu_coroutine_create(test_multi_co_mutex_entry, NULL);
 237         aio_co_schedule(ctx[i], co1);
 238     }
 239
 240     g_usleep(seconds * 1000000);
 241
 242     atomic_mb_set(&now_stopping, true);
 243     while (running > 0) {
 244         g_usleep(100000);
 245     }
 246
 247     join_aio_contexts();
 248     g_test_message("%d iterations/second\n", counter / seconds);
 249     g_assert_cmpint(counter, ==, atomic_counter);
 250 }
 251
 252 /* Testing with NUM_CONTEXTS threads focuses on the queue.  The mutex however
 253  * is too contended (and the threads spend too much time in aio_poll)
 254  * to actually stress the handoff protocol.
 255  */
 256 static void test_multi_co_mutex_1(void)
 257 {
 258     test_multi_co_mutex(NUM_CONTEXTS, 1);
 259 }
 260
 261 static void test_multi_co_mutex_10(void)
 262 {
 263     test_multi_co_mutex(NUM_CONTEXTS, 10);
 264 }
 265
 266 /* Testing with fewer threads stresses the handoff protocol too.  Still, the
 267  * case where the locker _can_ pick up a handoff is very rare, happening
 268  * about 10 times in 1 million, so increase the runtime a bit compared to
 269  * other "quick" testcases that only run for 1 second.
 270  */
 271 static void test_multi_co_mutex_2_3(void)
 272 {
 273     test_multi_co_mutex(2, 3);
 274 }
 275
 276 static void test_multi_co_mutex_2_30(void)
 277 {
 278     test_multi_co_mutex(2, 30);
 279 }
 280
 281 /* End of tests.  */
 282
 283 int main(int argc, char **argv)
 284 {
 285     init_clocks();
 286
 287     g_test_init(&argc, &argv, NULL);
 288     g_test_add_func("/aio/multi/lifecycle", test_lifecycle);
 289     if (g_test_quick()) {
 290         g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_1);
 291         g_test_add_func("/aio/multi/mutex/contended", test_multi_co_mutex_1);
 292         g_test_add_func("/aio/multi/mutex/handoff", test_multi_co_mutex_2_3);
 293     } else {
 294         g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_10);
 295         g_test_add_func("/aio/multi/mutex/contended", test_multi_co_mutex_10);
 296         g_test_add_func("/aio/multi/mutex/handoff", test_multi_co_mutex_2_30);
 297     }
 298     return g_test_run();
 299 }