1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29#include "qemu/osdep.h"
30#include "qemu/coroutine.h"
31#include "qemu/coroutine_int.h"
32#include "qemu/processor.h"
33#include "qemu/queue.h"
34#include "block/aio.h"
35#include "trace.h"
36
37void qemu_co_queue_init(CoQueue *queue)
38{
39 QSIMPLEQ_INIT(&queue->entries);
40}
41
42void coroutine_fn qemu_co_queue_wait_impl(CoQueue *queue, QemuLockable *lock)
43{
44 Coroutine *self = qemu_coroutine_self();
45 QSIMPLEQ_INSERT_TAIL(&queue->entries, self, co_queue_next);
46
47 if (lock) {
48 qemu_lockable_unlock(lock);
49 }
50
51
52
53
54
55
56 qemu_coroutine_yield();
57 assert(qemu_in_coroutine());
58
59
60
61
62
63
64
65 if (lock) {
66 qemu_lockable_lock(lock);
67 }
68}
69
70static bool qemu_co_queue_do_restart(CoQueue *queue, bool single)
71{
72 Coroutine *next;
73
74 if (QSIMPLEQ_EMPTY(&queue->entries)) {
75 return false;
76 }
77
78 while ((next = QSIMPLEQ_FIRST(&queue->entries)) != NULL) {
79 QSIMPLEQ_REMOVE_HEAD(&queue->entries, co_queue_next);
80 aio_co_wake(next);
81 if (single) {
82 break;
83 }
84 }
85 return true;
86}
87
88bool qemu_co_queue_next(CoQueue *queue)
89{
90 return qemu_co_queue_do_restart(queue, true);
91}
92
93void qemu_co_queue_restart_all(CoQueue *queue)
94{
95 qemu_co_queue_do_restart(queue, false);
96}
97
98bool qemu_co_enter_next_impl(CoQueue *queue, QemuLockable *lock)
99{
100 Coroutine *next;
101
102 next = QSIMPLEQ_FIRST(&queue->entries);
103 if (!next) {
104 return false;
105 }
106
107 QSIMPLEQ_REMOVE_HEAD(&queue->entries, co_queue_next);
108 if (lock) {
109 qemu_lockable_unlock(lock);
110 }
111 aio_co_wake(next);
112 if (lock) {
113 qemu_lockable_lock(lock);
114 }
115 return true;
116}
117
118bool qemu_co_queue_empty(CoQueue *queue)
119{
120 return QSIMPLEQ_FIRST(&queue->entries) == NULL;
121}
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142typedef struct CoWaitRecord {
143 Coroutine *co;
144 QSLIST_ENTRY(CoWaitRecord) next;
145} CoWaitRecord;
146
147static void push_waiter(CoMutex *mutex, CoWaitRecord *w)
148{
149 w->co = qemu_coroutine_self();
150 QSLIST_INSERT_HEAD_ATOMIC(&mutex->from_push, w, next);
151}
152
153static void move_waiters(CoMutex *mutex)
154{
155 QSLIST_HEAD(, CoWaitRecord) reversed;
156 QSLIST_MOVE_ATOMIC(&reversed, &mutex->from_push);
157 while (!QSLIST_EMPTY(&reversed)) {
158 CoWaitRecord *w = QSLIST_FIRST(&reversed);
159 QSLIST_REMOVE_HEAD(&reversed, next);
160 QSLIST_INSERT_HEAD(&mutex->to_pop, w, next);
161 }
162}
163
164static CoWaitRecord *pop_waiter(CoMutex *mutex)
165{
166 CoWaitRecord *w;
167
168 if (QSLIST_EMPTY(&mutex->to_pop)) {
169 move_waiters(mutex);
170 if (QSLIST_EMPTY(&mutex->to_pop)) {
171 return NULL;
172 }
173 }
174 w = QSLIST_FIRST(&mutex->to_pop);
175 QSLIST_REMOVE_HEAD(&mutex->to_pop, next);
176 return w;
177}
178
179static bool has_waiters(CoMutex *mutex)
180{
181 return QSLIST_EMPTY(&mutex->to_pop) || QSLIST_EMPTY(&mutex->from_push);
182}
183
184void qemu_co_mutex_init(CoMutex *mutex)
185{
186 memset(mutex, 0, sizeof(*mutex));
187}
188
189static void coroutine_fn qemu_co_mutex_wake(CoMutex *mutex, Coroutine *co)
190{
191
192
193
194 smp_read_barrier_depends();
195 mutex->ctx = co->ctx;
196 aio_co_wake(co);
197}
198
199static void coroutine_fn qemu_co_mutex_lock_slowpath(AioContext *ctx,
200 CoMutex *mutex)
201{
202 Coroutine *self = qemu_coroutine_self();
203 CoWaitRecord w;
204 unsigned old_handoff;
205
206 trace_qemu_co_mutex_lock_entry(mutex, self);
207 push_waiter(mutex, &w);
208
209
210
211
212 old_handoff = qatomic_mb_read(&mutex->handoff);
213 if (old_handoff &&
214 has_waiters(mutex) &&
215 qatomic_cmpxchg(&mutex->handoff, old_handoff, 0) == old_handoff) {
216
217
218
219 CoWaitRecord *to_wake = pop_waiter(mutex);
220 Coroutine *co = to_wake->co;
221 if (co == self) {
222
223 assert(to_wake == &w);
224 mutex->ctx = ctx;
225 return;
226 }
227
228 qemu_co_mutex_wake(mutex, co);
229 }
230
231 qemu_coroutine_yield();
232 trace_qemu_co_mutex_lock_return(mutex, self);
233}
234
235void coroutine_fn qemu_co_mutex_lock(CoMutex *mutex)
236{
237 AioContext *ctx = qemu_get_current_aio_context();
238 Coroutine *self = qemu_coroutine_self();
239 int waiters, i;
240
241
242
243
244
245
246
247
248 i = 0;
249retry_fast_path:
250 waiters = qatomic_cmpxchg(&mutex->locked, 0, 1);
251 if (waiters != 0) {
252 while (waiters == 1 && ++i < 1000) {
253 if (qatomic_read(&mutex->ctx) == ctx) {
254 break;
255 }
256 if (qatomic_read(&mutex->locked) == 0) {
257 goto retry_fast_path;
258 }
259 cpu_relax();
260 }
261 waiters = qatomic_fetch_inc(&mutex->locked);
262 }
263
264 if (waiters == 0) {
265
266 trace_qemu_co_mutex_lock_uncontended(mutex, self);
267 mutex->ctx = ctx;
268 } else {
269 qemu_co_mutex_lock_slowpath(ctx, mutex);
270 }
271 mutex->holder = self;
272 self->locks_held++;
273}
274
275void coroutine_fn qemu_co_mutex_unlock(CoMutex *mutex)
276{
277 Coroutine *self = qemu_coroutine_self();
278
279 trace_qemu_co_mutex_unlock_entry(mutex, self);
280
281 assert(mutex->locked);
282 assert(mutex->holder == self);
283 assert(qemu_in_coroutine());
284
285 mutex->ctx = NULL;
286 mutex->holder = NULL;
287 self->locks_held--;
288 if (qatomic_fetch_dec(&mutex->locked) == 1) {
289
290 return;
291 }
292
293 for (;;) {
294 CoWaitRecord *to_wake = pop_waiter(mutex);
295 unsigned our_handoff;
296
297 if (to_wake) {
298 qemu_co_mutex_wake(mutex, to_wake->co);
299 break;
300 }
301
302
303
304
305
306 if (++mutex->sequence == 0) {
307 mutex->sequence = 1;
308 }
309
310 our_handoff = mutex->sequence;
311 qatomic_mb_set(&mutex->handoff, our_handoff);
312 if (!has_waiters(mutex)) {
313
314
315
316 break;
317 }
318
319
320
321
322 if (qatomic_cmpxchg(&mutex->handoff, our_handoff, 0) != our_handoff) {
323 break;
324 }
325 }
326
327 trace_qemu_co_mutex_unlock_return(mutex, self);
328}
329
330struct CoRwTicket {
331 bool read;
332 Coroutine *co;
333 QSIMPLEQ_ENTRY(CoRwTicket) next;
334};
335
336void qemu_co_rwlock_init(CoRwlock *lock)
337{
338 qemu_co_mutex_init(&lock->mutex);
339 lock->owners = 0;
340 QSIMPLEQ_INIT(&lock->tickets);
341}
342
343
344static void qemu_co_rwlock_maybe_wake_one(CoRwlock *lock)
345{
346 CoRwTicket *tkt = QSIMPLEQ_FIRST(&lock->tickets);
347 Coroutine *co = NULL;
348
349
350
351
352
353
354 if (tkt) {
355 if (tkt->read) {
356 if (lock->owners >= 0) {
357 lock->owners++;
358 co = tkt->co;
359 }
360 } else {
361 if (lock->owners == 0) {
362 lock->owners = -1;
363 co = tkt->co;
364 }
365 }
366 }
367
368 if (co) {
369 QSIMPLEQ_REMOVE_HEAD(&lock->tickets, next);
370 qemu_co_mutex_unlock(&lock->mutex);
371 aio_co_wake(co);
372 } else {
373 qemu_co_mutex_unlock(&lock->mutex);
374 }
375}
376
377void qemu_co_rwlock_rdlock(CoRwlock *lock)
378{
379 Coroutine *self = qemu_coroutine_self();
380
381 qemu_co_mutex_lock(&lock->mutex);
382
383 if (lock->owners == 0 || (lock->owners > 0 && QSIMPLEQ_EMPTY(&lock->tickets))) {
384 lock->owners++;
385 qemu_co_mutex_unlock(&lock->mutex);
386 } else {
387 CoRwTicket my_ticket = { true, self };
388
389 QSIMPLEQ_INSERT_TAIL(&lock->tickets, &my_ticket, next);
390 qemu_co_mutex_unlock(&lock->mutex);
391 qemu_coroutine_yield();
392 assert(lock->owners >= 1);
393
394
395 qemu_co_mutex_lock(&lock->mutex);
396 qemu_co_rwlock_maybe_wake_one(lock);
397 }
398
399 self->locks_held++;
400}
401
402void qemu_co_rwlock_unlock(CoRwlock *lock)
403{
404 Coroutine *self = qemu_coroutine_self();
405
406 assert(qemu_in_coroutine());
407 self->locks_held--;
408
409 qemu_co_mutex_lock(&lock->mutex);
410 if (lock->owners > 0) {
411 lock->owners--;
412 } else {
413 assert(lock->owners == -1);
414 lock->owners = 0;
415 }
416
417 qemu_co_rwlock_maybe_wake_one(lock);
418}
419
420void qemu_co_rwlock_downgrade(CoRwlock *lock)
421{
422 qemu_co_mutex_lock(&lock->mutex);
423 assert(lock->owners == -1);
424 lock->owners = 1;
425
426
427 qemu_co_rwlock_maybe_wake_one(lock);
428}
429
430void qemu_co_rwlock_wrlock(CoRwlock *lock)
431{
432 Coroutine *self = qemu_coroutine_self();
433
434 qemu_co_mutex_lock(&lock->mutex);
435 if (lock->owners == 0) {
436 lock->owners = -1;
437 qemu_co_mutex_unlock(&lock->mutex);
438 } else {
439 CoRwTicket my_ticket = { false, qemu_coroutine_self() };
440
441 QSIMPLEQ_INSERT_TAIL(&lock->tickets, &my_ticket, next);
442 qemu_co_mutex_unlock(&lock->mutex);
443 qemu_coroutine_yield();
444 assert(lock->owners == -1);
445 }
446
447 self->locks_held++;
448}
449
450void qemu_co_rwlock_upgrade(CoRwlock *lock)
451{
452 qemu_co_mutex_lock(&lock->mutex);
453 assert(lock->owners > 0);
454
455 if (lock->owners == 1 && QSIMPLEQ_EMPTY(&lock->tickets)) {
456 lock->owners = -1;
457 qemu_co_mutex_unlock(&lock->mutex);
458 } else {
459 CoRwTicket my_ticket = { false, qemu_coroutine_self() };
460
461 lock->owners--;
462 QSIMPLEQ_INSERT_TAIL(&lock->tickets, &my_ticket, next);
463 qemu_co_rwlock_maybe_wake_one(lock);
464 qemu_coroutine_yield();
465 assert(lock->owners == -1);
466 }
467}
468