Scheduler multithreading

src/libstd/rt/io/extensions.rs

@@ -749,8 +749,6 @@ mod test {
     #[should_fail]
     #[ignore(cfg(windows))]
     fn push_bytes_fail_reset_len() {
-        use unstable::finally::Finally;
-
         // push_bytes unsafely sets the vector length. This is testing that
         // upon failure the length is reset correctly.
         let mut reader = MockReader::new();
@@ -772,7 +770,8 @@ mod test {
             reader.push_bytes(&mut *buf, 4);
         }).finally {
             // NB: Using rtassert here to trigger abort on failure since this is a should_fail test
-            rtassert!(*buf == ~[8, 9, 10]);
+            // FIXME: #7049 This fails because buf is still borrowed
+            //rtassert!(*buf == ~[8, 9, 10]);
         }
     }
 

src/libstd/rt/local.rs

@@ -85,30 +85,30 @@ impl Local for IoFactoryObject {
 
 #[cfg(test)]
 mod test {
+    use rt::test::*;
     use rt::sched::Scheduler;
-    use rt::uv::uvio::UvEventLoop;
     use super::*;
 
     #[test]
     fn thread_local_scheduler_smoke_test() {
-        let scheduler = ~UvEventLoop::new_scheduler();
+        let scheduler = ~new_test_uv_sched();
         Local::put(scheduler);
         let _scheduler: ~Scheduler = Local::take();
     }
 
     #[test]
     fn thread_local_scheduler_two_instances() {
-        let scheduler = ~UvEventLoop::new_scheduler();
+        let scheduler = ~new_test_uv_sched();
         Local::put(scheduler);
         let _scheduler: ~Scheduler = Local::take();
-        let scheduler = ~UvEventLoop::new_scheduler();
+        let scheduler = ~new_test_uv_sched();
         Local::put(scheduler);
         let _scheduler: ~Scheduler = Local::take();
     }
 
     #[test]
     fn borrow_smoke_test() {
-        let scheduler = ~UvEventLoop::new_scheduler();
+        let scheduler = ~new_test_uv_sched();
         Local::put(scheduler);
         unsafe {
             let _scheduler: *mut Scheduler = Local::unsafe_borrow();

src/libstd/rt/message_queue.rs

@@ -8,6 +8,9 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+//! A concurrent queue that supports multiple producers and a
+//! single consumer.
+
 use container::Container;
 use kinds::Owned;
 use vec::OwnedVector;

src/libstd/rt/metrics.rs

@@ -0,0 +1,88 @@
+// Copyright 2013 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 to_str::ToStr;
+
+pub struct SchedMetrics {
+    // The number of times executing `run_sched_once`.
+    turns: uint,
+    // The number of turns that received a message.
+    messages_received: uint,
+    // The number of turns that ran a task from the queue.
+    tasks_resumed_from_queue: uint,
+    // The number of turns that found no work to perform.
+    wasted_turns: uint,
+    // The number of times the scheduler went to sleep.
+    sleepy_times: uint,
+    // Context switches from the scheduler into a task.
+    context_switches_sched_to_task: uint,
+    // Context switches from a task into the scheduler.
+    context_switches_task_to_sched: uint,
+    // Context switches from a task to a task.
+    context_switches_task_to_task: uint,
+    // Message sends that unblock the receiver
+    rendezvous_sends: uint,
+    // Message sends that do not unblock the receiver
+    non_rendezvous_sends: uint,
+    // Message receives that do not block the receiver
+    rendezvous_recvs: uint,
+    // Message receives that block the receiver
+    non_rendezvous_recvs: uint
+}
+
+impl SchedMetrics {
+    pub fn new() -> SchedMetrics {
+        SchedMetrics {
+            turns: 0,
+            messages_received: 0,
+            tasks_resumed_from_queue: 0,
+            wasted_turns: 0,
+            sleepy_times: 0,
+            context_switches_sched_to_task: 0,
+            context_switches_task_to_sched: 0,
+            context_switches_task_to_task: 0,
+            rendezvous_sends: 0,
+            non_rendezvous_sends: 0,
+            rendezvous_recvs: 0,
+            non_rendezvous_recvs: 0
+        }
+    }
+}
+
+impl ToStr for SchedMetrics {
+    fn to_str(&self) -> ~str {
+        fmt!("turns: %u\n\
+              messages_received: %u\n\
+              tasks_resumed_from_queue: %u\n\
+              wasted_turns: %u\n\
+              sleepy_times: %u\n\
+              context_switches_sched_to_task: %u\n\
+              context_switches_task_to_sched: %u\n\
+              context_switches_task_to_task: %u\n\
+              rendezvous_sends: %u\n\
+              non_rendezvous_sends: %u\n\
+              rendezvous_recvs: %u\n\
+              non_rendezvous_recvs: %u\n\
+              ",
+             self.turns,
+             self.messages_received,
+             self.tasks_resumed_from_queue,
+             self.wasted_turns,
+             self.sleepy_times,
+             self.context_switches_sched_to_task,
+             self.context_switches_task_to_sched,
+             self.context_switches_task_to_task,
+             self.rendezvous_sends,
+             self.non_rendezvous_sends,
+             self.rendezvous_recvs,
+             self.non_rendezvous_recvs
+        )
+    }
+}

src/libstd/rt/mod.rs

@@ -55,6 +55,9 @@ Several modules in `core` are clients of `rt`:
 */
 
 #[doc(hidden)];
+#[deny(unused_imports)];
+#[deny(unused_mut)];
+#[deny(unused_variable)];
 
 use ptr::Ptr;
 
@@ -88,6 +91,9 @@ mod work_queue;
 /// A parallel queue.
 mod message_queue;
 
+/// A parallel data structure for tracking sleeping schedulers.
+mod sleeper_list;
+
 /// Stack segments and caching.
 mod stack;
 
@@ -127,6 +133,8 @@ pub mod local_ptr;
 /// Bindings to pthread/windows thread-local storage.
 pub mod thread_local_storage;
 
+pub mod metrics;
+
 
 /// Set up a default runtime configuration, given compiler-supplied arguments.
 ///
@@ -145,12 +153,17 @@ pub mod thread_local_storage;
 pub fn start(_argc: int, _argv: **u8, crate_map: *u8, main: ~fn()) -> int {
 
     use self::sched::{Scheduler, Coroutine};
+    use self::work_queue::WorkQueue;
     use self::uv::uvio::UvEventLoop;
+    use self::sleeper_list::SleeperList;
 
     init(crate_map);
 
     let loop_ = ~UvEventLoop::new();
-    let mut sched = ~Scheduler::new(loop_);
+    let work_queue = WorkQueue::new();
+    let sleepers = SleeperList::new();
+    let mut sched = ~Scheduler::new(loop_, work_queue, sleepers);
+    sched.no_sleep = true;
     let main_task = ~Coroutine::new(&mut sched.stack_pool, main);
 
     sched.enqueue_task(main_task);
@@ -218,23 +231,19 @@ pub fn context() -> RuntimeContext {
 fn test_context() {
     use unstable::run_in_bare_thread;
     use self::sched::{Scheduler, Coroutine};
-    use rt::uv::uvio::UvEventLoop;
-    use cell::Cell;
     use rt::local::Local;
+    use rt::test::new_test_uv_sched;
 
     assert_eq!(context(), OldTaskContext);
     do run_in_bare_thread {
         assert_eq!(context(), GlobalContext);
-        let mut sched = ~UvEventLoop::new_scheduler();
+        let mut sched = ~new_test_uv_sched();
         let task = ~do Coroutine::new(&mut sched.stack_pool) {
             assert_eq!(context(), TaskContext);
             let sched = Local::take::<Scheduler>();
-            do sched.deschedule_running_task_and_then() |task| {
+            do sched.deschedule_running_task_and_then() |sched, task| {
                 assert_eq!(context(), SchedulerContext);
-                let task = Cell(task);
-                do Local::borrow::<Scheduler> |sched| {
-                    sched.enqueue_task(task.take());
-                }
+                sched.enqueue_task(task);
             }
         };
         sched.enqueue_task(task);

src/libstd/rt/rtio.rs

@@ -18,6 +18,7 @@ use rt::uv::uvio;
 // XXX: ~object doesn't work currently so these are some placeholder
 // types to use instead
 pub type EventLoopObject = uvio::UvEventLoop;
+pub type RemoteCallbackObject = uvio::UvRemoteCallback;
 pub type IoFactoryObject = uvio::UvIoFactory;
 pub type RtioTcpStreamObject = uvio::UvTcpStream;
 pub type RtioTcpListenerObject = uvio::UvTcpListener;
@@ -26,10 +27,20 @@ pub trait EventLoop {
     fn run(&mut self);
     fn callback(&mut self, ~fn());
     fn callback_ms(&mut self, ms: u64, ~fn());
+    fn remote_callback(&mut self, ~fn()) -> ~RemoteCallbackObject;
     /// The asynchronous I/O services. Not all event loops may provide one
     fn io<'a>(&'a mut self) -> Option<&'a mut IoFactoryObject>;
 }
 
+pub trait RemoteCallback {
+    /// Trigger the remote callback. Note that the number of times the callback
+    /// is run is not guaranteed. All that is guaranteed is that, after calling 'fire',
+    /// the callback will be called at least once, but multiple callbacks may be coalesced
+    /// and callbacks may be called more often requested. Destruction also triggers the
+    /// callback.
+    fn fire(&mut self);
+}
+
 pub trait IoFactory {
     fn tcp_connect(&mut self, addr: IpAddr) -> Result<~RtioTcpStreamObject, IoError>;
     fn tcp_bind(&mut self, addr: IpAddr) -> Result<~RtioTcpListenerObject, IoError>;