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