parlib: Implement uthread mutexes with semaphores
[akaros.git] / user / parlib / mutex.c
1 /* Copyright (c) 2016-2017 Google, Inc.
2  * Barret Rhoden <brho@cs.berkeley.edu>
3  * See LICENSE for details. */
4
5 /* Generic Uthread Semaphores, Mutexes, CVs, and other synchronization
6  * functions.  2LSs implement their own sync objects (bottom of the file). */
7
8 #include <parlib/uthread.h>
9 #include <sys/queue.h>
10 #include <parlib/spinlock.h>
11 #include <malloc.h>
12
13
14 /************** Semaphores and Mutexes **************/
15
16 static void __uth_semaphore_init(void *arg)
17 {
18         struct uth_semaphore *sem = (struct uth_semaphore*)arg;
19
20         spin_pdr_init(&sem->lock);
21         sem->sync_obj = __uth_sync_alloc();
22         /* If we used a static initializer for a semaphore, count is already set.
23          * o/w it will be set by _alloc(). */
24 }
25
26 uth_semaphore_t *uth_semaphore_alloc(unsigned int count)
27 {
28         struct uth_semaphore *sem;
29
30         sem = malloc(sizeof(struct uth_semaphore));
31         assert(sem);
32         __uth_semaphore_init(sem);
33         sem->count = count;
34         /* The once is to make sure the object is initialized. */
35         parlib_set_ran_once(&sem->once_ctl);
36         return sem;
37 }
38
39 void uth_semaphore_free(uth_semaphore_t *sem)
40 {
41         /* Keep this in sync with uth_recurse_mutex_free(). */
42         __uth_sync_free(sem->sync_obj);
43         free(sem);
44 }
45
46 static void __semaphore_cb(struct uthread *uth, void *arg)
47 {
48         struct uth_semaphore *sem = (struct uth_semaphore*)arg;
49
50         /* We need to tell the 2LS that its thread blocked.  We need to do this
51          * before unlocking the sem, since as soon as we unlock, the sem could be
52          * released and our thread restarted.
53          *
54          * Also note the lock-ordering rule.  The sem lock is grabbed before any
55          * locks the 2LS might grab. */
56         uthread_has_blocked(uth, sem->sync_obj, UTH_EXT_BLK_MUTEX);
57         spin_pdr_unlock(&sem->lock);
58 }
59
60 void uth_semaphore_down(uth_semaphore_t *sem)
61 {
62         parlib_run_once(&sem->once_ctl, __uth_semaphore_init, sem);
63         spin_pdr_lock(&sem->lock);
64         if (sem->count > 0) {
65                 /* Only down if we got one.  This means a sem with no more counts is 0,
66                  * not negative (where -count == nr_waiters).  Doing it this way means
67                  * our timeout function works for sems and CVs. */
68                 sem->count--;
69                 spin_pdr_unlock(&sem->lock);
70                 return;
71         }
72         /* the unlock and sync enqueuing is done in the yield callback.  as always,
73          * we need to do this part in vcore context, since as soon as we unlock the
74          * uthread could restart.  (atomically yield and unlock). */
75         uthread_yield(TRUE, __semaphore_cb, sem);
76 }
77
78 bool uth_semaphore_trydown(uth_semaphore_t *sem)
79 {
80         bool ret = FALSE;
81
82         parlib_run_once(&sem->once_ctl, __uth_semaphore_init, sem);
83         spin_pdr_lock(&sem->lock);
84         if (sem->count > 0) {
85                 sem->count--;
86                 ret = TRUE;
87         }
88         spin_pdr_unlock(&sem->lock);
89         return ret;
90 }
91
92 void uth_semaphore_up(uth_semaphore_t *sem)
93 {
94         struct uthread *uth;
95
96         /* once-ing the 'up', unlike mtxs 'unlock', since sems can be special. */
97         parlib_run_once(&sem->once_ctl, __uth_semaphore_init, sem);
98         spin_pdr_lock(&sem->lock);
99         uth = __uth_sync_get_next(sem->sync_obj);
100         /* If there was a waiter, we pass our resource/count to them. */
101         if (!uth)
102                 sem->count++;
103         spin_pdr_unlock(&sem->lock);
104         if (uth)
105                 uthread_runnable(uth);
106 }
107
108 /* Takes a void * since it's called by parlib_run_once(), which enables us to
109  * statically initialize the mutex.  This init does everything not done by the
110  * static initializer.  Note we do not allow 'static' destruction.  (No one
111  * calls free). */
112 static void __uth_mutex_init(void *arg)
113 {
114         struct uth_semaphore *mtx = (struct uth_semaphore*)arg;
115
116         __uth_semaphore_init(mtx);
117         mtx->count = 1;
118 }
119
120 uth_mutex_t *uth_mutex_alloc(void)
121 {
122         struct uth_semaphore *mtx;
123
124         mtx = malloc(sizeof(struct uth_semaphore));
125         assert(mtx);
126         __uth_mutex_init(mtx);
127         parlib_set_ran_once(&mtx->once_ctl);
128         return mtx;
129 }
130
131 void uth_mutex_free(uth_mutex_t *mtx)
132 {
133         uth_semaphore_free(mtx);
134 }
135
136 void uth_mutex_lock(uth_mutex_t *mtx)
137 {
138         parlib_run_once(&mtx->once_ctl, __uth_mutex_init, mtx);
139         uth_semaphore_down(mtx);
140 }
141
142 bool uth_mutex_trylock(uth_mutex_t *mtx)
143 {
144         parlib_run_once(&mtx->once_ctl, __uth_mutex_init, mtx);
145         return uth_semaphore_trydown(mtx);
146 }
147
148 void uth_mutex_unlock(uth_mutex_t *mtx)
149 {
150         uth_semaphore_up(mtx);
151 }
152
153 /************** Recursive mutexes **************/
154
155 static void __uth_recurse_mutex_init(void *arg)
156 {
157         struct uth_recurse_mutex *r_mtx = (struct uth_recurse_mutex*)arg;
158
159         __uth_mutex_init(&r_mtx->mtx);
160         /* Since we always manually call __uth_mutex_init(), there's no reason to
161          * mess with the regular mutex's static initializer.  Just say it's been
162          * done. */
163         parlib_set_ran_once(&r_mtx->mtx.once_ctl);
164         r_mtx->lockholder = NULL;
165         r_mtx->count = 0;
166 }
167
168 uth_recurse_mutex_t *uth_recurse_mutex_alloc(void)
169 {
170         struct uth_recurse_mutex *r_mtx = malloc(sizeof(struct uth_recurse_mutex));
171
172         assert(r_mtx);
173         __uth_recurse_mutex_init(r_mtx);
174         parlib_set_ran_once(&r_mtx->once_ctl);
175         return r_mtx;
176 }
177
178 void uth_recurse_mutex_free(uth_recurse_mutex_t *r_mtx)
179 {
180         __uth_sync_free(r_mtx->mtx.sync_obj);
181         free(r_mtx);
182 }
183
184 void uth_recurse_mutex_lock(uth_recurse_mutex_t *r_mtx)
185 {
186         parlib_run_once(&r_mtx->once_ctl, __uth_recurse_mutex_init, r_mtx);
187         assert(!in_vcore_context());
188         assert(current_uthread);
189         /* We don't have to worry about races on current_uthread or count.  They are
190          * only written by the initial lockholder, and this check will only be true
191          * for the initial lockholder, which cannot concurrently call this function
192          * twice (a thread is single-threaded).
193          *
194          * A signal handler running for a thread should not attempt to grab a
195          * recursive mutex (that's probably a bug).  If we need to support that,
196          * we'll have to disable notifs temporarily. */
197         if (r_mtx->lockholder == current_uthread) {
198                 r_mtx->count++;
199                 return;
200         }
201         uth_mutex_lock(&r_mtx->mtx);
202         r_mtx->lockholder = current_uthread;
203         r_mtx->count = 1;
204 }
205
206 bool uth_recurse_mutex_trylock(uth_recurse_mutex_t *r_mtx)
207 {
208         bool ret;
209
210         parlib_run_once(&r_mtx->once_ctl, __uth_recurse_mutex_init, r_mtx);
211         assert(!in_vcore_context());
212         assert(current_uthread);
213         if (r_mtx->lockholder == current_uthread) {
214                 r_mtx->count++;
215                 return TRUE;
216         }
217         ret = uth_mutex_trylock(&r_mtx->mtx);
218         if (ret) {
219                 r_mtx->lockholder = current_uthread;
220                 r_mtx->count = 1;
221         }
222         return ret;
223 }
224
225 void uth_recurse_mutex_unlock(uth_recurse_mutex_t *r_mtx)
226 {
227         r_mtx->count--;
228         if (!r_mtx->count) {
229                 r_mtx->lockholder = NULL;
230                 uth_mutex_unlock(&r_mtx->mtx);
231         }
232 }
233
234
235 /************** Condition Variables **************/
236
237
238 static void __uth_cond_var_init(void *arg)
239 {
240         struct uth_cond_var *cv = (struct uth_cond_var*)arg;
241
242         spin_pdr_init(&cv->lock);
243         cv->sync_obj = __uth_sync_alloc();
244 }
245
246 uth_cond_var_t *uth_cond_var_alloc(void)
247 {
248         struct uth_cond_var *cv;
249
250         cv = malloc(sizeof(struct uth_cond_var));
251         assert(cv);
252         __uth_cond_var_init(cv);
253         parlib_set_ran_once(&cv->once_ctl);
254         return cv;
255 }
256
257 void uth_cond_var_free(uth_cond_var_t *cv)
258 {
259         __uth_sync_free(cv->sync_obj);
260         free(cv);
261 }
262
263 struct uth_cv_link {
264         struct uth_cond_var                     *cv;
265         struct uth_semaphore            *mtx;
266 };
267
268 static void __cv_wait_cb(struct uthread *uth, void *arg)
269 {
270         struct uth_cv_link *link = (struct uth_cv_link*)arg;
271         struct uth_cond_var *cv = link->cv;
272         struct uth_semaphore *mtx = link->mtx;
273
274         /* We need to tell the 2LS that its thread blocked.  We need to do this
275          * before unlocking the cv, since as soon as we unlock, the cv could be
276          * signalled and our thread restarted.
277          *
278          * Also note the lock-ordering rule.  The cv lock is grabbed before any
279          * locks the 2LS might grab. */
280         uthread_has_blocked(uth, cv->sync_obj, UTH_EXT_BLK_MUTEX);
281         spin_pdr_unlock(&cv->lock);
282         /* This looks dangerous, since both the CV and MTX could use the
283          * uth->sync_next TAILQ_ENTRY (or whatever the 2LS uses), but the uthread
284          * never sleeps on both at the same time.  We *hold* the mtx - we aren't
285          * *sleeping* on it.  Sleeping uses the sync_next.  Holding it doesn't.
286          *
287          * Next, consider what happens as soon as we unlock the CV.  Our thread
288          * could get woken up, and then immediately try to grab the mtx and go to
289          * sleep! (see below).  If that happens, the uthread is no longer sleeping
290          * on the CV, and the sync_next is free.  The invariant is that a uthread
291          * can only sleep on one sync_object at a time. */
292         uth_mutex_unlock(mtx);
293 }
294
295 /* Caller holds mtx.  We will 'atomically' release it and wait.  On return,
296  * caller holds mtx again.  Once our uth is on the CV's list, we can release the
297  * mtx without fear of missing a signal.
298  *
299  * POSIX refers to atomicity in this context as "atomically with respect to
300  * access by another thread to the mutex and then the condition variable"
301  *
302  * The idea is that we hold the mutex to protect some invariant; we check it,
303  * and decide to sleep.  Now we get on the list before releasing so that any
304  * changes to that invariant (e.g. a flag is now TRUE) happen after we're on the
305  * list, and so that we don't miss the signal.  To be more clear, the invariant
306  * in a basic wake-up flag scenario is: "whenever a flag is set from FALSE to
307  * TRUE, all waiters that saw FALSE are on the CV's waitqueue."  The mutex is
308  * required for this invariant.
309  *
310  * Note that signal/broadcasters do not *need* to hold the mutex, in general,
311  * but they do in the basic wake-up flag scenario.  If not, the race is this:
312  *
313  * Sleeper:                                                             Waker:
314  * -----------------------------------------------------------------
315  * Hold mutex
316  *   See flag is False
317  *   Decide to sleep
318  *                                                                              Set flag True
319  * PAUSE!                                                               Grab CV lock
320  *                                                                              See list is empty, unlock
321  *
322  *   Grab CV lock
323  *     Get put on list
324  *   Unlock CV lock
325  * Unlock mutex
326  * (Never wake up; we missed the signal)
327  *
328  * For those familiar with the kernel's CVs, we don't couple mutexes with CVs.
329  * cv_lock() actually grabs the spinlock inside the CV and uses *that* to
330  * protect the invariant.  The signallers always grab that lock, so the sleeper
331  * is not in danger of missing the signal.  The tradeoff is that the kernel CVs
332  * use a spinlock instead of a mutex for protecting its invariant; there might
333  * be some case that preferred blocking sync.
334  *
335  * The uthread CVs take a mutex, unlike the kernel CVs, to map more cleanly to
336  * POSIX CVs.  Maybe one approach or the other is a bad idea; we'll see.
337  *
338  * As far as lock ordering goes, once the sleeper holds the mutex and is on the
339  * CV's list, it can unlock in any order it wants.  However, unlocking a mutex
340  * actually requires grabbing its spinlock.  So as to not have a lock ordering
341  * between *spinlocks*, we let go of the CV's spinlock before unlocking the
342  * mutex.  There is an ordering between the mutex and the CV spinlock (mutex->cv
343  * spin), but there is no ordering between the mutex spin and cv spin.  And of
344  * course, we need to unlock the CV spinlock in the yield callback.
345  *
346  * Also note that we use the external API for the mutex operations.  A 2LS could
347  * have their own mutex ops but still use the generic cv ops. */
348 void uth_cond_var_wait(uth_cond_var_t *cv, uth_mutex_t *mtx)
349 {
350         struct uth_cv_link link;
351
352         parlib_run_once(&cv->once_ctl, __uth_cond_var_init, cv);
353         link.cv = cv;
354         link.mtx = mtx;
355         spin_pdr_lock(&cv->lock);
356         uthread_yield(TRUE, __cv_wait_cb, &link);
357         uth_mutex_lock(mtx);
358 }
359
360 void uth_cond_var_signal(uth_cond_var_t *cv)
361 {
362         struct uthread *uth;
363
364         parlib_run_once(&cv->once_ctl, __uth_cond_var_init, cv);
365         spin_pdr_lock(&cv->lock);
366         uth = __uth_sync_get_next(cv->sync_obj);
367         spin_pdr_unlock(&cv->lock);
368         if (uth)
369                 uthread_runnable(uth);
370 }
371
372 void uth_cond_var_broadcast(uth_cond_var_t *cv)
373 {
374         struct uth_tailq restartees = TAILQ_HEAD_INITIALIZER(restartees);
375         struct uthread *i, *safe;
376
377         parlib_run_once(&cv->once_ctl, __uth_cond_var_init, cv);
378         spin_pdr_lock(&cv->lock);
379         /* If this turns out to be slow or painful for 2LSs, we can implement a
380          * get_all or something (default used to use TAILQ_SWAP). */
381         while ((i = __uth_sync_get_next(cv->sync_obj))) {
382                 /* Once the uth is out of the sync obj, we can reuse sync_next. */
383                 TAILQ_INSERT_TAIL(&restartees, i, sync_next);
384         }
385         spin_pdr_unlock(&cv->lock);
386         /* Need the SAFE, since we can't touch the linkage once the uth could run */
387         TAILQ_FOREACH_SAFE(i, &restartees, sync_next, safe)
388                 uthread_runnable(i);
389 }
390
391
392 /************** Default Sync Obj Implementation **************/
393
394 static uth_sync_t uth_default_sync_alloc(void)
395 {
396         struct uth_tailq *tq;
397
398         tq = malloc(sizeof(struct uth_tailq));
399         assert(tq);
400         TAILQ_INIT(tq);
401         return (uth_sync_t)tq;
402 }
403
404 static void uth_default_sync_free(uth_sync_t sync)
405 {
406         struct uth_tailq *tq = (struct uth_tailq*)sync;
407
408         assert(TAILQ_EMPTY(tq));
409         free(tq);
410 }
411
412 static struct uthread *uth_default_sync_get_next(uth_sync_t sync)
413 {
414         struct uth_tailq *tq = (struct uth_tailq*)sync;
415         struct uthread *first;
416
417         first = TAILQ_FIRST(tq);
418         if (first)
419                 TAILQ_REMOVE(tq, first, sync_next);
420         return first;
421 }
422
423 static bool uth_default_sync_get_uth(uth_sync_t sync, struct uthread *uth)
424 {
425         struct uth_tailq *tq = (struct uth_tailq*)sync;
426         struct uthread *i;
427
428         TAILQ_FOREACH(i, tq, sync_next) {
429                 if (i == uth) {
430                         TAILQ_REMOVE(tq, i, sync_next);
431                         return TRUE;
432                 }
433         }
434         return FALSE;
435 }
436
437 /************** External uthread sync interface **************/
438
439 /* Called by the 2LS->has_blocked op, if they are using the default sync.*/
440 void __uth_default_sync_enqueue(struct uthread *uth, uth_sync_t sync)
441 {
442         struct uth_tailq *tq = (struct uth_tailq*)sync;
443
444         TAILQ_INSERT_TAIL(tq, uth, sync_next);
445 }
446
447 /* Called by 2LS-independent sync code when a sync object is created. */
448 uth_sync_t __uth_sync_alloc(void)
449 {
450         if (sched_ops->sync_alloc)
451                 return sched_ops->sync_alloc();
452         return uth_default_sync_alloc();
453 }
454
455 /* Called by 2LS-independent sync code when a sync object is destroyed. */
456 void __uth_sync_free(uth_sync_t sync)
457 {
458         if (sched_ops->sync_free) {
459                 sched_ops->sync_free(sync);
460                 return;
461         }
462         uth_default_sync_free(sync);
463 }
464
465 /* Called by 2LS-independent sync code when a thread needs to be woken. */
466 struct uthread *__uth_sync_get_next(uth_sync_t sync)
467 {
468         if (sched_ops->sync_get_next)
469                 return sched_ops->sync_get_next(sync);
470         return uth_default_sync_get_next(sync);
471 }
472
473 /* Called by 2LS-independent sync code when a specific thread needs to be woken.
474  * Returns TRUE if the uthread was blocked on the object, FALSE o/w. */
475 bool __uth_sync_get_uth(uth_sync_t sync, struct uthread *uth)
476 {
477         if (sched_ops->sync_get_uth)
478                 return sched_ops->sync_get_uth(sync, uth);
479         return uth_default_sync_get_uth(sync, uth);
480 }