V2: now with more locks!

Author: Blei

src/libcore/task.rs

@@ -1661,8 +1661,8 @@ extern mod rustrt {
 
     fn rust_get_sched_id() -> sched_id;
     fn rust_new_sched(num_threads: libc::uintptr_t) -> sched_id;
-    fn rust_max_sched_threads() -> libc::size_t;
     fn rust_sched_threads() -> libc::size_t;
+    fn rust_sched_current_nonlazy_threads() -> libc::size_t;
     fn rust_num_threads() -> libc::uintptr_t;
 
     fn get_task_id() -> task_id;
@@ -2436,7 +2436,7 @@ fn test_sched_thread_per_core() {
 
     do spawn_sched(ThreadPerCore) {
         let cores = rustrt::rust_num_threads();
-        let reported_threads = rustrt::rust_max_sched_threads();
+        let reported_threads = rustrt::rust_sched_threads();
         assert(cores as uint == reported_threads as uint);
         chan.send(());
     }
@@ -2449,9 +2449,9 @@ fn test_spawn_thread_on_demand() {
     let (chan, port) = pipes::stream();
 
     do spawn_sched(ManualThreads(2)) {
-        let max_threads = rustrt::rust_max_sched_threads();
+        let max_threads = rustrt::rust_sched_threads();
         assert(max_threads as int == 2);
-        let running_threads = rustrt::rust_sched_threads();
+        let running_threads = rustrt::rust_sched_current_nonlazy_threads();
         assert(running_threads as int == 1);
 
         let (chan2, port2) = pipes::stream();
@@ -2460,7 +2460,7 @@ fn test_spawn_thread_on_demand() {
             chan2.send(());
         }
 
-        let running_threads2 = rustrt::rust_sched_threads();
+        let running_threads2 = rustrt::rust_sched_current_nonlazy_threads();
         assert(running_threads2 as int == 2);
 
         port2.recv();

src/libstd/test.rs

@@ -26,7 +26,7 @@ export run_tests_console;
 
 #[abi = "cdecl"]
 extern mod rustrt {
-    fn rust_max_sched_threads() -> libc::size_t;
+    fn rust_sched_threads() -> libc::size_t;
 }
 
 // The name of a test. By convention this follows the rules for rust
@@ -327,7 +327,7 @@ const sched_overcommit : uint = 1u;
 const sched_overcommit : uint = 4u;
 
 fn get_concurrency() -> uint {
-    let threads = rustrt::rust_max_sched_threads() as uint;
+    let threads = rustrt::rust_sched_threads() as uint;
     if threads == 1u { 1u }
     else { threads * sched_overcommit }
 }

src/rt/rust_builtin.cpp

@@ -627,13 +627,13 @@ start_task(rust_task *target, fn_env_pair *f) {
 }
 
 extern "C" CDECL size_t
-rust_sched_threads() {
+rust_sched_current_nonlazy_threads() {
     rust_task *task = rust_get_current_task();
     return task->sched->number_of_threads();
 }
 
 extern "C" CDECL size_t
-rust_max_sched_threads() {
+rust_sched_threads() {
     rust_task *task = rust_get_current_task();
     return task->sched->max_number_of_threads();
 }

src/rt/rust_scheduler.cpp

@@ -23,6 +23,7 @@ rust_scheduler::rust_scheduler(rust_kernel *kernel,
     id(id)
 {
     // Create the first thread
+    scoped_lock with(lock);
     threads.push(create_task_thread(0));
 }
 
@@ -34,6 +35,7 @@ void rust_scheduler::delete_this() {
 
 rust_sched_launcher *
 rust_scheduler::create_task_thread(int id) {
+    lock.must_have_lock();
     live_threads++;
     rust_sched_launcher *thread = launchfac->create(this, id, killed);
     KLOG(kernel, kern, "created task thread: " PTR
@@ -50,6 +52,7 @@ rust_scheduler::destroy_task_thread(rust_sched_launcher *thread) {
 
 void
 rust_scheduler::destroy_task_threads() {
+    scoped_lock with(lock);
     for(size_t i = 0; i < threads.size(); ++i) {
         destroy_task_thread(threads[i]);
     }
@@ -58,6 +61,7 @@ rust_scheduler::destroy_task_threads() {
 void
 rust_scheduler::start_task_threads()
 {
+    scoped_lock with(lock);
     for(size_t i = 0; i < threads.size(); ++i) {
         rust_sched_launcher *thread = threads[i];
         thread->start();
@@ -67,6 +71,7 @@ rust_scheduler::start_task_threads()
 void
 rust_scheduler::join_task_threads()
 {
+    scoped_lock with(lock);
     for(size_t i = 0; i < threads.size(); ++i) {
         rust_sched_launcher *thread = threads[i];
         thread->join();
@@ -75,8 +80,16 @@ rust_scheduler::join_task_threads()
 
 void
 rust_scheduler::kill_all_tasks() {
-    for(size_t i = 0; i < threads.size(); ++i) {
-        rust_sched_launcher *thread = threads[i];
+    array_list<rust_sched_launcher *> copied_threads;
+    {
+        scoped_lock with(lock);
+        killed = true;
+        for (size_t i = 0; i < threads.size(); ++i) {
+            copied_threads.push(threads[i]);
+        }
+    }
+    for(size_t i = 0; i < copied_threads.size(); ++i) {
+        rust_sched_launcher *thread = copied_threads[i];
         thread->get_loop()->kill_all_tasks();
     }
 }
@@ -88,26 +101,16 @@ rust_scheduler::create_task(rust_task *spawner, const char *name) {
         scoped_lock with(lock);
         live_tasks++;
 
-        // Find unoccupied thread
-        for (thread_no = 0; thread_no < threads.size(); ++thread_no) {
-            if (threads[thread_no]->get_loop()->number_of_live_tasks() == 0)
-                break;
-        }
-
-        if (thread_no == threads.size()) {
-            if (threads.size() < max_num_threads) {
-                // Else create new thread
-                thread_no = threads.size();
-                rust_sched_launcher *thread = create_task_thread(thread_no);
-                thread->start();
-                threads.push(thread);
-            } else {
-                // Or use round robin allocation
-                thread_no = cur_thread++;
-                if (cur_thread >= max_num_threads)
-                    cur_thread = 0;
-            }
+        if (cur_thread < threads.size()) {
+            thread_no = cur_thread;
+        } else {
+            assert(threads.size() < max_num_threads);
+            thread_no = threads.size();
+            rust_sched_launcher *thread = create_task_thread(thread_no);
+            thread->start();
+            threads.push(thread);
         }
+        cur_thread = (thread_no + 1) % max_num_threads;
     }
     KLOG(kernel, kern, "Creating task %s, on thread %d.", name, thread_no);
     kernel->register_task();
@@ -135,6 +138,12 @@ void
 rust_scheduler::exit() {
     // Take a copy of the number of threads. After the last thread exits this
     // scheduler will get destroyed, and our fields will cease to exist.
+    //
+    // This is also the reason we can't use the lock here (as in the other
+    // cases when accessing `threads`), after the loop the lock won't exist
+    // anymore. This is safe because this method is only called when all the
+    // task are dead, so there is no chance of a task trying to create new
+    // threads.
     size_t current_num_threads = threads.size();
     for(size_t i = 0; i < current_num_threads; ++i) {
         threads[i]->get_loop()->exit();
@@ -148,6 +157,7 @@ rust_scheduler::max_number_of_threads() {
 
 size_t
 rust_scheduler::number_of_threads() {
+    scoped_lock with(lock);
     return threads.size();
 }
 

src/rt/rustrt.def.in

@@ -30,7 +30,6 @@ rand_new_seeded
 rand_next
 rand_seed
 rust_get_sched_id
-rust_max_sched_threads
 rust_new_sched
 rust_new_task_in_sched
 rust_num_threads
@@ -49,6 +48,7 @@ rust_port_size
 rust_process_wait
 rust_ptr_eq
 rust_run_program
+rust_sched_current_nonlazy_threads
 rust_sched_threads
 rust_set_exit_status
 rust_start

src/test/run-pass/morestack6.rs

@@ -8,15 +8,15 @@ extern mod rustrt {
     fn last_os_error() -> ~str;
     fn rust_getcwd() -> ~str;
     fn get_task_id() -> libc::intptr_t;
-    fn rust_max_sched_threads();
+    fn rust_sched_threads();
     fn rust_get_task();
 }
 
 fn calllink01() { rustrt::rust_get_sched_id(); }
 fn calllink02() { rustrt::last_os_error(); }
 fn calllink03() { rustrt::rust_getcwd(); }
 fn calllink08() { rustrt::get_task_id(); }
-fn calllink09() { rustrt::rust_max_sched_threads(); }
+fn calllink09() { rustrt::rust_sched_threads(); }
 fn calllink10() { rustrt::rust_get_task(); }
 
 fn runtest(f: fn~(), frame_backoff: u32) {