1
2
3
4
5
6
7
8
9
10
11
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
25
26
27
28
29enum {
30 POOL_MIN_BATCH_SIZE = 64,
31 POOL_INITIAL_MAX_SIZE = 64,
32};
33
34
35static QSLIST_HEAD(, Coroutine) release_pool = QSLIST_HEAD_INITIALIZER(pool);
36static unsigned int pool_max_size = POOL_INITIAL_MAX_SIZE;
37static unsigned int release_pool_size;
38
39typedef QSLIST_HEAD(, Coroutine) CoroutineQSList;
40QEMU_DEFINE_STATIC_CO_TLS(CoroutineQSList, alloc_pool);
41QEMU_DEFINE_STATIC_CO_TLS(unsigned int, alloc_pool_size);
42QEMU_DEFINE_STATIC_CO_TLS(Notifier, coroutine_pool_cleanup_notifier);
43
44static 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
56Coroutine *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
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
74
75
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
98static 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
118void 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
126 while (!QSIMPLEQ_EMPTY(&pending)) {
127 Coroutine *to = QSIMPLEQ_FIRST(&pending);
128 CoroutineAction ret;
129
130
131
132
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
143
144
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
161
162
163 smp_wmb();
164
165 ret = qemu_coroutine_switch(from, to, COROUTINE_ENTER);
166
167
168
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
186void qemu_coroutine_enter(Coroutine *co)
187{
188 qemu_aio_coroutine_enter(qemu_get_current_aio_context(), co);
189}
190
191void qemu_coroutine_enter_if_inactive(Coroutine *co)
192{
193 if (!qemu_coroutine_entered(co)) {
194 qemu_coroutine_enter(co);
195 }
196}
197
198void 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
214bool qemu_coroutine_entered(Coroutine *co)
215{
216 return co->caller;
217}
218
219AioContext *qemu_coroutine_get_aio_context(Coroutine *co)
220{
221 return co->ctx;
222}
223
224void qemu_coroutine_inc_pool_size(unsigned int additional_pool_size)
225{
226 qatomic_add(&pool_max_size, additional_pool_size);
227}
228
229void qemu_coroutine_dec_pool_size(unsigned int removing_pool_size)
230{
231 qatomic_sub(&pool_max_size, removing_pool_size);
232}
233