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