1/* Copyright 2016 Google Inc.
2
3 Licensed under the Apache License, Version 2.0 (the "License");
4 you may not use this file except in compliance with the License.
5 You may obtain a copy of the License at
6
7 http://www.apache.org/licenses/LICENSE-2.0
8
9 Unless required by applicable law or agreed to in writing, software
10 distributed under the License is distributed on an "AS IS" BASIS,
11 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 See the License for the specific language governing permissions and
13 limitations under the License. */
14
15#include "nsync_cpp.h"
16#include "platform.h"
17#include "compiler.h"
18#include "cputype.h"
19#include "nsync.h"
20#include "dll.h"
21#include "sem.h"
22#include "wait_internal.h"
23#include "common.h"
24#include "atomic.h"
25
26NSYNC_CPP_START_
27
28/* Initialize *cv. */
29void nsync_cv_init (nsync_cv *cv) {
30 memset ((void *) cv, 0, sizeof (*cv));
31}
32
33/* Wake the cv waiters in the circular list pointed to by
34 to_wake_list, which may not be NULL. If the waiter is associated with a
35 nsync_mu, the "wakeup" may consist of transferring the waiters to the nsync_mu's
36 queue. Requires that every waiter is associated with the same mutex.
37 all_readers indicates whether all the waiters on the list are readers. */
38static void wake_waiters (nsync_dll_list_ to_wake_list, int all_readers) {
39 nsync_dll_element_ *p = NULL;
40 nsync_dll_element_ *next = NULL;
41 nsync_dll_element_ *first_waiter = nsync_dll_first_ (to_wake_list);
42 struct nsync_waiter_s *first_nw = DLL_NSYNC_WAITER (first_waiter);
43 waiter *first_w = NULL;
44 nsync_mu *pmu = NULL;
45 if ((first_nw->flags & NSYNC_WAITER_FLAG_MUCV) != 0) {
46 first_w = DLL_WAITER (first_waiter);
47 pmu = first_w->cv_mu;
48 }
49 if (pmu != NULL) { /* waiter is associated with the nsync_mu *pmu. */
50 /* We will transfer elements of to_wake_list to *pmu if all of:
51 - some thread holds the lock, and
52 - *pmu's spinlock is not held, and
53 - either *pmu cannot be acquired in the mode of the first
54 waiter, or there's more than one thread on to_wake_list
55 and not all are readers, and
56 - we acquire the spinlock on the first try.
57 The spinlock acquisition also marks *pmu as having waiters.
58 The requirement that some thread holds the lock ensures
59 that at least one of the transferred waiters will be woken.
60 */
61 uint32_t old_mu_word = ATM_LOAD (&pmu->word);
62 int first_cant_acquire = ((old_mu_word & first_w->l_type->zero_to_acquire) != 0);
63 next = nsync_dll_next_ (to_wake_list, first_waiter);
64 if ((old_mu_word&MU_ANY_LOCK) != 0 &&
65 (old_mu_word&MU_SPINLOCK) == 0 &&
66 (first_cant_acquire || (next != NULL && !all_readers)) &&
67 ATM_CAS_ACQ (&pmu->word, old_mu_word,
68 (old_mu_word|MU_SPINLOCK|MU_WAITING) &
69 ~MU_ALL_FALSE)) {
70
71 uint32_t set_on_release = 0;
72
73 /* For any waiter that should be transferred, rather
74 than woken, move it from to_wake_list to pmu->waiters. */
75 int first_is_writer = first_w->l_type == nsync_writer_type_;
76 int transferred_a_writer = 0;
77 int woke_areader = 0;
78 /* Transfer the first waiter iff it can't acquire *pmu. */
79 if (first_cant_acquire) {
80 to_wake_list = nsync_dll_remove_ (to_wake_list, first_waiter);
81 pmu->waiters = nsync_dll_make_last_in_list_ (pmu->waiters, first_waiter);
82 /* tell nsync_cv_wait_with_deadline() that we
83 moved the waiter to *pmu's queue. */
84 first_w->cv_mu = NULL;
85 /* first_nw.waiting is already 1, from being on
86 cv's waiter queue. */
87 transferred_a_writer = first_is_writer;
88 } else {
89 woke_areader = !first_is_writer;
90 }
91 /* Now process the other waiters. */
92 for (p = next; p != NULL; p = next) {
93 int p_is_writer;
94 struct nsync_waiter_s *p_nw = DLL_NSYNC_WAITER (p);
95 waiter *p_w = NULL;
96 if ((p_nw->flags & NSYNC_WAITER_FLAG_MUCV) != 0) {
97 p_w = DLL_WAITER (p);
98 }
99 next = nsync_dll_next_ (to_wake_list, p);
100 p_is_writer = (p_w != NULL &&
101 DLL_WAITER (p)->l_type == nsync_writer_type_);
102 /* We transfer this element if any of:
103 - the first waiter can't acquire *pmu, or
104 - the first waiter is a writer, or
105 - this element is a writer. */
106 if (p_w == NULL) {
107 /* wake non-native waiter */
108 } else if (first_cant_acquire || first_is_writer || p_is_writer) {
109 to_wake_list = nsync_dll_remove_ (to_wake_list, p);
110 pmu->waiters = nsync_dll_make_last_in_list_ (pmu->waiters, p);
111 /* tell nsync_cv_wait_with_deadline()
112 that we moved the waiter to *pmu's
113 queue. */
114 p_w->cv_mu = NULL;
115 /* p_nw->waiting is already 1, from
116 being on cv's waiter queue. */
117 transferred_a_writer = transferred_a_writer || p_is_writer;
118 } else {
119 woke_areader = woke_areader || !p_is_writer;
120 }
121 }
122
123 /* Claim a waiting writer if we transferred one, except if we woke readers,
124 in which case we want those readers to be able to acquire immediately. */
125 if (transferred_a_writer && !woke_areader) {
126 set_on_release |= MU_WRITER_WAITING;
127 }
128
129 /* release *pmu's spinlock (MU_WAITING was set by CAS above) */
130 old_mu_word = ATM_LOAD (&pmu->word);
131 while (!ATM_CAS_REL (&pmu->word, old_mu_word,
132 (old_mu_word|set_on_release) & ~MU_SPINLOCK)) {
133 old_mu_word = ATM_LOAD (&pmu->word);
134 }
135 }
136 }
137
138 /* Wake any waiters we didn't manage to enqueue on the mu. */
139 for (p = nsync_dll_first_ (to_wake_list); p != NULL; p = next) {
140 struct nsync_waiter_s *p_nw = DLL_NSYNC_WAITER (p);
141 next = nsync_dll_next_ (to_wake_list, p);
142 to_wake_list = nsync_dll_remove_ (to_wake_list, p);
143 /* Wake the waiter. */
144 ATM_STORE_REL (&p_nw->waiting, 0); /* release store */
145 nsync_mu_semaphore_v (p_nw->sem);
146 }
147}
148
149/* ------------------------------------------ */
150
151/* Versions of nsync_mu_lock() and nsync_mu_unlock() that take "void *"
152 arguments, to avoid call through a function pointer of a different type,
153 which is undefined. */
154static void void_mu_lock (void *mu) {
155 nsync_mu_lock ((nsync_mu *) mu);
156}
157static void void_mu_unlock (void *mu) {
158 nsync_mu_unlock ((nsync_mu *) mu);
159}
160
161/* Atomically release *pmu (which must be held on entry)
162 and block the calling thread on *pcv. Then wait until awakened by a
163 call to nsync_cv_signal() or nsync_cv_broadcast() (or a spurious wakeup), or by the time
164 reaching abs_deadline, or by cancel_note being notified. In all cases,
165 reacquire *pmu, and return the reason for the call returned (0, ETIMEDOUT,
166 or ECANCELED). Callers should abs_deadline==nsync_time_no_deadline for no
167 deadline, and cancel_note==NULL for no cancellation. nsync_cv_wait_with_deadline()
168 should be used in a loop, as with all Mesa-style condition variables. See
169 examples above.
170
171 There are two reasons for using an absolute deadline, rather than a relative
172 timeout---these are why pthread_cond_timedwait() also uses an absolute
173 deadline. First, condition variable waits have to be used in a loop; with
174 an absolute times, the deadline does not have to be recomputed on each
175 iteration. Second, in most real programmes, some activity (such as an RPC
176 to a server, or when guaranteeing response time in a UI), there is a
177 deadline imposed by the specification or the caller/user; relative delays
178 can shift arbitrarily with scheduling delays, and so after multiple waits
179 might extend beyond the expected deadline. Relative delays tend to be more
180 convenient mostly in tests and trivial examples than they are in real
181 programmes. */
182int nsync_cv_wait_with_deadline_generic (nsync_cv *pcv, void *pmu,
183 void (*lock) (void *), void (*unlock) (void *),
184 nsync_time abs_deadline,
185 nsync_note cancel_note) {
186 nsync_mu *cv_mu = NULL;
187 int is_reader_mu;
188 uint32_t old_word;
189 uint32_t remove_count;
190 int sem_outcome;
191 unsigned attempts;
192 int outcome = 0;
193 waiter *w;
194 IGNORE_RACES_START ();
195 w = nsync_waiter_new_ ();
196 ATM_STORE (&w->nw.waiting, 1);
197 w->cond.f = NULL; /* Not using a conditional critical section. */
198 w->cond.v = NULL;
199 w->cond.eq = NULL;
200 if (lock == &void_mu_lock ||
201 lock == (void (*) (void *)) &nsync_mu_lock ||
202 lock == (void (*) (void *)) &nsync_mu_rlock) {
203 cv_mu = (nsync_mu *) pmu;
204 }
205 w->cv_mu = cv_mu; /* If *pmu is an nsync_mu, record its address, else record NULL. */
206 is_reader_mu = 0; /* If true, an nsync_mu in reader mode. */
207 if (cv_mu == NULL) {
208 w->l_type = NULL;
209 } else {
210 uint32_t old_mu_word = ATM_LOAD (&cv_mu->word);
211 int is_writer = (old_mu_word & MU_WHELD_IF_NON_ZERO) != 0;
212 int is_reader = (old_mu_word & MU_RHELD_IF_NON_ZERO) != 0;
213 if (is_writer) {
214 if (is_reader) {
215 nsync_panic_ ("mu held in reader and writer mode simultaneously "
216 "on entry to nsync_cv_wait_with_deadline()\n");
217 }
218 w->l_type = nsync_writer_type_;
219 } else if (is_reader) {
220 w->l_type = nsync_reader_type_;
221 is_reader_mu = 1;
222 } else {
223 nsync_panic_ ("mu not held on entry to nsync_cv_wait_with_deadline()\n");
224 }
225 }
226
227 /* acquire spinlock, set non-empty */
228 old_word = nsync_spin_test_and_set_ (&pcv->word, CV_SPINLOCK, CV_SPINLOCK|CV_NON_EMPTY, 0);
229 pcv->waiters = nsync_dll_make_last_in_list_ (pcv->waiters, &w->nw.q);
230 remove_count = ATM_LOAD (&w->remove_count);
231 /* Release the spin lock. */
232 ATM_STORE_REL (&pcv->word, old_word|CV_NON_EMPTY); /* release store */
233
234 /* Release *pmu. */
235 if (is_reader_mu) {
236 nsync_mu_runlock (cv_mu);
237 } else {
238 (*unlock) (pmu);
239 }
240
241 /* wait until awoken or a timeout. */
242 sem_outcome = 0;
243 attempts = 0;
244 while (ATM_LOAD_ACQ (&w->nw.waiting) != 0) { /* acquire load */
245 if (sem_outcome == 0) {
246 sem_outcome = nsync_sem_wait_with_cancel_ (w, abs_deadline, cancel_note);
247 }
248
249 if (sem_outcome != 0 && ATM_LOAD (&w->nw.waiting) != 0) {
250 /* A timeout or cancellation occurred, and no wakeup.
251 Acquire *pcv's spinlock, and confirm. */
252 old_word = nsync_spin_test_and_set_ (&pcv->word, CV_SPINLOCK,
253 CV_SPINLOCK, 0);
254 /* Check that w wasn't removed from the queue after we
255 checked above, but before we acquired the spinlock.
256 The test of remove_count confirms that the waiter *w
257 is still governed by *pcv's spinlock; otherwise, some
258 other thread is about to set w.waiting==0. */
259 if (ATM_LOAD (&w->nw.waiting) != 0) {
260 if (remove_count == ATM_LOAD (&w->remove_count)) {
261 uint32_t old_value;
262 /* still in cv waiter queue */
263 /* Not woken, so remove *w from cv
264 queue, and declare a
265 timeout/cancellation. */
266 outcome = sem_outcome;
267 pcv->waiters = nsync_dll_remove_ (pcv->waiters,
268 &w->nw.q);
269 do {
270 old_value = ATM_LOAD (&w->remove_count);
271 } while (!ATM_CAS (&w->remove_count, old_value, old_value+1));
272 if (nsync_dll_is_empty_ (pcv->waiters)) {
273 old_word &= ~(CV_NON_EMPTY);
274 }
275 ATM_STORE_REL (&w->nw.waiting, 0); /* release store */
276 }
277 }
278 /* Release spinlock. */
279 ATM_STORE_REL (&pcv->word, old_word); /* release store */
280 }
281
282 if (ATM_LOAD (&w->nw.waiting) != 0) {
283 /* The delay here causes this thread ultimately to
284 yield to another that has dequeued this thread, but
285 has not yet set the waiting field to zero; a
286 cancellation or timeout may prevent this thread
287 from blocking above on the semaphore. */
288 attempts = nsync_spin_delay_ (attempts);
289 }
290 }
291
292 if (cv_mu != NULL && w->cv_mu == NULL) { /* waiter was moved to *pmu's queue, and woken. */
293 /* Requeue on *pmu using existing waiter struct; current thread
294 is the designated waker. */
295 nsync_mu_lock_slow_ (cv_mu, w, MU_DESIG_WAKER, w->l_type);
296 nsync_waiter_free_ (w);
297 } else {
298 /* Traditional case: We've woken from the cv, and need to reacquire *pmu. */
299 nsync_waiter_free_ (w);
300 if (is_reader_mu) {
301 nsync_mu_rlock (cv_mu);
302 } else {
303 (*lock) (pmu);
304 }
305 }
306 IGNORE_RACES_END ();
307 return (outcome);
308}
309
310/* Wake at least one thread if any are currently blocked on *pcv. If
311 the chosen thread is a reader on an nsync_mu, wake all readers and, if
312 possible, a writer. */
313void nsync_cv_signal (nsync_cv *pcv) {
314 IGNORE_RACES_START ();
315 if ((ATM_LOAD_ACQ (&pcv->word) & CV_NON_EMPTY) != 0) { /* acquire load */
316 nsync_dll_list_ to_wake_list = NULL; /* waiters that we will wake */
317 int all_readers = 0;
318 /* acquire spinlock */
319 uint32_t old_word = nsync_spin_test_and_set_ (&pcv->word, CV_SPINLOCK,
320 CV_SPINLOCK, 0);
321 if (!nsync_dll_is_empty_ (pcv->waiters)) {
322 /* Point to first waiter that enqueued itself, and
323 detach it from all others. */
324 struct nsync_waiter_s *first_nw;
325 nsync_dll_element_ *first = nsync_dll_first_ (pcv->waiters);
326 pcv->waiters = nsync_dll_remove_ (pcv->waiters, first);
327 first_nw = DLL_NSYNC_WAITER (first);
328 if ((first_nw->flags & NSYNC_WAITER_FLAG_MUCV) != 0) {
329 uint32_t old_value;
330 do {
331 old_value =
332 ATM_LOAD (&DLL_WAITER (first)->remove_count);
333 } while (!ATM_CAS (&DLL_WAITER (first)->remove_count,
334 old_value, old_value+1));
335 }
336 to_wake_list = nsync_dll_make_last_in_list_ (to_wake_list, first);
337 if ((first_nw->flags & NSYNC_WAITER_FLAG_MUCV) != 0 &&
338 DLL_WAITER (first)->l_type == nsync_reader_type_) {
339 int woke_writer;
340 /* If the first waiter is a reader, wake all readers, and
341 if it's possible, one writer. This allows reader-regions
342 to be added to a monitor without invalidating code in which
343 a client has optimized broadcast calls by converting them to
344 signal calls. In particular, we wake a writer when waking
345 readers because the readers will not invalidate the condition
346 that motivated the client to call nsync_cv_signal(). But we
347 wake at most one writer because the first writer may invalidate
348 the condition; the client is expecting only one writer to be
349 able make use of the wakeup, or he would have called
350 nsync_cv_broadcast(). */
351 nsync_dll_element_ *p = NULL;
352 nsync_dll_element_ *next = NULL;
353 all_readers = 1;
354 woke_writer = 0;
355 for (p = nsync_dll_first_ (pcv->waiters); p != NULL; p = next) {
356 struct nsync_waiter_s *p_nw = DLL_NSYNC_WAITER (p);
357 int should_wake;
358 next = nsync_dll_next_ (pcv->waiters, p);
359 should_wake = 0;
360 if ((p_nw->flags & NSYNC_WAITER_FLAG_MUCV) != 0 &&
361 DLL_WAITER (p)->l_type == nsync_reader_type_) {
362 should_wake = 1;
363 } else if (!woke_writer) {
364 woke_writer = 1;
365 all_readers = 0;
366 should_wake = 1;
367 }
368 if (should_wake) {
369 pcv->waiters = nsync_dll_remove_ (pcv->waiters, p);
370 if ((p_nw->flags & NSYNC_WAITER_FLAG_MUCV) != 0) {
371 uint32_t old_value;
372 do {
373 old_value = ATM_LOAD (
374 &DLL_WAITER (p)->remove_count);
375 } while (!ATM_CAS (&DLL_WAITER (p)->remove_count,
376 old_value, old_value+1));
377 }
378 to_wake_list = nsync_dll_make_last_in_list_ (
379 to_wake_list, p);
380 }
381 }
382 }
383 if (nsync_dll_is_empty_ (pcv->waiters)) {
384 old_word &= ~(CV_NON_EMPTY);
385 }
386 }
387 /* Release spinlock. */
388 ATM_STORE_REL (&pcv->word, old_word); /* release store */
389 if (!nsync_dll_is_empty_ (to_wake_list)) {
390 wake_waiters (to_wake_list, all_readers);
391 }
392 }
393 IGNORE_RACES_END ();
394}
395
396/* Wake all threads currently blocked on *pcv. */
397void nsync_cv_broadcast (nsync_cv *pcv) {
398 IGNORE_RACES_START ();
399 if ((ATM_LOAD_ACQ (&pcv->word) & CV_NON_EMPTY) != 0) { /* acquire load */
400 nsync_dll_element_ *p;
401 nsync_dll_element_ *next;
402 int all_readers;
403 nsync_dll_list_ to_wake_list = NULL; /* waiters that we will wake */
404 /* acquire spinlock */
405 nsync_spin_test_and_set_ (&pcv->word, CV_SPINLOCK, CV_SPINLOCK, 0);
406 p = NULL;
407 next = NULL;
408 all_readers = 1;
409 /* Wake entire waiter list, which we leave empty. */
410 for (p = nsync_dll_first_ (pcv->waiters); p != NULL; p = next) {
411 struct nsync_waiter_s *p_nw = DLL_NSYNC_WAITER (p);
412 next = nsync_dll_next_ (pcv->waiters, p);
413 all_readers = all_readers && (p_nw->flags & NSYNC_WAITER_FLAG_MUCV) != 0 &&
414 (DLL_WAITER (p)->l_type == nsync_reader_type_);
415 pcv->waiters = nsync_dll_remove_ (pcv->waiters, p);
416 if ((p_nw->flags & NSYNC_WAITER_FLAG_MUCV) != 0) {
417 uint32_t old_value;
418 do {
419 old_value = ATM_LOAD (&DLL_WAITER (p)->remove_count);
420 } while (!ATM_CAS (&DLL_WAITER (p)->remove_count,
421 old_value, old_value+1));
422 }
423 to_wake_list = nsync_dll_make_last_in_list_ (to_wake_list, p);
424 }
425 /* Release spinlock and mark queue empty. */
426 ATM_STORE_REL (&pcv->word, 0); /* release store */
427 if (!nsync_dll_is_empty_ (to_wake_list)) { /* Wake them. */
428 wake_waiters (to_wake_list, all_readers);
429 }
430 }
431 IGNORE_RACES_END ();
432}
433
434/* Wait with deadline, using an nsync_mu. */
435int nsync_cv_wait_with_deadline (nsync_cv *pcv, nsync_mu *pmu,
436 nsync_time abs_deadline,
437 nsync_note cancel_note) {
438 return (nsync_cv_wait_with_deadline_generic (pcv, pmu, &void_mu_lock,
439 &void_mu_unlock,
440 abs_deadline, cancel_note));
441}
442
443/* Atomically release *pmu and block the caller on *pcv. Wait
444 until awakened by a call to nsync_cv_signal() or nsync_cv_broadcast(), or a spurious
445 wakeup. Then reacquires *pmu, and return. The call is equivalent to a call
446 to nsync_cv_wait_with_deadline() with abs_deadline==nsync_time_no_deadline, and a NULL
447 cancel_note. It should be used in a loop, as with all standard Mesa-style
448 condition variables. See examples above. */
449void nsync_cv_wait (nsync_cv *pcv, nsync_mu *pmu) {
450 nsync_cv_wait_with_deadline (pcv, pmu, nsync_time_no_deadline, NULL);
451}
452
453static nsync_time cv_ready_time (void *v UNUSED, struct nsync_waiter_s *nw) {
454 nsync_time r;
455 r = (nw == NULL || ATM_LOAD_ACQ (&nw->waiting) != 0? nsync_time_no_deadline : nsync_time_zero);
456 return (r);
457}
458
459static int cv_enqueue (void *v, struct nsync_waiter_s *nw) {
460 nsync_cv *pcv = (nsync_cv *) v;
461 /* acquire spinlock */
462 uint32_t old_word = nsync_spin_test_and_set_ (&pcv->word, CV_SPINLOCK, CV_SPINLOCK, 0);
463 pcv->waiters = nsync_dll_make_last_in_list_ (pcv->waiters, &nw->q);
464 ATM_STORE (&nw->waiting, 1);
465 /* Release spinlock. */
466 ATM_STORE_REL (&pcv->word, old_word | CV_NON_EMPTY); /* release store */
467 return (1);
468}
469
470static int cv_dequeue (void *v, struct nsync_waiter_s *nw) {
471 nsync_cv *pcv = (nsync_cv *) v;
472 int was_queued = 0;
473 /* acquire spinlock */
474 uint32_t old_word = nsync_spin_test_and_set_ (&pcv->word, CV_SPINLOCK, CV_SPINLOCK, 0);
475 if (ATM_LOAD_ACQ (&nw->waiting) != 0) {
476 pcv->waiters = nsync_dll_remove_ (pcv->waiters, &nw->q);
477 ATM_STORE (&nw->waiting, 0);
478 was_queued = 1;
479 }
480 if (nsync_dll_is_empty_ (pcv->waiters)) {
481 old_word &= ~(CV_NON_EMPTY);
482 }
483 /* Release spinlock. */
484 ATM_STORE_REL (&pcv->word, old_word); /* release store */
485 return (was_queued);
486}
487
488const struct nsync_waitable_funcs_s nsync_cv_waitable_funcs = {
489 &cv_ready_time,
490 &cv_enqueue,
491 &cv_dequeue
492};
493
494NSYNC_CPP_END_
495