Implement a lock-free work-stealing deque

This adds an implementation of the Chase-Lev work-stealing deque to libstd
under std::rt::deque. I've been unable to break the implementation of the deque
itself, and it's not super highly optimized just yet (everything uses a SeqCst
memory ordering).

The major snag in implementing the chase-lev deque is that the buffers used to
store data internally cannot get deallocated back to the OS. In the meantime, a
shared buffer pool (synchronized by a normal mutex) is used to
deallocate/allocate buffers from. This is done in hope of not overcommitting too
much memory. It is in theory possible to eventually free the buffers, but one
must be very careful in doing so.

I was unable to get some good numbers from src/test/bench tests (I don't think
many of them are slamming the work queue that much), but I was able to get some
good numbers from one of my own tests. In a recent rewrite of select::select(),
I found that my implementation was incredibly slow due to contention on the
shared work queue. Upon switching to the parallel deque, I saw the contention
drop to 0 and the runtime go from 1.6s to 0.9s with the most amount of time
spent in libuv awakening the schedulers (plus allocations).

Closes #4877

Author: alexcrichton

src/librustuv/net.rs

@@ -1076,21 +1076,22 @@ mod test {
         use std::rt::task::Task;
         use std::rt::task::UnwindResult;
         use std::rt::thread::Thread;
-        use std::rt::work_queue::WorkQueue;
+        use std::rt::deque::BufferPool;
         use std::unstable::run_in_bare_thread;
         use uvio::UvEventLoop;
 
         do run_in_bare_thread {
             let sleepers = SleeperList::new();
-            let work_queue1 = WorkQueue::new();
-            let work_queue2 = WorkQueue::new();
-            let queues = ~[work_queue1.clone(), work_queue2.clone()];
+            let mut pool = BufferPool::init();
+            let (worker1, stealer1) = pool.deque();
+            let (worker2, stealer2) = pool.deque();
+            let queues = ~[stealer1, stealer2];
 
             let loop1 = ~UvEventLoop::new() as ~EventLoop;
-            let mut sched1 = ~Scheduler::new(loop1, work_queue1, queues.clone(),
+            let mut sched1 = ~Scheduler::new(loop1, worker1, queues.clone(),
                                              sleepers.clone());
             let loop2 = ~UvEventLoop::new() as ~EventLoop;
-            let mut sched2 = ~Scheduler::new(loop2, work_queue2, queues.clone(),
+            let mut sched2 = ~Scheduler::new(loop2, worker2, queues.clone(),
                                              sleepers.clone());
 
             let handle1 = Cell::new(sched1.make_handle());

src/libstd/rt/deque.rs

@@ -75,7 +75,6 @@ use vec::{OwnedVector, MutableVector, ImmutableVector};
 use vec;
 
 use self::thread::Thread;
-use self::work_queue::WorkQueue;
 
 // the os module needs to reach into this helper, so allow general access
 // through this reexport.
@@ -130,9 +129,6 @@ pub mod rtio;
 /// or task-local storage.
 pub mod local;
 
-/// A parallel work-stealing deque.
-pub mod work_queue;
-
 /// A parallel queue.
 pub mod message_queue;
 
@@ -142,6 +138,9 @@ mod mpsc_queue;
 /// A lock-free multi-producer, multi-consumer bounded queue.
 mod mpmc_bounded_queue;
 
+/// A parallel work-stealing deque
+pub mod deque;
+
 /// A parallel data structure for tracking sleeping schedulers.
 pub mod sleeper_list;
 
@@ -287,22 +286,24 @@ fn run_(main: proc(), use_main_sched: bool) -> int {
 
     // Create a work queue for each scheduler, ntimes. Create an extra
     // for the main thread if that flag is set. We won't steal from it.
-    let work_queues: ~[WorkQueue<~Task>] = vec::from_fn(nscheds, |_| WorkQueue::new());
+    let mut pool = deque::BufferPool::init();
+    let arr = vec::from_fn(nscheds, |_| pool.deque());
+    let (workers, stealers) = vec::unzip(arr.move_iter());
 
     // The schedulers.
     let mut scheds = ~[];
     // Handles to the schedulers. When the main task ends these will be
     // sent the Shutdown message to terminate the schedulers.
     let mut handles = ~[];
 
-    for work_queue in work_queues.iter() {
+    for worker in workers.move_iter() {
         rtdebug!("inserting a regular scheduler");
 
         // Every scheduler is driven by an I/O event loop.
         let loop_ = new_event_loop();
         let mut sched = ~Scheduler::new(loop_,
-                                        work_queue.clone(),
-                                        work_queues.clone(),
+                                        worker,
+                                        stealers.clone(),
                                         sleepers.clone());
         let handle = sched.make_handle();
 
@@ -321,12 +322,12 @@ fn run_(main: proc(), use_main_sched: bool) -> int {
 
         // This scheduler needs a queue that isn't part of the stealee
         // set.
-        let work_queue = WorkQueue::new();
+        let (worker, _) = pool.deque();
 
         let main_loop = new_event_loop();
         let mut main_sched = ~Scheduler::new_special(main_loop,
-                                                     work_queue,
-                                                     work_queues.clone(),
+                                                     worker,
+                                                     stealers.clone(),
                                                      sleepers.clone(),
                                                      false,
                                                      Some(friend_handle));

src/libstd/rt/mod.rs

@@ -13,13 +13,13 @@ use cast::{transmute, transmute_mut_region, transmute_mut_unsafe};
 use clone::Clone;
 use unstable::raw;
 use super::sleeper_list::SleeperList;
-use super::work_queue::WorkQueue;
 use super::stack::{StackPool};
 use super::rtio::EventLoop;
 use super::context::Context;
 use super::task::{Task, AnySched, Sched};
 use super::message_queue::MessageQueue;
 use rt::kill::BlockedTask;
+use rt::deque;
 use rt::local_ptr;
 use rt::local::Local;
 use rt::rtio::{RemoteCallback, PausibleIdleCallback, Callback};
@@ -39,14 +39,14 @@ use vec::{OwnedVector};
 /// in too much allocation and too many events.
 pub struct Scheduler {
     /// There are N work queues, one per scheduler.
-    priv work_queue: WorkQueue<~Task>,
+    work_queue: deque::Worker<~Task>,
     /// Work queues for the other schedulers. These are created by
     /// cloning the core work queues.
-    work_queues: ~[WorkQueue<~Task>],
+    work_queues: ~[deque::Stealer<~Task>],
     /// The queue of incoming messages from other schedulers.
     /// These are enqueued by SchedHandles after which a remote callback
     /// is triggered to handle the message.
-    priv message_queue: MessageQueue<SchedMessage>,
+    message_queue: MessageQueue<SchedMessage>,
     /// A shared list of sleeping schedulers. We'll use this to wake
     /// up schedulers when pushing work onto the work queue.
     sleeper_list: SleeperList,
@@ -56,33 +56,33 @@ pub struct Scheduler {
     /// not active since there are multiple event sources that may
     /// wake the scheduler. It just prevents the scheduler from pushing
     /// multiple handles onto the sleeper list.
-    priv sleepy: bool,
+    sleepy: bool,
     /// A flag to indicate we've received the shutdown message and should
     /// no longer try to go to sleep, but exit instead.
     no_sleep: bool,
     stack_pool: StackPool,
     /// The scheduler runs on a special task. When it is not running
     /// it is stored here instead of the work queue.
-    priv sched_task: Option<~Task>,
+    sched_task: Option<~Task>,
     /// An action performed after a context switch on behalf of the
     /// code running before the context switch
-    priv cleanup_job: Option<CleanupJob>,
+    cleanup_job: Option<CleanupJob>,
     /// Should this scheduler run any task, or only pinned tasks?
     run_anything: bool,
     /// If the scheduler shouldn't run some tasks, a friend to send
     /// them to.
-    priv friend_handle: Option<SchedHandle>,
+    friend_handle: Option<SchedHandle>,
     /// A fast XorShift rng for scheduler use
     rng: XorShiftRng,
     /// A toggleable idle callback
-    priv idle_callback: Option<~PausibleIdleCallback>,
+    idle_callback: Option<~PausibleIdleCallback>,
     /// A countdown that starts at a random value and is decremented
     /// every time a yield check is performed. When it hits 0 a task
     /// will yield.
-    priv yield_check_count: uint,
+    yield_check_count: uint,
     /// A flag to tell the scheduler loop it needs to do some stealing
     /// in order to introduce randomness as part of a yield
-    priv steal_for_yield: bool,
+    steal_for_yield: bool,
 
     // n.b. currently destructors of an object are run in top-to-bottom in order
     //      of field declaration. Due to its nature, the pausible idle callback
@@ -115,8 +115,8 @@ impl Scheduler {
     // * Initialization Functions
 
     pub fn new(event_loop: ~EventLoop,
-               work_queue: WorkQueue<~Task>,
-               work_queues: ~[WorkQueue<~Task>],
+               work_queue: deque::Worker<~Task>,
+               work_queues: ~[deque::Stealer<~Task>],
                sleeper_list: SleeperList)
         -> Scheduler {
 
@@ -127,8 +127,8 @@ impl Scheduler {
     }
 
     pub fn new_special(event_loop: ~EventLoop,
-                       work_queue: WorkQueue<~Task>,
-                       work_queues: ~[WorkQueue<~Task>],
+                       work_queue: deque::Worker<~Task>,
+                       work_queues: ~[deque::Stealer<~Task>],
                        sleeper_list: SleeperList,
                        run_anything: bool,
                        friend: Option<SchedHandle>)
@@ -440,11 +440,11 @@ impl Scheduler {
         let start_index = self.rng.gen_range(0, len);
         for index in range(0, len).map(|i| (i + start_index) % len) {
             match work_queues[index].steal() {
-                Some(task) => {
+                deque::Data(task) => {
                     rtdebug!("found task by stealing");
                     return Some(task)
                 }
-                None => ()
+                _ => ()
             }
         };
         rtdebug!("giving up on stealing");
@@ -889,6 +889,7 @@ mod test {
     use borrow::to_uint;
     use rt::sched::{Scheduler};
     use cell::Cell;
+    use rt::deque::BufferPool;
     use rt::thread::Thread;
     use rt::task::{Task, Sched};
     use rt::basic;
@@ -994,22 +995,22 @@ mod test {
     #[test]
     fn test_schedule_home_states() {
         use rt::sleeper_list::SleeperList;
-        use rt::work_queue::WorkQueue;
         use rt::sched::Shutdown;
         use borrow;
         use rt::comm::*;
 
         do run_in_bare_thread {
 
             let sleepers = SleeperList::new();
-            let normal_queue = WorkQueue::new();
-            let special_queue = WorkQueue::new();
-            let queues = ~[normal_queue.clone(), special_queue.clone()];
+            let mut pool = BufferPool::init();
+            let (normal_worker, normal_stealer) = pool.deque();
+            let (special_worker, special_stealer) = pool.deque();
+            let queues = ~[normal_stealer, special_stealer];
 
             // Our normal scheduler
             let mut normal_sched = ~Scheduler::new(
                 basic::event_loop(),
-                normal_queue,
+                normal_worker,
                 queues.clone(),
                 sleepers.clone());
 
@@ -1020,7 +1021,7 @@ mod test {
             // Our special scheduler
             let mut special_sched = ~Scheduler::new_special(
                 basic::event_loop(),
-                special_queue.clone(),
+                special_worker,
                 queues.clone(),
                 sleepers.clone(),
                 false,
@@ -1169,7 +1170,6 @@ mod test {
     // Used to deadlock because Shutdown was never recvd.
     #[test]
     fn no_missed_messages() {
-        use rt::work_queue::WorkQueue;
         use rt::sleeper_list::SleeperList;
         use rt::stack::StackPool;
         use rt::sched::{Shutdown, TaskFromFriend};
@@ -1178,13 +1178,13 @@ mod test {
         do run_in_bare_thread {
             stress_factor().times(|| {
                 let sleepers = SleeperList::new();
-                let queue = WorkQueue::new();
-                let queues = ~[queue.clone()];
+                let mut pool = BufferPool::init();
+                let (worker, stealer) = pool.deque();
 
                 let mut sched = ~Scheduler::new(
                     basic::event_loop(),
-                    queue,
-                    queues.clone(),
+                    worker,
+                    ~[stealer],
                     sleepers.clone());
 
                 let mut handle = sched.make_handle();

src/libstd/rt/sched.rs

@@ -23,24 +23,25 @@ use rand;
 use result::{Result, Ok, Err};
 use rt::basic;
 use rt::comm::oneshot;
+use rt::deque::BufferPool;
 use rt::new_event_loop;
 use rt::sched::Scheduler;
 use rt::sleeper_list::SleeperList;
 use rt::task::Task;
 use rt::task::UnwindResult;
 use rt::thread::Thread;
-use rt::work_queue::WorkQueue;
 use unstable::{run_in_bare_thread};
+use vec;
 use vec::{OwnedVector, MutableVector, ImmutableVector};
 
 pub fn new_test_uv_sched() -> Scheduler {
 
-    let queue = WorkQueue::new();
-    let queues = ~[queue.clone()];
+    let mut pool = BufferPool::init();
+    let (worker, stealer) = pool.deque();
 
     let mut sched = Scheduler::new(new_event_loop(),
-                                   queue,
-                                   queues,
+                                   worker,
+                                   ~[stealer],
                                    SleeperList::new());
 
     // Don't wait for the Shutdown message
@@ -50,13 +51,12 @@ pub fn new_test_uv_sched() -> Scheduler {
 }
 
 pub fn new_test_sched() -> Scheduler {
-
-    let queue = WorkQueue::new();
-    let queues = ~[queue.clone()];
+    let mut pool = BufferPool::init();
+    let (worker, stealer) = pool.deque();
 
     let mut sched = Scheduler::new(basic::event_loop(),
-                                   queue,
-                                   queues,
+                                   worker,
+                                   ~[stealer],
                                    SleeperList::new());
 
     // Don't wait for the Shutdown message
@@ -227,18 +227,16 @@ pub fn run_in_mt_newsched_task(f: proc()) {
 
         let mut handles = ~[];
         let mut scheds = ~[];
-        let mut work_queues = ~[];
 
-        for _ in range(0u, nthreads) {
-            let work_queue = WorkQueue::new();
-            work_queues.push(work_queue);
-        }
+        let mut pool = BufferPool::<~Task>::init();
+        let workers = range(0, nthreads).map(|_| pool.deque());
+        let (workers, stealers) = vec::unzip(workers);
 
-        for i in range(0u, nthreads) {
+        for worker in workers.move_iter() {
             let loop_ = new_event_loop();
             let mut sched = ~Scheduler::new(loop_,
-                                            work_queues[i].clone(),
-                                            work_queues.clone(),
+                                            worker,
+                                            stealers.clone(),
                                             sleepers.clone());
             let handle = sched.make_handle();
 

src/libstd/rt/test.rs

@@ -84,7 +84,6 @@ use rt::sched::{Scheduler, Shutdown, TaskFromFriend};
 use rt::task::{Task, Sched};
 use rt::task::UnwindResult;
 use rt::thread::Thread;
-use rt::work_queue::WorkQueue;
 use rt::{in_green_task_context, new_event_loop};
 use task::SingleThreaded;
 use task::TaskOpts;
@@ -111,11 +110,11 @@ pub fn spawn_raw(mut opts: TaskOpts, f: proc()) {
             // Since this is a 1:1 scheduler we create a queue not in
             // the stealee set. The run_anything flag is set false
             // which will disable stealing.
-            let work_queue = WorkQueue::new();
+            let (worker, _stealer) = (*sched).work_queue.pool().deque();
 
             // Create a new scheduler to hold the new task
             let mut new_sched = ~Scheduler::new_special(new_event_loop(),
-                                                        work_queue,
+                                                        worker,
                                                         (*sched).work_queues.clone(),
                                                         (*sched).sleeper_list.clone(),
                                                         false,