src/libstd/task.rs

   1 // Copyright 2012-2013 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 //! Task creation
  12 //!
  13 //! An executing Rust program consists of a collection of tasks, each
  14 //! with their own stack and local state. A Rust task is typically
  15 //! backed by an operating system thread, making tasks 'just threads',
  16 //! but may also be implemented via other strategies as well
  17 //! (e.g. Rust comes with the [`green`](../../green/index.html)
  18 //! scheduling crate for creating tasks backed by green threads).
  19 //!
  20 //! Tasks generally have their memory *isolated* from each other by
  21 //! virtue of Rust's owned types (which of course may only be owned by
  22 //! a single task at a time). Communication between tasks is primarily
  23 //! done through [channels](../../std/comm/index.html), Rust's
  24 //! message-passing types, though [other forms of task
  25 //! synchronization](../../std/sync/index.html) are often employed to
  26 //! achieve particular performance goals. In particular, types that
  27 //! are guaranteed to be threadsafe are easily shared between threads
  28 //! using the atomically-reference-counted container,
  29 //! [`Arc`](../../std/sync/struct.Arc.html).
  30 //!
  31 //! Fatal logic errors in Rust cause *task failure*, during which
  32 //! a task will unwind the stack, running destructors and freeing
  33 //! owned resources. Task failure is unrecoverable from within
  34 //! the failing task (i.e. there is no 'try/catch' in Rust), but
  35 //! failure may optionally be detected from a different task. If
  36 //! the main task fails the application will exit with a non-zero
  37 //! exit code.
  38 //!
  39 //! # Basic task scheduling
  40 //!
  41 //! By default, every task is created with the same "flavor" as the calling task.
  42 //! This flavor refers to the scheduling mode, with two possibilities currently
  43 //! being 1:1 and M:N modes. Green (M:N) tasks are cooperatively scheduled and
  44 //! native (1:1) tasks are scheduled by the OS kernel.
  45 //!
  46 //! ## Example
  47 //!
  48 //! ```rust
  49 //! spawn(proc() {
  50 //!     println!("Hello, World!");
  51 //! })
  52 //! ```
  53 //!
  54 //! # Advanced task scheduling
  55 //!
  56 //! Task spawning can also be configured to use a particular scheduler, to
  57 //! redirect the new task's output, or to yield a `future` representing the
  58 //! task's final result. The configuration is established using the
  59 //! `TaskBuilder` API:
  60 //!
  61 //! ## Example
  62 //!
  63 //! ```rust
  64 //! extern crate green;
  65 //! extern crate native;
  66 //!
  67 //! use std::task::TaskBuilder;
  68 //! use green::{SchedPool, PoolConfig, GreenTaskBuilder};
  69 //! use native::NativeTaskBuilder;
  70 //!
  71 //! # fn main() {
  72 //! // Create a green scheduler pool with the default configuration
  73 //! let mut pool = SchedPool::new(PoolConfig::new());
  74 //!
  75 //! // Spawn a task in the green pool
  76 //! let mut fut_green = TaskBuilder::new().green(&mut pool).try_future(proc() {
  77 //!     /* ... */
  78 //! });
  79 //!
  80 //! // Spawn a native task
  81 //! let mut fut_native = TaskBuilder::new().native().try_future(proc() {
  82 //!     /* ... */
  83 //! });
  84 //!
  85 //! // Wait for both tasks to finish, recording their outcome
  86 //! let res_green  = fut_green.unwrap();
  87 //! let res_native = fut_native.unwrap();
  88 //!
  89 //! // Shut down the green scheduler pool
  90 //! pool.shutdown();
  91 //! # }
  92 //! ```
  93
  94 #![experimental]
  95
  96 use any::Any;
  97 use comm::channel;
  98 use io::{Writer, stdio};
  99 use kinds::{Send, marker};
 100 use option::{None, Some, Option};
 101 use owned::Box;
 102 use result::Result;
 103 use rt::local::Local;
 104 use rt::task;
 105 use rt::task::Task;
 106 use str::{Str, SendStr, IntoMaybeOwned};
 107 use sync::Future;
 108
 109 /// A means of spawning a task
 110 pub trait Spawner {
 111     /// Spawn a task, given low-level task options.
 112     fn spawn(self, opts: task::TaskOpts, f: proc():Send);
 113 }
 114
 115 /// The default task spawner, which spawns siblings to the current task.
 116 pub struct SiblingSpawner;
 117
 118 impl Spawner for SiblingSpawner {
 119     fn spawn(self, opts: task::TaskOpts, f: proc():Send) {
 120         // bind tb to provide type annotation
 121         let tb: Option<Box<Task>> = Local::try_take();
 122         match tb {
 123             Some(t) => t.spawn_sibling(opts, f),
 124             None => fail!("need a local task to spawn a sibling task"),
 125         };
 126     }
 127 }
 128
 129 /// The task builder type.
 130 ///
 131 /// Provides detailed control over the properties and behavior of new tasks.
 132
 133 // NB: Builders are designed to be single-use because they do stateful
 134 // things that get weird when reusing - e.g. if you create a result future
 135 // it only applies to a single task, so then you have to maintain Some
 136 // potentially tricky state to ensure that everything behaves correctly
 137 // when you try to reuse the builder to spawn a new task. We'll just
 138 // sidestep that whole issue by making builders uncopyable and making
 139 // the run function move them in.
 140 pub struct TaskBuilder<S = SiblingSpawner> {
 141     // A name for the task-to-be, for identification in failure messages
 142     name: Option<SendStr>,
 143     // The size of the stack for the spawned task
 144     stack_size: Option<uint>,
 145     // Task-local stdout
 146     stdout: Option<Box<Writer + Send>>,
 147     // Task-local stderr
 148     stderr: Option<Box<Writer + Send>>,
 149     // The mechanics of actually spawning the task (i.e.: green or native)
 150     spawner: S,
 151     // Optionally wrap the eventual task body
 152     gen_body: Option<proc(v: proc():Send):Send -> proc():Send>,
 153     nocopy: marker::NoCopy,
 154 }
 155
 156 impl TaskBuilder<SiblingSpawner> {
 157     /// Generate the base configuration for spawning a task, off of which more
 158     /// configuration methods can be chained.
 159     pub fn new() -> TaskBuilder<SiblingSpawner> {
 160         TaskBuilder {
 161             name: None,
 162             stack_size: None,
 163             stdout: None,
 164             stderr: None,
 165             spawner: SiblingSpawner,
 166             gen_body: None,
 167             nocopy: marker::NoCopy,
 168         }
 169     }
 170 }
 171
 172 impl<S: Spawner> TaskBuilder<S> {
 173     /// Name the task-to-be. Currently the name is used for identification
 174     /// only in failure messages.
 175     pub fn named<T: IntoMaybeOwned<'static>>(mut self, name: T) -> TaskBuilder<S> {
 176         self.name = Some(name.into_maybe_owned());
 177         self
 178     }
 179
 180     /// Set the size of the stack for the new task.
 181     pub fn stack_size(mut self, size: uint) -> TaskBuilder<S> {
 182         self.stack_size = Some(size);
 183         self
 184     }
 185
 186     /// Redirect task-local stdout.
 187     pub fn stdout(mut self, stdout: Box<Writer + Send>) -> TaskBuilder<S> {
 188         self.stdout = Some(stdout);
 189         self
 190     }
 191
 192     /// Redirect task-local stderr.
 193     pub fn stderr(mut self, stderr: Box<Writer + Send>) -> TaskBuilder<S> {
 194         self.stderr = Some(stderr);
 195         self
 196     }
 197
 198     /// Set the spawning mechanism for the task.
 199     ///
 200     /// The `TaskBuilder` API configures a task to be spawned, but defers to the
 201     /// "spawner" to actually create and spawn the task. The `spawner` method
 202     /// should not be called directly by `TaskBuiler` clients. It is intended
 203     /// for use by downstream crates (like `native` and `green`) that implement
 204     /// tasks. These downstream crates then add extension methods to the
 205     /// builder, like `.native()` and `.green(pool)`, that actually set the
 206     /// spawner.
 207     pub fn spawner<T: Spawner>(self, spawner: T) -> TaskBuilder<T> {
 208         // repackage the entire TaskBuilder since its type is changing.
 209         let TaskBuilder {
 210             name, stack_size, stdout, stderr, spawner: _, gen_body, nocopy
 211         } = self;
 212         TaskBuilder {
 213             name: name,
 214             stack_size: stack_size,
 215             stdout: stdout,
 216             stderr: stderr,
 217             spawner: spawner,
 218             gen_body: gen_body,
 219             nocopy: nocopy,
 220         }
 221     }
 222
 223     /// Add a wrapper to the body of the spawned task.
 224     ///
 225     /// Before the task is spawned it is passed through a 'body generator'
 226     /// function that may perform local setup operations as well as wrap
 227     /// the task body in remote setup operations. With this the behavior
 228     /// of tasks can be extended in simple ways.
 229     ///
 230     /// This function augments the current body generator with a new body
 231     /// generator by applying the task body which results from the
 232     /// existing body generator to the new body generator.
 233     #[deprecated = "this function will be removed soon"]
 234     pub fn with_wrapper(mut self, wrapper: proc(v: proc():Send):Send -> proc():Send)
 235                         -> TaskBuilder<S> {
 236         self.gen_body = match self.gen_body.take() {
 237             Some(prev) => Some(proc(body) { wrapper(prev(body)) }),
 238             None => Some(wrapper)
 239         };
 240         self
 241     }
 242
 243     // Where spawning actually happens (whether yielding a future or not)
 244     fn spawn_internal(self, f: proc():Send,
 245                       on_exit: Option<proc(Result<(), Box<Any + Send>>):Send>) {
 246         let TaskBuilder {
 247             name, stack_size, stdout, stderr, spawner, mut gen_body, nocopy: _
 248         } = self;
 249         let f = match gen_body.take() {
 250             Some(gen) => gen(f),
 251             None => f
 252         };
 253         let opts = task::TaskOpts {
 254             on_exit: on_exit,
 255             name: name,
 256             stack_size: stack_size,
 257         };
 258         if stdout.is_some() || stderr.is_some() {
 259             spawner.spawn(opts, proc() {
 260                 let _ = stdout.map(stdio::set_stdout);
 261                 let _ = stderr.map(stdio::set_stderr);
 262                 f();
 263             })
 264         } else {
 265             spawner.spawn(opts, f)
 266         }
 267     }
 268
 269     /// Creates and executes a new child task.
 270     ///
 271     /// Sets up a new task with its own call stack and schedules it to run
 272     /// the provided proc. The task has the properties and behavior
 273     /// specified by the `TaskBuilder`.
 274     pub fn spawn(self, f: proc():Send) {
 275         self.spawn_internal(f, None)
 276     }
 277
 278     /// Execute a proc in a newly-spawned task and return a future representing
 279     /// the task's result. The task has the properties and behavior
 280     /// specified by the `TaskBuilder`.
 281     ///
 282     /// Taking the value of the future will block until the child task
 283     /// terminates.
 284     ///
 285     /// # Return value
 286     ///
 287     /// If the child task executes successfully (without failing) then the
 288     /// future returns `result::Ok` containing the value returned by the
 289     /// function. If the child task fails then the future returns `result::Err`
 290     /// containing the argument to `fail!(...)` as an `Any` trait object.
 291     pub fn try_future<T:Send>(self, f: proc():Send -> T)
 292                               -> Future<Result<T, Box<Any + Send>>> {
 293         // currently, the on_exit proc provided by librustrt only works for unit
 294         // results, so we use an additional side-channel to communicate the
 295         // result.
 296
 297         let (tx_done, rx_done) = channel(); // signal that task has exited
 298         let (tx_retv, rx_retv) = channel(); // return value from task
 299
 300         let on_exit = proc(res) { let _ = tx_done.send_opt(res); };
 301         self.spawn_internal(proc() { let _ = tx_retv.send_opt(f()); },
 302                             Some(on_exit));
 303
 304         Future::from_fn(proc() {
 305             rx_done.recv().map(|_| rx_retv.recv())
 306         })
 307     }
 308
 309     /// Execute a function in a newly-spawnedtask and block until the task
 310     /// completes or fails. Equivalent to `.try_future(f).unwrap()`.
 311     pub fn try<T:Send>(self, f: proc():Send -> T) -> Result<T, Box<Any + Send>> {
 312         self.try_future(f).unwrap()
 313     }
 314 }
 315
 316 /* Convenience functions */
 317
 318 /// Creates and executes a new child task
 319 ///
 320 /// Sets up a new task with its own call stack and schedules it to run
 321 /// the provided unique closure.
 322 ///
 323 /// This function is equivalent to `TaskBuilder::new().spawn(f)`.
 324 pub fn spawn(f: proc(): Send) {
 325     TaskBuilder::new().spawn(f)
 326 }
 327
 328 /// Execute a function in a newly-spawned task and return either the return
 329 /// value of the function or an error if the task failed.
 330 ///
 331 /// This is equivalent to `TaskBuilder::new().try`.
 332 pub fn try<T: Send>(f: proc(): Send -> T) -> Result<T, Box<Any + Send>> {
 333     TaskBuilder::new().try(f)
 334 }
 335
 336 /// Execute a function in another task and return a future representing the
 337 /// task's result.
 338 ///
 339 /// This is equivalent to `TaskBuilder::new().try_future`.
 340 pub fn try_future<T:Send>(f: proc():Send -> T) -> Future<Result<T, Box<Any + Send>>> {
 341     TaskBuilder::new().try_future(f)
 342 }
 343
 344
 345 /* Lifecycle functions */
 346
 347 /// Read the name of the current task.
 348 pub fn with_task_name<U>(blk: |Option<&str>| -> U) -> U {
 349     use rt::task::Task;
 350
 351     let task = Local::borrow(None::<Task>);
 352     match task.name {
 353         Some(ref name) => blk(Some(name.as_slice())),
 354         None => blk(None)
 355     }
 356 }
 357
 358 /// Yield control to the task scheduler.
 359 pub fn deschedule() {
 360     use rt::local::Local;
 361
 362     // FIXME(#7544): Optimize this, since we know we won't block.
 363     let task: Box<Task> = Local::take();
 364     task.yield_now();
 365 }
 366
 367 /// True if the running task is currently failing (e.g. will return `true` inside a
 368 /// destructor that is run while unwinding the stack after a call to `fail!()`).
 369 pub fn failing() -> bool {
 370     use rt::task::Task;
 371     Local::borrow(None::<Task>).unwinder.unwinding()
 372 }
 373
 374 #[cfg(test)]
 375 mod test {
 376     use any::{Any, AnyRefExt};
 377     use owned::AnyOwnExt;
 378     use result;
 379     use result::{Ok, Err};
 380     use str::StrAllocating;
 381     use string::String;
 382     use std::io::{ChanReader, ChanWriter};
 383     use prelude::*;
 384     use super::*;
 385
 386     // !!! These tests are dangerous. If something is buggy, they will hang, !!!
 387     // !!! instead of exiting cleanly. This might wedge the buildbots.       !!!
 388
 389     #[test]
 390     fn test_unnamed_task() {
 391         spawn(proc() {
 392             with_task_name(|name| {
 393                 assert!(name.is_none());
 394             })
 395         })
 396     }
 397
 398     #[test]
 399     fn test_owned_named_task() {
 400         TaskBuilder::new().named("ada lovelace".to_string()).spawn(proc() {
 401             with_task_name(|name| {
 402                 assert!(name.unwrap() == "ada lovelace");
 403             })
 404         })
 405     }
 406
 407     #[test]
 408     fn test_static_named_task() {
 409         TaskBuilder::new().named("ada lovelace").spawn(proc() {
 410             with_task_name(|name| {
 411                 assert!(name.unwrap() == "ada lovelace");
 412             })
 413         })
 414     }
 415
 416     #[test]
 417     fn test_send_named_task() {
 418         TaskBuilder::new().named("ada lovelace".into_maybe_owned()).spawn(proc() {
 419             with_task_name(|name| {
 420                 assert!(name.unwrap() == "ada lovelace");
 421             })
 422         })
 423     }
 424
 425     #[test]
 426     fn test_run_basic() {
 427         let (tx, rx) = channel();
 428         TaskBuilder::new().spawn(proc() {
 429             tx.send(());
 430         });
 431         rx.recv();
 432     }
 433
 434     #[test]
 435     fn test_with_wrapper() {
 436         let (tx, rx) = channel();
 437         TaskBuilder::new().with_wrapper(proc(body) {
 438             let result: proc():Send = proc() {
 439                 body();
 440                 tx.send(());
 441             };
 442             result
 443         }).spawn(proc() { });
 444         rx.recv();
 445     }
 446
 447     #[test]
 448     fn test_try_future() {
 449         let result = TaskBuilder::new().try_future(proc() {});
 450         assert!(result.unwrap().is_ok());
 451
 452         let result = TaskBuilder::new().try_future(proc() -> () {
 453             fail!();
 454         });
 455         assert!(result.unwrap().is_err());
 456     }
 457
 458     #[test]
 459     fn test_try_success() {
 460         match try(proc() {
 461             "Success!".to_string()
 462         }).as_ref().map(|s| s.as_slice()) {
 463             result::Ok("Success!") => (),
 464             _ => fail!()
 465         }
 466     }
 467
 468     #[test]
 469     fn test_try_fail() {
 470         match try(proc() {
 471             fail!()
 472         }) {
 473             result::Err(_) => (),
 474             result::Ok(()) => fail!()
 475         }
 476     }
 477
 478     #[test]
 479     fn test_spawn_sched() {
 480         use clone::Clone;
 481
 482         let (tx, rx) = channel();
 483
 484         fn f(i: int, tx: Sender<()>) {
 485             let tx = tx.clone();
 486             spawn(proc() {
 487                 if i == 0 {
 488                     tx.send(());
 489                 } else {
 490                     f(i - 1, tx);
 491                 }
 492             });
 493
 494         }
 495         f(10, tx);
 496         rx.recv();
 497     }
 498
 499     #[test]
 500     fn test_spawn_sched_childs_on_default_sched() {
 501         let (tx, rx) = channel();
 502
 503         spawn(proc() {
 504             spawn(proc() {
 505                 tx.send(());
 506             });
 507         });
 508
 509         rx.recv();
 510     }
 511
 512     fn avoid_copying_the_body(spawnfn: |v: proc():Send|) {
 513         let (tx, rx) = channel::<uint>();
 514
 515         let x = box 1;
 516         let x_in_parent = (&*x) as *const int as uint;
 517
 518         spawnfn(proc() {
 519             let x_in_child = (&*x) as *const int as uint;
 520             tx.send(x_in_child);
 521         });
 522
 523         let x_in_child = rx.recv();
 524         assert_eq!(x_in_parent, x_in_child);
 525     }
 526
 527     #[test]
 528     fn test_avoid_copying_the_body_spawn() {
 529         avoid_copying_the_body(spawn);
 530     }
 531
 532     #[test]
 533     fn test_avoid_copying_the_body_task_spawn() {
 534         avoid_copying_the_body(|f| {
 535             let builder = TaskBuilder::new();
 536             builder.spawn(proc() {
 537                 f();
 538             });
 539         })
 540     }
 541
 542     #[test]
 543     fn test_avoid_copying_the_body_try() {
 544         avoid_copying_the_body(|f| {
 545             let _ = try(proc() {
 546                 f()
 547             });
 548         })
 549     }
 550
 551     #[test]
 552     fn test_child_doesnt_ref_parent() {
 553         // If the child refcounts the parent task, this will stack overflow when
 554         // climbing the task tree to dereference each ancestor. (See #1789)
 555         // (well, it would if the constant were 8000+ - I lowered it to be more
 556         // valgrind-friendly. try this at home, instead..!)
 557         static generations: uint = 16;
 558         fn child_no(x: uint) -> proc(): Send {
 559             return proc() {
 560                 if x < generations {
 561                     TaskBuilder::new().spawn(child_no(x+1));
 562                 }
 563             }
 564         }
 565         TaskBuilder::new().spawn(child_no(0));
 566     }
 567
 568     #[test]
 569     fn test_simple_newsched_spawn() {
 570         spawn(proc()())
 571     }
 572
 573     #[test]
 574     fn test_try_fail_message_static_str() {
 575         match try(proc() {
 576             fail!("static string");
 577         }) {
 578             Err(e) => {
 579                 type T = &'static str;
 580                 assert!(e.is::<T>());
 581                 assert_eq!(*e.move::<T>().unwrap(), "static string");
 582             }
 583             Ok(()) => fail!()
 584         }
 585     }
 586
 587     #[test]
 588     fn test_try_fail_message_owned_str() {
 589         match try(proc() {
 590             fail!("owned string".to_string());
 591         }) {
 592             Err(e) => {
 593                 type T = String;
 594                 assert!(e.is::<T>());
 595                 assert_eq!(*e.move::<T>().unwrap(), "owned string".to_string());
 596             }
 597             Ok(()) => fail!()
 598         }
 599     }
 600
 601     #[test]
 602     fn test_try_fail_message_any() {
 603         match try(proc() {
 604             fail!(box 413u16 as Box<Any + Send>);
 605         }) {
 606             Err(e) => {
 607                 type T = Box<Any + Send>;
 608                 assert!(e.is::<T>());
 609                 let any = e.move::<T>().unwrap();
 610                 assert!(any.is::<u16>());
 611                 assert_eq!(*any.move::<u16>().unwrap(), 413u16);
 612             }
 613             Ok(()) => fail!()
 614         }
 615     }
 616
 617     #[test]
 618     fn test_try_fail_message_unit_struct() {
 619         struct Juju;
 620
 621         match try(proc() {
 622             fail!(Juju)
 623         }) {
 624             Err(ref e) if e.is::<Juju>() => {}
 625             Err(_) | Ok(()) => fail!()
 626         }
 627     }
 628
 629     #[test]
 630     fn test_stdout() {
 631         let (tx, rx) = channel();
 632         let mut reader = ChanReader::new(rx);
 633         let stdout = ChanWriter::new(tx);
 634
 635         let r = TaskBuilder::new().stdout(box stdout as Box<Writer + Send>)
 636                                   .try(proc() {
 637                 print!("Hello, world!");
 638             });
 639         assert!(r.is_ok());
 640
 641         let output = reader.read_to_str().unwrap();
 642         assert_eq!(output, "Hello, world!".to_string());
 643     }
 644
 645     // NOTE: the corresponding test for stderr is in run-pass/task-stderr, due
 646     // to the test harness apparently interfering with stderr configuration.
 647 }
 648
 649 #[test]
 650 fn task_abort_no_kill_runtime() {
 651     use std::io::timer;
 652     use mem;
 653
 654     let tb = TaskBuilder::new();
 655     let rx = tb.try_future(proc() {});
 656     mem::drop(rx);
 657     timer::sleep(1000);
 658 }