rust-lang/rust
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

rollup merge of #20615: aturon/stab-2-thread

Browse Source 
This commit takes a first pass at stabilizing `std::thread`:

* It removes the `detach` method in favor of two constructors -- `spawn`
  for detached threads, `scoped` for "scoped" (i.e., must-join)
  threads. This addresses some of the surprise/frustrating debug
  sessions with the previous API, in which `spawn` produced a guard that
  on destruction joined the thread (unless `detach` was called).

  The reason to have the division in part is that `Send` will soon not
  imply `'static`, which means that `scoped` thread creation can take a
  closure over *shared stack data* of the parent thread. On the other
  hand, this means that the parent must not pop the relevant stack
  frames while the child thread is running. The `JoinGuard` is used to
  prevent this from happening by joining on drop (if you have not
  already explicitly `join`ed.) The APIs around `scoped` are
  future-proofed for the `Send` changes by taking an additional lifetime
  parameter. With the current definition of `Send`, this is forced to be
  `'static`, but when `Send` changes these APIs will gain their full
  flexibility immediately.

  Threads that are `spawn`ed, on the other hand, are detached from the
  start and do not yield an RAII guard.

  The hope is that, by making `scoped` an explicit opt-in with a very
  suggestive name, it will be drastically less likely to be caught by a
  surprising deadlock due to an implicit join at the end of a scope.

* The module itself is marked stable.

* Existing methods other than `spawn` and `scoped` are marked stable.

The migration path is:

```rust
Thread::spawn(f).detached()
```

becomes

```rust
Thread::spawn(f)
```

while

```rust
let res = Thread::spawn(f);
res.join()
```

becomes

```rust
let res = Thread::scoped(f);
res.join()
```

[breaking-change]
This commit is contained in:
Alex Crichton
2015-01-06 15:38:38 -08:00
parent 0631b466c2 caca9b2e71
commit 36f5d122b8
97 changed files with 361 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/libstd/sync/barrier.rs
View File

@@ -26,7 +26,7 @@ use sync::{Mutex, Condvar};
/// println!("before wait");
/// c.wait();
/// println!("after wait");
/// }).detach();
/// });
/// }
/// ```
#[stable]
@@ -126,7 +126,7 @@ mod tests {
let tx = tx.clone();
Thread::spawn(move|| {
tx.send(c.wait().is_leader()).unwrap();
}).detach();
});
}
// At this point, all spawned tasks should be blocked,

4
src/libstd/sync/condvar.rs
View File

@@ -48,7 +48,7 @@ use sync::{mutex, MutexGuard};
/// let mut started = lock.lock().unwrap();
/// *started = true;
/// cvar.notify_one();
/// }).detach();
/// });
///
/// // wait for the thread to start up
/// let &(ref lock, ref cvar) = &*pair;
@@ -338,7 +338,7 @@ mod tests {
cnt = cond.wait(cnt).unwrap();
}
tx.send(()).unwrap();
}).detach();
});
}
drop(tx);

2
src/libstd/sync/future.rs
View File

@@ -141,7 +141,7 @@ impl<A:Send> Future<A> {
Thread::spawn(move |:| {
// Don't panic if the other end has hung up
let _ = tx.send(blk());
}).detach();
});
Future::from_receiver(rx)
}

68
src/libstd/sync/mpsc/mod.rs
View File

@@ -60,7 +60,7 @@
//! let (tx, rx) = channel();
//! Thread::spawn(move|| {
//! tx.send(10i).unwrap();
//! }).detach();
//! });
//! assert_eq!(rx.recv().unwrap(), 10i);
//! ```
//!
@@ -78,7 +78,7 @@
//! let tx = tx.clone();
//! Thread::spawn(move|| {
//! tx.send(i).unwrap();
//! }).detach()
//! });
//! }
//!
//! for _ in range(0i, 10i) {
@@ -109,7 +109,7 @@
//! Thread::spawn(move|| {
//! // This will wait for the parent task to start receiving
//! tx.send(53).unwrap();
//! }).detach();
//! });
//! rx.recv().unwrap();
//! ```
//!
@@ -476,7 +476,7 @@ impl<T> UnsafeFlavor<T> for Receiver<T> {
/// Thread::spawn(move|| {
/// # fn expensive_computation() {}
/// tx.send(expensive_computation()).unwrap();
/// }).detach();
/// });
///
/// // Do some useful work for awhile
///
@@ -518,7 +518,7 @@ pub fn channel<T: Send>() -> (Sender<T>, Receiver<T>) {
/// Thread::spawn(move|| {
/// // this will block until the previous message has been received
/// tx.send(2i).unwrap();
/// }).detach();
/// });
///
/// assert_eq!(rx.recv().unwrap(), 1i);
/// assert_eq!(rx.recv().unwrap(), 2i);
@@ -1144,7 +1144,7 @@ mod test {
#[test]
fn stress() {
let (tx, rx) = channel::<int>();
let t = Thread::spawn(move|| {
let t = Thread::scoped(move|| {
for _ in range(0u, 10000) { tx.send(1i).unwrap(); }
});
for _ in range(0u, 10000) {
@@ -1159,7 +1159,7 @@ mod test {
static NTHREADS: uint = 8;
let (tx, rx) = channel::<int>();
let t = Thread::spawn(move|| {
let t = Thread::scoped(move|| {
for _ in range(0, AMT * NTHREADS) {
assert_eq!(rx.recv().unwrap(), 1);
}
@@ -1173,7 +1173,7 @@ mod test {
let tx = tx.clone();
Thread::spawn(move|| {
for _ in range(0, AMT) { tx.send(1).unwrap(); }
}).detach();
});
}
drop(tx);
t.join().ok().unwrap();
@@ -1183,14 +1183,14 @@ mod test {
fn send_from_outside_runtime() {
let (tx1, rx1) = channel::<()>();
let (tx2, rx2) = channel::<int>();
let t1 = Thread::spawn(move|| {
let t1 = Thread::scoped(move|| {
tx1.send(()).unwrap();
for _ in range(0i, 40) {
assert_eq!(rx2.recv().unwrap(), 1);
}
});
rx1.recv().unwrap();
let t2 = Thread::spawn(move|| {
let t2 = Thread::scoped(move|| {
for _ in range(0i, 40) {
tx2.send(1).unwrap();
}
@@ -1202,7 +1202,7 @@ mod test {
#[test]
fn recv_from_outside_runtime() {
let (tx, rx) = channel::<int>();
let t = Thread::spawn(move|| {
let t = Thread::scoped(move|| {
for _ in range(0i, 40) {
assert_eq!(rx.recv().unwrap(), 1);
}
@@ -1217,11 +1217,11 @@ mod test {
fn no_runtime() {
let (tx1, rx1) = channel::<int>();
let (tx2, rx2) = channel::<int>();
let t1 = Thread::spawn(move|| {
let t1 = Thread::scoped(move|| {
assert_eq!(rx1.recv().unwrap(), 1);
tx2.send(2).unwrap();
});
let t2 = Thread::spawn(move|| {
let t2 = Thread::scoped(move|| {
tx1.send(1).unwrap();
assert_eq!(rx2.recv().unwrap(), 2);
});
@@ -1254,7 +1254,7 @@ mod test {
#[test]
fn oneshot_single_thread_recv_chan_close() {
// Receiving on a closed chan will panic
let res = Thread::spawn(move|| {
let res = Thread::scoped(move|| {
let (tx, rx) = channel::<int>();
drop(tx);
rx.recv().unwrap();
@@ -1336,7 +1336,7 @@ mod test {
let _t = Thread::spawn(move|| {
drop(tx);
});
let res = Thread::spawn(move|| {
let res = Thread::scoped(move|| {
assert!(rx.recv().unwrap() == box 10);
}).join();
assert!(res.is_err());
@@ -1360,7 +1360,7 @@ mod test {
let _t = Thread::spawn(move|| {
drop(rx);
});
let _ = Thread::spawn(move|| {
let _ = Thread::scoped(move|| {
tx.send(1).unwrap();
}).join();
}
@@ -1371,15 +1371,15 @@ mod test {
for _ in range(0, stress_factor()) {
let (tx, rx) = channel::<int>();
Thread::spawn(move|| {
let res = Thread::spawn(move|| {
let res = Thread::scoped(move|| {
rx.recv().unwrap();
}).join();
assert!(res.is_err());
}).detach();
});
let _t = Thread::spawn(move|| {
Thread::spawn(move|| {
drop(tx);
}).detach();
});
});
}
}
@@ -1409,7 +1409,7 @@ mod test {
Thread::spawn(move|| {
tx.send(box i).unwrap();
send(tx, i + 1);
}).detach();
});
}
fn recv(rx: Receiver<Box<int>>, i: int) {
@@ -1418,7 +1418,7 @@ mod test {
Thread::spawn(move|| {
assert!(rx.recv().unwrap() == box i);
recv(rx, i + 1);
}).detach();
});
}
}
}
@@ -1439,7 +1439,7 @@ mod test {
let tx = tx.clone();
Thread::spawn(move|| {
tx.send(()).unwrap();
}).detach();
});
}
for _ in range(0, total) {
@@ -1644,7 +1644,7 @@ mod sync_tests {
Thread::spawn(move|| {
tx.send(1).unwrap();
tx.send(1).unwrap();
}).detach();
});
while rx.recv().is_ok() {}
}
@@ -1653,7 +1653,7 @@ mod sync_tests {
let (tx, rx) = sync_channel::<int>(0);
Thread::spawn(move|| {
for _ in range(0u, 10000) { tx.send(1).unwrap(); }
}).detach();
});
for _ in range(0u, 10000) {
assert_eq!(rx.recv().unwrap(), 1);
}
@@ -1675,13 +1675,13 @@ mod sync_tests {
_ => {}
}
dtx.send(()).unwrap();
}).detach();
});
for _ in range(0, NTHREADS) {
let tx = tx.clone();
Thread::spawn(move|| {
for _ in range(0, AMT) { tx.send(1).unwrap(); }
}).detach();
});
}
drop(tx);
drx.recv().unwrap();
@@ -1712,7 +1712,7 @@ mod sync_tests {
#[test]
fn oneshot_single_thread_recv_chan_close() {
// Receiving on a closed chan will panic
let res = Thread::spawn(move|| {
let res = Thread::scoped(move|| {
let (tx, rx) = sync_channel::<int>(0);
drop(tx);
rx.recv().unwrap();
@@ -1800,7 +1800,7 @@ mod sync_tests {
let _t = Thread::spawn(move|| {
drop(tx);
});
let res = Thread::spawn(move|| {
let res = Thread::scoped(move|| {
assert!(rx.recv().unwrap() == box 10);
}).join();
assert!(res.is_err());
@@ -1824,7 +1824,7 @@ mod sync_tests {
let _t = Thread::spawn(move|| {
drop(rx);
});
let _ = Thread::spawn(move || {
let _ = Thread::scoped(move || {
tx.send(1).unwrap();
}).join();
}
@@ -1835,7 +1835,7 @@ mod sync_tests {
for _ in range(0, stress_factor()) {
let (tx, rx) = sync_channel::<int>(0);
let _t = Thread::spawn(move|| {
let res = Thread::spawn(move|| {
let res = Thread::scoped(move|| {
rx.recv().unwrap();
}).join();
assert!(res.is_err());
@@ -1843,7 +1843,7 @@ mod sync_tests {
let _t = Thread::spawn(move|| {
Thread::spawn(move|| {
drop(tx);
}).detach();
});
});
}
}
@@ -1873,7 +1873,7 @@ mod sync_tests {
Thread::spawn(move|| {
tx.send(box i).unwrap();
send(tx, i + 1);
}).detach();
});
}
fn recv(rx: Receiver<Box<int>>, i: int) {
@@ -1882,7 +1882,7 @@ mod sync_tests {
Thread::spawn(move|| {
assert!(rx.recv().unwrap() == box i);
recv(rx, i + 1);
}).detach();
});
}
}
}
@@ -1903,7 +1903,7 @@ mod sync_tests {
let tx = tx.clone();
Thread::spawn(move|| {
tx.send(()).unwrap();
}).detach();
});
}
for _ in range(0, total) {

2
src/libstd/sync/mpsc/mpsc_queue.rs
View File

@@ -188,7 +188,7 @@ mod tests {
q.push(i);
}
tx.send(()).unwrap();
}).detach();
});
}
let mut i = 0u;

12
src/libstd/sync/mutex.rs
View File

@@ -75,7 +75,7 @@ use sys_common::mutex as sys;
/// tx.send(()).unwrap();
/// }
/// // the lock is unlocked here when `data` goes out of scope.
/// }).detach();
/// });
/// }
///
/// rx.recv().unwrap();
@@ -90,7 +90,7 @@ use sys_common::mutex as sys;
/// let lock = Arc::new(Mutex::new(0u));
/// let lock2 = lock.clone();
///
/// let _ = Thread::spawn(move || -> () {
/// let _ = Thread::scoped(move || -> () {
/// // This thread will acquire the mutex first, unwrapping the result of
/// // `lock` because the lock has not been poisoned.
/// let _lock = lock2.lock().unwrap();
@@ -376,9 +376,9 @@ mod test {
let (tx, rx) = channel();
for _ in range(0, K) {
let tx2 = tx.clone();
Thread::spawn(move|| { inc(); tx2.send(()).unwrap(); }).detach();
Thread::spawn(move|| { inc(); tx2.send(()).unwrap(); });
let tx2 = tx.clone();
Thread::spawn(move|| { inc(); tx2.send(()).unwrap(); }).detach();
Thread::spawn(move|| { inc(); tx2.send(()).unwrap(); });
}
drop(tx);
@@ -453,7 +453,7 @@ mod test {
fn test_mutex_arc_poison() {
let arc = Arc::new(Mutex::new(1i));
let arc2 = arc.clone();
let _ = Thread::spawn(move|| {
let _ = Thread::scoped(move|| {
let lock = arc2.lock().unwrap();
assert_eq!(*lock, 2);
}).join();
@@ -480,7 +480,7 @@ mod test {
fn test_mutex_arc_access_in_unwind() {
let arc = Arc::new(Mutex::new(1i));
let arc2 = arc.clone();
let _ = Thread::spawn(move|| -> () {
let _ = Thread::scoped(move|| -> () {
struct Unwinder {
i: Arc<Mutex<int>>,
}

2
src/libstd/sync/once.rs
View File

@@ -159,7 +159,7 @@ mod test {
assert!(run);
}
tx.send(()).unwrap();
}).detach();
});
}
unsafe {

16
src/libstd/sync/rwlock.rs
View File

@@ -411,7 +411,7 @@ mod tests {
}
}
drop(tx);
}).detach();
});
}
drop(tx);
let _ = rx.recv();
@@ -422,7 +422,7 @@ mod tests {
fn test_rw_arc_poison_wr() {
let arc = Arc::new(RwLock::new(1i));
let arc2 = arc.clone();
let _: Result<uint, _> = Thread::spawn(move|| {
let _: Result<uint, _> = Thread::scoped(move|| {
let _lock = arc2.write().unwrap();
panic!();
}).join();
@@ -433,7 +433,7 @@ mod tests {
fn test_rw_arc_poison_ww() {
let arc = Arc::new(RwLock::new(1i));
let arc2 = arc.clone();
let _: Result<uint, _> = Thread::spawn(move|| {
let _: Result<uint, _> = Thread::scoped(move|| {
let _lock = arc2.write().unwrap();
panic!();
}).join();
@@ -444,7 +444,7 @@ mod tests {
fn test_rw_arc_no_poison_rr() {
let arc = Arc::new(RwLock::new(1i));
let arc2 = arc.clone();
let _: Result<uint, _> = Thread::spawn(move|| {
let _: Result<uint, _> = Thread::scoped(move|| {
let _lock = arc2.read().unwrap();
panic!();
}).join();
@@ -455,7 +455,7 @@ mod tests {
fn test_rw_arc_no_poison_rw() {
let arc = Arc::new(RwLock::new(1i));
let arc2 = arc.clone();
let _: Result<uint, _> = Thread::spawn(move|| {
let _: Result<uint, _> = Thread::scoped(move|| {
let _lock = arc2.read().unwrap();
panic!()
}).join();
@@ -478,13 +478,13 @@ mod tests {
*lock = tmp + 1;
}
tx.send(()).unwrap();
}).detach();
});
// Readers try to catch the writer in the act
let mut children = Vec::new();
for _ in range(0u, 5) {
let arc3 = arc.clone();
children.push(Thread::spawn(move|| {
children.push(Thread::scoped(move|| {
let lock = arc3.read().unwrap();
assert!(*lock >= 0);
}));
@@ -505,7 +505,7 @@ mod tests {
fn test_rw_arc_access_in_unwind() {
let arc = Arc::new(RwLock::new(1i));
let arc2 = arc.clone();
let _ = Thread::spawn(move|| -> () {
let _ = Thread::scoped(move|| -> () {
struct Unwinder {
i: Arc<RwLock<int>>,
}

2
src/libstd/sync/semaphore.rs
View File

@@ -193,7 +193,7 @@ mod tests {
tx.send(()).unwrap();
drop(s2.access());
tx.send(()).unwrap();
}).detach();
});
rx.recv().unwrap(); // wait for child to come alive
}
rx.recv().unwrap(); // wait for child to be done

2
src/libstd/sync/task_pool.rs
View File

@@ -132,7 +132,7 @@ fn spawn_in_pool(jobs: Arc<Mutex<Receiver<Thunk>>>) {
}
sentinel.cancel();
}).detach();
});
}
#[cfg(test)]