src/librustc/ty/query/job.rs

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