(index<- ) ./libsync/raw.rs
git branch: * master 5200215 auto merge of #14035 : alexcrichton/rust/experimental, r=huonw
modified: Fri May 9 13:02:28 2014
2 // file at the top-level directory of this distribution and at
3 // http://rust-lang.org/COPYRIGHT.
4 //
5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
8 // option. This file may not be copied, modified, or distributed
9 // except according to those terms.
10
11 //! Raw concurrency primitives you know and love.
12 //!
13 //! These primitives are not recommended for general use, but are provided for
14 //! flavorful use-cases. It is recommended to use the types at the top of the
15 //! `sync` crate which wrap values directly and provide safer abstractions for
16 //! containing data.
17
18 use std::cast;
19 use std::kinds::marker;
20 use std::mem::replace;
21 use std::sync::atomics;
22 use std::unstable::finally::Finally;
23
24 use mutex;
25
26 /****************************************************************************
27 * Internals
28 ****************************************************************************/
29
30 // Each waiting task receives on one of these.
31 type WaitEnd = Receiver<()>;
32 type SignalEnd = Sender<()>;
33 // A doubly-ended queue of waiting tasks.
34 struct WaitQueue {
35 head: Receiver<SignalEnd>,
36 tail: Sender<SignalEnd>,
37 }
38
39 impl WaitQueue {
40 fn new() -> WaitQueue {
41 let (block_tail, block_head) = channel();
42 WaitQueue { head: block_head, tail: block_tail }
43 }
44
45 // Signals one live task from the queue.
46 fn signal(&self) -> bool {
47 match self.head.try_recv() {
48 Ok(ch) => {
49 // Send a wakeup signal. If the waiter was killed, its port will
50 // have closed. Keep trying until we get a live task.
51 if ch.send_opt(()).is_ok() {
52 true
53 } else {
54 self.signal()
55 }
56 }
57 _ => false
58 }
59 }
60
61 fn broadcast(&self) -> uint {
62 let mut count = 0;
63 loop {
64 match self.head.try_recv() {
65 Ok(ch) => {
66 if ch.send_opt(()).is_ok() {
67 count += 1;
68 }
69 }
70 _ => break
71 }
72 }
73 count
74 }
75
76 fn wait_end(&self) -> WaitEnd {
77 let (signal_end, wait_end) = channel();
78 self.tail.send(signal_end);
79 wait_end
80 }
81 }
82
83 // The building-block used to make semaphores, mutexes, and rwlocks.
84 struct Sem<Q> {
85 lock: mutex::Mutex,
86 // n.b, we need Sem to be `Share`, but the WaitQueue type is not send/share
87 // (for good reason). We have an internal invariant on this semaphore,
88 // however, that the queue is never accessed outside of a locked
89 // context. For this reason, we shove these behind a pointer which will
90 // be inferred to be `Share`.
91 //
92 // FIXME: this requires an extra allocation, which is bad.
93 inner: *()
94 }
95
96 struct SemInner<Q> {
97 count: int,
98 waiters: WaitQueue,
99 // Can be either unit or another waitqueue. Some sems shouldn't come with
100 // a condition variable attached, others should.
101 blocked: Q,
102 }
103
104 #[must_use]
105 struct SemGuard<'a, Q> {
106 sem: &'a Sem<Q>,
107 }
108
109 impl<Q: Send> Sem<Q> {
110 fn new(count: int, q: Q) -> Sem<Q> {
111 let inner = unsafe {
112 cast::transmute(box SemInner {
113 waiters: WaitQueue::new(),
114 count: count,
115 blocked: q,
116 })
117 };
118 Sem {
119 lock: mutex::Mutex::new(),
120 inner: inner,
121 }
122 }
123
124 unsafe fn with(&self, f: |&mut SemInner<Q>|) {
125 let _g = self.lock.lock();
126 f(&mut *(self.inner as *mut SemInner<Q>))
127 }
128
129 pub fn acquire(&self) {
130 unsafe {
131 let mut waiter_nobe = None;
132 self.with(|state| {
133 state.count -= 1;
134 if state.count < 0 {
135 // Create waiter nobe, enqueue ourself, and tell
136 // outer scope we need to block.
137 waiter_nobe = Some(state.waiters.wait_end());
138 }
139 });
140 // Uncomment if you wish to test for sem races. Not
141 // valgrind-friendly.
142 /* for _ in range(0, 1000) { task::deschedule(); } */
143 // Need to wait outside the exclusive.
144 if waiter_nobe.is_some() {
145 let _ = waiter_nobe.unwrap().recv();
146 }
147 }
148 }
149
150 pub fn release(&self) {
151 unsafe {
152 self.with(|state| {
153 state.count += 1;
154 if state.count <= 0 {
155 state.waiters.signal();
156 }
157 })
158 }
159 }
160
161 pub fn access<'a>(&'a self) -> SemGuard<'a, Q> {
162 self.acquire();
163 SemGuard { sem: self }
164 }
165 }
166
167 #[unsafe_destructor]
168 impl<Q: Send> Drop for Sem<Q> {
169 fn drop(&mut self) {
170 let _waiters: Box<SemInner<Q>> = unsafe {
171 cast::transmute(self.inner)
172 };
173 self.inner = 0 as *();
174 }
175 }
176
177 #[unsafe_destructor]
178 impl<'a, Q: Send> Drop for SemGuard<'a, Q> {
179 fn drop(&mut self) {
180 self.sem.release();
181 }
182 }
183
184 impl Sem<Vec<WaitQueue>> {
185 fn new_and_signal(count: int, num_condvars: uint) -> Sem<Vec<WaitQueue>> {
186 let mut queues = Vec::new();
187 for _ in range(0, num_condvars) { queues.push(WaitQueue::new()); }
188 Sem::new(count, queues)
189 }
190
191 // The only other places that condvars get built are rwlock.write_cond()
192 // and rwlock_write_mode.
193 pub fn access_cond<'a>(&'a self) -> SemCondGuard<'a> {
194 SemCondGuard {
195 guard: self.access(),
196 cvar: Condvar { sem: self, order: Nothing, nocopy: marker::NoCopy },
197 }
198 }
199 }
200
201 // FIXME(#3598): Want to use an Option down below, but we need a custom enum
202 // that's not polymorphic to get around the fact that lifetimes are invariant
203 // inside of type parameters.
204 enum ReacquireOrderLock<'a> {
205 Nothing, // c.c
206 Just(&'a Semaphore),
207 }
208
209 /// A mechanism for atomic-unlock-and-deschedule blocking and signalling.
210 pub struct Condvar<'a> {
211 // The 'Sem' object associated with this condvar. This is the one that's
212 // atomically-unlocked-and-descheduled upon and reacquired during wakeup.
213 sem: &'a Sem<Vec<WaitQueue> >,
214 // This is (can be) an extra semaphore which is held around the reacquire
215 // operation on the first one. This is only used in cvars associated with
216 // rwlocks, and is needed to ensure that, when a downgrader is trying to
217 // hand off the access lock (which would be the first field, here), a 2nd
218 // writer waking up from a cvar wait can't race with a reader to steal it,
219 // See the comment in write_cond for more detail.
220 order: ReacquireOrderLock<'a>,
221 // Make sure condvars are non-copyable.
222 nocopy: marker::NoCopy,
223 }
224
225 impl<'a> Condvar<'a> {
226 /// Atomically drop the associated lock, and block until a signal is sent.
227 ///
228 /// # Failure
229 ///
230 /// A task which is killed while waiting on a condition variable will wake
231 /// up, fail, and unlock the associated lock as it unwinds.
232 pub fn wait(&self) { self.wait_on(0) }
233
234 /// As wait(), but can specify which of multiple condition variables to
235 /// wait on. Only a signal_on() or broadcast_on() with the same condvar_id
236 /// will wake this thread.
237 ///
238 /// The associated lock must have been initialised with an appropriate
239 /// number of condvars. The condvar_id must be between 0 and num_condvars-1
240 /// or else this call will fail.
241 ///
242 /// wait() is equivalent to wait_on(0).
243 pub fn wait_on(&self, condvar_id: uint) {
244 let mut wait_end = None;
245 let mut out_of_bounds = None;
246 // Release lock, 'atomically' enqueuing ourselves in so doing.
247 unsafe {
248 self.sem.with(|state| {
249 if condvar_id < state.blocked.len() {
250 // Drop the lock.
251 state.count += 1;
252 if state.count <= 0 {
253 state.waiters.signal();
254 }
255 // Create waiter nobe, and enqueue ourself to
256 // be woken up by a signaller.
257 wait_end = Some(state.blocked.get(condvar_id).wait_end());
258 } else {
259 out_of_bounds = Some(state.blocked.len());
260 }
261 })
262 }
263
264 // If deschedule checks start getting inserted anywhere, we can be
265 // killed before or after enqueueing.
266 check_cvar_bounds(out_of_bounds, condvar_id, "cond.wait_on()", || {
267 // Unconditionally "block". (Might not actually block if a
268 // signaller already sent -- I mean 'unconditionally' in contrast
269 // with acquire().)
270 (|| {
271 let _ = wait_end.take_unwrap().recv();
272 }).finally(|| {
273 // Reacquire the condvar.
274 match self.order {
275 Just(lock) => {
276 let _g = lock.access();
277 self.sem.acquire();
278 }
279 Nothing => self.sem.acquire(),
280 }
281 })
282 })
283 }
284
285 /// Wake up a blocked task. Returns false if there was no blocked task.
286 pub fn signal(&self) -> bool { self.signal_on(0) }
287
288 /// As signal, but with a specified condvar_id. See wait_on.
289 pub fn signal_on(&self, condvar_id: uint) -> bool {
290 unsafe {
291 let mut out_of_bounds = None;
292 let mut result = false;
293 self.sem.with(|state| {
294 if condvar_id < state.blocked.len() {
295 result = state.blocked.get(condvar_id).signal();
296 } else {
297 out_of_bounds = Some(state.blocked.len());
298 }
299 });
300 check_cvar_bounds(out_of_bounds,
301 condvar_id,
302 "cond.signal_on()",
303 || result)
304 }
305 }
306
307 /// Wake up all blocked tasks. Returns the number of tasks woken.
308 pub fn broadcast(&self) -> uint { self.broadcast_on(0) }
309
310 /// As broadcast, but with a specified condvar_id. See wait_on.
311 pub fn broadcast_on(&self, condvar_id: uint) -> uint {
312 let mut out_of_bounds = None;
313 let mut queue = None;
314 unsafe {
315 self.sem.with(|state| {
316 if condvar_id < state.blocked.len() {
317 // To avoid :broadcast_heavy, we make a new waitqueue,
318 // swap it out with the old one, and broadcast on the
319 // old one outside of the little-lock.
320 queue = Some(replace(state.blocked.get_mut(condvar_id),
321 WaitQueue::new()));
322 } else {
323 out_of_bounds = Some(state.blocked.len());
324 }
325 });
326 check_cvar_bounds(out_of_bounds,
327 condvar_id,
328 "cond.signal_on()",
329 || {
330 queue.take_unwrap().broadcast()
331 })
332 }
333 }
334 }
335
336 // Checks whether a condvar ID was out of bounds, and fails if so, or does
337 // something else next on success.
338 #[inline]
339 fn check_cvar_bounds<U>(
340 out_of_bounds: Option<uint>,
341 id: uint,
342 act: &str,
343 blk: || -> U)
344 -> U {
345 match out_of_bounds {
346 Some(0) =>
347 fail!("{} with illegal ID {} - this lock has no condvars!", act, id),
348 Some(length) =>
349 fail!("{} with illegal ID {} - ID must be less than {}", act, id, length),
350 None => blk()
351 }
352 }
353
354 #[must_use]
355 struct SemCondGuard<'a> {
356 guard: SemGuard<'a, Vec<WaitQueue>>,
357 cvar: Condvar<'a>,
358 }
359
360 /****************************************************************************
361 * Semaphores
362 ****************************************************************************/
363
364 /// A counting, blocking, bounded-waiting semaphore.
365 pub struct Semaphore {
366 sem: Sem<()>,
367 }
368
369 /// An RAII guard used to represent an acquired resource to a semaphore. When
370 /// dropped, this value will release the resource back to the semaphore.
371 #[must_use]
372 pub struct SemaphoreGuard<'a> {
373 guard: SemGuard<'a, ()>,
374 }
375
376 impl Semaphore {
377 /// Create a new semaphore with the specified count.
378 pub fn new(count: int) -> Semaphore {
379 Semaphore { sem: Sem::new(count, ()) }
380 }
381
382 /// Acquire a resource represented by the semaphore. Blocks if necessary
383 /// until resource(s) become available.
384 pub fn acquire(&self) { self.sem.acquire() }
385
386 /// Release a held resource represented by the semaphore. Wakes a blocked
387 /// contending task, if any exist. Won't block the caller.
388 pub fn release(&self) { self.sem.release() }
389
390 /// Acquire a resource of this semaphore, returning an RAII guard which will
391 /// release the resource when dropped.
392 pub fn access<'a>(&'a self) -> SemaphoreGuard<'a> {
393 SemaphoreGuard { guard: self.sem.access() }
394 }
395 }
396
397 /****************************************************************************
398 * Mutexes
399 ****************************************************************************/
400
401 /// A blocking, bounded-waiting, mutual exclusion lock with an associated
402 /// FIFO condition variable.
403 ///
404 /// # Failure
405 /// A task which fails while holding a mutex will unlock the mutex as it
406 /// unwinds.
407 pub struct Mutex {
408 sem: Sem<Vec<WaitQueue>>,
409 }
410
411 /// An RAII structure which is used to gain access to a mutex's condition
412 /// variable. Additionally, when a value of this type is dropped, the
413 /// corresponding mutex is also unlocked.
414 #[must_use]
415 pub struct MutexGuard<'a> {
416 guard: SemGuard<'a, Vec<WaitQueue>>,
417 /// Inner condition variable which is connected to the outer mutex, and can
418 /// be used for atomic-unlock-and-deschedule.
419 pub cond: Condvar<'a>,
420 }
421
422 impl Mutex {
423 /// Create a new mutex, with one associated condvar.
424 pub fn new() -> Mutex { Mutex::new_with_condvars(1) }
425
426 /// Create a new mutex, with a specified number of associated condvars. This
427 /// will allow calling wait_on/signal_on/broadcast_on with condvar IDs
428 /// between 0 and num_condvars-1. (If num_condvars is 0, lock_cond will be
429 /// allowed but any operations on the condvar will fail.)
430 pub fn new_with_condvars(num_condvars: uint) -> Mutex {
431 Mutex { sem: Sem::new_and_signal(1, num_condvars) }
432 }
433
434 /// Acquires ownership of this mutex, returning an RAII guard which will
435 /// unlock the mutex when dropped. The associated condition variable can
436 /// also be accessed through the returned guard.
437 pub fn lock<'a>(&'a self) -> MutexGuard<'a> {
438 let SemCondGuard { guard, cvar } = self.sem.access_cond();
439 MutexGuard { guard: guard, cond: cvar }
440 }
441 }
442
443 /****************************************************************************
444 * Reader-writer locks
445 ****************************************************************************/
446
447 // NB: Wikipedia - Readers-writers_problem#The_third_readers-writers_problem
448
449 /// A blocking, no-starvation, reader-writer lock with an associated condvar.
450 ///
451 /// # Failure
452 ///
453 /// A task which fails while holding an rwlock will unlock the rwlock as it
454 /// unwinds.
455 pub struct RWLock {
456 order_lock: Semaphore,
457 access_lock: Sem<Vec<WaitQueue>>,
458
459 // The only way the count flag is ever accessed is with xadd. Since it is
460 // a read-modify-write operation, multiple xadds on different cores will
461 // always be consistent with respect to each other, so a monotonic/relaxed
462 // consistency ordering suffices (i.e., no extra barriers are needed).
463 //
464 // FIXME(#6598): The atomics module has no relaxed ordering flag, so I use
465 // acquire/release orderings superfluously. Change these someday.
466 read_count: atomics::AtomicUint,
467 }
468
469 /// An RAII helper which is created by acquiring a read lock on an RWLock. When
470 /// dropped, this will unlock the RWLock.
471 #[must_use]
472 pub struct RWLockReadGuard<'a> {
473 lock: &'a RWLock,
474 }
475
476 /// An RAII helper which is created by acquiring a write lock on an RWLock. When
477 /// dropped, this will unlock the RWLock.
478 ///
479 /// A value of this type can also be consumed to downgrade to a read-only lock.
480 #[must_use]
481 pub struct RWLockWriteGuard<'a> {
482 lock: &'a RWLock,
483 /// Inner condition variable that is connected to the write-mode of the
484 /// outer rwlock.
485 pub cond: Condvar<'a>,
486 }
487
488 impl RWLock {
489 /// Create a new rwlock, with one associated condvar.
490 pub fn new() -> RWLock { RWLock::new_with_condvars(1) }
491
492 /// Create a new rwlock, with a specified number of associated condvars.
493 /// Similar to mutex_with_condvars.
494 pub fn new_with_condvars(num_condvars: uint) -> RWLock {
495 RWLock {
496 order_lock: Semaphore::new(1),
497 access_lock: Sem::new_and_signal(1, num_condvars),
498 read_count: atomics::AtomicUint::new(0),
499 }
500 }
501
502 /// Acquires a read-lock, returning an RAII guard that will unlock the lock
503 /// when dropped. Calls to 'read' from other tasks may run concurrently with
504 /// this one.
505 pub fn read<'a>(&'a self) -> RWLockReadGuard<'a> {
506 let _guard = self.order_lock.access();
507 let old_count = self.read_count.fetch_add(1, atomics::Acquire);
508 if old_count == 0 {
509 self.access_lock.acquire();
510 }
511 RWLockReadGuard { lock: self }
512 }
513
514 /// Acquire a write-lock, returning an RAII guard that will unlock the lock
515 /// when dropped. No calls to 'read' or 'write' from other tasks will run
516 /// concurrently with this one.
517 ///
518 /// You can also downgrade a write to a read by calling the `downgrade`
519 /// method on the returned guard. Additionally, the guard will contain a
520 /// `Condvar` attached to this lock.
521 ///
522 /// # Example
523 ///
524 /// ```rust
525 /// use sync::raw::RWLock;
526 ///
527 /// let lock = RWLock::new();
528 /// let write = lock.write();
529 /// // ... exclusive access ...
530 /// let read = write.downgrade();
531 /// // ... shared access ...
532 /// drop(read);
533 /// ```
534 pub fn write<'a>(&'a self) -> RWLockWriteGuard<'a> {
535 let _g = self.order_lock.access();
536 self.access_lock.acquire();
537
538 // It's important to thread our order lock into the condvar, so that
539 // when a cond.wait() wakes up, it uses it while reacquiring the
540 // access lock. If we permitted a waking-up writer to "cut in line",
541 // there could arise a subtle race when a downgrader attempts to hand
542 // off the reader cloud lock to a waiting reader. This race is tested
543 // in arc.rs (test_rw_write_cond_downgrade_read_race) and looks like:
544 // T1 (writer) T2 (downgrader) T3 (reader)
545 // [in cond.wait()]
546 // [locks for writing]
547 // [holds access_lock]
548 // [is signalled, perhaps by
549 // downgrader or a 4th thread]
550 // tries to lock access(!)
551 // lock order_lock
552 // xadd read_count[0->1]
553 // tries to lock access
554 // [downgrade]
555 // xadd read_count[1->2]
556 // unlock access
557 // Since T1 contended on the access lock before T3 did, it will steal
558 // the lock handoff. Adding order_lock in the condvar reacquire path
559 // solves this because T1 will hold order_lock while waiting on access,
560 // which will cause T3 to have to wait until T1 finishes its write,
561 // which can't happen until T2 finishes the downgrade-read entirely.
562 // The astute reader will also note that making waking writers use the
563 // order_lock is better for not starving readers.
564 RWLockWriteGuard {
565 lock: self,
566 cond: Condvar {
567 sem: &self.access_lock,
568 order: Just(&self.order_lock),
569 nocopy: marker::NoCopy,
570 }
571 }
572 }
573 }
574
575 impl<'a> RWLockWriteGuard<'a> {
576 /// Consumes this write lock and converts it into a read lock.
577 pub fn downgrade(self) -> RWLockReadGuard<'a> {
578 let lock = self.lock;
579 // Don't run the destructor of the write guard, we're in charge of
580 // things from now on
581 unsafe { cast::forget(self) }
582
583 let old_count = lock.read_count.fetch_add(1, atomics::Release);
584 // If another reader was already blocking, we need to hand-off
585 // the "reader cloud" access lock to them.
586 if old_count != 0 {
587 // Guaranteed not to let another writer in, because
588 // another reader was holding the order_lock. Hence they
589 // must be the one to get the access_lock (because all
590 // access_locks are acquired with order_lock held). See
591 // the comment in write_cond for more justification.
592 lock.access_lock.release();
593 }
594 RWLockReadGuard { lock: lock }
595 }
596 }
597
598 #[unsafe_destructor]
599 impl<'a> Drop for RWLockWriteGuard<'a> {
600 fn drop(&mut self) {
601 self.lock.access_lock.release();
602 }
603 }
604
605 #[unsafe_destructor]
606 impl<'a> Drop for RWLockReadGuard<'a> {
607 fn drop(&mut self) {
608 let old_count = self.lock.read_count.fetch_sub(1, atomics::Release);
609 assert!(old_count > 0);
610 if old_count == 1 {
611 // Note: this release used to be outside of a locked access
612 // to exclusive-protected state. If this code is ever
613 // converted back to such (instead of using atomic ops),
614 // this access MUST NOT go inside the exclusive access.
615 self.lock.access_lock.release();
616 }
617 }
618 }
619
620 /****************************************************************************
621 * Tests
622 ****************************************************************************/
623
624 #[cfg(test)]
625 mod tests {
626 use arc::Arc;
627 use super::{Semaphore, Mutex, RWLock, Condvar};
628
629 use std::cast;
630 use std::result;
631 use std::task;
632
633 /************************************************************************
634 * Semaphore tests
635 ************************************************************************/
636 #[test]
637 fn test_sem_acquire_release() {
638 let s = Semaphore::new(1);
639 s.acquire();
640 s.release();
641 s.acquire();
642 }
643 #[test]
644 fn test_sem_basic() {
645 let s = Semaphore::new(1);
646 let _g = s.access();
647 }
648 #[test]
649 fn test_sem_as_mutex() {
650 let s = Arc::new(Semaphore::new(1));
651 let s2 = s.clone();
652 task::spawn(proc() {
653 let _g = s2.access();
654 for _ in range(0, 5) { task::deschedule(); }
655 });
656 let _g = s.access();
657 for _ in range(0, 5) { task::deschedule(); }
658 }
659 #[test]
660 fn test_sem_as_cvar() {
661 /* Child waits and parent signals */
662 let (tx, rx) = channel();
663 let s = Arc::new(Semaphore::new(0));
664 let s2 = s.clone();
665 task::spawn(proc() {
666 s2.acquire();
667 tx.send(());
668 });
669 for _ in range(0, 5) { task::deschedule(); }
670 s.release();
671 let _ = rx.recv();
672
673 /* Parent waits and child signals */
674 let (tx, rx) = channel();
675 let s = Arc::new(Semaphore::new(0));
676 let s2 = s.clone();
677 task::spawn(proc() {
678 for _ in range(0, 5) { task::deschedule(); }
679 s2.release();
680 let _ = rx.recv();
681 });
682 s.acquire();
683 tx.send(());
684 }
685 #[test]
686 fn test_sem_multi_resource() {
687 // Parent and child both get in the critical section at the same
688 // time, and shake hands.
689 let s = Arc::new(Semaphore::new(2));
690 let s2 = s.clone();
691 let (tx1, rx1) = channel();
692 let (tx2, rx2) = channel();
693 task::spawn(proc() {
694 let _g = s2.access();
695 let _ = rx2.recv();
696 tx1.send(());
697 });
698 let _g = s.access();
699 tx2.send(());
700 let _ = rx1.recv();
701 }
702 #[test]
703 fn test_sem_runtime_friendly_blocking() {
704 // Force the runtime to schedule two threads on the same sched_loop.
705 // When one blocks, it should schedule the other one.
706 let s = Arc::new(Semaphore::new(1));
707 let s2 = s.clone();
708 let (tx, rx) = channel();
709 {
710 let _g = s.access();
711 task::spawn(proc() {
712 tx.send(());
713 drop(s2.access());
714 tx.send(());
715 });
716 rx.recv(); // wait for child to come alive
717 for _ in range(0, 5) { task::deschedule(); } // let the child contend
718 }
719 rx.recv(); // wait for child to be done
720 }
721 /************************************************************************
722 * Mutex tests
723 ************************************************************************/
724 #[test]
725 fn test_mutex_lock() {
726 // Unsafely achieve shared state, and do the textbook
727 // "load tmp = move ptr; inc tmp; store ptr <- tmp" dance.
728 let (tx, rx) = channel();
729 let m = Arc::new(Mutex::new());
730 let m2 = m.clone();
731 let mut sharedstate = box 0;
732 {
733 let ptr: *mut int = &mut *sharedstate;
734 task::spawn(proc() {
735 access_shared(ptr, &m2, 10);
736 tx.send(());
737 });
738 }
739 {
740 access_shared(&mut *sharedstate, &m, 10);
741 let _ = rx.recv();
742
743 assert_eq!(*sharedstate, 20);
744 }
745
746 fn access_shared(sharedstate: *mut int, m: &Arc<Mutex>, n: uint) {
747 for _ in range(0, n) {
748 let _g = m.lock();
749 let oldval = unsafe { *sharedstate };
750 task::deschedule();
751 unsafe { *sharedstate = oldval + 1; }
752 }
753 }
754 }
755 #[test]
756 fn test_mutex_cond_wait() {
757 let m = Arc::new(Mutex::new());
758
759 // Child wakes up parent
760 {
761 let lock = m.lock();
762 let m2 = m.clone();
763 task::spawn(proc() {
764 let lock = m2.lock();
765 let woken = lock.cond.signal();
766 assert!(woken);
767 });
768 lock.cond.wait();
769 }
770 // Parent wakes up child
771 let (tx, rx) = channel();
772 let m3 = m.clone();
773 task::spawn(proc() {
774 let lock = m3.lock();
775 tx.send(());
776 lock.cond.wait();
777 tx.send(());
778 });
779 rx.recv(); // Wait until child gets in the mutex
780 {
781 let lock = m.lock();
782 let woken = lock.cond.signal();
783 assert!(woken);
784 }
785 rx.recv(); // Wait until child wakes up
786 }
787
788 fn test_mutex_cond_broadcast_helper(num_waiters: uint) {
789 let m = Arc::new(Mutex::new());
790 let mut rxs = Vec::new();
791
792 for _ in range(0, num_waiters) {
793 let mi = m.clone();
794 let (tx, rx) = channel();
795 rxs.push(rx);
796 task::spawn(proc() {
797 let lock = mi.lock();
798 tx.send(());
799 lock.cond.wait();
800 tx.send(());
801 });
802 }
803
804 // wait until all children get in the mutex
805 for rx in rxs.mut_iter() { rx.recv(); }
806 {
807 let lock = m.lock();
808 let num_woken = lock.cond.broadcast();
809 assert_eq!(num_woken, num_waiters);
810 }
811 // wait until all children wake up
812 for rx in rxs.mut_iter() { rx.recv(); }
813 }
814 #[test]
815 fn test_mutex_cond_broadcast() {
816 test_mutex_cond_broadcast_helper(12);
817 }
818 #[test]
819 fn test_mutex_cond_broadcast_none() {
820 test_mutex_cond_broadcast_helper(0);
821 }
822 #[test]
823 fn test_mutex_cond_no_waiter() {
824 let m = Arc::new(Mutex::new());
825 let m2 = m.clone();
826 let _ = task::try(proc() {
827 drop(m.lock());
828 });
829 let lock = m2.lock();
830 assert!(!lock.cond.signal());
831 }
832 #[test]
833 fn test_mutex_killed_simple() {
834 use std::any::Any;
835
836 // Mutex must get automatically unlocked if failed/killed within.
837 let m = Arc::new(Mutex::new());
838 let m2 = m.clone();
839
840 let result: result::Result<(), Box<Any:Send>> = task::try(proc() {
841 let _lock = m2.lock();
842 fail!();
843 });
844 assert!(result.is_err());
845 // child task must have finished by the time try returns
846 drop(m.lock());
847 }
848 #[test]
849 fn test_mutex_cond_signal_on_0() {
850 // Tests that signal_on(0) is equivalent to signal().
851 let m = Arc::new(Mutex::new());
852 let lock = m.lock();
853 let m2 = m.clone();
854 task::spawn(proc() {
855 let lock = m2.lock();
856 lock.cond.signal_on(0);
857 });
858 lock.cond.wait();
859 }
860 #[test]
861 fn test_mutex_no_condvars() {
862 let result = task::try(proc() {
863 let m = Mutex::new_with_condvars(0);
864 m.lock().cond.wait();
865 });
866 assert!(result.is_err());
867 let result = task::try(proc() {
868 let m = Mutex::new_with_condvars(0);
869 m.lock().cond.signal();
870 });
871 assert!(result.is_err());
872 let result = task::try(proc() {
873 let m = Mutex::new_with_condvars(0);
874 m.lock().cond.broadcast();
875 });
876 assert!(result.is_err());
877 }
878 /************************************************************************
879 * Reader/writer lock tests
880 ************************************************************************/
881 #[cfg(test)]
882 pub enum RWLockMode { Read, Write, Downgrade, DowngradeRead }
883 #[cfg(test)]
884 fn lock_rwlock_in_mode(x: &Arc<RWLock>, mode: RWLockMode, blk: ||) {
885 match mode {
886 Read => { let _g = x.read(); blk() }
887 Write => { let _g = x.write(); blk() }
888 Downgrade => { let _g = x.write(); blk() }
889 DowngradeRead => { let _g = x.write().downgrade(); blk() }
890 }
891 }
892 #[cfg(test)]
893 fn test_rwlock_exclusion(x: Arc<RWLock>,
894 mode1: RWLockMode,
895 mode2: RWLockMode) {
896 // Test mutual exclusion between readers and writers. Just like the
897 // mutex mutual exclusion test, a ways above.
898 let (tx, rx) = channel();
899 let x2 = x.clone();
900 let mut sharedstate = box 0;
901 {
902 let ptr: *int = &*sharedstate;
903 task::spawn(proc() {
904 let sharedstate: &mut int =
905 unsafe { cast::transmute(ptr) };
906 access_shared(sharedstate, &x2, mode1, 10);
907 tx.send(());
908 });
909 }
910 {
911 access_shared(sharedstate, &x, mode2, 10);
912 let _ = rx.recv();
913
914 assert_eq!(*sharedstate, 20);
915 }
916
917 fn access_shared(sharedstate: &mut int, x: &Arc<RWLock>,
918 mode: RWLockMode, n: uint) {
919 for _ in range(0, n) {
920 lock_rwlock_in_mode(x, mode, || {
921 let oldval = *sharedstate;
922 task::deschedule();
923 *sharedstate = oldval + 1;
924 })
925 }
926 }
927 }
928 #[test]
929 fn test_rwlock_readers_wont_modify_the_data() {
930 test_rwlock_exclusion(Arc::new(RWLock::new()), Read, Write);
931 test_rwlock_exclusion(Arc::new(RWLock::new()), Write, Read);
932 test_rwlock_exclusion(Arc::new(RWLock::new()), Read, Downgrade);
933 test_rwlock_exclusion(Arc::new(RWLock::new()), Downgrade, Read);
934 test_rwlock_exclusion(Arc::new(RWLock::new()), Write, DowngradeRead);
935 test_rwlock_exclusion(Arc::new(RWLock::new()), DowngradeRead, Write);
936 }
937 #[test]
938 fn test_rwlock_writers_and_writers() {
939 test_rwlock_exclusion(Arc::new(RWLock::new()), Write, Write);
940 test_rwlock_exclusion(Arc::new(RWLock::new()), Write, Downgrade);
941 test_rwlock_exclusion(Arc::new(RWLock::new()), Downgrade, Write);
942 test_rwlock_exclusion(Arc::new(RWLock::new()), Downgrade, Downgrade);
943 }
944 #[cfg(test)]
945 fn test_rwlock_handshake(x: Arc<RWLock>,
946 mode1: RWLockMode,
947 mode2: RWLockMode,
948 make_mode2_go_first: bool) {
949 // Much like sem_multi_resource.
950 let x2 = x.clone();
951 let (tx1, rx1) = channel();
952 let (tx2, rx2) = channel();
953 task::spawn(proc() {
954 if !make_mode2_go_first {
955 rx2.recv(); // parent sends to us once it locks, or ...
956 }
957 lock_rwlock_in_mode(&x2, mode2, || {
958 if make_mode2_go_first {
959 tx1.send(()); // ... we send to it once we lock
960 }
961 rx2.recv();
962 tx1.send(());
963 })
964 });
965 if make_mode2_go_first {
966 rx1.recv(); // child sends to us once it locks, or ...
967 }
968 lock_rwlock_in_mode(&x, mode1, || {
969 if !make_mode2_go_first {
970 tx2.send(()); // ... we send to it once we lock
971 }
972 tx2.send(());
973 rx1.recv();
974 })
975 }
976 #[test]
977 fn test_rwlock_readers_and_readers() {
978 test_rwlock_handshake(Arc::new(RWLock::new()), Read, Read, false);
979 // The downgrader needs to get in before the reader gets in, otherwise
980 // they cannot end up reading at the same time.
981 test_rwlock_handshake(Arc::new(RWLock::new()), DowngradeRead, Read, false);
982 test_rwlock_handshake(Arc::new(RWLock::new()), Read, DowngradeRead, true);
983 // Two downgrade_reads can never both end up reading at the same time.
984 }
985 #[test]
986 fn test_rwlock_downgrade_unlock() {
987 // Tests that downgrade can unlock the lock in both modes
988 let x = Arc::new(RWLock::new());
989 lock_rwlock_in_mode(&x, Downgrade, || { });
990 test_rwlock_handshake(x, Read, Read, false);
991 let y = Arc::new(RWLock::new());
992 lock_rwlock_in_mode(&y, DowngradeRead, || { });
993 test_rwlock_exclusion(y, Write, Write);
994 }
995 #[test]
996 fn test_rwlock_read_recursive() {
997 let x = RWLock::new();
998 let _g1 = x.read();
999 let _g2 = x.read();
1000 }
1001 #[test]
1002 fn test_rwlock_cond_wait() {
1003 // As test_mutex_cond_wait above.
1004 let x = Arc::new(RWLock::new());
1005
1006 // Child wakes up parent
1007 {
1008 let lock = x.write();
1009 let x2 = x.clone();
1010 task::spawn(proc() {
1011 let lock = x2.write();
1012 assert!(lock.cond.signal());
1013 });
1014 lock.cond.wait();
1015 }
1016 // Parent wakes up child
1017 let (tx, rx) = channel();
1018 let x3 = x.clone();
1019 task::spawn(proc() {
1020 let lock = x3.write();
1021 tx.send(());
1022 lock.cond.wait();
1023 tx.send(());
1024 });
1025 rx.recv(); // Wait until child gets in the rwlock
1026 drop(x.read()); // Must be able to get in as a reader
1027 {
1028 let x = x.write();
1029 assert!(x.cond.signal());
1030 }
1031 rx.recv(); // Wait until child wakes up
1032 drop(x.read()); // Just for good measure
1033 }
1034 #[cfg(test)]
1035 fn test_rwlock_cond_broadcast_helper(num_waiters: uint) {
1036 // Much like the mutex broadcast test. Downgrade-enabled.
1037 fn lock_cond(x: &Arc<RWLock>, blk: |c: &Condvar|) {
1038 let lock = x.write();
1039 blk(&lock.cond);
1040 }
1041
1042 let x = Arc::new(RWLock::new());
1043 let mut rxs = Vec::new();
1044
1045 for _ in range(0, num_waiters) {
1046 let xi = x.clone();
1047 let (tx, rx) = channel();
1048 rxs.push(rx);
1049 task::spawn(proc() {
1050 lock_cond(&xi, |cond| {
1051 tx.send(());
1052 cond.wait();
1053 tx.send(());
1054 })
1055 });
1056 }
1057
1058 // wait until all children get in the mutex
1059 for rx in rxs.mut_iter() { let _ = rx.recv(); }
1060 lock_cond(&x, |cond| {
1061 let num_woken = cond.broadcast();
1062 assert_eq!(num_woken, num_waiters);
1063 });
1064 // wait until all children wake up
1065 for rx in rxs.mut_iter() { let _ = rx.recv(); }
1066 }
1067 #[test]
1068 fn test_rwlock_cond_broadcast() {
1069 test_rwlock_cond_broadcast_helper(0);
1070 test_rwlock_cond_broadcast_helper(12);
1071 }
1072 #[cfg(test)]
1073 fn rwlock_kill_helper(mode1: RWLockMode, mode2: RWLockMode) {
1074 use std::any::Any;
1075
1076 // Mutex must get automatically unlocked if failed/killed within.
1077 let x = Arc::new(RWLock::new());
1078 let x2 = x.clone();
1079
1080 let result: result::Result<(), Box<Any:Send>> = task::try(proc() {
1081 lock_rwlock_in_mode(&x2, mode1, || {
1082 fail!();
1083 })
1084 });
1085 assert!(result.is_err());
1086 // child task must have finished by the time try returns
1087 lock_rwlock_in_mode(&x, mode2, || { })
1088 }
1089 #[test]
1090 fn test_rwlock_reader_killed_writer() {
1091 rwlock_kill_helper(Read, Write);
1092 }
1093 #[test]
1094 fn test_rwlock_writer_killed_reader() {
1095 rwlock_kill_helper(Write, Read);
1096 }
1097 #[test]
1098 fn test_rwlock_reader_killed_reader() {
1099 rwlock_kill_helper(Read, Read);
1100 }
1101 #[test]
1102 fn test_rwlock_writer_killed_writer() {
1103 rwlock_kill_helper(Write, Write);
1104 }
1105 #[test]
1106 fn test_rwlock_kill_downgrader() {
1107 rwlock_kill_helper(Downgrade, Read);
1108 rwlock_kill_helper(Read, Downgrade);
1109 rwlock_kill_helper(Downgrade, Write);
1110 rwlock_kill_helper(Write, Downgrade);
1111 rwlock_kill_helper(DowngradeRead, Read);
1112 rwlock_kill_helper(Read, DowngradeRead);
1113 rwlock_kill_helper(DowngradeRead, Write);
1114 rwlock_kill_helper(Write, DowngradeRead);
1115 rwlock_kill_helper(DowngradeRead, Downgrade);
1116 rwlock_kill_helper(DowngradeRead, Downgrade);
1117 rwlock_kill_helper(Downgrade, DowngradeRead);
1118 rwlock_kill_helper(Downgrade, DowngradeRead);
1119 }
1120 }
libsync/raw.rs:33:42-33:42 -struct- definition:
// A doubly-ended queue of waiting tasks.
struct WaitQueue {
head: Receiver<SignalEnd>,
references:- 1141: let (block_tail, block_head) = channel();
42: WaitQueue { head: block_head, tail: block_tail }
43: }
--
184: impl Sem<Vec<WaitQueue>> {
185: fn new_and_signal(count: int, num_condvars: uint) -> Sem<Vec<WaitQueue>> {
--
407: pub struct Mutex {
408: sem: Sem<Vec<WaitQueue>>,
409: }
--
415: pub struct MutexGuard<'a> {
416: guard: SemGuard<'a, Vec<WaitQueue>>,
417: /// Inner condition variable which is connected to the outer mutex, and can
--
456: order_lock: Semaphore,
457: access_lock: Sem<Vec<WaitQueue>>,
libsync/raw.rs:354:12-354:12 -struct- definition:
struct SemCondGuard<'a> {
guard: SemGuard<'a, Vec<WaitQueue>>,
cvar: Condvar<'a>,
references:- 3192: // and rwlock_write_mode.
193: pub fn access_cond<'a>(&'a self) -> SemCondGuard<'a> {
194: SemCondGuard {
--
437: pub fn lock<'a>(&'a self) -> MutexGuard<'a> {
438: let SemCondGuard { guard, cvar } = self.sem.access_cond();
439: MutexGuard { guard: guard, cond: cvar }
libsync/raw.rs:95:1-95:1 -struct- definition:
struct SemInner<Q> {
count: int,
waiters: WaitQueue,
references:- 4111: let inner = unsafe {
112: cast::transmute(box SemInner {
113: waiters: WaitQueue::new(),
--
124: unsafe fn with(&self, f: |&mut SemInner<Q>|) {
125: let _g = self.lock.lock();
--
169: fn drop(&mut self) {
170: let _waiters: Box<SemInner<Q>> = unsafe {
171: cast::transmute(self.inner)
libsync/raw.rs:364:53-364:53 -struct- definition:
/// A counting, blocking, bounded-waiting semaphore.
pub struct Semaphore {
sem: Sem<()>,
references:- 5377: /// Create a new semaphore with the specified count.
378: pub fn new(count: int) -> Semaphore {
379: Semaphore { sem: Sem::new(count, ()) }
--
455: pub struct RWLock {
456: order_lock: Semaphore,
457: access_lock: Sem<Vec<WaitQueue>>,
libsync/raw.rs:338:10-338:10 -fn- definition:
fn check_cvar_bounds<U>(
out_of_bounds: Option<uint>,
id: uint,
references:- 3299: });
300: check_cvar_bounds(out_of_bounds,
301: condvar_id,
--
325: });
326: check_cvar_bounds(out_of_bounds,
327: condvar_id,
libsync/raw.rs:480:12-480:12 -struct- definition:
pub struct RWLockWriteGuard<'a> {
lock: &'a RWLock,
/// Inner condition variable that is connected to the write-mode of the
references:- 5563: // order_lock is better for not starving readers.
564: RWLockWriteGuard {
565: lock: self,
--
599: impl<'a> Drop for RWLockWriteGuard<'a> {
600: fn drop(&mut self) {
libsync/lock.rs:
66: InnerMutex(raw::MutexGuard<'a>),
67: InnerRWLock(raw::RWLockWriteGuard<'a>),
68: }
libsync/raw.rs:
575: impl<'a> RWLockWriteGuard<'a> {
576: /// Consumes this write lock and converts it into a read lock.
libsync/raw.rs:31:29-31:29 -NK_AS_STR_TODO- definition:
type WaitEnd = Receiver<()>;
type SignalEnd = Sender<()>;
// A doubly-ended queue of waiting tasks.
references:- 234: struct WaitQueue {
35: head: Receiver<SignalEnd>,
36: tail: Sender<SignalEnd>,
37: }
libsync/raw.rs:406:13-406:13 -struct- definition:
/// unwinds.
pub struct Mutex {
sem: Sem<Vec<WaitQueue>>,
references:- 5430: pub fn new_with_condvars(num_condvars: uint) -> Mutex {
431: Mutex { sem: Sem::new_and_signal(1, num_condvars) }
432: }
libsync/lock.rs:
168: pub struct Mutex<T> {
169: lock: raw::Mutex,
170: failed: Unsafe<bool>,
libsync/raw.rs:371:12-371:12 -struct- definition:
pub struct SemaphoreGuard<'a> {
guard: SemGuard<'a, ()>,
}
references:- 2391: /// release the resource when dropped.
392: pub fn access<'a>(&'a self) -> SemaphoreGuard<'a> {
393: SemaphoreGuard { guard: self.sem.access() }
394: }
libsync/raw.rs:83:69-83:69 -struct- definition:
// The building-block used to make semaphores, mutexes, and rwlocks.
struct Sem<Q> {
lock: mutex::Mutex,
references:- 11117: };
118: Sem {
119: lock: mutex::Mutex::new(),
--
365: pub struct Semaphore {
366: sem: Sem<()>,
367: }
--
407: pub struct Mutex {
408: sem: Sem<Vec<WaitQueue>>,
409: }
--
456: order_lock: Semaphore,
457: access_lock: Sem<Vec<WaitQueue>>,
libsync/raw.rs:104:12-104:12 -struct- definition:
struct SemGuard<'a, Q> {
sem: &'a Sem<Q>,
}
references:- 6162: self.acquire();
163: SemGuard { sem: self }
164: }
--
372: pub struct SemaphoreGuard<'a> {
373: guard: SemGuard<'a, ()>,
374: }
--
415: pub struct MutexGuard<'a> {
416: guard: SemGuard<'a, Vec<WaitQueue>>,
417: /// Inner condition variable which is connected to the outer mutex, and can
libsync/raw.rs:209:74-209:74 -struct- definition:
/// A mechanism for atomic-unlock-and-deschedule blocking and signalling.
pub struct Condvar<'a> {
// The 'Sem' object associated with this condvar. This is the one that's
references:- 7565: lock: self,
566: cond: Condvar {
567: sem: &self.access_lock,
libsync/lock.rs:
70: impl<'b> Inner<'b> {
71: fn cond<'a>(&'a self) -> &'a raw::Condvar<'b> {
72: match *self {
libsync/raw.rs:
195: guard: self.access(),
196: cvar: Condvar { sem: self, order: Nothing, nocopy: marker::NoCopy },
197: }
libsync/raw.rs:414:12-414:12 -struct- definition:
pub struct MutexGuard<'a> {
guard: SemGuard<'a, Vec<WaitQueue>>,
/// Inner condition variable which is connected to the outer mutex, and can
references:- 3436: /// also be accessed through the returned guard.
437: pub fn lock<'a>(&'a self) -> MutexGuard<'a> {
438: let SemCondGuard { guard, cvar } = self.sem.access_cond();
libsync/lock.rs:
65: enum Inner<'a> {
66: InnerMutex(raw::MutexGuard<'a>),
67: InnerRWLock(raw::RWLockWriteGuard<'a>),
libsync/raw.rs:
438: let SemCondGuard { guard, cvar } = self.sem.access_cond();
439: MutexGuard { guard: guard, cond: cvar }
440: }
libsync/raw.rs:454:13-454:13 -struct- definition:
/// unwinds.
pub struct RWLock {
order_lock: Semaphore,
references:- 7494: pub fn new_with_condvars(num_condvars: uint) -> RWLock {
495: RWLock {
496: order_lock: Semaphore::new(1),
libsync/lock.rs:
266: pub struct RWLock<T> {
267: lock: raw::RWLock,
268: failed: Unsafe<bool>,
libsync/raw.rs:471:12-471:12 -struct- definition:
pub struct RWLockReadGuard<'a> {
lock: &'a RWLock,
}
references:- 6593: }
594: RWLockReadGuard { lock: lock }
595: }
--
606: impl<'a> Drop for RWLockReadGuard<'a> {
607: fn drop(&mut self) {
libsync/lock.rs:
284: data: &'a T,
285: guard: raw::RWLockReadGuard<'a>,
286: }
libsync/raw.rs:
510: }
511: RWLockReadGuard { lock: self }
512: }