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 borrow;
  17 use cast::transmute;
  18 use cleanup;
  19 use libc::{c_void, uintptr_t};
  20 use ptr;
  21 use prelude::*;
  22 use option::{Option, Some, None};
  23 use rt::env;
  24 use rt::kill::Death;
  25 use rt::local::Local;
  26 use rt::logging::StdErrLogger;
  27 use super::local_heap::LocalHeap;
  28 use rt::sched::{Scheduler, SchedHandle};
  29 use rt::stack::{StackSegment, StackPool};
  30 use rt::context::Context;
  31 use unstable::finally::Finally;
  32 use task::spawn::Taskgroup;
  33 use cell::Cell;
  34
  35 // The Task struct represents all state associated with a rust
  36 // task. There are at this point two primary "subtypes" of task,
  37 // however instead of using a subtype we just have a "task_type" field
  38 // in the struct. This contains a pointer to another struct that holds
  39 // the type-specific state.
  40
  41 pub struct Task {
  42     heap: LocalHeap,
  43     gc: GarbageCollector,
  44     storage: LocalStorage,
  45     logger: StdErrLogger,
  46     unwinder: Unwinder,
  47     taskgroup: Option<Taskgroup>,
  48     death: Death,
  49     destroyed: bool,
  50     // FIXME(#6874/#7599) use StringRef to save on allocations
  51     name: Option<~str>,
  52     coroutine: Option<Coroutine>,
  53     sched: Option<~Scheduler>,
  54     task_type: TaskType
  55 }
  56
  57 pub enum TaskType {
  58     GreenTask(Option<~SchedHome>),
  59     SchedTask
  60 }
  61
  62 /// A coroutine is nothing more than a (register context, stack) pair.
  63 pub struct Coroutine {
  64     /// The segment of stack on which the task is currently running or
  65     /// if the task is blocked, on which the task will resume
  66     /// execution.
  67     priv current_stack_segment: StackSegment,
  68     /// Always valid if the task is alive and not running.
  69     saved_context: Context
  70 }
  71
  72 /// Some tasks have a deciated home scheduler that they must run on.
  73 pub enum SchedHome {
  74     AnySched,
  75     Sched(SchedHandle)
  76 }
  77
  78 pub struct GarbageCollector;
  79 pub struct LocalStorage(*c_void, Option<extern "Rust" fn(*c_void)>);
  80
  81 pub struct Unwinder {
  82     unwinding: bool,
  83 }
  84
  85 impl Task {
  86
  87     // A helper to build a new task using the dynamically found
  88     // scheduler and task. Only works in GreenTask context.
  89     pub fn build_homed_child(stack_size: Option<uint>, f: ~fn(), home: SchedHome) -> ~Task {
  90         let f = Cell::new(f);
  91         let home = Cell::new(home);
  92         do Local::borrow::<Task, ~Task> |running_task| {
  93             let mut sched = running_task.sched.take_unwrap();
  94             let new_task = ~running_task.new_child_homed(&mut sched.stack_pool,
  95                                                          stack_size,
  96                                                          home.take(),
  97                                                          f.take());
  98             running_task.sched = Some(sched);
  99             new_task
 100         }
 101     }
 102
 103     pub fn build_child(stack_size: Option<uint>, f: ~fn()) -> ~Task {
 104         Task::build_homed_child(stack_size, f, AnySched)
 105     }
 106
 107     pub fn build_homed_root(stack_size: Option<uint>, f: ~fn(), home: SchedHome) -> ~Task {
 108         let f = Cell::new(f);
 109         let home = Cell::new(home);
 110         do Local::borrow::<Task, ~Task> |running_task| {
 111             let mut sched = running_task.sched.take_unwrap();
 112             let new_task = ~Task::new_root_homed(&mut sched.stack_pool,
 113                                                  stack_size,
 114                                                  home.take(),
 115                                                  f.take());
 116             running_task.sched = Some(sched);
 117             new_task
 118         }
 119     }
 120
 121     pub fn build_root(stack_size: Option<uint>, f: ~fn()) -> ~Task {
 122         Task::build_homed_root(stack_size, f, AnySched)
 123     }
 124
 125     pub fn new_sched_task() -> Task {
 126         Task {
 127             heap: LocalHeap::new(),
 128             gc: GarbageCollector,
 129             storage: LocalStorage(ptr::null(), None),
 130             logger: StdErrLogger,
 131             unwinder: Unwinder { unwinding: false },
 132             taskgroup: None,
 133             death: Death::new(),
 134             destroyed: false,
 135             coroutine: Some(Coroutine::empty()),
 136             name: None,
 137             sched: None,
 138             task_type: SchedTask
 139         }
 140     }
 141
 142     pub fn new_root(stack_pool: &mut StackPool,
 143                     stack_size: Option<uint>,
 144                     start: ~fn()) -> Task {
 145         Task::new_root_homed(stack_pool, stack_size, AnySched, start)
 146     }
 147
 148     pub fn new_child(&mut self,
 149                      stack_pool: &mut StackPool,
 150                      stack_size: Option<uint>,
 151                      start: ~fn()) -> Task {
 152         self.new_child_homed(stack_pool, stack_size, AnySched, start)
 153     }
 154
 155     pub fn new_root_homed(stack_pool: &mut StackPool,
 156                           stack_size: Option<uint>,
 157                           home: SchedHome,
 158                           start: ~fn()) -> Task {
 159         Task {
 160             heap: LocalHeap::new(),
 161             gc: GarbageCollector,
 162             storage: LocalStorage(ptr::null(), None),
 163             logger: StdErrLogger,
 164             unwinder: Unwinder { unwinding: false },
 165             taskgroup: None,
 166             death: Death::new(),
 167             destroyed: false,
 168             name: None,
 169             coroutine: Some(Coroutine::new(stack_pool, stack_size, start)),
 170             sched: None,
 171             task_type: GreenTask(Some(~home))
 172         }
 173     }
 174
 175     pub fn new_child_homed(&mut self,
 176                            stack_pool: &mut StackPool,
 177                            stack_size: Option<uint>,
 178                            home: SchedHome,
 179                            start: ~fn()) -> Task {
 180         Task {
 181             heap: LocalHeap::new(),
 182             gc: GarbageCollector,
 183             storage: LocalStorage(ptr::null(), None),
 184             logger: StdErrLogger,
 185             unwinder: Unwinder { unwinding: false },
 186             taskgroup: None,
 187             // FIXME(#7544) make watching optional
 188             death: self.death.new_child(),
 189             destroyed: false,
 190             name: None,
 191             coroutine: Some(Coroutine::new(stack_pool, stack_size, start)),
 192             sched: None,
 193             task_type: GreenTask(Some(~home))
 194         }
 195     }
 196
 197     pub fn give_home(&mut self, new_home: SchedHome) {
 198         match self.task_type {
 199             GreenTask(ref mut home) => {
 200                 *home = Some(~new_home);
 201             }
 202             SchedTask => {
 203                 rtabort!("type error: used SchedTask as GreenTask");
 204             }
 205         }
 206     }
 207
 208     pub fn take_unwrap_home(&mut self) -> SchedHome {
 209         match self.task_type {
 210             GreenTask(ref mut home) => {
 211                 let out = home.take_unwrap();
 212                 return *out;
 213             }
 214             SchedTask => {
 215                 rtabort!("type error: used SchedTask as GreenTask");
 216             }
 217         }
 218     }
 219
 220     pub fn run(&mut self, f: &fn()) {
 221         rtdebug!("run called on task: %u", borrow::to_uint(self));
 222
 223         // The only try/catch block in the world. Attempt to run the task's
 224         // client-specified code and catch any failures.
 225         do self.unwinder.try {
 226
 227             // Run the task main function, then do some cleanup.
 228             do f.finally {
 229
 230                 // Destroy task-local storage. This may run user dtors.
 231                 match self.storage {
 232                     LocalStorage(ptr, Some(ref dtor)) => {
 233                         (*dtor)(ptr)
 234                     }
 235                     _ => ()
 236                 }
 237
 238                 // FIXME #8302: Dear diary. I'm so tired and confused.
 239                 // There's some interaction in rustc between the box
 240                 // annihilator and the TLS dtor by which TLS is
 241                 // accessed from annihilated box dtors *after* TLS is
 242                 // destroyed. Somehow setting TLS back to null, as the
 243                 // old runtime did, makes this work, but I don't currently
 244                 // understand how. I would expect that, if the annihilator
 245                 // reinvokes TLS while TLS is uninitialized, that
 246                 // TLS would be reinitialized but never destroyed,
 247                 // but somehow this works. I have no idea what's going
 248                 // on but this seems to make things magically work. FML.
 249                 self.storage = LocalStorage(ptr::null(), None);
 250
 251                 // Destroy remaining boxes. Also may run user dtors.
 252                 unsafe { cleanup::annihilate(); }
 253             }
 254         }
 255
 256         // FIXME(#7544): We pass the taskgroup into death so that it can be
 257         // dropped while the unkillable counter is set. This should not be
 258         // necessary except for an extraneous clone() in task/spawn.rs that
 259         // causes a killhandle to get dropped, which mustn't receive a kill
 260         // signal since we're outside of the unwinder's try() scope.
 261         // { let _ = self.taskgroup.take(); }
 262         self.death.collect_failure(!self.unwinder.unwinding, self.taskgroup.take());
 263         self.destroyed = true;
 264     }
 265
 266     // New utility functions for homes.
 267
 268     pub fn is_home_no_tls(&self, sched: &~Scheduler) -> bool {
 269         match self.task_type {
 270             GreenTask(Some(~AnySched)) => { false }
 271             GreenTask(Some(~Sched(SchedHandle { sched_id: ref id, _}))) => {
 272                 *id == sched.sched_id()
 273             }
 274             GreenTask(None) => {
 275                 rtabort!("task without home");
 276             }
 277             SchedTask => {
 278                 // Awe yea
 279                 rtabort!("type error: expected: GreenTask, found: SchedTask");
 280             }
 281         }
 282     }
 283
 284     pub fn homed(&self) -> bool {
 285         match self.task_type {
 286             GreenTask(Some(~AnySched)) => { false }
 287             GreenTask(Some(~Sched(SchedHandle { _ }))) => { true }
 288             GreenTask(None) => {
 289                 rtabort!("task without home");
 290             }
 291             SchedTask => {
 292                 rtabort!("type error: expected: GreenTask, found: SchedTask");
 293             }
 294         }
 295     }
 296
 297     // Grab both the scheduler and the task from TLS and check if the
 298     // task is executing on an appropriate scheduler.
 299     pub fn on_appropriate_sched() -> bool {
 300         do Local::borrow::<Task,bool> |task| {
 301             let sched_id = task.sched.get_ref().sched_id();
 302             let sched_run_anything = task.sched.get_ref().run_anything;
 303             match task.task_type {
 304                 GreenTask(Some(~AnySched)) => {
 305                     rtdebug!("anysched task in sched check ****");
 306                     sched_run_anything
 307                 }
 308                 GreenTask(Some(~Sched(SchedHandle { sched_id: ref id, _ }))) => {
 309                     rtdebug!("homed task in sched check ****");
 310                     *id == sched_id
 311                 }
 312                 GreenTask(None) => {
 313                     rtabort!("task without home");
 314                 }
 315                 SchedTask => {
 316                     rtabort!("type error: expected: GreenTask, found: SchedTask");
 317                 }
 318             }
 319         }
 320     }
 321 }
 322
 323 impl Drop for Task {
 324     fn drop(&self) {
 325         rtdebug!("called drop for a task: %u", borrow::to_uint(self));
 326         rtassert!(self.destroyed)
 327     }
 328 }
 329
 330 // Coroutines represent nothing more than a context and a stack
 331 // segment.
 332
 333 impl Coroutine {
 334
 335     pub fn new(stack_pool: &mut StackPool, stack_size: Option<uint>, start: ~fn()) -> Coroutine {
 336         let stack_size = match stack_size {
 337             Some(size) => size,
 338             None => env::min_stack()
 339         };
 340         let start = Coroutine::build_start_wrapper(start);
 341         let mut stack = stack_pool.take_segment(stack_size);
 342         let initial_context = Context::new(start, &mut stack);
 343         Coroutine {
 344             current_stack_segment: stack,
 345             saved_context: initial_context
 346         }
 347     }
 348
 349     pub fn empty() -> Coroutine {
 350         Coroutine {
 351             current_stack_segment: StackSegment::new(0),
 352             saved_context: Context::empty()
 353         }
 354     }
 355
 356     fn build_start_wrapper(start: ~fn()) -> ~fn() {
 357         let start_cell = Cell::new(start);
 358         let wrapper: ~fn() = || {
 359             // First code after swap to this new context. Run our
 360             // cleanup job.
 361             unsafe {
 362
 363                 // Again - might work while safe, or it might not.
 364                 do Local::borrow::<Scheduler,()> |sched| {
 365                     (sched).run_cleanup_job();
 366                 }
 367
 368                 // To call the run method on a task we need a direct
 369                 // reference to it. The task is in TLS, so we can
 370                 // simply unsafe_borrow it to get this reference. We
 371                 // need to still have the task in TLS though, so we
 372                 // need to unsafe_borrow.
 373                 let task = Local::unsafe_borrow::<Task>();
 374
 375                 do (*task).run {
 376                     // N.B. Removing `start` from the start wrapper
 377                     // closure by emptying a cell is critical for
 378                     // correctness. The ~Task pointer, and in turn the
 379                     // closure used to initialize the first call
 380                     // frame, is destroyed in the scheduler context,
 381                     // not task context. So any captured closures must
 382                     // not contain user-definable dtors that expect to
 383                     // be in task context. By moving `start` out of
 384                     // the closure, all the user code goes our of
 385                     // scope while the task is still running.
 386                     let start = start_cell.take();
 387                     start();
 388                 };
 389             }
 390
 391             // We remove the sched from the Task in TLS right now.
 392             let sched = Local::take::<Scheduler>();
 393             // ... allowing us to give it away when performing a
 394             // scheduling operation.
 395             sched.terminate_current_task()
 396         };
 397         return wrapper;
 398     }
 399
 400     /// Destroy coroutine and try to reuse stack segment.
 401     pub fn recycle(self, stack_pool: &mut StackPool) {
 402         match self {
 403             Coroutine { current_stack_segment, _ } => {
 404                 stack_pool.give_segment(current_stack_segment);
 405             }
 406         }
 407     }
 408
 409 }
 410
 411
 412 // Just a sanity check to make sure we are catching a Rust-thrown exception
 413 static UNWIND_TOKEN: uintptr_t = 839147;
 414
 415 impl Unwinder {
 416     pub fn try(&mut self, f: &fn()) {
 417         use unstable::raw::Closure;
 418
 419         unsafe {
 420             let closure: Closure = transmute(f);
 421             let code = transmute(closure.code);
 422             let env = transmute(closure.env);
 423
 424             let token = rust_try(try_fn, code, env);
 425             assert!(token == 0 || token == UNWIND_TOKEN);
 426         }
 427
 428         extern fn try_fn(code: *c_void, env: *c_void) {
 429             unsafe {
 430                 let closure: Closure = Closure {
 431                     code: transmute(code),
 432                     env: transmute(env),
 433                 };
 434                 let closure: &fn() = transmute(closure);
 435                 closure();
 436             }
 437         }
 438
 439         extern {
 440             #[rust_stack]
 441             fn rust_try(f: *u8, code: *c_void, data: *c_void) -> uintptr_t;
 442         }
 443     }
 444
 445     pub fn begin_unwind(&mut self) -> ! {
 446         self.unwinding = true;
 447         unsafe {
 448             rust_begin_unwind(UNWIND_TOKEN);
 449             return transmute(());
 450         }
 451         extern {
 452             fn rust_begin_unwind(token: uintptr_t);
 453         }
 454     }
 455 }
 456
 457 #[cfg(test)]
 458 mod test {
 459     use rt::test::*;
 460
 461     #[test]
 462     fn local_heap() {
 463         do run_in_newsched_task() {
 464             let a = @5;
 465             let b = a;
 466             assert!(*a == 5);
 467             assert!(*b == 5);
 468         }
 469     }
 470
 471     #[test]
 472     fn tls() {
 473         use local_data;
 474         do run_in_newsched_task() {
 475             static key: local_data::Key<@~str> = &local_data::Key;
 476             local_data::set(key, @~"data");
 477             assert!(*local_data::get(key, |k| k.map_move(|k| *k)).unwrap() == ~"data");
 478             static key2: local_data::Key<@~str> = &local_data::Key;
 479             local_data::set(key2, @~"data");
 480             assert!(*local_data::get(key2, |k| k.map_move(|k| *k)).unwrap() == ~"data");
 481         }
 482     }
 483
 484     #[test]
 485     fn unwind() {
 486         do run_in_newsched_task() {
 487             let result = spawntask_try(||());
 488             rtdebug!("trying first assert");
 489             assert!(result.is_ok());
 490             let result = spawntask_try(|| fail!());
 491             rtdebug!("trying second assert");
 492             assert!(result.is_err());
 493         }
 494     }
 495
 496     #[test]
 497     fn rng() {
 498         do run_in_newsched_task() {
 499             use rand::{rng, Rng};
 500             let mut r = rng();
 501             let _ = r.next();
 502         }
 503     }
 504
 505     #[test]
 506     fn logging() {
 507         do run_in_newsched_task() {
 508             info!("here i am. logging in a newsched task");
 509         }
 510     }
 511
 512     #[test]
 513     fn comm_oneshot() {
 514         use comm::*;
 515
 516         do run_in_newsched_task {
 517             let (port, chan) = oneshot();
 518             send_one(chan, 10);
 519             assert!(recv_one(port) == 10);
 520         }
 521     }
 522
 523     #[test]
 524     fn comm_stream() {
 525         use comm::*;
 526
 527         do run_in_newsched_task() {
 528             let (port, chan) = stream();
 529             chan.send(10);
 530             assert!(port.recv() == 10);
 531         }
 532     }
 533
 534     #[test]
 535     fn comm_shared_chan() {
 536         use comm::*;
 537
 538         do run_in_newsched_task() {
 539             let (port, chan) = stream();
 540             let chan = SharedChan::new(chan);
 541             chan.send(10);
 542             assert!(port.recv() == 10);
 543         }
 544     }
 545
 546     #[test]
 547     fn linked_failure() {
 548         do run_in_newsched_task() {
 549             let res = do spawntask_try {
 550                 spawntask_random(|| fail!());
 551             };
 552             assert!(res.is_err());
 553         }
 554     }
 555
 556     #[test]
 557     fn heap_cycles() {
 558         use option::{Option, Some, None};
 559
 560         do run_in_newsched_task {
 561             struct List {
 562                 next: Option<@mut List>,
 563             }
 564
 565             let a = @mut List { next: None };
 566             let b = @mut List { next: Some(a) };
 567
 568             a.next = Some(b);
 569         }
 570     }
 571
 572     // XXX: This is a copy of test_future_result in std::task.
 573     // It can be removed once the scheduler is turned on by default.
 574     #[test]
 575     fn future_result() {
 576         do run_in_newsched_task {
 577             use option::{Some, None};
 578             use task::*;
 579
 580             let mut result = None;
 581             let mut builder = task();
 582             builder.future_result(|r| result = Some(r));
 583             do builder.spawn {}
 584             assert_eq!(result.unwrap().recv(), Success);
 585
 586             result = None;
 587             let mut builder = task();
 588             builder.future_result(|r| result = Some(r));
 589             builder.unlinked();
 590             do builder.spawn {
 591                 fail!();
 592             }
 593             assert_eq!(result.unwrap().recv(), Failure);
 594         }
 595     }
 596 }
 597