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.
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.
14 use rustc_data_structures::fx::FxHashSet;
15 use rustc_data_structures::sync::{Lock, LockGuard, Lrc, Weak};
16 use rustc_data_structures::OnDrop;
20 use ty::query::plumbing::CycleError;
21 use ty::context::TyCtxt;
22 use errors::Diagnostic;
26 #[cfg(parallel_queries)]
29 parking_lot::{Mutex, Condvar},
30 std::sync::atomic::Ordering,
33 std::iter::FromIterator,
35 rustc_data_structures::stable_hasher::{StableHasherResult, StableHasher, HashStable},
38 /// Indicates the state of a query for a given key in a query map
39 pub(super) enum QueryResult<'tcx> {
40 /// An already executing query. The query job can be used to await for its completion
41 Started(Lrc<QueryJob<'tcx>>),
43 /// The query panicked. Queries trying to wait on this will raise a fatal error / silently panic
47 /// A span and a query key
48 #[derive(Clone, Debug)]
49 pub struct QueryInfo<'tcx> {
50 /// The span for a reason this query was required
52 pub query: Query<'tcx>,
55 /// A object representing an active query job.
56 pub struct QueryJob<'tcx> {
57 pub info: QueryInfo<'tcx>,
59 /// The parent query job which created this job and is implicitly waiting on it.
60 pub parent: Option<Lrc<QueryJob<'tcx>>>,
62 /// Diagnostic messages which are emitted while the query executes
63 pub diagnostics: Lock<Vec<Diagnostic>>,
65 /// The latch which is used to wait on this job
66 #[cfg(parallel_queries)]
67 latch: QueryLatch<'tcx>,
70 impl<'tcx> QueryJob<'tcx> {
71 /// Creates a new query job
72 pub fn new(info: QueryInfo<'tcx>, parent: Option<Lrc<QueryJob<'tcx>>>) -> Self {
74 diagnostics: Lock::new(Vec::new()),
77 #[cfg(parallel_queries)]
78 latch: QueryLatch::new(),
82 /// Awaits for the query job to complete.
84 /// For single threaded rustc there's no concurrent jobs running, so if we are waiting for any
85 /// query that means that there is a query cycle, thus this always running a cycle error.
86 pub(super) fn await<'lcx>(
88 tcx: TyCtxt<'_, 'tcx, 'lcx>,
90 ) -> Result<(), CycleError<'tcx>> {
91 #[cfg(not(parallel_queries))]
93 self.find_cycle_in_stack(tcx, span)
96 #[cfg(parallel_queries)]
98 tls::with_related_context(tcx, move |icx| {
99 let mut waiter = Lrc::new(QueryWaiter {
100 query: icx.query.clone(),
102 cycle: Lock::new(None),
103 condvar: Condvar::new(),
105 self.latch.await(&waiter);
107 match Lrc::get_mut(&mut waiter).unwrap().cycle.get_mut().take() {
109 Some(cycle) => Err(cycle)
115 #[cfg(not(parallel_queries))]
116 fn find_cycle_in_stack<'lcx>(
118 tcx: TyCtxt<'_, 'tcx, 'lcx>,
120 ) -> Result<(), CycleError<'tcx>> {
121 // Get the current executing query (waiter) and find the waitee amongst its parents
122 let mut current_job = tls::with_related_context(tcx, |icx| icx.query.clone());
123 let mut cycle = Vec::new();
125 while let Some(job) = current_job {
126 cycle.insert(0, job.info.clone());
128 if ptr::eq(&*job, self) {
129 // This is the end of the cycle
130 // The span entry we included was for the usage
131 // of the cycle itself, and not part of the cycle
132 // Replace it with the span which caused the cycle to form
133 cycle[0].span = span;
134 // Find out why the cycle itself was used
135 let usage = job.parent.as_ref().map(|parent| {
136 (job.info.span, parent.info.query.clone())
138 return Err(CycleError { usage, cycle });
141 current_job = job.parent.clone();
144 panic!("did not find a cycle")
147 /// Signals to waiters that the query is complete.
149 /// This does nothing for single threaded rustc,
150 /// as there are no concurrent jobs which could be waiting on us
151 pub fn signal_complete(&self) {
152 #[cfg(parallel_queries)]
156 fn as_ptr(&self) -> *const QueryJob<'tcx> {
161 #[cfg(parallel_queries)]
162 struct QueryWaiter<'tcx> {
163 query: Option<Lrc<QueryJob<'tcx>>>,
166 cycle: Lock<Option<CycleError<'tcx>>>,
169 #[cfg(parallel_queries)]
170 impl<'tcx> QueryWaiter<'tcx> {
171 fn notify(&self, registry: &rayon_core::Registry) {
172 rayon_core::mark_unblocked(registry);
173 self.condvar.notify_one();
177 #[cfg(parallel_queries)]
178 struct QueryLatchInfo<'tcx> {
180 waiters: Vec<Lrc<QueryWaiter<'tcx>>>,
183 #[cfg(parallel_queries)]
184 struct QueryLatch<'tcx> {
185 info: Mutex<QueryLatchInfo<'tcx>>,
188 #[cfg(parallel_queries)]
189 impl<'tcx> QueryLatch<'tcx> {
192 info: Mutex::new(QueryLatchInfo {
199 /// Awaits the caller on this latch by blocking the current thread.
200 fn await(&self, waiter: &Lrc<QueryWaiter<'tcx>>) {
201 let mut info = self.info.lock();
203 // We push the waiter on to the `waiters` list. It can be accessed inside
204 // the `wait` call below, by 1) the `set` method or 2) by deadlock detection.
205 // Both of these will remove it from the `waiters` list before resuming
207 info.waiters.push(waiter.clone());
209 // If this detects a deadlock and the deadlock handler wants to resume this thread
210 // we have to be in the `wait` call. This is ensured by the deadlock handler
211 // getting the self.info lock.
212 rayon_core::mark_blocked();
213 waiter.condvar.wait(&mut info);
217 /// Sets the latch and resumes all waiters on it
219 let mut info = self.info.lock();
220 debug_assert!(!info.complete);
221 info.complete = true;
222 let registry = rayon_core::Registry::current();
223 for waiter in info.waiters.drain(..) {
224 waiter.notify(®istry);
228 /// Remove a single waiter from the list of waiters.
229 /// This is used to break query cycles.
233 ) -> Lrc<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)
241 /// A resumable waiter of a query. The usize is the index into waiters in the query's latch
242 #[cfg(parallel_queries)]
243 type Waiter<'tcx> = (Lrc<QueryJob<'tcx>>, usize);
245 /// Visits all the non-resumable and resumable waiters of a query.
246 /// Only waiters in a query are visited.
247 /// `visit` is called for every waiter and is passed a query waiting on `query_ref`
248 /// and a span indicating the reason the query waited on `query_ref`.
249 /// If `visit` returns Some, this function returns.
250 /// For visits of non-resumable waiters it returns the return value of `visit`.
251 /// For visits of resumable waiters it returns Some(Some(Waiter)) which has the
252 /// required information to resume the waiter.
253 /// If all `visit` calls returns None, this function also returns None.
254 #[cfg(parallel_queries)]
255 fn visit_waiters<'tcx, F>(query: Lrc<QueryJob<'tcx>>, mut visit: F) -> Option<Option<Waiter<'tcx>>>
257 F: FnMut(Span, Lrc<QueryJob<'tcx>>) -> Option<Option<Waiter<'tcx>>>
259 // Visit the parent query which is a non-resumable waiter since it's on the same stack
260 if let Some(ref parent) = query.parent {
261 if let Some(cycle) = visit(query.info.span, parent.clone()) {
266 // Visit the explicit waiters which use condvars and are resumable
267 for (i, waiter) in query.latch.info.lock().waiters.iter().enumerate() {
268 if let Some(ref waiter_query) = waiter.query {
269 if visit(waiter.span, waiter_query.clone()).is_some() {
270 // Return a value which indicates that this waiter can be resumed
271 return Some(Some((query.clone(), i)));
278 /// Look for query cycles by doing a depth first search starting at `query`.
279 /// `span` is the reason for the `query` to execute. This is initially DUMMY_SP.
280 /// If a cycle is detected, this initial value is replaced with the span causing
282 #[cfg(parallel_queries)]
283 fn cycle_check<'tcx>(query: Lrc<QueryJob<'tcx>>,
285 stack: &mut Vec<(Span, Lrc<QueryJob<'tcx>>)>,
286 visited: &mut FxHashSet<*const QueryJob<'tcx>>
287 ) -> Option<Option<Waiter<'tcx>>> {
288 if visited.contains(&query.as_ptr()) {
289 return if let Some(p) = stack.iter().position(|q| q.1.as_ptr() == query.as_ptr()) {
290 // We detected a query cycle, fix up the initial span and return Some
292 // Remove previous stack entries
293 stack.splice(0..p, iter::empty());
294 // Replace the span for the first query with the cycle cause
302 // Mark this query is visited and add it to the stack
303 visited.insert(query.as_ptr());
304 stack.push((span, query.clone()));
306 // Visit all the waiters
307 let r = visit_waiters(query, |span, successor| {
308 cycle_check(successor, span, stack, visited)
311 // Remove the entry in our stack if we didn't find a cycle
319 /// Finds out if there's a path to the compiler root (aka. code which isn't in a query)
320 /// from `query` without going through any of the queries in `visited`.
321 /// This is achieved with a depth first search.
322 #[cfg(parallel_queries)]
323 fn connected_to_root<'tcx>(
324 query: Lrc<QueryJob<'tcx>>,
325 visited: &mut FxHashSet<*const QueryJob<'tcx>>
327 // We already visited this or we're deliberately ignoring it
328 if visited.contains(&query.as_ptr()) {
332 // This query is connected to the root (it has no query parent), return true
333 if query.parent.is_none() {
337 visited.insert(query.as_ptr());
339 let mut connected = false;
341 visit_waiters(query, |_, successor| {
342 if connected_to_root(successor, visited) {
350 /// Looks for query cycles starting from the last query in `jobs`.
351 /// If a cycle is found, all queries in the cycle is removed from `jobs` and
352 /// the function return true.
353 /// If a cycle was not found, the starting query is removed from `jobs` and
354 /// the function returns false.
355 #[cfg(parallel_queries)]
356 fn remove_cycle<'tcx>(
357 jobs: &mut Vec<Lrc<QueryJob<'tcx>>>,
358 wakelist: &mut Vec<Lrc<QueryWaiter<'tcx>>>,
359 tcx: TyCtxt<'_, 'tcx, '_>
361 let mut visited = FxHashSet::default();
362 let mut stack = Vec::new();
363 // Look for a cycle starting with the last query in `jobs`
364 if let Some(waiter) = cycle_check(jobs.pop().unwrap(),
368 // Reverse the stack so earlier entries require later entries
371 // Extract the spans and queries into separate arrays
372 let mut spans: Vec<_> = stack.iter().map(|e| e.0).collect();
373 let queries = stack.into_iter().map(|e| e.1);
375 // Shift the spans so that queries are matched with the span for their waitee
376 let last = spans.pop().unwrap();
377 spans.insert(0, last);
379 // Zip them back together
380 let mut stack: Vec<_> = spans.into_iter().zip(queries).collect();
382 // Remove the queries in our cycle from the list of jobs to look at
384 if let Some(pos) = jobs.iter().position(|j| j.as_ptr() == r.1.as_ptr()) {
389 // Find the queries in the cycle which are
390 // connected to queries outside the cycle
391 let entry_points: Vec<Lrc<QueryJob<'tcx>>> = stack.iter().filter_map(|query| {
392 // Mark all the other queries in the cycle as already visited
393 let mut visited = FxHashSet::from_iter(stack.iter().filter_map(|q| {
394 if q.1.as_ptr() != query.1.as_ptr() {
401 if connected_to_root(query.1.clone(), &mut visited) {
402 Some(query.1.clone())
408 // Deterministically pick an entry point
409 // FIXME: Sort this instead
410 let mut hcx = tcx.create_stable_hashing_context();
411 let entry_point = entry_points.iter().min_by_key(|q| {
412 let mut stable_hasher = StableHasher::<u64>::new();
413 q.info.query.hash_stable(&mut hcx, &mut stable_hasher);
414 stable_hasher.finish()
415 }).unwrap().as_ptr();
417 // Shift the stack until our entry point is first
418 while stack[0].1.as_ptr() != entry_point {
419 let last = stack.pop().unwrap();
420 stack.insert(0, last);
423 // Create the cycle error
424 let mut error = CycleError {
426 cycle: stack.iter().map(|&(s, ref q)| QueryInfo {
428 query: q.info.query.clone(),
432 // We unwrap `waiter` here since there must always be one
433 // edge which is resumeable / waited using a query latch
434 let (waitee_query, waiter_idx) = waiter.unwrap();
436 // Extract the waiter we want to resume
437 let waiter = waitee_query.latch.extract_waiter(waiter_idx);
439 // Set the cycle error so it will be picked up when resumed
440 *waiter.cycle.lock() = Some(error);
442 // Put the waiter on the list of things to resume
443 wakelist.push(waiter);
451 /// Creates a new thread and forwards information in thread locals to it.
452 /// The new thread runs the deadlock handler.
453 /// Must only be called when a deadlock is about to happen.
454 #[cfg(parallel_queries)]
455 pub unsafe fn handle_deadlock() {
459 let registry = rayon_core::Registry::current();
461 let gcx_ptr = tls::GCX_PTR.with(|gcx_ptr| {
464 let gcx_ptr = &*gcx_ptr;
466 let syntax_globals = syntax::GLOBALS.with(|syntax_globals| {
467 syntax_globals as *const _
469 let syntax_globals = &*syntax_globals;
471 let syntax_pos_globals = syntax_pos::GLOBALS.with(|syntax_pos_globals| {
472 syntax_pos_globals as *const _
474 let syntax_pos_globals = &*syntax_pos_globals;
475 thread::spawn(move || {
476 tls::GCX_PTR.set(gcx_ptr, || {
477 syntax_pos::GLOBALS.set(syntax_pos_globals, || {
478 syntax_pos::GLOBALS.set(syntax_pos_globals, || {
479 tls::with_thread_locals(|| {
480 tls::with_global(|tcx| deadlock(tcx, ®istry))
488 /// Detects query cycles by using depth first search over all active query jobs.
489 /// If a query cycle is found it will break the cycle by finding an edge which
490 /// uses a query latch and then resuming that waiter.
491 /// There may be multiple cycles involved in a deadlock, so this searches
492 /// all active queries for cycles before finally resuming all the waiters at once.
493 #[cfg(parallel_queries)]
494 fn deadlock(tcx: TyCtxt<'_, '_, '_>, registry: &rayon_core::Registry) {
495 let on_panic = OnDrop(|| {
496 eprintln!("deadlock handler panicked, aborting process");
500 let mut wakelist = Vec::new();
501 let mut jobs: Vec<_> = tcx.queries.collect_active_jobs();
503 let mut found_cycle = false;
505 while jobs.len() > 0 {
506 if remove_cycle(&mut jobs, &mut wakelist, tcx) {
511 // Check that a cycle was found. It is possible for a deadlock to occur without
512 // a query cycle if a query which can be waited on uses Rayon to do multithreading
513 // internally. Such a query (X) may be executing on 2 threads (A and B) and A may
514 // wait using Rayon on B. Rayon may then switch to executing another query (Y)
515 // which in turn will wait on X causing a deadlock. We have a false dependency from
516 // X to Y due to Rayon waiting and a true dependency from Y to X. The algorithm here
517 // only considers the true dependency and won't detect a cycle.
518 assert!(found_cycle);
520 // FIXME: Ensure this won't cause a deadlock before we return
521 for waiter in wakelist.into_iter() {
522 waiter.notify(registry);