Add jobserver support to sccache

Cargo.toml

@@ -28,12 +28,14 @@ futures = "0.1.11"
 futures-cpupool = "0.1"
 hyper = { version = "0.11", optional = true }
 hyper-tls = { version = "0.1", optional = true }
+jobserver = "0.1"
 jsonwebtoken = { version = "2.0", optional = true }
 libc = "0.2.10"
 local-encoding = "0.2.0"
 log = "0.3.6"
 lru-disk-cache = { path = "lru-disk-cache", version = "0.1.0" }
 native-tls = "0.1"
+num_cpus = "1.0"
 number_prefix = "0.2.5"
 openssl = { version = "0.9", optional = true }
 redis = { version = "0.8.0", optional = true }

src/cache/disk.rs

@@ -18,7 +18,6 @@ use cache::{
     CacheWrite,
     Storage,
 };
-use futures::Future;
 use futures_cpupool::CpuPool;
 use lru_disk_cache::LruDiskCache;
 use lru_disk_cache::Error as LruError;
@@ -62,7 +61,7 @@ impl Storage for DiskCache {
         let path = make_key_path(key);
         let lru = self.lru.clone();
         let key = key.to_owned();
-        self.pool.spawn_fn(move || {
+        Box::new(self.pool.spawn_fn(move || {
             let mut lru = lru.lock().unwrap();
             let f = match lru.get(&path) {
                 Ok(f) => f,
@@ -78,7 +77,7 @@ impl Storage for DiskCache {
             };
             let hit = CacheRead::from(f)?;
             Ok(Cache::Hit(hit))
-        }).boxed()
+        }))
     }
 
     fn put(&self, key: &str, entry: CacheWrite) -> SFuture<Duration> {
@@ -87,12 +86,12 @@ impl Storage for DiskCache {
         trace!("DiskCache::finish_put({})", key);
         let lru = self.lru.clone();
         let key = make_key_path(key);
-        self.pool.spawn_fn(move || {
+        Box::new(self.pool.spawn_fn(move || {
             let start = Instant::now();
             let v = entry.finish()?;
             lru.lock().unwrap().insert_bytes(key, &v)?;
             Ok(start.elapsed())
-        }).boxed()
+        }))
     }
 
     fn location(&self) -> String {

src/commands.rs

@@ -18,6 +18,7 @@ use client::{
     ServerConnection,
 };
 use cmdline::{Command, StatsFormat};
+use jobserver::Client;
 use log::LogLevel::Trace;
 use mock_command::{
     CommandCreatorSync,
@@ -601,9 +602,10 @@ pub fn run_command(cmd: Command) -> Result<i32> {
         }
         Command::Compile { exe, cmdline, cwd, env_vars } => {
             trace!("Command::Compile {{ {:?}, {:?}, {:?} }}", exe, cmdline, cwd);
+            let jobserver = unsafe { Client::new() };
             let conn = connect_or_start_server(get_port())?;
             let mut core = Core::new()?;
-            let res = do_compile(ProcessCommandCreator::new(&core.handle()),
+            let res = do_compile(ProcessCommandCreator::new(&core.handle(), &jobserver),
                                  &mut core,
                                  conn,
                                  exe.as_ref(),

src/compiler/compiler.rs

@@ -470,14 +470,13 @@ fn detect_compiler<T>(creator: &T, executable: &Path, pool: &CpuPool)
     };
     let is_rustc = if filename.to_string_lossy().to_lowercase() == "rustc" {
         // Sanity check that it's really rustc.
+        let executable = executable.to_path_buf();
         let child = creator.clone().new_command_sync(&executable)
             .stdout(Stdio::piped())
             .stderr(Stdio::null())
             .args(&["--version"])
-            .spawn().chain_err(|| {
-                format!("failed to execute {:?}", executable)
-            });
-        let output = child.into_future().and_then(move |child| {
+            .spawn();
+        let output = child.and_then(move |child| {
             child.wait_with_output()
                 .chain_err(|| "failed to read child output")
         });
@@ -530,10 +529,7 @@ gcc
     let output = write.and_then(move |(tempdir, src)| {
         cmd.arg("-E").arg(src);
         trace!("compiler {:?}", cmd);
-        let child = cmd.spawn().chain_err(|| {
-            format!("failed to execute {:?}", cmd)
-        });
-        child.into_future().and_then(|child| {
+        cmd.spawn().and_then(|child| {
             child.wait_with_output().chain_err(|| "failed to read child output")
         }).map(|e| {
             drop(tempdir);
@@ -724,11 +720,9 @@ mod test {
         let o = obj.clone();
         next_command_calls(&creator, move |_| {
             // Pretend to compile something.
-            match File::create(&o)
-                .and_then(|mut f| f.write_all(b"file contents")) {
-                    Ok(_) => Ok(MockChild::new(exit_status(0), COMPILER_STDOUT, COMPILER_STDERR)),
-                    Err(e) => Err(e),
-                }
+            let mut f = File::create(&o)?;
+            f.write_all(b"file contents")?;
+            Ok(MockChild::new(exit_status(0), COMPILER_STDOUT, COMPILER_STDERR))
         });
         let cwd = f.tempdir.path();
         let arguments = ovec!["-c", "foo.c", "-o", "foo.o"];
@@ -805,11 +799,9 @@ mod test {
         let o = obj.clone();
         next_command_calls(&creator, move |_| {
             // Pretend to compile something.
-            match File::create(&o)
-                .and_then(|mut f| f.write_all(b"file contents")) {
-                    Ok(_) => Ok(MockChild::new(exit_status(0), COMPILER_STDOUT, COMPILER_STDERR)),
-                    Err(e) => Err(e),
-                }
+            let mut f = File::create(&o)?;
+            f.write_all(b"file contents")?;
+            Ok(MockChild::new(exit_status(0), COMPILER_STDOUT, COMPILER_STDERR))
         });
         let cwd = f.tempdir.path();
         let arguments = ovec!["-c", "foo.c", "-o", "foo.o"];
@@ -887,11 +879,9 @@ mod test {
         let o = obj.clone();
         next_command_calls(&creator, move |_| {
             // Pretend to compile something.
-            match File::create(&o)
-                .and_then(|mut f| f.write_all(b"file contents")) {
-                    Ok(_) => Ok(MockChild::new(exit_status(0), COMPILER_STDOUT, COMPILER_STDERR)),
-                    Err(e) => Err(e),
-                }
+            let mut f = File::create(&o)?;
+            f.write_all(b"file contents")?;
+            Ok(MockChild::new(exit_status(0), COMPILER_STDOUT, COMPILER_STDERR))
         });
         let cwd = f.tempdir.path();
         let arguments = ovec!["-c", "foo.c", "-o", "foo.o"];
@@ -954,11 +944,9 @@ mod test {
             let o = obj.clone();
             next_command_calls(&creator, move |_| {
                 // Pretend to compile something.
-                match File::create(&o)
-                    .and_then(|mut f| f.write_all(b"file contents")) {
-                        Ok(_) => Ok(MockChild::new(exit_status(0), COMPILER_STDOUT, COMPILER_STDERR)),
-                        Err(e) => Err(e),
-                    }
+                let mut f = File::create(&o)?;
+                f.write_all(b"file contents")?;
+                Ok(MockChild::new(exit_status(0), COMPILER_STDOUT, COMPILER_STDERR))
             });
         }
         let cwd = f.tempdir.path();

src/jobserver.rs

@@ -0,0 +1,71 @@
+extern crate jobserver;
+
+use std::io;
+use std::process::Command;
+use std::sync::Arc;
+
+use futures::prelude::*;
+use futures::sync::mpsc;
+use futures::sync::oneshot;
+use num_cpus;
+
+use errors::*;
+
+pub use self::jobserver::Acquired;
+
+#[derive(Clone)]
+pub struct Client {
+    helper: Arc<jobserver::HelperThread>,
+    inner: jobserver::Client,
+    tx: mpsc::UnboundedSender<oneshot::Sender<io::Result<Acquired>>>
+}
+
+impl Client {
+    // unsafe because `from_env` is unsafe (can use the wrong fds)
+    pub unsafe fn new() -> Client {
+        match jobserver::Client::from_env() {
+            Some(c) => Client::_new(c),
+            None => Client::new_num(num_cpus::get()),
+        }
+    }
+
+    pub fn new_num(num: usize) -> Client {
+        let inner = jobserver::Client::new(num)
+                .expect("failed to create jobserver");
+        Client::_new(inner)
+    }
+
+    fn _new(inner: jobserver::Client) -> Client {
+        let (tx, rx) = mpsc::unbounded::<oneshot::Sender<_>>();
+        let mut rx = rx.wait();
+        let helper = inner.clone().into_helper_thread(move |token| {
+            if let Some(Ok(sender)) = rx.next() {
+                drop(sender.send(token));
+            }
+        }).expect("failed to spawn helper thread");
+
+        Client {
+            inner: inner,
+            helper: Arc::new(helper),
+            tx: tx,
+        }
+    }
+
+    /// Configures this jobserver to be inherited by the specified command
+    pub fn configure(&self, cmd: &mut Command) {
+        self.inner.configure(cmd)
+    }
+
+    /// Returns a future that represents an acquired jobserver token.
+    ///
+    /// This should be invoked before any "work" is spawend (for whatever the
+    /// defnition of "work" is) to ensure that the system is properly
+    /// rate-limiting itself.
+    pub fn acquire(&self) -> SFuture<Acquired> {
+        let (tx, rx) = oneshot::channel();
+        self.helper.request_token();
+        self.tx.unbounded_send(tx).unwrap();
+        Box::new(rx.chain_err(|| "jobserver helper panicked")
+                   .and_then(|t| t.chain_err(|| "failed to acquire jobserver token")))
+    }
+}

src/main.rs

@@ -52,6 +52,7 @@ extern crate libc;
 #[cfg(windows)]
 extern crate mio_named_pipes;
 extern crate native_tls;
+extern crate num_cpus;
 extern crate number_prefix;
 #[cfg(feature = "openssl")]
 extern crate openssl;
@@ -93,6 +94,7 @@ mod client;
 mod cmdline;
 mod commands;
 mod compiler;
+mod jobserver;
 mod mock_command;
 mod protocol;
 mod server;