src/librustc/ty/maps/job.rs

   1 // Copyright 2017 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 #![allow(warnings)]
  12
  13 use std::mem;
  14 use rustc_data_structures::sync::{Lock, LockGuard, Lrc, Weak};
  15 use rustc_data_structures::OnDrop;
  16 use syntax_pos::Span;
  17 use ty::tls;
  18 use ty::maps::Query;
  19 use ty::maps::plumbing::CycleError;
  20 use ty::context::TyCtxt;
  21 use errors::Diagnostic;
  22 use std::process;
  23 use std::fmt;
  24 use std::collections::HashSet;
  25 #[cfg(parallel_queries)]
  26 use {
  27     rayon_core,
  28     parking_lot::{Mutex, Condvar},
  29     std::sync::atomic::Ordering,
  30     std::thread,
  31     std::iter,
  32     std::iter::FromIterator,
  33     syntax_pos::DUMMY_SP,
  34     rustc_data_structures::stable_hasher::{StableHasherResult, StableHasher, HashStable},
  35 };
  36
  37 /// Indicates the state of a query for a given key in a query map
  38 pub(super) enum QueryResult<'tcx> {
  39     /// An already executing query. The query job can be used to await for its completion
  40     Started(Lrc<QueryJob<'tcx>>),
  41
  42     /// The query panicked. Queries trying to wait on this will raise a fatal error / silently panic
  43     Poisoned,
  44 }
  45
  46 /// A span and a query key
  47 #[derive(Clone, Debug)]
  48 pub struct QueryInfo<'tcx> {
  49     /// The span for a reason this query was required
  50     pub span: Span,
  51     pub query: Query<'tcx>,
  52 }
  53
  54 /// A object representing an active query job.
  55 pub struct QueryJob<'tcx> {
  56     pub info: QueryInfo<'tcx>,
  57
  58     /// The parent query job which created this job and is implicitly waiting on it.
  59     pub parent: Option<Lrc<QueryJob<'tcx>>>,
  60
  61     /// Diagnostic messages which are emitted while the query executes
  62     pub diagnostics: Lock<Vec<Diagnostic>>,
  63
  64     #[cfg(parallel_queries)]
  65     latch: QueryLatch<'tcx>,
  66 }
  67
  68 impl<'tcx> QueryJob<'tcx> {
  69     /// Creates a new query job
  70     pub fn new(info: QueryInfo<'tcx>, parent: Option<Lrc<QueryJob<'tcx>>>) -> Self {
  71         QueryJob {
  72             diagnostics: Lock::new(Vec::new()),
  73             info,
  74             parent,
  75             #[cfg(parallel_queries)]
  76             latch: QueryLatch::new(),
  77         }
  78     }
  79
  80     /// Awaits for the query job to complete.
  81     ///
  82     /// For single threaded rustc there's no concurrent jobs running, so if we are waiting for any
  83     /// query that means that there is a query cycle, thus this always running a cycle error.
  84     pub(super) fn await<'lcx>(
  85         &self,
  86         tcx: TyCtxt<'_, 'tcx, 'lcx>,
  87         span: Span,
  88     ) -> Result<(), CycleError<'tcx>> {
  89         #[cfg(not(parallel_queries))]
  90         {
  91             self.find_cycle_in_stack(tcx, span)
  92         }
  93
  94         #[cfg(parallel_queries)]
  95         {
  96             tls::with_related_context(tcx, move |icx| {
  97                 let mut waiter = QueryWaiter {
  98                     query: &icx.query,
  99                     span,
 100                     cycle: None,
 101                     condvar: Condvar::new(),
 102                 };
 103                 self.latch.await(&mut waiter);
 104
 105                 match waiter.cycle {
 106                     None => Ok(()),
 107                     Some(cycle) => Err(cycle)
 108                 }
 109             })
 110         }
 111     }
 112
 113     #[cfg(not(parallel_queries))]
 114     fn find_cycle_in_stack<'lcx>(
 115         &self,
 116         tcx: TyCtxt<'_, 'tcx, 'lcx>,
 117         span: Span,
 118     ) -> Result<(), CycleError<'tcx>> {
 119         // Get the current executing query (waiter) and find the waitee amongst its parents
 120         let mut current_job = tls::with_related_context(tcx, |icx| icx.query.clone());
 121         let mut cycle = Vec::new();
 122
 123         while let Some(job) = current_job {
 124             cycle.insert(0, job.info.clone());
 125
 126             if &*job as *const _ == self as *const _ {
 127                 // This is the end of the cycle
 128                 // The span entry we included was for the usage
 129                 // of the cycle itself, and not part of the cycle
 130                 // Replace it with the span which caused the cycle to form
 131                 cycle[0].span = span;
 132                 // Find out why the cycle itself was used
 133                 let usage = job.parent.as_ref().map(|parent| {
 134                     (job.info.span, parent.info.query.clone())
 135                 });
 136                 return Err(CycleError { usage, cycle });
 137             }
 138
 139             current_job = job.parent.clone();
 140         }
 141
 142         panic!("did not find a cycle")
 143     }
 144
 145     /// Signals to waiters that the query is complete.
 146     ///
 147     /// This does nothing for single threaded rustc,
 148     /// as there are no concurrent jobs which could be waiting on us
 149     pub fn signal_complete(&self) {
 150         #[cfg(parallel_queries)]
 151         self.latch.set();
 152     }
 153 }
 154
 155 #[cfg(parallel_queries)]
 156 struct QueryWaiter<'tcx> {
 157     query: *const Option<Lrc<QueryJob<'tcx>>>,
 158     condvar: Condvar,
 159     span: Span,
 160     cycle: Option<CycleError<'tcx>>,
 161 }
 162
 163 #[cfg(parallel_queries)]
 164 impl<'tcx> QueryWaiter<'tcx> {
 165     fn notify(&self, registry: &rayon_core::Registry) {
 166         rayon_core::mark_unblocked(registry);
 167         self.condvar.notify_one();
 168     }
 169 }
 170
 171 #[cfg(parallel_queries)]
 172 struct QueryLatchInfo<'tcx> {
 173     complete: bool,
 174     waiters: Vec<*mut QueryWaiter<'tcx>>,
 175 }
 176
 177 // Required because of raw pointers
 178 #[cfg(parallel_queries)]
 179 unsafe impl<'tcx> Send for QueryLatchInfo<'tcx> {}
 180
 181 #[cfg(parallel_queries)]
 182 struct QueryLatch<'tcx> {
 183     info: Mutex<QueryLatchInfo<'tcx>>,
 184 }
 185
 186 #[cfg(parallel_queries)]
 187 impl<'tcx> QueryLatch<'tcx> {
 188     fn new() -> Self {
 189         QueryLatch {
 190             info: Mutex::new(QueryLatchInfo {
 191                 complete: false,
 192                 waiters: Vec::new(),
 193             }),
 194         }
 195     }
 196
 197     /// Awaits the caller on this latch by blocking the current thread.
 198     fn await(&self, waiter: &mut QueryWaiter<'tcx>) {
 199         let mut info = self.info.lock();
 200         if !info.complete {
 201             // We push the waiter on to the `waiters` list. It can be accessed inside
 202             // the `wait` call below, by 1) the `set` method or 2) by deadlock detection.
 203             // Both of these will remove it from the `waiters` list before resuming
 204             // this thread.
 205             info.waiters.push(waiter);
 206
 207             // If this detects a deadlock and the deadlock handler want to resume this thread
 208             // we have to be in the `wait` call. This is ensured by the deadlock handler
 209             // getting the self.info lock.
 210             rayon_core::mark_blocked();
 211             waiter.condvar.wait(&mut info);
 212         }
 213     }
 214
 215     /// Sets the latch and resumes all waiters on it
 216     fn set(&self) {
 217         let mut info = self.info.lock();
 218         debug_assert!(!info.complete);
 219         info.complete = true;
 220         let registry = rayon_core::Registry::current();
 221         for waiter in info.waiters.drain(..) {
 222             unsafe {
 223                 (*waiter).notify(&registry);
 224             }
 225         }
 226     }
 227
 228     /// Remove a single waiter from the list of waiters.
 229     /// This is used to break query cycles.
 230     fn extract_waiter(
 231         &self,
 232         waiter: usize,
 233     ) -> *mut QueryWaiter<'tcx> {
 234         let mut info = self.info.lock();
 235         debug_assert!(!info.complete);
 236         // Remove the waiter from the list of waiters
 237         info.waiters.remove(waiter)
 238     }
 239 }
 240
 241 /// A pointer to an active query job. This is used to give query jobs an identity.
 242 #[cfg(parallel_queries)]
 243 type Ref<'tcx> = *const QueryJob<'tcx>;
 244
 245 /// A resumable waiter of a query. The usize is the index into waiters in the query's latch
 246 #[cfg(parallel_queries)]
 247 type Waiter<'tcx> = (Ref<'tcx>, usize);
 248
 249 /// Visits all the non-resumable and resumable waiters of a query.
 250 /// Only waiters in a query are visited.
 251 /// `visit` is called for every waiter and is passed a query waiting on `query_ref`
 252 /// and a span indicating the reason the query waited on `query_ref`.
 253 /// If `visit` returns Some, this function returns.
 254 /// For visits of non-resumable waiters it returns the return value of `visit`.
 255 /// For visits of resumable waiters it returns Some(Some(Waiter)) which has the
 256 /// required information to resume the waiter.
 257 /// If all `visit` calls returns None, this function also returns None.
 258 #[cfg(parallel_queries)]
 259 fn visit_waiters<'tcx, F>(query_ref: Ref<'tcx>, mut visit: F) -> Option<Option<Waiter<'tcx>>>
 260 where
 261     F: FnMut(Span, Ref<'tcx>) -> Option<Option<Waiter<'tcx>>>
 262 {
 263     let query = unsafe { &*query_ref };
 264
 265     // Visit the parent query which is a non-resumable waiter since it's on the same stack
 266     if let Some(ref parent) = query.parent {
 267         if let Some(cycle) = visit(query.info.span, &**parent as Ref) {
 268             return Some(cycle);
 269         }
 270     }
 271
 272     // Visit the explict waiters which use condvars and are resumable
 273     for (i, &waiter) in query.latch.info.lock().waiters.iter().enumerate() {
 274         unsafe {
 275             if let Some(ref waiter_query) = *(*waiter).query {
 276                 if visit((*waiter).span, &**waiter_query as Ref).is_some() {
 277                     // Return a value which indicates that this waiter can be resumed
 278                     return Some(Some((query_ref, i)));
 279                 }
 280             }
 281         }
 282     }
 283     None
 284 }
 285
 286 /// Look for query cycles by doing a depth first search starting at `query`.
 287 /// `span` is the reason for the `query` to execute. This is initially DUMMY_SP.
 288 /// If a cycle is detected, this initial value is replaced with the span causing
 289 /// the cycle.
 290 #[cfg(parallel_queries)]
 291 fn cycle_check<'tcx>(query: Ref<'tcx>,
 292                      span: Span,
 293                      stack: &mut Vec<(Span, Ref<'tcx>)>,
 294                      visited: &mut HashSet<Ref<'tcx>>) -> Option<Option<Waiter<'tcx>>> {
 295     if visited.contains(&query) {
 296         return if let Some(p) = stack.iter().position(|q| q.1 == query) {
 297             // We detected a query cycle, fix up the initial span and return Some
 298
 299             // Remove previous stack entries
 300             stack.splice(0..p, iter::empty());
 301             // Replace the span for the first query with the cycle cause
 302             stack[0].0 = span;
 303             Some(None)
 304         } else {
 305             None
 306         }
 307     }
 308
 309     // Mark this query is visited and add it to the stack
 310     visited.insert(query);
 311     stack.push((span, query));
 312
 313     // Visit all the waiters
 314     let r = visit_waiters(query, |span, successor| {
 315         cycle_check(successor, span, stack, visited)
 316     });
 317
 318     // Remove the entry in our stack if we didn't find a cycle
 319     if r.is_none() {
 320         stack.pop();
 321     }
 322
 323     r
 324 }
 325
 326 /// Finds out if there's a path to the compiler root (aka. code which isn't in a query)
 327 /// from `query` without going through any of the queries in `visited`.
 328 /// This is achieved with a depth first search.
 329 #[cfg(parallel_queries)]
 330 fn connected_to_root<'tcx>(query: Ref<'tcx>, visited: &mut HashSet<Ref<'tcx>>) -> bool {
 331     // We already visited this or we're deliberately ignoring it
 332     if visited.contains(&query) {
 333         return false;
 334     }
 335
 336     // This query is connected to the root (it has no query parent), return true
 337     if unsafe { (*query).parent.is_none() } {
 338         return true;
 339     }
 340
 341     visited.insert(query);
 342
 343     let mut connected = false;
 344
 345     visit_waiters(query, |_, successor| {
 346         if connected_to_root(successor, visited) {
 347             Some(None)
 348         } else {
 349             None
 350         }
 351     }).is_some()
 352 }
 353
 354 /// Looks for query cycles starting from the last query in `jobs`.
 355 /// If a cycle is found, all queries in the cycle is removed from `jobs` and
 356 /// the function return true.
 357 /// If a cycle was not found, the starting query is removed from `jobs` and
 358 /// the function returns false.
 359 #[cfg(parallel_queries)]
 360 fn remove_cycle<'tcx>(
 361     jobs: &mut Vec<Ref<'tcx>>,
 362     wakelist: &mut Vec<*mut QueryWaiter<'tcx>>,
 363     tcx: TyCtxt<'_, 'tcx, '_>
 364 ) -> bool {
 365     let mut visited = HashSet::new();
 366     let mut stack = Vec::new();
 367     // Look for a cycle starting with the last query in `jobs`
 368     if let Some(waiter) = cycle_check(jobs.pop().unwrap(),
 369                                       DUMMY_SP,
 370                                       &mut stack,
 371                                       &mut visited) {
 372         // Reverse the stack so earlier entries require later entries
 373         stack.reverse();
 374
 375         // Extract the spans and queries into separate arrays
 376         let mut spans: Vec<_> = stack.iter().map(|e| e.0).collect();
 377         let queries = stack.iter().map(|e| e.1);
 378
 379         // Shift the spans so that queries are matched with the span for their waitee
 380         let last = spans.pop().unwrap();
 381         spans.insert(0, last);
 382
 383         // Zip them back together
 384         let mut stack: Vec<_> = spans.into_iter().zip(queries).collect();
 385
 386         // Remove the queries in our cycle from the list of jobs to look at
 387         for r in &stack {
 388             jobs.remove_item(&r.1);
 389         }
 390
 391         // Find the queries in the cycle which are
 392         // connected to queries outside the cycle
 393         let entry_points: Vec<Ref<'_>> = stack.iter().filter_map(|query| {
 394             // Mark all the other queries in the cycle as already visited
 395             let mut visited = HashSet::from_iter(stack.iter().filter_map(|q| {
 396                 if q.1 != query.1 {
 397                     Some(q.1)
 398                 } else {
 399                     None
 400                 }
 401             }));
 402
 403             if connected_to_root(query.1, &mut visited) {
 404                 Some(query.1)
 405             } else {
 406                 None
 407             }
 408         }).collect();
 409
 410         // Deterministically pick an entry point
 411         // FIXME: Sort this instead
 412         let mut hcx = tcx.create_stable_hashing_context();
 413         let entry_point = *entry_points.iter().min_by_key(|&&q| {
 414             let mut stable_hasher = StableHasher::<u64>::new();
 415             unsafe { (*q).info.query.hash_stable(&mut hcx, &mut stable_hasher); }
 416             stable_hasher.finish()
 417         }).unwrap();
 418
 419         // Shift the stack until our entry point is first
 420         while stack[0].1 != entry_point {
 421             let last = stack.pop().unwrap();
 422             stack.insert(0, last);
 423         }
 424
 425         // Create the cycle error
 426         let mut error = CycleError {
 427             usage: None,
 428             cycle: stack.iter().map(|&(s, q)| QueryInfo {
 429                 span: s,
 430                 query: unsafe { (*q).info.query.clone() },
 431             } ).collect(),
 432         };
 433
 434         // We unwrap `waiter` here since there must always be one
 435         // edge which is resumeable / waited using a query latch
 436         let (waitee_query, waiter_idx) = waiter.unwrap();
 437         let waitee_query = unsafe { &*waitee_query };
 438
 439         // Extract the waiter we want to resume
 440         let waiter = waitee_query.latch.extract_waiter(waiter_idx);
 441
 442         // Set the cycle error it will be picked it up when resumed
 443         unsafe {
 444             (*waiter).cycle = Some(error);
 445         }
 446
 447         // Put the waiter on the list of things to resume
 448         wakelist.push(waiter);
 449
 450         true
 451     } else {
 452         false
 453     }
 454 }
 455
 456 /// Creates a new thread and forwards information in thread locals to it.
 457 /// The new thread runs the deadlock handler.
 458 #[cfg(parallel_queries)]
 459 pub fn handle_deadlock() {
 460     use syntax;
 461     use syntax_pos;
 462
 463     let registry = rayon_core::Registry::current();
 464
 465     let gcx_ptr = tls::GCX_PTR.with(|gcx_ptr| {
 466         gcx_ptr as *const _
 467     });
 468     let gcx_ptr = unsafe { &*gcx_ptr };
 469
 470     let syntax_globals = syntax::GLOBALS.with(|syntax_globals| {
 471         syntax_globals as *const _
 472     });
 473     let syntax_globals = unsafe { &*syntax_globals };
 474
 475     let syntax_pos_globals = syntax_pos::GLOBALS.with(|syntax_pos_globals| {
 476         syntax_pos_globals as *const _
 477     });
 478     let syntax_pos_globals = unsafe { &*syntax_pos_globals };
 479     thread::spawn(move || {
 480         tls::GCX_PTR.set(gcx_ptr, || {
 481             syntax_pos::GLOBALS.set(syntax_pos_globals, || {
 482                 syntax_pos::GLOBALS.set(syntax_pos_globals, || {
 483                     tls::with_thread_locals(|| {
 484                         unsafe {
 485                             tls::with_global(|tcx| deadlock(tcx, &registry))
 486                         }
 487                     })
 488                 })
 489             })
 490         })
 491     });
 492 }
 493
 494 /// Detects query cycles by using depth first search over all active query jobs.
 495 /// If a query cycle is found it will break the cycle by finding an edge which
 496 /// uses a query latch and then resuming that waiter.
 497 /// There may be multiple cycles involved in a deadlock, so this searches
 498 /// all active queries for cycles before finally resuming all the waiters at once.
 499 #[cfg(parallel_queries)]
 500 fn deadlock(tcx: TyCtxt<'_, '_, '_>, registry: &rayon_core::Registry) {
 501     let on_panic = OnDrop(|| {
 502         eprintln!("deadlock handler panicked, aborting process");
 503         process::abort();
 504     });
 505
 506     let mut wakelist = Vec::new();
 507     let mut jobs: Vec<_> = tcx.maps.collect_active_jobs().iter().map(|j| &**j as Ref).collect();
 508
 509     let mut found_cycle = false;
 510
 511     while jobs.len() > 0 {
 512         if remove_cycle(&mut jobs, &mut wakelist, tcx) {
 513             found_cycle = true;
 514         }
 515     }
 516
 517     // Check that a cycle was found. It is possible for a deadlock to occur without
 518     // a query cycle if a query which can be waited on uses Rayon to do multithreading
 519     // internally. Such a query (X) may be executing on 2 threads (A and B) and A may
 520     // wait using Rayon on B. Rayon may then switch to executing another query (Y)
 521     // which in turn will wait on X causing a deadlock. We have a false dependency from
 522     // X to Y due to Rayon waiting and a true dependency from Y to X. The algorithm here
 523     // only considers the true dependency and won't detect a cycle.
 524     assert!(found_cycle);
 525
 526     // FIXME: Ensure this won't cause a deadlock before we return
 527     for waiter in wakelist.into_iter() {
 528         unsafe {
 529             (*waiter).notify(registry);
 530         }
 531     }
 532
 533     on_panic.disable();
 534 }