src/librustc/dep_graph/thread.rs

   1 // Copyright 2012-2015 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 //! Manages the communication between the compiler's main thread and
  12 //! the thread that constructs the dependency graph. The basic idea is
  13 //! to use double buffering to lower the cost of producing a message.
  14 //! In the compiler thread, we accumulate messages in a vector until
  15 //! the vector is full, or until we want to query the graph, and then
  16 //! we send that vector over to the depgraph thread. At the same time,
  17 //! we receive an empty vector from the depgraph thread that we can use
  18 //! to accumulate more messages. This way we only ever have two vectors
  19 //! allocated (and both have a fairly large capacity).
  20
  21 use hir::def_id::DefId;
  22 use rustc_data_structures::veccell::VecCell;
  23 use std::sync::mpsc::{self, Sender, Receiver};
  24 use std::thread;
  25
  26 use super::DepGraphQuery;
  27 use super::DepNode;
  28 use super::edges::DepGraphEdges;
  29 use super::shadow::ShadowGraph;
  30
  31 #[derive(Debug)]
  32 pub enum DepMessage {
  33     Read(DepNode<DefId>),
  34     Write(DepNode<DefId>),
  35     PushTask(DepNode<DefId>),
  36     PopTask(DepNode<DefId>),
  37     PushIgnore,
  38     PopIgnore,
  39     Query,
  40 }
  41
  42 pub struct DepGraphThreadData {
  43     enabled: bool,
  44
  45     // The "shadow graph" is a debugging aid. We give it each message
  46     // in real time as it arrives and it checks for various errors
  47     // (for example, a read/write when there is no current task; it
  48     // can also apply user-defined filters; see `shadow` module for
  49     // details). This only occurs if debug-assertions are enabled.
  50     //
  51     // Note that in some cases the same errors will occur when the
  52     // data is processed off the main thread, but that's annoying
  53     // because it lacks precision about the source of the error.
  54     shadow_graph: ShadowGraph,
  55
  56     // current buffer, where we accumulate messages
  57     messages: VecCell<DepMessage>,
  58
  59     // whence to receive new buffer when full
  60     swap_in: Receiver<Vec<DepMessage>>,
  61
  62     // where to send buffer when full
  63     swap_out: Sender<Vec<DepMessage>>,
  64
  65     // where to receive query results
  66     query_in: Receiver<DepGraphQuery<DefId>>,
  67 }
  68
  69 const INITIAL_CAPACITY: usize = 2048;
  70
  71 impl DepGraphThreadData {
  72     pub fn new(enabled: bool) -> DepGraphThreadData {
  73         let (tx1, rx1) = mpsc::channel();
  74         let (tx2, rx2) = mpsc::channel();
  75         let (txq, rxq) = mpsc::channel();
  76
  77         if enabled {
  78             thread::spawn(move || main(rx1, tx2, txq));
  79         }
  80
  81         DepGraphThreadData {
  82             enabled: enabled,
  83             shadow_graph: ShadowGraph::new(),
  84             messages: VecCell::with_capacity(INITIAL_CAPACITY),
  85             swap_in: rx2,
  86             swap_out: tx1,
  87             query_in: rxq,
  88         }
  89     }
  90
  91     /// True if we are actually building the full dep-graph.
  92     #[inline]
  93     pub fn is_fully_enabled(&self) -> bool {
  94         self.enabled
  95     }
  96
  97     /// True if (a) we are actually building the full dep-graph, or (b) we are
  98     /// only enqueuing messages in order to sanity-check them (which happens
  99     /// when debug assertions are enabled).
 100     #[inline]
 101     pub fn is_enqueue_enabled(&self) -> bool {
 102         self.is_fully_enabled() || self.shadow_graph.enabled()
 103     }
 104
 105     /// Sends the current batch of messages to the thread. Installs a
 106     /// new vector of messages.
 107     fn swap(&self) {
 108         assert!(self.is_fully_enabled(), "should never swap if not fully enabled");
 109
 110         // should be a buffer waiting for us (though of course we may
 111         // have to wait for depgraph thread to finish processing the
 112         // old messages)
 113         let new_messages = self.swap_in.recv().unwrap();
 114         assert!(new_messages.is_empty());
 115
 116         // swap in the empty buffer and extract the full one
 117         let old_messages = self.messages.swap(new_messages);
 118
 119         // send full buffer to depgraph thread to be processed
 120         self.swap_out.send(old_messages).unwrap();
 121     }
 122
 123     pub fn query(&self) -> DepGraphQuery<DefId> {
 124         assert!(self.is_fully_enabled(), "should never query if not fully enabled");
 125         self.enqueue(DepMessage::Query);
 126         self.swap();
 127         self.query_in.recv().unwrap()
 128     }
 129
 130     /// Enqueue a message to be sent when things are next swapped. (If
 131     /// the buffer is full, this may swap.)
 132     #[inline]
 133     pub fn enqueue(&self, message: DepMessage) {
 134         assert!(self.is_enqueue_enabled(), "should never enqueue if not enqueue-enabled");
 135         self.shadow_graph.enqueue(&message);
 136         if self.is_fully_enabled() {
 137             self.enqueue_enabled(message);
 138         }
 139     }
 140
 141     // Outline this fn since I expect it may want to be inlined
 142     // separately.
 143     fn enqueue_enabled(&self, message: DepMessage) {
 144         let len = self.messages.push(message);
 145         if len == INITIAL_CAPACITY {
 146             self.swap();
 147         }
 148     }
 149 }
 150
 151 /// Definition of the depgraph thread.
 152 pub fn main(swap_in: Receiver<Vec<DepMessage>>,
 153             swap_out: Sender<Vec<DepMessage>>,
 154             query_out: Sender<DepGraphQuery<DefId>>) {
 155     let mut edges = DepGraphEdges::new();
 156
 157     // the compiler thread always expects a fresh buffer to be
 158     // waiting, so queue one up
 159     swap_out.send(Vec::with_capacity(INITIAL_CAPACITY)).unwrap();
 160
 161     // process the buffers from compiler thread as we receive them
 162     for mut messages in swap_in {
 163         for msg in messages.drain(..) {
 164             match msg {
 165                 DepMessage::Read(node) => edges.read(node),
 166                 DepMessage::Write(node) => edges.write(node),
 167                 DepMessage::PushTask(node) => edges.push_task(node),
 168                 DepMessage::PopTask(node) => edges.pop_task(node),
 169                 DepMessage::PushIgnore => edges.push_ignore(),
 170                 DepMessage::PopIgnore => edges.pop_ignore(),
 171                 DepMessage::Query => query_out.send(edges.query()).unwrap(),
 172             }
 173         }
 174         if let Err(_) = swap_out.send(messages) {
 175             // the receiver must have been dropped already
 176             break;
 177         }
 178     }
 179 }