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