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