src/librustc/ty/query/job.rs

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