Finalize the green::Pool type

The scheduler pool now has a much more simplified interface. There is now a
clear distinction between creating the pool and then interacting the pool. When
a pool is created, all schedulers are not active, and only later if a spawn is
done does activity occur.

There are four operations that you can do on a pool:

1. Create a new pool. The only argument to this function is the configuration
   for the scheduler pool. Currently the only configuration parameter is the
   number of threads to initially spawn.

2. Spawn a task into this pool. This takes a procedure and task configuration
   options and spawns a new task into the pool of schedulers.

3. Spawn a new scheduler into the pool. This will return a handle on which to
   communicate with the scheduler in order to do something like a pinned task.

4. Shut down the scheduler pool. This will consume the scheduler pool, request
   all of the schedulers to shut down, and then wait on all the scheduler
   threads. Currently this will block the invoking OS thread, but I plan on
   making 'Thread::join' not a thread-blocking call.

These operations can be used to encode all current usage of M:N schedulers, as
well as providing a simple interface through which a pool can be modified. There
is currently no way to remove a scheduler from a pool of scheduler, as there's
no way to guarantee that a scheduler has exited. This may be added in the
future, however (as necessary).

1 files changed, 19 insertions, 1 deletions

diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs
index c0164891cd4..91e285b1061 100644
--- a/src/libstd/rt/task.rs
+++ b/src/libstd/rt/task.rs
@@ -33,12 +33,16 @@ use rt::rtio::LocalIo;
 use rt::unwind::Unwinder;
 use send_str::SendStr;
 use sync::arc::UnsafeArc;
-use sync::atomics::{AtomicUint, SeqCst};
+use sync::atomics::{AtomicUint, SeqCst, INIT_ATOMIC_UINT};
 use task::{TaskResult, TaskOpts};
 use unstable::finally::Finally;
+use unstable::mutex::{Mutex, MUTEX_INIT};
 
 #[cfg(stage0)] pub use rt::unwind::begin_unwind;
 
+static mut TASK_COUNT: AtomicUint = INIT_ATOMIC_UINT;
+static mut TASK_LOCK: Mutex = MUTEX_INIT;
+
 // The Task struct represents all state associated with a rust
 // task. There are at this point two primary "subtypes" of task,
 // however instead of using a subtype we just have a "task_type" field
@@ -117,6 +121,7 @@ impl Task {
             *cast::transmute::<&~Task, &*mut Task>(&self)
         };
         Local::put(self);
+        unsafe { TASK_COUNT.fetch_add(1, SeqCst); }
 
         // The only try/catch block in the world. Attempt to run the task's
         // client-specified code and catch any failures.
@@ -180,6 +185,11 @@ impl Task {
         unsafe {
             let me: *mut Task = Local::unsafe_borrow();
             (*me).death.collect_failure((*me).unwinder.result());
+            if TASK_COUNT.fetch_sub(1, SeqCst) == 1 {
+                TASK_LOCK.lock();
+                TASK_LOCK.signal();
+                TASK_LOCK.unlock();
+            }
         }
         let mut me: ~Task = Local::take();
         me.destroyed = true;
@@ -376,6 +386,14 @@ impl Drop for Death {
     }
 }
 
+pub unsafe fn wait_for_completion() {
+    TASK_LOCK.lock();
+    while TASK_COUNT.load(SeqCst) > 0 {
+        TASK_LOCK.wait();
+    }
+    TASK_LOCK.unlock();
+}
+
 #[cfg(test)]
 mod test {
     use super::*;