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