4 * Copyright IBM, Corp. 2011
7 * Stefan Hajnoczi <stefanha@linux.vnet.ibm.com>
8 * Kevin Wolf <kwolf@redhat.com>
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.
15 #include "qemu/osdep.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"
24 /** Initial batch size is 64, and is increased on demand */
26 POOL_INITIAL_BATCH_SIZE
= 64,
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
;
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
);
39 static void coroutine_pool_cleanup(Notifier
*n
, void *value
)
43 CoroutineQSList
*alloc_pool
= get_ptr_alloc_pool();
45 QSLIST_FOREACH_SAFE(co
, alloc_pool
, pool_next
, tmp
) {
46 QSLIST_REMOVE_HEAD(alloc_pool
, pool_next
);
47 qemu_coroutine_delete(co
);
51 Coroutine
*qemu_coroutine_create(CoroutineEntry
*entry
, void *opaque
)
55 if (CONFIG_COROUTINE_POOL
) {
56 CoroutineQSList
*alloc_pool
= get_ptr_alloc_pool();
58 co
= QSLIST_FIRST(alloc_pool
);
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
);
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.
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
);
78 QSLIST_REMOVE_HEAD(alloc_pool
, pool_next
);
79 set_alloc_pool_size(get_alloc_pool_size() - 1);
84 co
= qemu_coroutine_new();
88 co
->entry_arg
= opaque
;
89 QSIMPLEQ_INIT(&co
->co_queue_wakeup
);
93 static void coroutine_delete(Coroutine
*co
)
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
);
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);
110 qemu_coroutine_delete(co
);
113 void qemu_aio_coroutine_enter(AioContext
*ctx
, Coroutine
*co
)
115 QSIMPLEQ_HEAD(, Coroutine
) pending
= QSIMPLEQ_HEAD_INITIALIZER(pending
);
116 Coroutine
*from
= qemu_coroutine_self();
118 QSIMPLEQ_INSERT_TAIL(&pending
, co
, co_queue_next
);
120 /* Run co and any queued coroutines */
121 while (!QSIMPLEQ_EMPTY(&pending
)) {
122 Coroutine
*to
= QSIMPLEQ_FIRST(&pending
);
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
);
129 QSIMPLEQ_REMOVE_HEAD(&pending
, co_queue_next
);
131 trace_qemu_aio_coroutine_enter(ctx
, from
, to
, to
->entry_arg
);
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
138 "%s: Co-routine was already scheduled in '%s'\n",
139 __func__
, scheduled
);
144 fprintf(stderr
, "Co-routine re-entered recursively\n");
151 /* Store to->ctx before anything that stores to. Matches
152 * barrier in aio_co_wake and qemu_co_mutex_wake.
156 ret
= qemu_coroutine_switch(from
, to
, COROUTINE_ENTER
);
158 /* Queued coroutines are run depth-first; previously pending coroutines
159 * run after those queued more recently.
161 QSIMPLEQ_PREPEND(&pending
, &to
->co_queue_wakeup
);
164 case COROUTINE_YIELD
:
166 case COROUTINE_TERMINATE
:
167 assert(!to
->locks_held
);
168 trace_qemu_coroutine_terminate(to
);
169 coroutine_delete(to
);
177 void qemu_coroutine_enter(Coroutine
*co
)
179 qemu_aio_coroutine_enter(qemu_get_current_aio_context(), co
);
182 void qemu_coroutine_enter_if_inactive(Coroutine
*co
)
184 if (!qemu_coroutine_entered(co
)) {
185 qemu_coroutine_enter(co
);
189 void coroutine_fn
qemu_coroutine_yield(void)
191 Coroutine
*self
= qemu_coroutine_self();
192 Coroutine
*to
= self
->caller
;
194 trace_qemu_coroutine_yield(self
, to
);
197 fprintf(stderr
, "Co-routine is yielding to no one\n");
202 qemu_coroutine_switch(self
, to
, COROUTINE_YIELD
);
205 bool qemu_coroutine_entered(Coroutine
*co
)
210 AioContext
*coroutine_fn
qemu_coroutine_get_aio_context(Coroutine
*co
)
215 void qemu_coroutine_increase_pool_batch_size(unsigned int additional_pool_size
)
217 qatomic_add(&pool_batch_size
, additional_pool_size
);
220 void qemu_coroutine_decrease_pool_batch_size(unsigned int removing_pool_size
)
222 qatomic_sub(&pool_batch_size
, removing_pool_size
);