ppc/pnv: Add support for POWER8+ LPC Controller
[qemu.git] / util / 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 * The lock-free mutex implementation is based on OSv
25 * (core/lfmutex.cc, include/lockfree/mutex.hh).
26 * Copyright (C) 2013 Cloudius Systems, Ltd.
27 */
28
29 #include "qemu/osdep.h"
30 #include "qemu-common.h"
31 #include "qemu/coroutine.h"
32 #include "qemu/coroutine_int.h"
33 #include "qemu/processor.h"
34 #include "qemu/queue.h"
35 #include "block/aio.h"
36 #include "trace.h"
37
38 void qemu_co_queue_init(CoQueue *queue)
39 {
40 QSIMPLEQ_INIT(&queue->entries);
41 }
42
43 void coroutine_fn qemu_co_queue_wait(CoQueue *queue, CoMutex *mutex)
44 {
45 Coroutine *self = qemu_coroutine_self();
46 QSIMPLEQ_INSERT_TAIL(&queue->entries, self, co_queue_next);
47
48 if (mutex) {
49 qemu_co_mutex_unlock(mutex);
50 }
51
52 /* There is no race condition here. Other threads will call
53 * aio_co_schedule on our AioContext, which can reenter this
54 * coroutine but only after this yield and after the main loop
55 * has gone through the next iteration.
56 */
57 qemu_coroutine_yield();
58 assert(qemu_in_coroutine());
59
60 /* TODO: OSv implements wait morphing here, where the wakeup
61 * primitive automatically places the woken coroutine on the
62 * mutex's queue. This avoids the thundering herd effect.
63 */
64 if (mutex) {
65 qemu_co_mutex_lock(mutex);
66 }
67 }
68
69 /**
70 * qemu_co_queue_run_restart:
71 *
72 * Enter each coroutine that was previously marked for restart by
73 * qemu_co_queue_next() or qemu_co_queue_restart_all(). This function is
74 * invoked by the core coroutine code when the current coroutine yields or
75 * terminates.
76 */
77 void qemu_co_queue_run_restart(Coroutine *co)
78 {
79 Coroutine *next;
80
81 trace_qemu_co_queue_run_restart(co);
82 while ((next = QSIMPLEQ_FIRST(&co->co_queue_wakeup))) {
83 QSIMPLEQ_REMOVE_HEAD(&co->co_queue_wakeup, co_queue_next);
84 qemu_coroutine_enter(next);
85 }
86 }
87
88 static bool qemu_co_queue_do_restart(CoQueue *queue, bool single)
89 {
90 Coroutine *next;
91
92 if (QSIMPLEQ_EMPTY(&queue->entries)) {
93 return false;
94 }
95
96 while ((next = QSIMPLEQ_FIRST(&queue->entries)) != NULL) {
97 QSIMPLEQ_REMOVE_HEAD(&queue->entries, co_queue_next);
98 aio_co_wake(next);
99 if (single) {
100 break;
101 }
102 }
103 return true;
104 }
105
106 bool coroutine_fn qemu_co_queue_next(CoQueue *queue)
107 {
108 assert(qemu_in_coroutine());
109 return qemu_co_queue_do_restart(queue, true);
110 }
111
112 void coroutine_fn qemu_co_queue_restart_all(CoQueue *queue)
113 {
114 assert(qemu_in_coroutine());
115 qemu_co_queue_do_restart(queue, false);
116 }
117
118 bool qemu_co_enter_next(CoQueue *queue)
119 {
120 Coroutine *next;
121
122 next = QSIMPLEQ_FIRST(&queue->entries);
123 if (!next) {
124 return false;
125 }
126
127 QSIMPLEQ_REMOVE_HEAD(&queue->entries, co_queue_next);
128 qemu_coroutine_enter(next);
129 return true;
130 }
131
132 bool qemu_co_queue_empty(CoQueue *queue)
133 {
134 return QSIMPLEQ_FIRST(&queue->entries) == NULL;
135 }
136
137 /* The wait records are handled with a multiple-producer, single-consumer
138 * lock-free queue. There cannot be two concurrent pop_waiter() calls
139 * because pop_waiter() can only be called while mutex->handoff is zero.
140 * This can happen in three cases:
141 * - in qemu_co_mutex_unlock, before the hand-off protocol has started.
142 * In this case, qemu_co_mutex_lock will see mutex->handoff == 0 and
143 * not take part in the handoff.
144 * - in qemu_co_mutex_lock, if it steals the hand-off responsibility from
145 * qemu_co_mutex_unlock. In this case, qemu_co_mutex_unlock will fail
146 * the cmpxchg (it will see either 0 or the next sequence value) and
147 * exit. The next hand-off cannot begin until qemu_co_mutex_lock has
148 * woken up someone.
149 * - in qemu_co_mutex_unlock, if it takes the hand-off token itself.
150 * In this case another iteration starts with mutex->handoff == 0;
151 * a concurrent qemu_co_mutex_lock will fail the cmpxchg, and
152 * qemu_co_mutex_unlock will go back to case (1).
153 *
154 * The following functions manage this queue.
155 */
156 typedef struct CoWaitRecord {
157 Coroutine *co;
158 QSLIST_ENTRY(CoWaitRecord) next;
159 } CoWaitRecord;
160
161 static void push_waiter(CoMutex *mutex, CoWaitRecord *w)
162 {
163 w->co = qemu_coroutine_self();
164 QSLIST_INSERT_HEAD_ATOMIC(&mutex->from_push, w, next);
165 }
166
167 static void move_waiters(CoMutex *mutex)
168 {
169 QSLIST_HEAD(, CoWaitRecord) reversed;
170 QSLIST_MOVE_ATOMIC(&reversed, &mutex->from_push);
171 while (!QSLIST_EMPTY(&reversed)) {
172 CoWaitRecord *w = QSLIST_FIRST(&reversed);
173 QSLIST_REMOVE_HEAD(&reversed, next);
174 QSLIST_INSERT_HEAD(&mutex->to_pop, w, next);
175 }
176 }
177
178 static CoWaitRecord *pop_waiter(CoMutex *mutex)
179 {
180 CoWaitRecord *w;
181
182 if (QSLIST_EMPTY(&mutex->to_pop)) {
183 move_waiters(mutex);
184 if (QSLIST_EMPTY(&mutex->to_pop)) {
185 return NULL;
186 }
187 }
188 w = QSLIST_FIRST(&mutex->to_pop);
189 QSLIST_REMOVE_HEAD(&mutex->to_pop, next);
190 return w;
191 }
192
193 static bool has_waiters(CoMutex *mutex)
194 {
195 return QSLIST_EMPTY(&mutex->to_pop) || QSLIST_EMPTY(&mutex->from_push);
196 }
197
198 void qemu_co_mutex_init(CoMutex *mutex)
199 {
200 memset(mutex, 0, sizeof(*mutex));
201 }
202
203 static void coroutine_fn qemu_co_mutex_wake(CoMutex *mutex, Coroutine *co)
204 {
205 /* Read co before co->ctx; pairs with smp_wmb() in
206 * qemu_coroutine_enter().
207 */
208 smp_read_barrier_depends();
209 mutex->ctx = co->ctx;
210 aio_co_wake(co);
211 }
212
213 static void coroutine_fn qemu_co_mutex_lock_slowpath(AioContext *ctx,
214 CoMutex *mutex)
215 {
216 Coroutine *self = qemu_coroutine_self();
217 CoWaitRecord w;
218 unsigned old_handoff;
219
220 trace_qemu_co_mutex_lock_entry(mutex, self);
221 w.co = self;
222 push_waiter(mutex, &w);
223
224 /* This is the "Responsibility Hand-Off" protocol; a lock() picks from
225 * a concurrent unlock() the responsibility of waking somebody up.
226 */
227 old_handoff = atomic_mb_read(&mutex->handoff);
228 if (old_handoff &&
229 has_waiters(mutex) &&
230 atomic_cmpxchg(&mutex->handoff, old_handoff, 0) == old_handoff) {
231 /* There can be no concurrent pops, because there can be only
232 * one active handoff at a time.
233 */
234 CoWaitRecord *to_wake = pop_waiter(mutex);
235 Coroutine *co = to_wake->co;
236 if (co == self) {
237 /* We got the lock ourselves! */
238 assert(to_wake == &w);
239 mutex->ctx = ctx;
240 return;
241 }
242
243 qemu_co_mutex_wake(mutex, co);
244 }
245
246 qemu_coroutine_yield();
247 trace_qemu_co_mutex_lock_return(mutex, self);
248 }
249
250 void coroutine_fn qemu_co_mutex_lock(CoMutex *mutex)
251 {
252 AioContext *ctx = qemu_get_current_aio_context();
253 Coroutine *self = qemu_coroutine_self();
254 int waiters, i;
255
256 /* Running a very small critical section on pthread_mutex_t and CoMutex
257 * shows that pthread_mutex_t is much faster because it doesn't actually
258 * go to sleep. What happens is that the critical section is shorter
259 * than the latency of entering the kernel and thus FUTEX_WAIT always
260 * fails. With CoMutex there is no such latency but you still want to
261 * avoid wait and wakeup. So introduce it artificially.
262 */
263 i = 0;
264 retry_fast_path:
265 waiters = atomic_cmpxchg(&mutex->locked, 0, 1);
266 if (waiters != 0) {
267 while (waiters == 1 && ++i < 1000) {
268 if (atomic_read(&mutex->ctx) == ctx) {
269 break;
270 }
271 if (atomic_read(&mutex->locked) == 0) {
272 goto retry_fast_path;
273 }
274 cpu_relax();
275 }
276 waiters = atomic_fetch_inc(&mutex->locked);
277 }
278
279 if (waiters == 0) {
280 /* Uncontended. */
281 trace_qemu_co_mutex_lock_uncontended(mutex, self);
282 mutex->ctx = ctx;
283 } else {
284 qemu_co_mutex_lock_slowpath(ctx, mutex);
285 }
286 mutex->holder = self;
287 self->locks_held++;
288 }
289
290 void coroutine_fn qemu_co_mutex_unlock(CoMutex *mutex)
291 {
292 Coroutine *self = qemu_coroutine_self();
293
294 trace_qemu_co_mutex_unlock_entry(mutex, self);
295
296 assert(mutex->locked);
297 assert(mutex->holder == self);
298 assert(qemu_in_coroutine());
299
300 mutex->ctx = NULL;
301 mutex->holder = NULL;
302 self->locks_held--;
303 if (atomic_fetch_dec(&mutex->locked) == 1) {
304 /* No waiting qemu_co_mutex_lock(). Pfew, that was easy! */
305 return;
306 }
307
308 for (;;) {
309 CoWaitRecord *to_wake = pop_waiter(mutex);
310 unsigned our_handoff;
311
312 if (to_wake) {
313 qemu_co_mutex_wake(mutex, to_wake->co);
314 break;
315 }
316
317 /* Some concurrent lock() is in progress (we know this because
318 * mutex->locked was >1) but it hasn't yet put itself on the wait
319 * queue. Pick a sequence number for the handoff protocol (not 0).
320 */
321 if (++mutex->sequence == 0) {
322 mutex->sequence = 1;
323 }
324
325 our_handoff = mutex->sequence;
326 atomic_mb_set(&mutex->handoff, our_handoff);
327 if (!has_waiters(mutex)) {
328 /* The concurrent lock has not added itself yet, so it
329 * will be able to pick our handoff.
330 */
331 break;
332 }
333
334 /* Try to do the handoff protocol ourselves; if somebody else has
335 * already taken it, however, we're done and they're responsible.
336 */
337 if (atomic_cmpxchg(&mutex->handoff, our_handoff, 0) != our_handoff) {
338 break;
339 }
340 }
341
342 trace_qemu_co_mutex_unlock_return(mutex, self);
343 }
344
345 void qemu_co_rwlock_init(CoRwlock *lock)
346 {
347 memset(lock, 0, sizeof(*lock));
348 qemu_co_queue_init(&lock->queue);
349 qemu_co_mutex_init(&lock->mutex);
350 }
351
352 void qemu_co_rwlock_rdlock(CoRwlock *lock)
353 {
354 Coroutine *self = qemu_coroutine_self();
355
356 qemu_co_mutex_lock(&lock->mutex);
357 /* For fairness, wait if a writer is in line. */
358 while (lock->pending_writer) {
359 qemu_co_queue_wait(&lock->queue, &lock->mutex);
360 }
361 lock->reader++;
362 qemu_co_mutex_unlock(&lock->mutex);
363
364 /* The rest of the read-side critical section is run without the mutex. */
365 self->locks_held++;
366 }
367
368 void qemu_co_rwlock_unlock(CoRwlock *lock)
369 {
370 Coroutine *self = qemu_coroutine_self();
371
372 assert(qemu_in_coroutine());
373 if (!lock->reader) {
374 /* The critical section started in qemu_co_rwlock_wrlock. */
375 qemu_co_queue_restart_all(&lock->queue);
376 } else {
377 self->locks_held--;
378
379 qemu_co_mutex_lock(&lock->mutex);
380 lock->reader--;
381 assert(lock->reader >= 0);
382 /* Wakeup only one waiting writer */
383 if (!lock->reader) {
384 qemu_co_queue_next(&lock->queue);
385 }
386 }
387 qemu_co_mutex_unlock(&lock->mutex);
388 }
389
390 void qemu_co_rwlock_wrlock(CoRwlock *lock)
391 {
392 qemu_co_mutex_lock(&lock->mutex);
393 lock->pending_writer++;
394 while (lock->reader) {
395 qemu_co_queue_wait(&lock->queue, &lock->mutex);
396 }
397 lock->pending_writer--;
398
399 /* The rest of the write-side critical section is run with
400 * the mutex taken, so that lock->reader remains zero.
401 * There is no need to update self->locks_held.
402 */
403 }