util/qemu-coroutine.c

   1 /*
   2  * QEMU coroutines
   3  *
   4  * Copyright IBM, Corp. 2011
   5  *
   6  * Authors:
   7  *  Stefan Hajnoczi    <stefanha@linux.vnet.ibm.com>
   8  *  Kevin Wolf         <kwolf@redhat.com>
   9  *
  10  * This work is licensed under the terms of the GNU LGPL, version 2 or later.
  11  * See the COPYING.LIB file in the top-level directory.
  12  *
  13  */
  14
  15 #include "qemu/osdep.h"
  16 #include "trace.h"
  17 #include "qemu/thread.h"
  18 #include "qemu/atomic.h"
  19 #include "qemu/coroutine.h"
  20 #include "qemu/coroutine_int.h"
  21 #include "qemu/coroutine-tls.h"
  22 #include "block/aio.h"
  23
  24 /** Initial batch size is 64, and is increased on demand */
  25 enum {
  26     POOL_INITIAL_BATCH_SIZE = 64,
  27 };
  28
  29 /** Free list to speed up creation */
  30 static QSLIST_HEAD(, Coroutine) release_pool = QSLIST_HEAD_INITIALIZER(pool);
  31 static unsigned int pool_batch_size = POOL_INITIAL_BATCH_SIZE;
  32 static unsigned int release_pool_size;
  33
  34 typedef QSLIST_HEAD(, Coroutine) CoroutineQSList;
  35 QEMU_DEFINE_STATIC_CO_TLS(CoroutineQSList, alloc_pool);
  36 QEMU_DEFINE_STATIC_CO_TLS(unsigned int, alloc_pool_size);
  37 QEMU_DEFINE_STATIC_CO_TLS(Notifier, coroutine_pool_cleanup_notifier);
  38
  39 static void coroutine_pool_cleanup(Notifier *n, void *value)
  40 {
  41     Coroutine *co;
  42     Coroutine *tmp;
  43     CoroutineQSList *alloc_pool = get_ptr_alloc_pool();
  44
  45     QSLIST_FOREACH_SAFE(co, alloc_pool, pool_next, tmp) {
  46         QSLIST_REMOVE_HEAD(alloc_pool, pool_next);
  47         qemu_coroutine_delete(co);
  48     }
  49 }
  50
  51 Coroutine *qemu_coroutine_create(CoroutineEntry *entry, void *opaque)
  52 {
  53     Coroutine *co = NULL;
  54
  55     if (CONFIG_COROUTINE_POOL) {
  56         CoroutineQSList *alloc_pool = get_ptr_alloc_pool();
  57
  58         co = QSLIST_FIRST(alloc_pool);
  59         if (!co) {
  60             if (release_pool_size > qatomic_read(&pool_batch_size)) {
  61                 /* Slow path; a good place to register the destructor, too.  */
  62                 Notifier *notifier = get_ptr_coroutine_pool_cleanup_notifier();
  63                 if (!notifier->notify) {
  64                     notifier->notify = coroutine_pool_cleanup;
  65                     qemu_thread_atexit_add(notifier);
  66                 }
  67
  68                 /* This is not exact; there could be a little skew between
  69                  * release_pool_size and the actual size of release_pool.  But
  70                  * it is just a heuristic, it does not need to be perfect.
  71                  */
  72                 set_alloc_pool_size(qatomic_xchg(&release_pool_size, 0));
  73                 QSLIST_MOVE_ATOMIC(alloc_pool, &release_pool);
  74                 co = QSLIST_FIRST(alloc_pool);
  75             }
  76         }
  77         if (co) {
  78             QSLIST_REMOVE_HEAD(alloc_pool, pool_next);
  79             set_alloc_pool_size(get_alloc_pool_size() - 1);
  80         }
  81     }
  82
  83     if (!co) {
  84         co = qemu_coroutine_new();
  85     }
  86
  87     co->entry = entry;
  88     co->entry_arg = opaque;
  89     QSIMPLEQ_INIT(&co->co_queue_wakeup);
  90     return co;
  91 }
  92
  93 static void coroutine_delete(Coroutine *co)
  94 {
  95     co->caller = NULL;
  96
  97     if (CONFIG_COROUTINE_POOL) {
  98         if (release_pool_size < qatomic_read(&pool_batch_size) * 2) {
  99             QSLIST_INSERT_HEAD_ATOMIC(&release_pool, co, pool_next);
 100             qatomic_inc(&release_pool_size);
 101             return;
 102         }
 103         if (get_alloc_pool_size() < qatomic_read(&pool_batch_size)) {
 104             QSLIST_INSERT_HEAD(get_ptr_alloc_pool(), co, pool_next);
 105             set_alloc_pool_size(get_alloc_pool_size() + 1);
 106             return;
 107         }
 108     }
 109
 110     qemu_coroutine_delete(co);
 111 }
 112
 113 void qemu_aio_coroutine_enter(AioContext *ctx, Coroutine *co)
 114 {
 115     QSIMPLEQ_HEAD(, Coroutine) pending = QSIMPLEQ_HEAD_INITIALIZER(pending);
 116     Coroutine *from = qemu_coroutine_self();
 117
 118     QSIMPLEQ_INSERT_TAIL(&pending, co, co_queue_next);
 119
 120     /* Run co and any queued coroutines */
 121     while (!QSIMPLEQ_EMPTY(&pending)) {
 122         Coroutine *to = QSIMPLEQ_FIRST(&pending);
 123         CoroutineAction ret;
 124
 125         /* Cannot rely on the read barrier for to in aio_co_wake(), as there are
 126          * callers outside of aio_co_wake() */
 127         const char *scheduled = qatomic_mb_read(&to->scheduled);
 128
 129         QSIMPLEQ_REMOVE_HEAD(&pending, co_queue_next);
 130
 131         trace_qemu_aio_coroutine_enter(ctx, from, to, to->entry_arg);
 132
 133         /* if the Coroutine has already been scheduled, entering it again will
 134          * cause us to enter it twice, potentially even after the coroutine has
 135          * been deleted */
 136         if (scheduled) {
 137             fprintf(stderr,
 138                     "%s: Co-routine was already scheduled in '%s'\n",
 139                     __func__, scheduled);
 140             abort();
 141         }
 142
 143         if (to->caller) {
 144             fprintf(stderr, "Co-routine re-entered recursively\n");
 145             abort();
 146         }
 147
 148         to->caller = from;
 149         to->ctx = ctx;
 150
 151         /* Store to->ctx before anything that stores to.  Matches
 152          * barrier in aio_co_wake and qemu_co_mutex_wake.
 153          */
 154         smp_wmb();
 155
 156         ret = qemu_coroutine_switch(from, to, COROUTINE_ENTER);
 157
 158         /* Queued coroutines are run depth-first; previously pending coroutines
 159          * run after those queued more recently.
 160          */
 161         QSIMPLEQ_PREPEND(&pending, &to->co_queue_wakeup);
 162
 163         switch (ret) {
 164         case COROUTINE_YIELD:
 165             break;
 166         case COROUTINE_TERMINATE:
 167             assert(!to->locks_held);
 168             trace_qemu_coroutine_terminate(to);
 169             coroutine_delete(to);
 170             break;
 171         default:
 172             abort();
 173         }
 174     }
 175 }
 176
 177 void qemu_coroutine_enter(Coroutine *co)
 178 {
 179     qemu_aio_coroutine_enter(qemu_get_current_aio_context(), co);
 180 }
 181
 182 void qemu_coroutine_enter_if_inactive(Coroutine *co)
 183 {
 184     if (!qemu_coroutine_entered(co)) {
 185         qemu_coroutine_enter(co);
 186     }
 187 }
 188
 189 void coroutine_fn qemu_coroutine_yield(void)
 190 {
 191     Coroutine *self = qemu_coroutine_self();
 192     Coroutine *to = self->caller;
 193
 194     trace_qemu_coroutine_yield(self, to);
 195
 196     if (!to) {
 197         fprintf(stderr, "Co-routine is yielding to no one\n");
 198         abort();
 199     }
 200
 201     self->caller = NULL;
 202     qemu_coroutine_switch(self, to, COROUTINE_YIELD);
 203 }
 204
 205 bool qemu_coroutine_entered(Coroutine *co)
 206 {
 207     return co->caller;
 208 }
 209
 210 AioContext *coroutine_fn qemu_coroutine_get_aio_context(Coroutine *co)
 211 {
 212     return co->ctx;
 213 }
 214
 215 void qemu_coroutine_inc_pool_size(unsigned int additional_pool_size)
 216 {
 217     qatomic_add(&pool_batch_size, additional_pool_size);
 218 }
 219
 220 void qemu_coroutine_dec_pool_size(unsigned int removing_pool_size)
 221 {
 222     qatomic_sub(&pool_batch_size, removing_pool_size);
 223 }