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