src/librustc/ty/query/job.rs

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