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