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 } 217