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 /*!
  12  * Utilities for managing and scheduling tasks
  13  *
  14  * An executing Rust program consists of a collection of tasks, each with their
  15  * own stack, and sole ownership of their allocated heap data. Tasks communicate
  16  * with each other using channels (see `std::comm` for more info about how
  17  * communication works).
  18  *
  19  * Failure in one task does not propagate to any others (not to parent, not to
  20  * child).  Failure propagation is instead handled by using the channel send()
  21  * and recv() methods which will fail if the other end has hung up already.
  22  *
  23  * 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
  39 use any::Any;
  40 use comm::{Sender, Receiver, channel};
  41 use io::Writer;
  42 use kinds::{Send, marker};
  43 use option::{None, Some, Option};
  44 use result::{Result, Ok, Err};
  45 use rt::local::Local;
  46 use rt::task::Task;
  47 use str::{Str, SendStr, IntoMaybeOwned};
  48
  49 #[cfg(test)] use any::{AnyOwnExt, AnyRefExt};
  50 #[cfg(test)] use result;
  51 #[cfg(test)] use str::StrSlice;
  52
  53 /// Indicates the manner in which a task exited.
  54 ///
  55 /// A task that completes without failing is considered to exit successfully.
  56 ///
  57 /// If you wish for this result's delivery to block until all
  58 /// children tasks complete, recommend using a result future.
  59 pub type TaskResult = Result<(), ~Any:Send>;
  60
  61 /// Task configuration options
  62 pub struct TaskOpts {
  63     /// Enable lifecycle notifications on the given channel
  64     pub notify_chan: Option<Sender<TaskResult>>,
  65     /// A name for the task-to-be, for identification in failure messages
  66     pub name: Option<SendStr>,
  67     /// The size of the stack for the spawned task
  68     pub stack_size: Option<uint>,
  69     /// Task-local stdout
  70     pub stdout: Option<~Writer:Send>,
  71     /// Task-local stderr
  72     pub stderr: Option<~Writer:Send>,
  73 }
  74
  75 /**
  76  * The task builder type.
  77  *
  78  * Provides detailed control over the properties and behavior of new tasks.
  79  */
  80 // NB: Builders are designed to be single-use because they do stateful
  81 // things that get weird when reusing - e.g. if you create a result future
  82 // it only applies to a single task, so then you have to maintain Some
  83 // potentially tricky state to ensure that everything behaves correctly
  84 // when you try to reuse the builder to spawn a new task. We'll just
  85 // sidestep that whole issue by making builders uncopyable and making
  86 // the run function move them in.
  87 pub struct TaskBuilder {
  88     /// Options to spawn the new task with
  89     pub opts: TaskOpts,
  90     gen_body: Option<proc(v: proc():Send):Send -> proc():Send>,
  91     nocopy: Option<marker::NoCopy>,
  92 }
  93
  94 impl TaskBuilder {
  95      /// Generate the base configuration for spawning a task, off of which more
  96      /// configuration methods can be chained.
  97     pub fn new() -> TaskBuilder {
  98         TaskBuilder {
  99             opts: TaskOpts::new(),
 100             gen_body: None,
 101             nocopy: None,
 102         }
 103     }
 104
 105     /// Get a future representing the exit status of the task.
 106     ///
 107     /// Taking the value of the future will block until the child task
 108     /// terminates. The future result return value will be created *before* the task is
 109     /// spawned; as such, do not invoke .get() on it directly;
 110     /// rather, store it in an outer variable/list for later use.
 111     ///
 112     /// # Failure
 113     /// Fails if a future_result was already set for this task.
 114     pub fn future_result(&mut self) -> Receiver<TaskResult> {
 115         // FIXME (#3725): Once linked failure and notification are
 116         // handled in the library, I can imagine implementing this by just
 117         // registering an arbitrary number of task::on_exit handlers and
 118         // sending out messages.
 119
 120         if self.opts.notify_chan.is_some() {
 121             fail!("Can't set multiple future_results for one task!");
 122         }
 123
 124         // Construct the future and give it to the caller.
 125         let (tx, rx) = channel();
 126
 127         // Reconfigure self to use a notify channel.
 128         self.opts.notify_chan = Some(tx);
 129
 130         rx
 131     }
 132
 133     /// Name the task-to-be. Currently the name is used for identification
 134     /// only in failure messages.
 135     pub fn named<S: IntoMaybeOwned<'static>>(mut self, name: S) -> TaskBuilder {
 136         self.opts.name = Some(name.into_maybe_owned());
 137         self
 138     }
 139
 140     /**
 141      * Add a wrapper to the body of the spawned task.
 142      *
 143      * Before the task is spawned it is passed through a 'body generator'
 144      * function that may perform local setup operations as well as wrap
 145      * the task body in remote setup operations. With this the behavior
 146      * of tasks can be extended in simple ways.
 147      *
 148      * This function augments the current body generator with a new body
 149      * generator by applying the task body which results from the
 150      * existing body generator to the new body generator.
 151      */
 152     pub fn with_wrapper(mut self,
 153                         wrapper: proc(v: proc():Send):Send -> proc():Send)
 154         -> TaskBuilder
 155     {
 156         self.gen_body = match self.gen_body.take() {
 157             Some(prev) => Some(proc(body) { wrapper(prev(body)) }),
 158             None => Some(wrapper)
 159         };
 160         self
 161     }
 162
 163     /**
 164      * Creates and executes a new child task
 165      *
 166      * Sets up a new task with its own call stack and schedules it to run
 167      * the provided unique closure. The task has the properties and behavior
 168      * specified by the task_builder.
 169      */
 170     pub fn spawn(mut self, f: proc():Send) {
 171         let gen_body = self.gen_body.take();
 172         let f = match gen_body {
 173             Some(gen) => gen(f),
 174             None => f
 175         };
 176         let t: ~Task = Local::take();
 177         t.spawn_sibling(self.opts, f);
 178     }
 179
 180     /**
 181      * Execute a function in another task and return either the return value
 182      * of the function or result::err.
 183      *
 184      * # Return value
 185      *
 186      * If the function executed successfully then try returns result::ok
 187      * containing the value returned by the function. If the function fails
 188      * then try returns result::err containing nil.
 189      *
 190      * # Failure
 191      * Fails if a future_result was already set for this task.
 192      */
 193     pub fn try<T:Send>(mut self, f: proc():Send -> T) -> Result<T, ~Any:Send> {
 194         let (tx, rx) = channel();
 195
 196         let result = self.future_result();
 197
 198         self.spawn(proc() {
 199             tx.send(f());
 200         });
 201
 202         match result.recv() {
 203             Ok(())     => Ok(rx.recv()),
 204             Err(cause) => Err(cause)
 205         }
 206     }
 207 }
 208
 209 /* Task construction */
 210
 211 impl TaskOpts {
 212     pub fn new() -> TaskOpts {
 213         /*!
 214          * The default task options
 215          */
 216
 217         TaskOpts {
 218             notify_chan: None,
 219             name: None,
 220             stack_size: None,
 221             stdout: None,
 222             stderr: None,
 223         }
 224     }
 225 }
 226
 227 /* Spawn convenience functions */
 228
 229 /// Creates and executes a new child task
 230 ///
 231 /// Sets up a new task with its own call stack and schedules it to run
 232 /// the provided unique closure.
 233 ///
 234 /// This function is equivalent to `TaskBuilder::new().spawn(f)`.
 235 pub fn spawn(f: proc():Send) {
 236     TaskBuilder::new().spawn(f)
 237 }
 238
 239 /// Execute a function in another task and return either the return value of
 240 /// the function or an error if the task failed
 241 ///
 242 /// This is equivalent to TaskBuilder::new().try
 243 pub fn try<T:Send>(f: proc():Send -> T) -> Result<T, ~Any:Send> {
 244     TaskBuilder::new().try(f)
 245 }
 246
 247
 248 /* Lifecycle functions */
 249
 250 /// Read the name of the current task.
 251 pub fn with_task_name<U>(blk: |Option<&str>| -> U) -> U {
 252     use rt::task::Task;
 253
 254     let task = Local::borrow(None::<Task>);
 255     match task.name {
 256         Some(ref name) => blk(Some(name.as_slice())),
 257         None => blk(None)
 258     }
 259 }
 260
 261 pub fn deschedule() {
 262     //! Yield control to the task scheduler
 263
 264     use rt::local::Local;
 265
 266     // FIXME(#7544): Optimize this, since we know we won't block.
 267     let task: ~Task = Local::take();
 268     task.yield_now();
 269 }
 270
 271 pub fn failing() -> bool {
 272     //! True if the running task has failed
 273     use rt::task::Task;
 274     Local::borrow(None::<Task>).unwinder.unwinding()
 275 }
 276
 277 // The following 8 tests test the following 2^3 combinations:
 278 // {un,}linked {un,}supervised failure propagation {up,down}wards.
 279
 280 // !!! These tests are dangerous. If Something is buggy, they will hang, !!!
 281 // !!! instead of exiting cleanly. This might wedge the buildbots.       !!!
 282
 283 #[test]
 284 fn test_unnamed_task() {
 285     spawn(proc() {
 286         with_task_name(|name| {
 287             assert!(name.is_none());
 288         })
 289     })
 290 }
 291
 292 #[test]
 293 fn test_owned_named_task() {
 294     TaskBuilder::new().named("ada lovelace".to_owned()).spawn(proc() {
 295         with_task_name(|name| {
 296             assert!(name.unwrap() == "ada lovelace");
 297         })
 298     })
 299 }
 300
 301 #[test]
 302 fn test_static_named_task() {
 303     TaskBuilder::new().named("ada lovelace").spawn(proc() {
 304         with_task_name(|name| {
 305             assert!(name.unwrap() == "ada lovelace");
 306         })
 307     })
 308 }
 309
 310 #[test]
 311 fn test_send_named_task() {
 312     TaskBuilder::new().named("ada lovelace".into_maybe_owned()).spawn(proc() {
 313         with_task_name(|name| {
 314             assert!(name.unwrap() == "ada lovelace");
 315         })
 316     })
 317 }
 318
 319 #[test]
 320 fn test_run_basic() {
 321     let (tx, rx) = channel();
 322     TaskBuilder::new().spawn(proc() {
 323         tx.send(());
 324     });
 325     rx.recv();
 326 }
 327
 328 #[test]
 329 fn test_with_wrapper() {
 330     let (tx, rx) = channel();
 331     TaskBuilder::new().with_wrapper(proc(body) {
 332         let result: proc():Send = proc() {
 333             body();
 334             tx.send(());
 335         };
 336         result
 337     }).spawn(proc() { });
 338     rx.recv();
 339 }
 340
 341 #[test]
 342 fn test_future_result() {
 343     let mut builder = TaskBuilder::new();
 344     let result = builder.future_result();
 345     builder.spawn(proc() {});
 346     assert!(result.recv().is_ok());
 347
 348     let mut builder = TaskBuilder::new();
 349     let result = builder.future_result();
 350     builder.spawn(proc() {
 351         fail!();
 352     });
 353     assert!(result.recv().is_err());
 354 }
 355
 356 #[test] #[should_fail]
 357 fn test_back_to_the_future_result() {
 358     let mut builder = TaskBuilder::new();
 359     builder.future_result();
 360     builder.future_result();
 361 }
 362
 363 #[test]
 364 fn test_try_success() {
 365     match try(proc() {
 366         "Success!".to_owned()
 367     }).as_ref().map(|s| s.as_slice()) {
 368         result::Ok("Success!") => (),
 369         _ => fail!()
 370     }
 371 }
 372
 373 #[test]
 374 fn test_try_fail() {
 375     match try(proc() {
 376         fail!()
 377     }) {
 378         result::Err(_) => (),
 379         result::Ok(()) => fail!()
 380     }
 381 }
 382
 383 #[test]
 384 fn test_spawn_sched() {
 385     use clone::Clone;
 386
 387     let (tx, rx) = channel();
 388
 389     fn f(i: int, tx: Sender<()>) {
 390         let tx = tx.clone();
 391         spawn(proc() {
 392             if i == 0 {
 393                 tx.send(());
 394             } else {
 395                 f(i - 1, tx);
 396             }
 397         });
 398
 399     }
 400     f(10, tx);
 401     rx.recv();
 402 }
 403
 404 #[test]
 405 fn test_spawn_sched_childs_on_default_sched() {
 406     let (tx, rx) = channel();
 407
 408     spawn(proc() {
 409         spawn(proc() {
 410             tx.send(());
 411         });
 412     });
 413
 414     rx.recv();
 415 }
 416
 417 #[cfg(test)]
 418 fn avoid_copying_the_body(spawnfn: |v: proc():Send|) {
 419     let (tx, rx) = channel::<uint>();
 420
 421     let x = ~1;
 422     let x_in_parent = (&*x) as *int as uint;
 423
 424     spawnfn(proc() {
 425         let x_in_child = (&*x) as *int as uint;
 426         tx.send(x_in_child);
 427     });
 428
 429     let x_in_child = rx.recv();
 430     assert_eq!(x_in_parent, x_in_child);
 431 }
 432
 433 #[test]
 434 fn test_avoid_copying_the_body_spawn() {
 435     avoid_copying_the_body(spawn);
 436 }
 437
 438 #[test]
 439 fn test_avoid_copying_the_body_task_spawn() {
 440     avoid_copying_the_body(|f| {
 441         let builder = TaskBuilder::new();
 442         builder.spawn(proc() {
 443             f();
 444         });
 445     })
 446 }
 447
 448 #[test]
 449 fn test_avoid_copying_the_body_try() {
 450     avoid_copying_the_body(|f| {
 451         let _ = try(proc() {
 452             f()
 453         });
 454     })
 455 }
 456
 457 #[test]
 458 fn test_child_doesnt_ref_parent() {
 459     // If the child refcounts the parent task, this will stack overflow when
 460     // climbing the task tree to dereference each ancestor. (See #1789)
 461     // (well, it would if the constant were 8000+ - I lowered it to be more
 462     // valgrind-friendly. try this at home, instead..!)
 463     static generations: uint = 16;
 464     fn child_no(x: uint) -> proc():Send {
 465         return proc() {
 466             if x < generations {
 467                 TaskBuilder::new().spawn(child_no(x+1));
 468             }
 469         }
 470     }
 471     TaskBuilder::new().spawn(child_no(0));
 472 }
 473
 474 #[test]
 475 fn test_simple_newsched_spawn() {
 476     spawn(proc()())
 477 }
 478
 479 #[test]
 480 fn test_try_fail_message_static_str() {
 481     match try(proc() {
 482         fail!("static string");
 483     }) {
 484         Err(e) => {
 485             type T = &'static str;
 486             assert!(e.is::<T>());
 487             assert_eq!(*e.move::<T>().unwrap(), "static string");
 488         }
 489         Ok(()) => fail!()
 490     }
 491 }
 492
 493 #[test]
 494 fn test_try_fail_message_owned_str() {
 495     match try(proc() {
 496         fail!("owned string".to_owned());
 497     }) {
 498         Err(e) => {
 499             type T = ~str;
 500             assert!(e.is::<T>());
 501             assert_eq!(*e.move::<T>().unwrap(), "owned string".to_owned());
 502         }
 503         Ok(()) => fail!()
 504     }
 505 }
 506
 507 #[test]
 508 fn test_try_fail_message_any() {
 509     match try(proc() {
 510         fail!(~413u16 as ~Any:Send);
 511     }) {
 512         Err(e) => {
 513             type T = ~Any:Send;
 514             assert!(e.is::<T>());
 515             let any = e.move::<T>().unwrap();
 516             assert!(any.is::<u16>());
 517             assert_eq!(*any.move::<u16>().unwrap(), 413u16);
 518         }
 519         Ok(()) => fail!()
 520     }
 521 }
 522
 523 #[test]
 524 fn test_try_fail_message_unit_struct() {
 525     struct Juju;
 526
 527     match try(proc() {
 528         fail!(Juju)
 529     }) {
 530         Err(ref e) if e.is::<Juju>() => {}
 531         Err(_) | Ok(()) => fail!()
 532     }
 533 }