qemu-coroutine-lock.c

   1 /*
   2  * coroutine queues and locks
   3  *
   4  * Copyright (c) 2011 Kevin Wolf <kwolf@redhat.com>
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a copy
   7  * of this software and associated documentation files (the "Software"), to deal
   8  * in the Software without restriction, including without limitation the rights
   9  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  10  * copies of the Software, and to permit persons to whom the Software is
  11  * furnished to do so, subject to the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be included in
  14  * all copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
  19  * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  21  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  22  * THE SOFTWARE.
  23  */
  24
  25 #include "qemu-common.h"
  26 #include "block/coroutine.h"
  27 #include "block/coroutine_int.h"
  28 #include "qemu/queue.h"
  29 #include "block/aio.h"
  30 #include "trace.h"
  31
  32 /* Coroutines are awoken from a BH to allow the current coroutine to complete
  33  * its flow of execution.  The BH may run after the CoQueue has been destroyed,
  34  * so keep BH data in a separate heap-allocated struct.
  35  */
  36 typedef struct {
  37     QEMUBH *bh;
  38     QTAILQ_HEAD(, Coroutine) entries;
  39 } CoQueueNextData;
  40
  41 static void qemu_co_queue_next_bh(void *opaque)
  42 {
  43     CoQueueNextData *data = opaque;
  44     Coroutine *next;
  45
  46     trace_qemu_co_queue_next_bh();
  47     while ((next = QTAILQ_FIRST(&data->entries))) {
  48         QTAILQ_REMOVE(&data->entries, next, co_queue_next);
  49         qemu_coroutine_enter(next, NULL);
  50     }
  51
  52     qemu_bh_delete(data->bh);
  53     g_slice_free(CoQueueNextData, data);
  54 }
  55
  56 void qemu_co_queue_init(CoQueue *queue)
  57 {
  58     QTAILQ_INIT(&queue->entries);
  59
  60     /* This will be exposed to callers once there are multiple AioContexts */
  61     queue->ctx = qemu_get_aio_context();
  62 }
  63
  64 void coroutine_fn qemu_co_queue_wait(CoQueue *queue)
  65 {
  66     Coroutine *self = qemu_coroutine_self();
  67     QTAILQ_INSERT_TAIL(&queue->entries, self, co_queue_next);
  68     qemu_coroutine_yield();
  69     assert(qemu_in_coroutine());
  70 }
  71
  72 void coroutine_fn qemu_co_queue_wait_insert_head(CoQueue *queue)
  73 {
  74     Coroutine *self = qemu_coroutine_self();
  75     QTAILQ_INSERT_HEAD(&queue->entries, self, co_queue_next);
  76     qemu_coroutine_yield();
  77     assert(qemu_in_coroutine());
  78 }
  79
  80 static bool qemu_co_queue_do_restart(CoQueue *queue, bool single)
  81 {
  82     Coroutine *next;
  83     CoQueueNextData *data;
  84
  85     if (QTAILQ_EMPTY(&queue->entries)) {
  86         return false;
  87     }
  88
  89     data = g_slice_new(CoQueueNextData);
  90     data->bh = aio_bh_new(queue->ctx, qemu_co_queue_next_bh, data);
  91     QTAILQ_INIT(&data->entries);
  92     qemu_bh_schedule(data->bh);
  93
  94     while ((next = QTAILQ_FIRST(&queue->entries)) != NULL) {
  95         QTAILQ_REMOVE(&queue->entries, next, co_queue_next);
  96         QTAILQ_INSERT_TAIL(&data->entries, next, co_queue_next);
  97         trace_qemu_co_queue_next(next);
  98         if (single) {
  99             break;
 100         }
 101     }
 102     return true;
 103 }
 104
 105 bool qemu_co_queue_next(CoQueue *queue)
 106 {
 107     return qemu_co_queue_do_restart(queue, true);
 108 }
 109
 110 void qemu_co_queue_restart_all(CoQueue *queue)
 111 {
 112     qemu_co_queue_do_restart(queue, false);
 113 }
 114
 115 bool qemu_co_queue_empty(CoQueue *queue)
 116 {
 117     return (QTAILQ_FIRST(&queue->entries) == NULL);
 118 }
 119
 120 void qemu_co_mutex_init(CoMutex *mutex)
 121 {
 122     memset(mutex, 0, sizeof(*mutex));
 123     qemu_co_queue_init(&mutex->queue);
 124 }
 125
 126 void coroutine_fn qemu_co_mutex_lock(CoMutex *mutex)
 127 {
 128     Coroutine *self = qemu_coroutine_self();
 129
 130     trace_qemu_co_mutex_lock_entry(mutex, self);
 131
 132     while (mutex->locked) {
 133         qemu_co_queue_wait(&mutex->queue);
 134     }
 135
 136     mutex->locked = true;
 137
 138     trace_qemu_co_mutex_lock_return(mutex, self);
 139 }
 140
 141 void coroutine_fn qemu_co_mutex_unlock(CoMutex *mutex)
 142 {
 143     Coroutine *self = qemu_coroutine_self();
 144
 145     trace_qemu_co_mutex_unlock_entry(mutex, self);
 146
 147     assert(mutex->locked == true);
 148     assert(qemu_in_coroutine());
 149
 150     mutex->locked = false;
 151     qemu_co_queue_next(&mutex->queue);
 152
 153     trace_qemu_co_mutex_unlock_return(mutex, self);
 154 }
 155
 156 void qemu_co_rwlock_init(CoRwlock *lock)
 157 {
 158     memset(lock, 0, sizeof(*lock));
 159     qemu_co_queue_init(&lock->queue);
 160 }
 161
 162 void qemu_co_rwlock_rdlock(CoRwlock *lock)
 163 {
 164     while (lock->writer) {
 165         qemu_co_queue_wait(&lock->queue);
 166     }
 167     lock->reader++;
 168 }
 169
 170 void qemu_co_rwlock_unlock(CoRwlock *lock)
 171 {
 172     assert(qemu_in_coroutine());
 173     if (lock->writer) {
 174         lock->writer = false;
 175         qemu_co_queue_restart_all(&lock->queue);
 176     } else {
 177         lock->reader--;
 178         assert(lock->reader >= 0);
 179         /* Wakeup only one waiting writer */
 180         if (!lock->reader) {
 181             qemu_co_queue_next(&lock->queue);
 182         }
 183     }
 184 }
 185
 186 void qemu_co_rwlock_wrlock(CoRwlock *lock)
 187 {
 188     while (lock->writer || lock->reader) {
 189         qemu_co_queue_wait(&lock->queue);
 190     }
 191     lock->writer = true;
 192 }