src/libstd/rt/task.rs

   1 // Copyright 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 //! Language-level runtime services that should reasonably expected
  12 //! to be available 'everywhere'. Local heaps, GC, unwinding,
  13 //! local storage, and logging. Even a 'freestanding' Rust would likely want
  14 //! to implement this.
  15
  16 use any::AnyOwnExt;
  17 use borrow;
  18 use cast;
  19 use cleanup;
  20 use io::Writer;
  21 use iter::{Iterator, Take};
  22 use local_data;
  23 use ops::Drop;
  24 use option::{Option, Some, None};
  25 use prelude::drop;
  26 use result::{Result, Ok, Err};
  27 use rt::Runtime;
  28 use rt::borrowck::BorrowRecord;
  29 use rt::borrowck;
  30 use rt::local::Local;
  31 use rt::local_heap::LocalHeap;
  32 use rt::logging::StdErrLogger;
  33 use rt::rtio::LocalIo;
  34 use rt::unwind::Unwinder;
  35 use send_str::SendStr;
  36 use sync::arc::UnsafeArc;
  37 use sync::atomics::{AtomicUint, SeqCst};
  38 use task::{TaskResult, TaskOpts};
  39 use unstable::finally::Finally;
  40
  41 #[cfg(stage0)]
  42 pub use rt::unwind::begin_unwind;
  43
  44 // The Task struct represents all state associated with a rust
  45 // task. There are at this point two primary "subtypes" of task,
  46 // however instead of using a subtype we just have a "task_type" field
  47 // in the struct. This contains a pointer to another struct that holds
  48 // the type-specific state.
  49
  50 pub struct Task {
  51     heap: LocalHeap,
  52     gc: GarbageCollector,
  53     storage: LocalStorage,
  54     unwinder: Unwinder,
  55     death: Death,
  56     destroyed: bool,
  57     name: Option<SendStr>,
  58     // Dynamic borrowck debugging info
  59     borrow_list: Option<~[BorrowRecord]>,
  60
  61     logger: Option<StdErrLogger>,
  62     stdout_handle: Option<~Writer>,
  63
  64     priv imp: Option<~Runtime>,
  65 }
  66
  67 pub struct GarbageCollector;
  68 pub struct LocalStorage(Option<local_data::Map>);
  69
  70 /// A handle to a blocked task. Usually this means having the ~Task pointer by
  71 /// ownership, but if the task is killable, a killer can steal it at any time.
  72 pub enum BlockedTask {
  73     Owned(~Task),
  74     Shared(UnsafeArc<AtomicUint>),
  75 }
  76
  77 /// Per-task state related to task death, killing, failure, etc.
  78 pub struct Death {
  79     // Action to be done with the exit code. If set, also makes the task wait
  80     // until all its watched children exit before collecting the status.
  81     on_exit: Option<proc(TaskResult)>,
  82 }
  83
  84 pub struct BlockedTaskIterator {
  85     priv inner: UnsafeArc<AtomicUint>,
  86 }
  87
  88 impl Task {
  89     pub fn new() -> Task {
  90         Task {
  91             heap: LocalHeap::new(),
  92             gc: GarbageCollector,
  93             storage: LocalStorage(None),
  94             unwinder: Unwinder::new(),
  95             death: Death::new(),
  96             destroyed: false,
  97             name: None,
  98             borrow_list: None,
  99             logger: None,
 100             stdout_handle: None,
 101             imp: None,
 102         }
 103     }
 104
 105     /// Executes the given closure as if it's running inside this task. The task
 106     /// is consumed upon entry, and the destroyed task is returned from this
 107     /// function in order for the caller to free. This function is guaranteed to
 108     /// not unwind because the closure specified is run inside of a `rust_try`
 109     /// block. (this is the only try/catch block in the world).
 110     ///
 111     /// This function is *not* meant to be abused as a "try/catch" block. This
 112     /// is meant to be used at the absolute boundaries of a task's lifetime, and
 113     /// only for that purpose.
 114     pub fn run(~self, f: ||) -> ~Task {
 115         // Need to put ourselves into TLS, but also need access to the unwinder.
 116         // Unsafely get a handle to the task so we can continue to use it after
 117         // putting it in tls (so we can invoke the unwinder).
 118         let handle: *mut Task = unsafe {
 119             *cast::transmute::<&~Task, &*mut Task>(&self)
 120         };
 121         Local::put(self);
 122
 123         // The only try/catch block in the world. Attempt to run the task's
 124         // client-specified code and catch any failures.
 125         let try_block = || {
 126
 127             // Run the task main function, then do some cleanup.
 128             f.finally(|| {
 129                 fn flush(w: Option<~Writer>) {
 130                     match w {
 131                         Some(mut w) => { w.flush(); }
 132                         None => {}
 133                     }
 134                 }
 135
 136                 // First, destroy task-local storage. This may run user dtors.
 137                 //
 138                 // FIXME #8302: Dear diary. I'm so tired and confused.
 139                 // There's some interaction in rustc between the box
 140                 // annihilator and the TLS dtor by which TLS is
 141                 // accessed from annihilated box dtors *after* TLS is
 142                 // destroyed. Somehow setting TLS back to null, as the
 143                 // old runtime did, makes this work, but I don't currently
 144                 // understand how. I would expect that, if the annihilator
 145                 // reinvokes TLS while TLS is uninitialized, that
 146                 // TLS would be reinitialized but never destroyed,
 147                 // but somehow this works. I have no idea what's going
 148                 // on but this seems to make things magically work. FML.
 149                 //
 150                 // (added after initial comment) A possible interaction here is
 151                 // that the destructors for the objects in TLS themselves invoke
 152                 // TLS, or possibly some destructors for those objects being
 153                 // annihilated invoke TLS. Sadly these two operations seemed to
 154                 // be intertwined, and miraculously work for now...
 155                 let mut task = Local::borrow(None::<Task>);
 156                 let storage_map = {
 157                     let task = task.get();
 158                     let LocalStorage(ref mut optmap) = task.storage;
 159                     optmap.take()
 160                 };
 161                 drop(task);
 162                 drop(storage_map);
 163
 164                 // Destroy remaining boxes. Also may run user dtors.
 165                 unsafe { cleanup::annihilate(); }
 166
 167                 // Finally flush and destroy any output handles which the task
 168                 // owns. There are no boxes here, and no user destructors should
 169                 // run after this any more.
 170                 let mut task = Local::borrow(None::<Task>);
 171                 let stdout = task.get().stdout_handle.take();
 172                 let logger = task.get().logger.take();
 173                 drop(task);
 174
 175                 flush(stdout);
 176                 drop(logger);
 177             })
 178         };
 179
 180         unsafe { (*handle).unwinder.try(try_block); }
 181
 182         // Cleanup the dynamic borrowck debugging info
 183         borrowck::clear_task_borrow_list();
 184
 185         // Here we must unsafely borrow the task in order to not remove it from
 186         // TLS. When collecting failure, we may attempt to send on a channel (or
 187         // just run aribitrary code), so we must be sure to still have a local
 188         // task in TLS.
 189         unsafe {
 190             let me: *mut Task = Local::unsafe_borrow();
 191             (*me).death.collect_failure((*me).unwinder.result());
 192         }
 193         let mut me: ~Task = Local::take();
 194         me.destroyed = true;
 195         return me;
 196     }
 197
 198     /// Inserts a runtime object into this task, transferring ownership to the
 199     /// task. It is illegal to replace a previous runtime object in this task
 200     /// with this argument.
 201     pub fn put_runtime(&mut self, ops: ~Runtime) {
 202         assert!(self.imp.is_none());
 203         self.imp = Some(ops);
 204     }
 205
 206     /// Attempts to extract the runtime as a specific type. If the runtime does
 207     /// not have the provided type, then the runtime is not removed. If the
 208     /// runtime does have the specified type, then it is removed and returned
 209     /// (transfer of ownership).
 210     ///
 211     /// It is recommended to only use this method when *absolutely necessary*.
 212     /// This function may not be available in the future.
 213     pub fn maybe_take_runtime<T: 'static>(&mut self) -> Option<~T> {
 214         // This is a terrible, terrible function. The general idea here is to
 215         // take the runtime, cast it to ~Any, check if it has the right type,
 216         // and then re-cast it back if necessary. The method of doing this is
 217         // pretty sketchy and involves shuffling vtables of trait objects
 218         // around, but it gets the job done.
 219         //
 220         // XXX: This function is a serious code smell and should be avoided at
 221         //      all costs. I have yet to think of a method to avoid this
 222         //      function, and I would be saddened if more usage of the function
 223         //      crops up.
 224         unsafe {
 225             let imp = self.imp.take_unwrap();
 226             let &(vtable, _): &(uint, uint) = cast::transmute(&imp);
 227             match imp.wrap().move::<T>() {
 228                 Ok(t) => Some(t),
 229                 Err(t) => {
 230                     let (_, obj): (uint, uint) = cast::transmute(t);
 231                     let obj: ~Runtime = cast::transmute((vtable, obj));
 232                     self.put_runtime(obj);
 233                     None
 234                 }
 235             }
 236         }
 237     }
 238
 239     /// Spawns a sibling to this task. The newly spawned task is configured with
 240     /// the `opts` structure and will run `f` as the body of its code.
 241     pub fn spawn_sibling(mut ~self, opts: TaskOpts, f: proc()) {
 242         let ops = self.imp.take_unwrap();
 243         ops.spawn_sibling(self, opts, f)
 244     }
 245
 246     /// Deschedules the current task, invoking `f` `amt` times. It is not
 247     /// recommended to use this function directly, but rather communication
 248     /// primitives in `std::comm` should be used.
 249     pub fn deschedule(mut ~self, amt: uint,
 250                       f: |BlockedTask| -> Result<(), BlockedTask>) {
 251         let ops = self.imp.take_unwrap();
 252         ops.deschedule(amt, self, f)
 253     }
 254
 255     /// Wakes up a previously blocked task, optionally specifiying whether the
 256     /// current task can accept a change in scheduling. This function can only
 257     /// be called on tasks that were previously blocked in `deschedule`.
 258     pub fn reawaken(mut ~self, can_resched: bool) {
 259         let ops = self.imp.take_unwrap();
 260         ops.reawaken(self, can_resched);
 261     }
 262
 263     /// Yields control of this task to another task. This function will
 264     /// eventually return, but possibly not immediately. This is used as an
 265     /// opportunity to allow other tasks a chance to run.
 266     pub fn yield_now(mut ~self) {
 267         let ops = self.imp.take_unwrap();
 268         ops.yield_now(self);
 269     }
 270
 271     /// Similar to `yield_now`, except that this function may immediately return
 272     /// without yielding (depending on what the runtime decides to do).
 273     pub fn maybe_yield(mut ~self) {
 274         let ops = self.imp.take_unwrap();
 275         ops.maybe_yield(self);
 276     }
 277
 278     /// Acquires a handle to the I/O factory that this task contains, normally
 279     /// stored in the task's runtime. This factory may not always be available,
 280     /// which is why the return type is `Option`
 281     pub fn local_io<'a>(&'a mut self) -> Option<LocalIo<'a>> {
 282         self.imp.get_mut_ref().local_io()
 283     }
 284
 285     /// Returns the stack bounds for this task in (lo, hi) format. The stack
 286     /// bounds may not be known for all tasks, so the return value may be
 287     /// `None`.
 288     pub fn stack_bounds(&self) -> Option<(uint, uint)> {
 289         self.imp.get_ref().stack_bounds()
 290     }
 291 }
 292
 293 impl Drop for Task {
 294     fn drop(&mut self) {
 295         rtdebug!("called drop for a task: {}", borrow::to_uint(self));
 296         rtassert!(self.destroyed);
 297     }
 298 }
 299
 300 impl Iterator<BlockedTask> for BlockedTaskIterator {
 301     fn next(&mut self) -> Option<BlockedTask> {
 302         Some(Shared(self.inner.clone()))
 303     }
 304 }
 305
 306 impl BlockedTask {
 307     /// Returns Some if the task was successfully woken; None if already killed.
 308     pub fn wake(self) -> Option<~Task> {
 309         match self {
 310             Owned(task) => Some(task),
 311             Shared(arc) => unsafe {
 312                 match (*arc.get()).swap(0, SeqCst) {
 313                     0 => None,
 314                     n => Some(cast::transmute(n)),
 315                 }
 316             }
 317         }
 318     }
 319
 320     // This assertion has two flavours because the wake involves an atomic op.
 321     // In the faster version, destructors will fail dramatically instead.
 322     #[cfg(not(test))] pub fn trash(self) { }
 323     #[cfg(test)]      pub fn trash(self) { assert!(self.wake().is_none()); }
 324
 325     /// Create a blocked task, unless the task was already killed.
 326     pub fn block(task: ~Task) -> BlockedTask {
 327         Owned(task)
 328     }
 329
 330     /// Converts one blocked task handle to a list of many handles to the same.
 331     pub fn make_selectable(self, num_handles: uint) -> Take<BlockedTaskIterator>
 332     {
 333         let arc = match self {
 334             Owned(task) => {
 335                 let flag = unsafe { AtomicUint::new(cast::transmute(task)) };
 336                 UnsafeArc::new(flag)
 337             }
 338             Shared(arc) => arc.clone(),
 339         };
 340         BlockedTaskIterator{ inner: arc }.take(num_handles)
 341     }
 342
 343     /// Convert to an unsafe uint value. Useful for storing in a pipe's state
 344     /// flag.
 345     #[inline]
 346     pub unsafe fn cast_to_uint(self) -> uint {
 347         match self {
 348             Owned(task) => {
 349                 let blocked_task_ptr: uint = cast::transmute(task);
 350                 rtassert!(blocked_task_ptr & 0x1 == 0);
 351                 blocked_task_ptr
 352             }
 353             Shared(arc) => {
 354                 let blocked_task_ptr: uint = cast::transmute(~arc);
 355                 rtassert!(blocked_task_ptr & 0x1 == 0);
 356                 blocked_task_ptr | 0x1
 357             }
 358         }
 359     }
 360
 361     /// Convert from an unsafe uint value. Useful for retrieving a pipe's state
 362     /// flag.
 363     #[inline]
 364     pub unsafe fn cast_from_uint(blocked_task_ptr: uint) -> BlockedTask {
 365         if blocked_task_ptr & 0x1 == 0 {
 366             Owned(cast::transmute(blocked_task_ptr))
 367         } else {
 368             let ptr: ~UnsafeArc<AtomicUint> =
 369                 cast::transmute(blocked_task_ptr & !1);
 370             Shared(*ptr)
 371         }
 372     }
 373 }
 374
 375 impl Death {
 376     pub fn new() -> Death {
 377         Death { on_exit: None, }
 378     }
 379
 380     /// Collect failure exit codes from children and propagate them to a parent.
 381     pub fn collect_failure(&mut self, result: TaskResult) {
 382         match self.on_exit.take() {
 383             Some(f) => f(result),
 384             None => {}
 385         }
 386     }
 387 }
 388
 389 impl Drop for Death {
 390     fn drop(&mut self) {
 391         // make this type noncopyable
 392     }
 393 }
 394
 395 #[cfg(test)]
 396 mod test {
 397     use super::*;
 398     use prelude::*;
 399     use task;
 400
 401     #[test]
 402     fn local_heap() {
 403         let a = @5;
 404         let b = a;
 405         assert!(*a == 5);
 406         assert!(*b == 5);
 407     }
 408
 409     #[test]
 410     fn tls() {
 411         use local_data;
 412         local_data_key!(key: @~str)
 413         local_data::set(key, @~"data");
 414         assert!(*local_data::get(key, |k| k.map(|k| *k)).unwrap() == ~"data");
 415         local_data_key!(key2: @~str)
 416         local_data::set(key2, @~"data");
 417         assert!(*local_data::get(key2, |k| k.map(|k| *k)).unwrap() == ~"data");
 418     }
 419
 420     #[test]
 421     fn unwind() {
 422         let result = task::try(proc()());
 423         rtdebug!("trying first assert");
 424         assert!(result.is_ok());
 425         let result = task::try::<()>(proc() fail!());
 426         rtdebug!("trying second assert");
 427         assert!(result.is_err());
 428     }
 429
 430     #[test]
 431     fn rng() {
 432         use rand::{rng, Rng};
 433         let mut r = rng();
 434         let _ = r.next_u32();
 435     }
 436
 437     #[test]
 438     fn logging() {
 439         info!("here i am. logging in a newsched task");
 440     }
 441
 442     #[test]
 443     fn comm_stream() {
 444         let (port, chan) = Chan::new();
 445         chan.send(10);
 446         assert!(port.recv() == 10);
 447     }
 448
 449     #[test]
 450     fn comm_shared_chan() {
 451         let (port, chan) = SharedChan::new();
 452         chan.send(10);
 453         assert!(port.recv() == 10);
 454     }
 455
 456     #[test]
 457     fn heap_cycles() {
 458         use cell::RefCell;
 459         use option::{Option, Some, None};
 460
 461         struct List {
 462             next: Option<@RefCell<List>>,
 463         }
 464
 465         let a = @RefCell::new(List { next: None });
 466         let b = @RefCell::new(List { next: Some(a) });
 467
 468         {
 469             let mut a = a.borrow_mut();
 470             a.get().next = Some(b);
 471         }
 472     }
 473
 474     #[test]
 475     #[should_fail]
 476     fn test_begin_unwind() {
 477         use rt::unwind::begin_unwind;
 478         begin_unwind("cause", file!(), line!())
 479     }
 480
 481     // Task blocking tests
 482
 483     #[test]
 484     fn block_and_wake() {
 485         let task = ~Task::new();
 486         let mut task = BlockedTask::block(task).wake().unwrap();
 487         task.destroyed = true;
 488     }
 489 }