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