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