24  * The lock-free mutex implementation is based on OSv
32 #include "qemu/queue.h"
36 void qemu_co_queue_init(CoQueue *queue)  in qemu_co_queue_init()  argument
38     QSIMPLEQ_INIT(&queue->entries);  in qemu_co_queue_init()
41 void coroutine_fn qemu_co_queue_wait_impl(CoQueue *queue, QemuLockable *lock,  in qemu_co_queue_wait_impl()  argument
46         QSIMPLEQ_INSERT_HEAD(&queue->entries, self, co_queue_next);  in qemu_co_queue_wait_impl()
48         QSIMPLEQ_INSERT_TAIL(&queue->entries, self, co_queue_next);  in qemu_co_queue_wait_impl()
63     /* TODO: OSv implements wait morphing here, where the wakeup  in qemu_co_queue_wait_impl()
65      * mutex's queue.  This avoids the thundering herd effect.  in qemu_co_queue_wait_impl()
74 bool qemu_co_enter_next_impl(CoQueue *queue, QemuLockable *lock)  in qemu_co_enter_next_impl()  argument
78     next = QSIMPLEQ_FIRST(&queue->entries);  in qemu_co_enter_next_impl()
83     QSIMPLEQ_REMOVE_HEAD(&queue->entries, co_queue_next);  in qemu_co_enter_next_impl()
94 bool coroutine_fn qemu_co_queue_next(CoQueue *queue)  in qemu_co_queue_next()  argument
97     return qemu_co_enter_next_impl(queue, NULL);  in qemu_co_queue_next()
100 void qemu_co_enter_all_impl(CoQueue *queue, QemuLockable *lock)  in qemu_co_enter_all_impl()  argument
102     while (qemu_co_enter_next_impl(queue, lock)) {  in qemu_co_enter_all_impl()
107 void coroutine_fn qemu_co_queue_restart_all(CoQueue *queue)  in qemu_co_queue_restart_all()  argument
110     qemu_co_enter_all_impl(queue, NULL);  in qemu_co_queue_restart_all()
113 bool qemu_co_queue_empty(CoQueue *queue)  in qemu_co_queue_empty()  argument
115     return QSIMPLEQ_FIRST(&queue->entries) == NULL;  in qemu_co_queue_empty()
118 /* The wait records are handled with a multiple-producer, single-consumer
119  * lock-free queue.  There cannot be two concurrent pop_waiter() calls
120  * because pop_waiter() can only be called while mutex->handoff is zero.
122  * - in qemu_co_mutex_unlock, before the hand-off protocol has started.
123  *   In this case, qemu_co_mutex_lock will see mutex->handoff == 0 and
125  * - in qemu_co_mutex_lock, if it steals the hand-off responsibility from
128  *   exit.  The next hand-off cannot begin until qemu_co_mutex_lock has
130  * - in qemu_co_mutex_unlock, if it takes the hand-off token itself.
131  *   In this case another iteration starts with mutex->handoff == 0;
135  * The following functions manage this queue.
144     w->co = qemu_coroutine_self();  in push_waiter()
145     QSLIST_INSERT_HEAD_ATOMIC(&mutex->from_push, w, next);  in push_waiter()
151     QSLIST_MOVE_ATOMIC(&reversed, &mutex->from_push);  in move_waiters()
155         QSLIST_INSERT_HEAD(&mutex->to_pop, w, next);  in move_waiters()
163     if (QSLIST_EMPTY(&mutex->to_pop)) {  in pop_waiter()
165         if (QSLIST_EMPTY(&mutex->to_pop)) {  in pop_waiter()
169     w = QSLIST_FIRST(&mutex->to_pop);  in pop_waiter()
170     QSLIST_REMOVE_HEAD(&mutex->to_pop, next);  in pop_waiter()
176     return QSLIST_EMPTY(&mutex->to_pop) || QSLIST_EMPTY(&mutex->from_push);  in has_waiters()
186     /* Read co before co->ctx; pairs with smp_wmb() in  in qemu_co_mutex_wake()
190     mutex->ctx = co->ctx;  in qemu_co_mutex_wake()
205      * Add waiter before reading mutex->handoff.  Pairs with qatomic_set_mb  in qemu_co_mutex_lock_slowpath()
210     /* This is the "Responsibility Hand-Off" protocol; a lock() picks from  in qemu_co_mutex_lock_slowpath()
213     old_handoff = qatomic_read(&mutex->handoff);  in qemu_co_mutex_lock_slowpath()
216         qatomic_cmpxchg(&mutex->handoff, old_handoff, 0) == old_handoff) {  in qemu_co_mutex_lock_slowpath()
221         Coroutine *co = to_wake->co;  in qemu_co_mutex_lock_slowpath()
225             mutex->ctx = ctx;  in qemu_co_mutex_lock_slowpath()
247      * avoid wait and wakeup.  So introduce it artificially.  in qemu_co_mutex_lock()
251     waiters = qatomic_cmpxchg(&mutex->locked, 0, 1);  in qemu_co_mutex_lock()
254             if (qatomic_read(&mutex->ctx) == ctx) {  in qemu_co_mutex_lock()
257             if (qatomic_read(&mutex->locked) == 0) {  in qemu_co_mutex_lock()
262         waiters = qatomic_fetch_inc(&mutex->locked);  in qemu_co_mutex_lock()
268         mutex->ctx = ctx;  in qemu_co_mutex_lock()
272     mutex->holder = self;  in qemu_co_mutex_lock()
273     self->locks_held++;  in qemu_co_mutex_lock()
282     assert(mutex->locked);  in qemu_co_mutex_unlock()
283     assert(mutex->holder == self);  in qemu_co_mutex_unlock()
286     mutex->ctx = NULL;  in qemu_co_mutex_unlock()
287     mutex->holder = NULL;  in qemu_co_mutex_unlock()
288     self->locks_held--;  in qemu_co_mutex_unlock()
289     if (qatomic_fetch_dec(&mutex->locked) == 1) {  in qemu_co_mutex_unlock()
299             qemu_co_mutex_wake(mutex, to_wake->co);  in qemu_co_mutex_unlock()
304          * mutex->locked was >1) but it hasn't yet put itself on the wait  in qemu_co_mutex_unlock()
305          * queue.  Pick a sequence number for the handoff protocol (not 0).  in qemu_co_mutex_unlock()
307         if (++mutex->sequence == 0) {  in qemu_co_mutex_unlock()
308             mutex->sequence = 1;  in qemu_co_mutex_unlock()
311         our_handoff = mutex->sequence;  in qemu_co_mutex_unlock()
313         qatomic_set_mb(&mutex->handoff, our_handoff);  in qemu_co_mutex_unlock()
324         if (qatomic_cmpxchg(&mutex->handoff, our_handoff, 0) != our_handoff) {  in qemu_co_mutex_unlock()
340     qemu_co_mutex_init(&lock->mutex);  in qemu_co_rwlock_init()
341     lock->owners = 0;  in qemu_co_rwlock_init()
342     QSIMPLEQ_INIT(&lock->tickets);  in qemu_co_rwlock_init()
348     CoRwTicket *tkt = QSIMPLEQ_FIRST(&lock->tickets);  in qemu_co_rwlock_maybe_wake_one()
352      * Setting lock->owners here prevents rdlock and wrlock from  in qemu_co_rwlock_maybe_wake_one()
357         if (tkt->read) {  in qemu_co_rwlock_maybe_wake_one()
358             if (lock->owners >= 0) {  in qemu_co_rwlock_maybe_wake_one()
359                 lock->owners++;  in qemu_co_rwlock_maybe_wake_one()
360                 co = tkt->co;  in qemu_co_rwlock_maybe_wake_one()
363             if (lock->owners == 0) {  in qemu_co_rwlock_maybe_wake_one()
364                 lock->owners = -1;  in qemu_co_rwlock_maybe_wake_one()
365                 co = tkt->co;  in qemu_co_rwlock_maybe_wake_one()
371         QSIMPLEQ_REMOVE_HEAD(&lock->tickets, next);  in qemu_co_rwlock_maybe_wake_one()
372         qemu_co_mutex_unlock(&lock->mutex);  in qemu_co_rwlock_maybe_wake_one()
375         qemu_co_mutex_unlock(&lock->mutex);  in qemu_co_rwlock_maybe_wake_one()
383     qemu_co_mutex_lock(&lock->mutex);  in qemu_co_rwlock_rdlock()
384     /* For fairness, wait if a writer is in line.  */  in qemu_co_rwlock_rdlock()
385     if (lock->owners == 0 || (lock->owners > 0 && QSIMPLEQ_EMPTY(&lock->tickets))) {  in qemu_co_rwlock_rdlock()
386         lock->owners++;  in qemu_co_rwlock_rdlock()
387         qemu_co_mutex_unlock(&lock->mutex);  in qemu_co_rwlock_rdlock()
391         QSIMPLEQ_INSERT_TAIL(&lock->tickets, &my_ticket, next);  in qemu_co_rwlock_rdlock()
392         qemu_co_mutex_unlock(&lock->mutex);  in qemu_co_rwlock_rdlock()
394         assert(lock->owners >= 1);  in qemu_co_rwlock_rdlock()
397         qemu_co_mutex_lock(&lock->mutex);  in qemu_co_rwlock_rdlock()
401     self->locks_held++;  in qemu_co_rwlock_rdlock()
409     self->locks_held--;  in qemu_co_rwlock_unlock()
411     qemu_co_mutex_lock(&lock->mutex);  in qemu_co_rwlock_unlock()
412     if (lock->owners > 0) {  in qemu_co_rwlock_unlock()
413         lock->owners--;  in qemu_co_rwlock_unlock()
415         assert(lock->owners == -1);  in qemu_co_rwlock_unlock()
416         lock->owners = 0;  in qemu_co_rwlock_unlock()
424     qemu_co_mutex_lock(&lock->mutex);  in qemu_co_rwlock_downgrade()
425     assert(lock->owners == -1);  in qemu_co_rwlock_downgrade()
426     lock->owners = 1;  in qemu_co_rwlock_downgrade()
436     qemu_co_mutex_lock(&lock->mutex);  in qemu_co_rwlock_wrlock()
437     if (lock->owners == 0) {  in qemu_co_rwlock_wrlock()
438         lock->owners = -1;  in qemu_co_rwlock_wrlock()
439         qemu_co_mutex_unlock(&lock->mutex);  in qemu_co_rwlock_wrlock()
443         QSIMPLEQ_INSERT_TAIL(&lock->tickets, &my_ticket, next);  in qemu_co_rwlock_wrlock()
444         qemu_co_mutex_unlock(&lock->mutex);  in qemu_co_rwlock_wrlock()
446         assert(lock->owners == -1);  in qemu_co_rwlock_wrlock()
449     self->locks_held++;  in qemu_co_rwlock_wrlock()
454     qemu_co_mutex_lock(&lock->mutex);  in qemu_co_rwlock_upgrade()
455     assert(lock->owners > 0);  in qemu_co_rwlock_upgrade()
456     /* For fairness, wait if a writer is in line.  */  in qemu_co_rwlock_upgrade()
457     if (lock->owners == 1 && QSIMPLEQ_EMPTY(&lock->tickets)) {  in qemu_co_rwlock_upgrade()
458         lock->owners = -1;  in qemu_co_rwlock_upgrade()
459         qemu_co_mutex_unlock(&lock->mutex);  in qemu_co_rwlock_upgrade()
463         lock->owners--;  in qemu_co_rwlock_upgrade()
464         QSIMPLEQ_INSERT_TAIL(&lock->tickets, &my_ticket, next);  in qemu_co_rwlock_upgrade()
467         assert(lock->owners == -1);  in qemu_co_rwlock_upgrade()