Rewrite threading infrastructure, introducing Thunk to represent

boxed `FnOnce` closures.

Author: nikomatsakis

src/librustrt/at_exit_imp.rs

@@ -18,10 +18,11 @@ use alloc::boxed::Box;
 use collections::vec::Vec;
 use core::atomic;
 use core::mem;
+use thunk::{Thunk};
 
 use exclusive::Exclusive;
 
-type Queue = Exclusive<Vec<proc():Send>>;
+type Queue = Exclusive<Vec<Thunk>>;
 
 static QUEUE: atomic::AtomicUint = atomic::INIT_ATOMIC_UINT;
 static RUNNING: atomic::AtomicBool = atomic::INIT_ATOMIC_BOOL;
@@ -34,7 +35,7 @@ pub fn init() {
     }
 }
 
-pub fn push(f: proc():Send) {
+pub fn push(f: Thunk) {
     unsafe {
         // Note that the check against 0 for the queue pointer is not atomic at
         // all with respect to `run`, meaning that this could theoretically be a
@@ -59,6 +60,6 @@ pub fn run() {
     };
 
     for to_run in cur.into_iter() {
-        to_run();
+        to_run.invoke(());
     }
 }

src/librustrt/lib.rs

@@ -46,6 +46,7 @@ mod thread_local_storage;
 mod util;
 mod libunwind;
 mod stack_overflow;
+pub mod thunk;
 
 pub mod args;
 pub mod bookkeeping;
@@ -95,8 +96,8 @@ pub fn init(argc: int, argv: *const *const u8) {
 ///
 /// It is forbidden for procedures to register more `at_exit` handlers when they
 /// are running, and doing so will lead to a process abort.
-pub fn at_exit(f: proc():Send) {
-    at_exit_imp::push(f);
+pub fn at_exit<F:FnOnce()+Send>(f: F) {
+    at_exit_imp::push(thunk::Thunk::new(f));
 }
 
 /// One-time runtime cleanup.

src/librustrt/task.rs

@@ -21,6 +21,7 @@ use core::any::Any;
 use core::atomic::{AtomicUint, SeqCst};
 use core::iter::{IteratorExt, Take};
 use core::kinds::marker;
+use core::ops::FnOnce;
 use core::mem;
 use core::ops::FnMut;
 use core::prelude::{Clone, Drop, Err, Iterator, None, Ok, Option, Send, Some};
@@ -34,6 +35,7 @@ use stack;
 use unwind;
 use unwind::Unwinder;
 use collections::str::SendStr;
+use thunk::Thunk;
 
 /// State associated with Rust tasks.
 ///
@@ -67,7 +69,7 @@ enum TaskState {
 
 pub struct TaskOpts {
     /// Invoke this procedure with the result of the task when it finishes.
-    pub on_exit: Option<proc(Result): Send>,
+    pub on_exit: Option<Thunk<Result>>,
     /// A name for the task-to-be, for identification in panic messages
     pub name: Option<SendStr>,
     /// The size of the stack for the spawned task
@@ -92,7 +94,7 @@ pub enum BlockedTask {
 
 /// Per-task state related to task death, killing, panic, etc.
 pub struct Death {
-    pub on_exit: Option<proc(Result):Send>,
+    pub on_exit: Option<Thunk<Result>>,
     marker: marker::NoCopy,
 }
 
@@ -116,7 +118,13 @@ impl Task {
         }
     }
 
-    pub fn spawn(opts: TaskOpts, f: proc():Send) {
+    pub fn spawn<F>(opts: TaskOpts, f: F)
+        where F : FnOnce(), F : Send
+    {
+        Task::spawn_thunk(opts, Thunk::new(f))
+    }
+
+    fn spawn_thunk(opts: TaskOpts, f: Thunk) {
         let TaskOpts { name, stack_size, on_exit } = opts;
 
         let mut task = box Task::new(None, None);
@@ -138,7 +146,7 @@ impl Task {
         // because by the time that this function is executing we've already
         // consumed at least a little bit of stack (we don't know the exact byte
         // address at which our stack started).
-        Thread::spawn_stack(stack, proc() {
+        Thread::spawn_stack(stack, move|| {
             let something_around_the_top_of_the_stack = 1;
             let addr = &something_around_the_top_of_the_stack as *const int;
             let my_stack = addr as uint;
@@ -150,7 +158,7 @@ impl Task {
             task.stack_bounds = (my_stack - stack + 1024, my_stack);
 
             let mut f = Some(f);
-            drop(task.run(|| { f.take().unwrap()() }).destroy());
+            drop(task.run(|| { f.take().unwrap().invoke(()) }).destroy());
             drop(token);
         })
     }
@@ -241,7 +249,7 @@ impl Task {
         //        reconsideration to whether it's a reasonable thing to let a
         //        task to do or not.
         match what_to_do {
-            Some(f) => { f(result) }
+            Some(f) => { f.invoke(result) }
             None => { drop(result) }
         }
 
@@ -500,14 +508,13 @@ mod test {
     use super::*;
     use std::prelude::*;
     use std::task;
-    use unwind;
 
     #[test]
     fn unwind() {
-        let result = task::try(proc()());
+        let result = task::try(move|| ());
         rtdebug!("trying first assert");
         assert!(result.is_ok());
-        let result = task::try::<()>(proc() panic!());
+        let result = task::try(move|| -> () panic!());
         rtdebug!("trying second assert");
         assert!(result.is_err());
     }

src/librustrt/thread.rs

@@ -22,6 +22,7 @@ use alloc::boxed::Box;
 use core::mem;
 use core::uint;
 use libc;
+use thunk::{Thunk};
 
 use stack;
 use stack_overflow;
@@ -60,8 +61,8 @@ fn start_thread(main: *mut libc::c_void) -> imp::rust_thread_return {
     unsafe {
         stack::record_os_managed_stack_bounds(0, uint::MAX);
         let handler = stack_overflow::Handler::new();
-        let f: Box<proc()> = mem::transmute(main);
-        (*f)();
+        let f: Box<Thunk> = mem::transmute(main);
+        f.invoke(());
         drop(handler);
         mem::transmute(0 as imp::rust_thread_return)
     }
@@ -113,23 +114,29 @@ impl Thread<()> {
     /// to finish executing. This means that even if `join` is not explicitly
     /// called, when the `Thread` falls out of scope its destructor will block
     /// waiting for the OS thread.
-    pub fn start<T: Send>(main: proc():Send -> T) -> Thread<T> {
+    pub fn start<T,F>(main: F) -> Thread<T>
+        where T:Send, F:FnOnce() -> T, F:Send
+    {
         Thread::start_stack(DEFAULT_STACK_SIZE, main)
     }
 
     /// Performs the same functionality as `start`, but specifies an explicit
     /// stack size for the new thread.
-    pub fn start_stack<T: Send>(stack: uint, main: proc():Send -> T) -> Thread<T> {
-
+    pub fn start_stack<T, F>(stack: uint, main: F) -> Thread<T>
+        where T:Send, F:FnOnce() -> T, F:Send
+    {
         // We need the address of the packet to fill in to be stable so when
         // `main` fills it in it's still valid, so allocate an extra box to do
         // so.
         let packet = box None;
         let packet2: *mut Option<T> = unsafe {
             *mem::transmute::<&Box<Option<T>>, *const *mut Option<T>>(&packet)
         };
-        let main = proc() unsafe { *packet2 = Some(main()); };
-        let native = unsafe { imp::create(stack, box main) };
+        let native = unsafe {
+            imp::create(stack, Thunk::new(move |:| {
+                *packet2 = Some(main.call_once(()));
+            }))
+        };
 
         Thread {
             native: native,
@@ -144,15 +151,19 @@ impl Thread<()> {
     /// This corresponds to creating threads in the 'detached' state on unix
     /// systems. Note that platforms may not keep the main program alive even if
     /// there are detached thread still running around.
-    pub fn spawn(main: proc():Send) {
+    pub fn spawn<F>(main: F)
+        where F : FnOnce() + Send
+    {
         Thread::spawn_stack(DEFAULT_STACK_SIZE, main)
     }
 
     /// Performs the same functionality as `spawn`, but explicitly specifies a
     /// stack size for the new thread.
-    pub fn spawn_stack(stack: uint, main: proc():Send) {
+    pub fn spawn_stack<F>(stack: uint, main: F)
+        where F : FnOnce() + Send
+    {
         unsafe {
-            let handle = imp::create(stack, box main);
+            let handle = imp::create(stack, Thunk::new(main));
             imp::detach(handle);
         }
     }
@@ -190,8 +201,6 @@ impl<T: Send> Drop for Thread<T> {
 #[cfg(windows)]
 #[allow(non_snake_case)]
 mod imp {
-    use core::prelude::*;
-
     use alloc::boxed::Box;
     use core::cmp;
     use core::mem;
@@ -200,6 +209,7 @@ mod imp {
     use libc::types::os::arch::extra::{LPSECURITY_ATTRIBUTES, SIZE_T, BOOL,
                                        LPVOID, DWORD, LPDWORD, HANDLE};
     use stack::RED_ZONE;
+    use thunk::Thunk;
 
     pub type rust_thread = HANDLE;
     pub type rust_thread_return = DWORD;
@@ -217,8 +227,9 @@ mod imp {
         }
     }
 
-    pub unsafe fn create(stack: uint, p: Box<proc():Send>) -> rust_thread {
-        let arg: *mut libc::c_void = mem::transmute(p);
+    pub unsafe fn create(stack: uint, p: Thunk) -> rust_thread {
+        let arg: *mut libc::c_void = mem::transmute(box p);
+
         // FIXME On UNIX, we guard against stack sizes that are too small but
         // that's because pthreads enforces that stacks are at least
         // PTHREAD_STACK_MIN bytes big.  Windows has no such lower limit, it's
@@ -234,7 +245,7 @@ mod imp {
 
         if ret as uint == 0 {
             // be sure to not leak the closure
-            let _p: Box<proc():Send> = mem::transmute(arg);
+            let _p: Box<Thunk> = mem::transmute(arg);
             panic!("failed to spawn native thread: {}", ret);
         }
         return ret;
@@ -279,6 +290,7 @@ mod imp {
     use core::ptr;
     use libc::consts::os::posix01::{PTHREAD_CREATE_JOINABLE, PTHREAD_STACK_MIN};
     use libc;
+    use thunk::Thunk;
 
     use stack::RED_ZONE;
 
@@ -409,7 +421,7 @@ mod imp {
         }
     }
 
-    pub unsafe fn create(stack: uint, p: Box<proc():Send>) -> rust_thread {
+    pub unsafe fn create(stack: uint, p: Thunk) -> rust_thread {
         let mut native: libc::pthread_t = mem::zeroed();
         let mut attr: libc::pthread_attr_t = mem::zeroed();
         assert_eq!(pthread_attr_init(&mut attr), 0);
@@ -437,13 +449,13 @@ mod imp {
             },
         };
 
-        let arg: *mut libc::c_void = mem::transmute(p);
+        let arg: *mut libc::c_void = mem::transmute(box p); // must box since sizeof(p)=2*uint
         let ret = pthread_create(&mut native, &attr, super::thread_start, arg);
         assert_eq!(pthread_attr_destroy(&mut attr), 0);
 
         if ret != 0 {
             // be sure to not leak the closure
-            let _p: Box<proc():Send> = mem::transmute(arg);
+            let _p: Box<Box<FnOnce()+Send>> = mem::transmute(arg);
             panic!("failed to spawn native thread: {}", ret);
         }
         native
@@ -531,17 +543,17 @@ mod tests {
     use super::Thread;
 
     #[test]
-    fn smoke() { Thread::start(proc (){}).join(); }
+    fn smoke() { Thread::start(move|| {}).join(); }
 
     #[test]
-    fn data() { assert_eq!(Thread::start(proc () { 1i }).join(), 1); }
+    fn data() { assert_eq!(Thread::start(move|| { 1i }).join(), 1); }
 
     #[test]
-    fn detached() { Thread::spawn(proc () {}) }
+    fn detached() { Thread::spawn(move|| {}) }
 
     #[test]
     fn small_stacks() {
-        assert_eq!(42i, Thread::start_stack(0, proc () 42i).join());
-        assert_eq!(42i, Thread::start_stack(1, proc () 42i).join());
+        assert_eq!(42i, Thread::start_stack(0, move|| 42i).join());
+        assert_eq!(42i, Thread::start_stack(1, move|| 42i).join());
     }
 }

src/librustrt/thunk.rs

@@ -0,0 +1,52 @@
+// Copyright 2014 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use alloc::boxed::Box;
+use core::kinds::Send;
+use core::ops::FnOnce;
+
+pub struct Thunk<A=(),R=()> {
+    invoke: Box<Invoke<A,R>+Send>
+}
+
+impl<R> Thunk<(),R> {
+    pub fn new<F>(func: F) -> Thunk<(),R>
+        where F : FnOnce() -> R, F : Send
+    {
+        Thunk::with_arg(move|: ()| func())
+    }
+}
+
+impl<A,R> Thunk<A,R> {
+    pub fn with_arg<F>(func: F) -> Thunk<A,R>
+        where F : FnOnce(A) -> R, F : Send
+    {
+        Thunk {
+            invoke: box func
+        }
+    }
+
+    pub fn invoke(self, arg: A) -> R {
+        self.invoke.invoke(arg)
+    }
+}
+
+pub trait Invoke<A=(),R=()> {
+    fn invoke(self: Box<Self>, arg: A) -> R;
+}
+
+impl<A,R,F> Invoke<A,R> for F
+    where F : FnOnce(A) -> R
+{
+    fn invoke(self: Box<F>, arg: A) -> R {
+        let f = *self;
+        f(arg)
+    }
+}

src/libstd/lib.rs

@@ -171,6 +171,8 @@ pub use rustrt::c_str;
 
 pub use unicode::char;
 
+pub use rustrt::thunk;
+
 /* Exported macros */
 
 pub mod macros;