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 panic*, during which
  32 //! a task will unwind the stack, running destructors and freeing
  33 //! owned resources. Task panic is unrecoverable from within
  34 //! the panicking task (i.e. there is no 'try/catch' in Rust), but
  35 //! panic may optionally be detected from a different task. If
  36 //! the main task panics 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 #![stable]
  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 boxed::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 string::String;
 108 use sync::Future;
 109 use to_string::ToString;
 110
 111 /// A means of spawning a task
 112 pub trait Spawner {
 113     /// Spawn a task, given low-level task options.
 114     fn spawn(self, opts: task::TaskOpts, f: proc():Send);
 115 }
 116
 117 /// The default task spawner, which spawns siblings to the current task.
 118 pub struct SiblingSpawner;
 119
 120 impl Spawner for SiblingSpawner {
 121     fn spawn(self, opts: task::TaskOpts, f: proc():Send) {
 122         // bind tb to provide type annotation
 123         let tb: Option<Box<Task>> = Local::try_take();
 124         match tb {
 125             Some(t) => t.spawn_sibling(opts, f),
 126             None => panic!("need a local task to spawn a sibling task"),
 127         };
 128     }
 129 }
 130
 131 /// The task builder type.
 132 ///
 133 /// Provides detailed control over the properties and behavior of new tasks.
 134
 135 // NB: Builders are designed to be single-use because they do stateful
 136 // things that get weird when reusing - e.g. if you create a result future
 137 // it only applies to a single task, so then you have to maintain Some
 138 // potentially tricky state to ensure that everything behaves correctly
 139 // when you try to reuse the builder to spawn a new task. We'll just
 140 // sidestep that whole issue by making builders uncopyable and making
 141 // the run function move them in.
 142 pub struct TaskBuilder<S = SiblingSpawner> {
 143     // A name for the task-to-be, for identification in panic messages
 144     name: Option<SendStr>,
 145     // The size of the stack for the spawned task
 146     stack_size: Option<uint>,
 147     // Task-local stdout
 148     stdout: Option<Box<Writer + Send>>,
 149     // Task-local stderr
 150     stderr: Option<Box<Writer + Send>>,
 151     // The mechanics of actually spawning the task (i.e.: green or native)
 152     spawner: S,
 153     // Optionally wrap the eventual task body
 154     gen_body: Option<proc(v: proc():Send):Send -> proc():Send>,
 155     nocopy: marker::NoCopy,
 156 }
 157
 158 impl TaskBuilder<SiblingSpawner> {
 159     /// Generate the base configuration for spawning a task, off of which more
 160     /// configuration methods can be chained.
 161     pub fn new() -> TaskBuilder<SiblingSpawner> {
 162         TaskBuilder {
 163             name: None,
 164             stack_size: None,
 165             stdout: None,
 166             stderr: None,
 167             spawner: SiblingSpawner,
 168             gen_body: None,
 169             nocopy: marker::NoCopy,
 170         }
 171     }
 172 }
 173
 174 impl<S: Spawner> TaskBuilder<S> {
 175     /// Name the task-to-be. Currently the name is used for identification
 176     /// only in panic messages.
 177     #[unstable = "IntoMaybeOwned will probably change."]
 178     pub fn named<T: IntoMaybeOwned<'static>>(mut self, name: T) -> TaskBuilder<S> {
 179         self.name = Some(name.into_maybe_owned());
 180         self
 181     }
 182
 183     /// Set the size of the stack for the new task.
 184     pub fn stack_size(mut self, size: uint) -> TaskBuilder<S> {
 185         self.stack_size = Some(size);
 186         self
 187     }
 188
 189     /// Redirect task-local stdout.
 190     #[experimental = "May not want to make stdio overridable here."]
 191     pub fn stdout(mut self, stdout: Box<Writer + Send>) -> TaskBuilder<S> {
 192         self.stdout = Some(stdout);
 193         self
 194     }
 195
 196     /// Redirect task-local stderr.
 197     #[experimental = "May not want to make stdio overridable here."]
 198     pub fn stderr(mut self, stderr: Box<Writer + Send>) -> TaskBuilder<S> {
 199         self.stderr = Some(stderr);
 200         self
 201     }
 202
 203     /// Set the spawning mechanism for the task.
 204     ///
 205     /// The `TaskBuilder` API configures a task to be spawned, but defers to the
 206     /// "spawner" to actually create and spawn the task. The `spawner` method
 207     /// should not be called directly by `TaskBuiler` clients. It is intended
 208     /// for use by downstream crates (like `native` and `green`) that implement
 209     /// tasks. These downstream crates then add extension methods to the
 210     /// builder, like `.native()` and `.green(pool)`, that actually set the
 211     /// spawner.
 212     pub fn spawner<T: Spawner>(self, spawner: T) -> TaskBuilder<T> {
 213         // repackage the entire TaskBuilder since its type is changing.
 214         let TaskBuilder {
 215             name, stack_size, stdout, stderr, spawner: _, gen_body, nocopy
 216         } = self;
 217         TaskBuilder {
 218             name: name,
 219             stack_size: stack_size,
 220             stdout: stdout,
 221             stderr: stderr,
 222             spawner: spawner,
 223             gen_body: gen_body,
 224             nocopy: nocopy,
 225         }
 226     }
 227
 228     // Where spawning actually happens (whether yielding a future or not)
 229     fn spawn_internal(self, f: proc():Send,
 230                       on_exit: Option<proc(Result<(), Box<Any + Send>>):Send>) {
 231         let TaskBuilder {
 232             name, stack_size, stdout, stderr, spawner, mut gen_body, nocopy: _
 233         } = self;
 234         let f = match gen_body.take() {
 235             Some(gen) => gen(f),
 236             None => f
 237         };
 238         let opts = task::TaskOpts {
 239             on_exit: on_exit,
 240             name: name,
 241             stack_size: stack_size,
 242         };
 243         if stdout.is_some() || stderr.is_some() {
 244             spawner.spawn(opts, proc() {
 245                 let _ = stdout.map(stdio::set_stdout);
 246                 let _ = stderr.map(stdio::set_stderr);
 247                 f();
 248             })
 249         } else {
 250             spawner.spawn(opts, f)
 251         }
 252     }
 253
 254     /// Creates and executes a new child task.
 255     ///
 256     /// Sets up a new task with its own call stack and schedules it to run
 257     /// the provided proc. The task has the properties and behavior
 258     /// specified by the `TaskBuilder`.
 259     pub fn spawn(self, f: proc():Send) {
 260         self.spawn_internal(f, None)
 261     }
 262
 263     /// Execute a proc in a newly-spawned task and return a future representing
 264     /// the task's result. The task has the properties and behavior
 265     /// specified by the `TaskBuilder`.
 266     ///
 267     /// Taking the value of the future will block until the child task
 268     /// terminates.
 269     ///
 270     /// # Return value
 271     ///
 272     /// If the child task executes successfully (without panicking) then the
 273     /// future returns `result::Ok` containing the value returned by the
 274     /// function. If the child task panics then the future returns `result::Err`
 275     /// containing the argument to `panic!(...)` as an `Any` trait object.
 276     #[experimental = "Futures are experimental."]
 277     pub fn try_future<T:Send>(self, f: proc():Send -> T)
 278                               -> Future<Result<T, Box<Any + Send>>> {
 279         // currently, the on_exit proc provided by librustrt only works for unit
 280         // results, so we use an additional side-channel to communicate the
 281         // result.
 282
 283         let (tx_done, rx_done) = channel(); // signal that task has exited
 284         let (tx_retv, rx_retv) = channel(); // return value from task
 285
 286         let on_exit = proc(res) { let _ = tx_done.send_opt(res); };
 287         self.spawn_internal(proc() { let _ = tx_retv.send_opt(f()); },
 288                             Some(on_exit));
 289
 290         Future::from_fn(proc() {
 291             rx_done.recv().map(|_| rx_retv.recv())
 292         })
 293     }
 294
 295     /// Execute a function in a newly-spawnedtask and block until the task
 296     /// completes or panics. Equivalent to `.try_future(f).unwrap()`.
 297     #[unstable = "Error type may change."]
 298     pub fn try<T:Send>(self, f: proc():Send -> T) -> Result<T, Box<Any + Send>> {
 299         self.try_future(f).unwrap()
 300     }
 301 }
 302
 303 /* Convenience functions */
 304
 305 /// Creates and executes a new child task
 306 ///
 307 /// Sets up a new task with its own call stack and schedules it to run
 308 /// the provided unique closure.
 309 ///
 310 /// This function is equivalent to `TaskBuilder::new().spawn(f)`.
 311 pub fn spawn(f: proc(): Send) {
 312     TaskBuilder::new().spawn(f)
 313 }
 314
 315 /// Execute a function in a newly-spawned task and return either the return
 316 /// value of the function or an error if the task panicked.
 317 ///
 318 /// This is equivalent to `TaskBuilder::new().try`.
 319 #[unstable = "Error type may change."]
 320 pub fn try<T: Send>(f: proc(): Send -> T) -> Result<T, Box<Any + Send>> {
 321     TaskBuilder::new().try(f)
 322 }
 323
 324 /// Execute a function in another task and return a future representing the
 325 /// task's result.
 326 ///
 327 /// This is equivalent to `TaskBuilder::new().try_future`.
 328 #[experimental = "Futures are experimental."]
 329 pub fn try_future<T:Send>(f: proc():Send -> T) -> Future<Result<T, Box<Any + Send>>> {
 330     TaskBuilder::new().try_future(f)
 331 }
 332
 333
 334 /* Lifecycle functions */
 335
 336 /// Read the name of the current task.
 337 #[stable]
 338 pub fn name() -> Option<String> {
 339     use rt::task::Task;
 340
 341     let task = Local::borrow(None::<Task>);
 342     match task.name {
 343         Some(ref name) => Some(name.as_slice().to_string()),
 344         None => None
 345     }
 346 }
 347
 348 /// Yield control to the task scheduler.
 349 #[unstable = "Name will change."]
 350 pub fn deschedule() {
 351     use rt::local::Local;
 352
 353     // FIXME(#7544): Optimize this, since we know we won't block.
 354     let task: Box<Task> = Local::take();
 355     task.yield_now();
 356 }
 357
 358 /// True if the running task is currently panicking (e.g. will return `true` inside a
 359 /// destructor that is run while unwinding the stack after a call to `panic!()`).
 360 #[unstable = "May move to a different module."]
 361 pub fn failing() -> bool {
 362     use rt::task::Task;
 363     Local::borrow(None::<Task>).unwinder.unwinding()
 364 }
 365
 366 #[cfg(test)]
 367 mod test {
 368     use any::{Any, AnyRefExt};
 369     use boxed::BoxAny;
 370     use result;
 371     use result::{Ok, Err};
 372     use string::String;
 373     use std::io::{ChanReader, ChanWriter};
 374     use prelude::*;
 375     use super::*;
 376
 377     // !!! These tests are dangerous. If something is buggy, they will hang, !!!
 378     // !!! instead of exiting cleanly. This might wedge the buildbots.       !!!
 379
 380     #[test]
 381     fn test_unnamed_task() {
 382         try(proc() {
 383             assert!(name().is_none());
 384         }).map_err(|_| ()).unwrap();
 385     }
 386
 387     #[test]
 388     fn test_owned_named_task() {
 389         TaskBuilder::new().named("ada lovelace".to_string()).try(proc() {
 390             assert!(name().unwrap() == "ada lovelace".to_string());
 391         }).map_err(|_| ()).unwrap();
 392     }
 393
 394     #[test]
 395     fn test_static_named_task() {
 396         TaskBuilder::new().named("ada lovelace").try(proc() {
 397             assert!(name().unwrap() == "ada lovelace".to_string());
 398         }).map_err(|_| ()).unwrap();
 399     }
 400
 401     #[test]
 402     fn test_send_named_task() {
 403         TaskBuilder::new().named("ada lovelace".into_maybe_owned()).try(proc() {
 404             assert!(name().unwrap() == "ada lovelace".to_string());
 405         }).map_err(|_| ()).unwrap();
 406     }
 407
 408     #[test]
 409     fn test_run_basic() {
 410         let (tx, rx) = channel();
 411         TaskBuilder::new().spawn(proc() {
 412             tx.send(());
 413         });
 414         rx.recv();
 415     }
 416
 417     #[test]
 418     fn test_try_future() {
 419         let result = TaskBuilder::new().try_future(proc() {});
 420         assert!(result.unwrap().is_ok());
 421
 422         let result = TaskBuilder::new().try_future(proc() -> () {
 423             panic!();
 424         });
 425         assert!(result.unwrap().is_err());
 426     }
 427
 428     #[test]
 429     fn test_try_success() {
 430         match try(proc() {
 431             "Success!".to_string()
 432         }).as_ref().map(|s| s.as_slice()) {
 433             result::Ok("Success!") => (),
 434             _ => panic!()
 435         }
 436     }
 437
 438     #[test]
 439     fn test_try_panic() {
 440         match try(proc() {
 441             panic!()
 442         }) {
 443             result::Err(_) => (),
 444             result::Ok(()) => panic!()
 445         }
 446     }
 447
 448     #[test]
 449     fn test_spawn_sched() {
 450         use clone::Clone;
 451
 452         let (tx, rx) = channel();
 453
 454         fn f(i: int, tx: Sender<()>) {
 455             let tx = tx.clone();
 456             spawn(proc() {
 457                 if i == 0 {
 458                     tx.send(());
 459                 } else {
 460                     f(i - 1, tx);
 461                 }
 462             });
 463
 464         }
 465         f(10, tx);
 466         rx.recv();
 467     }
 468
 469     #[test]
 470     fn test_spawn_sched_childs_on_default_sched() {
 471         let (tx, rx) = channel();
 472
 473         spawn(proc() {
 474             spawn(proc() {
 475                 tx.send(());
 476             });
 477         });
 478
 479         rx.recv();
 480     }
 481
 482     fn avoid_copying_the_body(spawnfn: |v: proc():Send|) {
 483         let (tx, rx) = channel::<uint>();
 484
 485         let x = box 1;
 486         let x_in_parent = (&*x) as *const int as uint;
 487
 488         spawnfn(proc() {
 489             let x_in_child = (&*x) as *const int as uint;
 490             tx.send(x_in_child);
 491         });
 492
 493         let x_in_child = rx.recv();
 494         assert_eq!(x_in_parent, x_in_child);
 495     }
 496
 497     #[test]
 498     fn test_avoid_copying_the_body_spawn() {
 499         avoid_copying_the_body(spawn);
 500     }
 501
 502     #[test]
 503     fn test_avoid_copying_the_body_task_spawn() {
 504         avoid_copying_the_body(|f| {
 505             let builder = TaskBuilder::new();
 506             builder.spawn(proc() {
 507                 f();
 508             });
 509         })
 510     }
 511
 512     #[test]
 513     fn test_avoid_copying_the_body_try() {
 514         avoid_copying_the_body(|f| {
 515             let _ = try(proc() {
 516                 f()
 517             });
 518         })
 519     }
 520
 521     #[test]
 522     fn test_child_doesnt_ref_parent() {
 523         // If the child refcounts the parent task, this will stack overflow when
 524         // climbing the task tree to dereference each ancestor. (See #1789)
 525         // (well, it would if the constant were 8000+ - I lowered it to be more
 526         // valgrind-friendly. try this at home, instead..!)
 527         static GENERATIONS: uint = 16;
 528         fn child_no(x: uint) -> proc(): Send {
 529             return proc() {
 530                 if x < GENERATIONS {
 531                     TaskBuilder::new().spawn(child_no(x+1));
 532                 }
 533             }
 534         }
 535         TaskBuilder::new().spawn(child_no(0));
 536     }
 537
 538     #[test]
 539     fn test_simple_newsched_spawn() {
 540         spawn(proc()())
 541     }
 542
 543     #[test]
 544     fn test_try_panic_message_static_str() {
 545         match try(proc() {
 546             panic!("static string");
 547         }) {
 548             Err(e) => {
 549                 type T = &'static str;
 550                 assert!(e.is::<T>());
 551                 assert_eq!(*e.downcast::<T>().unwrap(), "static string");
 552             }
 553             Ok(()) => panic!()
 554         }
 555     }
 556
 557     #[test]
 558     fn test_try_panic_message_owned_str() {
 559         match try(proc() {
 560             panic!("owned string".to_string());
 561         }) {
 562             Err(e) => {
 563                 type T = String;
 564                 assert!(e.is::<T>());
 565                 assert_eq!(*e.downcast::<T>().unwrap(), "owned string".to_string());
 566             }
 567             Ok(()) => panic!()
 568         }
 569     }
 570
 571     #[test]
 572     fn test_try_panic_message_any() {
 573         match try(proc() {
 574             panic!(box 413u16 as Box<Any + Send>);
 575         }) {
 576             Err(e) => {
 577                 type T = Box<Any + Send>;
 578                 assert!(e.is::<T>());
 579                 let any = e.downcast::<T>().unwrap();
 580                 assert!(any.is::<u16>());
 581                 assert_eq!(*any.downcast::<u16>().unwrap(), 413u16);
 582             }
 583             Ok(()) => panic!()
 584         }
 585     }
 586
 587     #[test]
 588     fn test_try_panic_message_unit_struct() {
 589         struct Juju;
 590
 591         match try(proc() {
 592             panic!(Juju)
 593         }) {
 594             Err(ref e) if e.is::<Juju>() => {}
 595             Err(_) | Ok(()) => panic!()
 596         }
 597     }
 598
 599     #[test]
 600     fn test_stdout() {
 601         let (tx, rx) = channel();
 602         let mut reader = ChanReader::new(rx);
 603         let stdout = ChanWriter::new(tx);
 604
 605         let r = TaskBuilder::new().stdout(box stdout as Box<Writer + Send>)
 606                                   .try(proc() {
 607                 print!("Hello, world!");
 608             });
 609         assert!(r.is_ok());
 610
 611         let output = reader.read_to_string().unwrap();
 612         assert_eq!(output, "Hello, world!".to_string());
 613     }
 614
 615     // NOTE: the corresponding test for stderr is in run-pass/task-stderr, due
 616     // to the test harness apparently interfering with stderr configuration.
 617 }
 618
 619 #[test]
 620 fn task_abort_no_kill_runtime() {
 621     use std::io::timer;
 622     use time::Duration;
 623     use mem;
 624
 625     let tb = TaskBuilder::new();
 626     let rx = tb.try_future(proc() {});
 627     mem::drop(rx);
 628     timer::sleep(Duration::milliseconds(1000));
 629 }