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 /* Cannot rely on the read barrier for to in aio_co_wake(), as there are 131 * callers outside of aio_co_wake() */ 132 const char *scheduled = qatomic_mb_read(&to->scheduled); 133 134 QSIMPLEQ_REMOVE_HEAD(&pending, co_queue_next); 135 136 trace_qemu_aio_coroutine_enter(ctx, from, to, to->entry_arg); 137 138 /* if the Coroutine has already been scheduled, entering it again will 139 * cause us to enter it twice, potentially even after the coroutine has 140 * been deleted */ 141 if (scheduled) { 142 fprintf(stderr, 143 "%s: Co-routine was already scheduled in '%s'\n", 144 __func__, scheduled); 145 abort(); 146 } 147 148 if (to->caller) { 149 fprintf(stderr, "Co-routine re-entered recursively\n"); 150 abort(); 151 } 152 153 to->caller = from; 154 to->ctx = ctx; 155 156 /* Store to->ctx before anything that stores to. Matches 157 * barrier in aio_co_wake and qemu_co_mutex_wake. 158 */ 159 smp_wmb(); 160 161 ret = qemu_coroutine_switch(from, to, COROUTINE_ENTER); 162 163 /* Queued coroutines are run depth-first; previously pending coroutines 164 * run after those queued more recently. 165 */ 166 QSIMPLEQ_PREPEND(&pending, &to->co_queue_wakeup); 167 168 switch (ret) { 169 case COROUTINE_YIELD: 170 break; 171 case COROUTINE_TERMINATE: 172 assert(!to->locks_held); 173 trace_qemu_coroutine_terminate(to); 174 coroutine_delete(to); 175 break; 176 default: 177 abort(); 178 } 179 } 180 } 181 182 void qemu_coroutine_enter(Coroutine *co) 183 { 184 qemu_aio_coroutine_enter(qemu_get_current_aio_context(), co); 185 } 186 187 void qemu_coroutine_enter_if_inactive(Coroutine *co) 188 { 189 if (!qemu_coroutine_entered(co)) { 190 qemu_coroutine_enter(co); 191 } 192 } 193 194 void coroutine_fn qemu_coroutine_yield(void) 195 { 196 Coroutine *self = qemu_coroutine_self(); 197 Coroutine *to = self->caller; 198 199 trace_qemu_coroutine_yield(self, to); 200 201 if (!to) { 202 fprintf(stderr, "Co-routine is yielding to no one\n"); 203 abort(); 204 } 205 206 self->caller = NULL; 207 qemu_coroutine_switch(self, to, COROUTINE_YIELD); 208 } 209 210 bool qemu_coroutine_entered(Coroutine *co) 211 { 212 return co->caller; 213 } 214 215 AioContext *qemu_coroutine_get_aio_context(Coroutine *co) 216 { 217 return co->ctx; 218 } 219 220 void qemu_coroutine_inc_pool_size(unsigned int additional_pool_size) 221 { 222 qatomic_add(&pool_max_size, additional_pool_size); 223 } 224 225 void qemu_coroutine_dec_pool_size(unsigned int removing_pool_size) 226 { 227 qatomic_sub(&pool_max_size, removing_pool_size); 228 } 229