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 /* End of tests.  */
 200
 201 int main(int argc, char **argv)
 202 {
 203     init_clocks();
 204
 205     g_test_init(&argc, &argv, NULL);
 206     g_test_add_func("/aio/multi/lifecycle", test_lifecycle);
 207     if (g_test_quick()) {
 208         g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_1);
 209     } else {
 210         g_test_add_func("/aio/multi/schedule", test_multi_co_schedule_10);
 211     }
 212     return g_test_run();
 213 }