From ef24de24a82fe681581cc130f342363c47c0969a Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Fri, 7 Jun 2024 07:48:48 +0200 Subject: Merging upstream version 1.75.0+dfsg1. Signed-off-by: Daniel Baumann --- vendor/jobserver/src/error.rs | 84 +++ vendor/jobserver/src/lib.rs | 1140 ++++++++++++++++++++------------------- vendor/jobserver/src/unix.rs | 908 ++++++++++++++++--------------- vendor/jobserver/src/wasm.rs | 191 +++---- vendor/jobserver/src/windows.rs | 536 +++++++++--------- 5 files changed, 1523 insertions(+), 1336 deletions(-) create mode 100644 vendor/jobserver/src/error.rs (limited to 'vendor/jobserver/src') diff --git a/vendor/jobserver/src/error.rs b/vendor/jobserver/src/error.rs new file mode 100644 index 000000000..9f95f18b8 --- /dev/null +++ b/vendor/jobserver/src/error.rs @@ -0,0 +1,84 @@ +#[cfg(unix)] +type RawFd = std::os::fd::RawFd; +#[cfg(not(unix))] +type RawFd = std::convert::Infallible; + +/// Error type for `from_env_ext` function. +#[derive(Debug)] +pub struct FromEnvError { + pub(crate) inner: FromEnvErrorInner, +} + +/// Kind of an error returned from `from_env_ext` function. +#[derive(Debug)] +#[non_exhaustive] +pub enum FromEnvErrorKind { + /// There is no environment variable that describes jobserver to inherit. + NoEnvVar, + /// There is no jobserver in the environment variable. + /// Variables associated with Make can be used for passing data other than jobserver info. + NoJobserver, + /// Cannot parse jobserver environment variable value, incorrect format. + CannotParse, + /// Cannot open path or name from the jobserver environment variable value. + CannotOpenPath, + /// Cannot open file descriptor from the jobserver environment variable value. + CannotOpenFd, + /// File descriptor from the jobserver environment variable value is not a pipe. + NotAPipe, + /// Jobserver inheritance is not supported on this platform. + Unsupported, +} + +impl FromEnvError { + /// Get the error kind. + pub fn kind(&self) -> FromEnvErrorKind { + match self.inner { + FromEnvErrorInner::NoEnvVar => FromEnvErrorKind::NoEnvVar, + FromEnvErrorInner::NoJobserver => FromEnvErrorKind::NoJobserver, + FromEnvErrorInner::CannotParse(_) => FromEnvErrorKind::CannotParse, + FromEnvErrorInner::CannotOpenPath(..) => FromEnvErrorKind::CannotOpenPath, + FromEnvErrorInner::CannotOpenFd(..) => FromEnvErrorKind::CannotOpenFd, + FromEnvErrorInner::NotAPipe(..) => FromEnvErrorKind::NotAPipe, + FromEnvErrorInner::Unsupported => FromEnvErrorKind::Unsupported, + } + } +} + +impl std::fmt::Display for FromEnvError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match &self.inner { + FromEnvErrorInner::NoEnvVar => write!(f, "there is no environment variable that describes jobserver to inherit"), + FromEnvErrorInner::NoJobserver => write!(f, "there is no `--jobserver-fds=` or `--jobserver-auth=` in the environment variable"), + FromEnvErrorInner::CannotParse(s) => write!(f, "cannot parse jobserver environment variable value: {s}"), + FromEnvErrorInner::CannotOpenPath(s, err) => write!(f, "cannot open path or name {s} from the jobserver environment variable value: {err}"), + FromEnvErrorInner::CannotOpenFd(fd, err) => write!(f, "cannot open file descriptor {fd} from the jobserver environment variable value: {err}"), + FromEnvErrorInner::NotAPipe(fd, None) => write!(f, "file descriptor {fd} from the jobserver environment variable value is not a pipe"), + FromEnvErrorInner::NotAPipe(fd, Some(err)) => write!(f, "file descriptor {fd} from the jobserver environment variable value is not a pipe: {err}"), + FromEnvErrorInner::Unsupported => write!(f, "jobserver inheritance is not supported on this platform"), + } + } +} +impl std::error::Error for FromEnvError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match &self.inner { + FromEnvErrorInner::CannotOpenPath(_, err) => Some(err), + FromEnvErrorInner::NotAPipe(_, Some(err)) | FromEnvErrorInner::CannotOpenFd(_, err) => { + Some(err) + } + _ => None, + } + } +} + +#[allow(dead_code)] +#[derive(Debug)] +pub(crate) enum FromEnvErrorInner { + NoEnvVar, + NoJobserver, + CannotParse(String), + CannotOpenPath(String, std::io::Error), + CannotOpenFd(RawFd, std::io::Error), + NotAPipe(RawFd, Option), + Unsupported, +} diff --git a/vendor/jobserver/src/lib.rs b/vendor/jobserver/src/lib.rs index cd0cdd749..bfa7980c4 100644 --- a/vendor/jobserver/src/lib.rs +++ b/vendor/jobserver/src/lib.rs @@ -1,544 +1,596 @@ -//! An implementation of the GNU make jobserver. -//! -//! This crate is an implementation, in Rust, of the GNU `make` jobserver for -//! CLI tools that are interoperating with make or otherwise require some form -//! of parallelism limiting across process boundaries. This was originally -//! written for usage in Cargo to both (a) work when `cargo` is invoked from -//! `make` (using `make`'s jobserver) and (b) work when `cargo` invokes build -//! scripts, exporting a jobserver implementation for `make` processes to -//! transitively use. -//! -//! The jobserver implementation can be found in [detail online][docs] but -//! basically boils down to a cross-process semaphore. On Unix this is -//! implemented with the `pipe` syscall and read/write ends of a pipe and on -//! Windows this is implemented literally with IPC semaphores. Starting from -//! GNU `make` version 4.4, named pipe becomes the default way in communication -//! on Unix. This crate also supports that feature in the sense of inheriting -//! and forwarding the correct environment. -//! -//! The jobserver protocol in `make` also dictates when tokens are acquired to -//! run child work, and clients using this crate should take care to implement -//! such details to ensure correct interoperation with `make` itself. -//! -//! ## Examples -//! -//! Connect to a jobserver that was set up by `make` or a different process: -//! -//! ```no_run -//! use jobserver::Client; -//! -//! // See API documentation for why this is `unsafe` -//! let client = match unsafe { Client::from_env() } { -//! Some(client) => client, -//! None => panic!("client not configured"), -//! }; -//! ``` -//! -//! Acquire and release token from a jobserver: -//! -//! ```no_run -//! use jobserver::Client; -//! -//! let client = unsafe { Client::from_env().unwrap() }; -//! let token = client.acquire().unwrap(); // blocks until it is available -//! drop(token); // releases the token when the work is done -//! ``` -//! -//! Create a new jobserver and configure a child process to have access: -//! -//! ``` -//! use std::process::Command; -//! use jobserver::Client; -//! -//! let client = Client::new(4).expect("failed to create jobserver"); -//! let mut cmd = Command::new("make"); -//! client.configure(&mut cmd); -//! ``` -//! -//! ## Caveats -//! -//! This crate makes no attempt to release tokens back to a jobserver on -//! abnormal exit of a process. If a process which acquires a token is killed -//! with ctrl-c or some similar signal then tokens will not be released and the -//! jobserver may be in a corrupt state. -//! -//! Note that this is typically ok as ctrl-c means that an entire build process -//! is being torn down, but it's worth being aware of at least! -//! -//! ## Windows caveats -//! -//! There appear to be two implementations of `make` on Windows. On MSYS2 one -//! typically comes as `mingw32-make` and the other as `make` itself. I'm not -//! personally too familiar with what's going on here, but for jobserver-related -//! information the `mingw32-make` implementation uses Windows semaphores -//! whereas the `make` program does not. The `make` program appears to use file -//! descriptors and I'm not really sure how it works, so this crate is not -//! compatible with `make` on Windows. It is, however, compatible with -//! `mingw32-make`. -//! -//! [docs]: http://make.mad-scientist.net/papers/jobserver-implementation/ - -#![deny(missing_docs, missing_debug_implementations)] -#![doc(html_root_url = "https://docs.rs/jobserver/0.1")] - -use std::env; -use std::io; -use std::process::Command; -use std::sync::{Arc, Condvar, Mutex, MutexGuard}; - -#[cfg(unix)] -#[path = "unix.rs"] -mod imp; -#[cfg(windows)] -#[path = "windows.rs"] -mod imp; -#[cfg(not(any(unix, windows)))] -#[path = "wasm.rs"] -mod imp; - -/// A client of a jobserver -/// -/// This structure is the main type exposed by this library, and is where -/// interaction to a jobserver is configured through. Clients are either created -/// from scratch in which case the internal semphore is initialied on the spot, -/// or a client is created from the environment to connect to a jobserver -/// already created. -/// -/// Some usage examples can be found in the crate documentation for using a -/// client. -/// -/// Note that a `Client` implements the `Clone` trait, and all instances of a -/// `Client` refer to the same jobserver instance. -#[derive(Clone, Debug)] -pub struct Client { - inner: Arc, -} - -/// An acquired token from a jobserver. -/// -/// This token will be released back to the jobserver when it is dropped and -/// otherwise represents the ability to spawn off another thread of work. -#[derive(Debug)] -pub struct Acquired { - client: Arc, - data: imp::Acquired, - disabled: bool, -} - -impl Acquired { - /// This drops the `Acquired` token without releasing the associated token. - /// - /// This is not generally useful, but can be helpful if you do not have the - /// ability to store an Acquired token but need to not yet release it. - /// - /// You'll typically want to follow this up with a call to `release_raw` or - /// similar to actually release the token later on. - pub fn drop_without_releasing(mut self) { - self.disabled = true; - } -} - -#[derive(Default, Debug)] -struct HelperState { - lock: Mutex, - cvar: Condvar, -} - -#[derive(Default, Debug)] -struct HelperInner { - requests: usize, - producer_done: bool, - consumer_done: bool, -} - -impl Client { - /// Creates a new jobserver initialized with the given parallelism limit. - /// - /// A client to the jobserver created will be returned. This client will - /// allow at most `limit` tokens to be acquired from it in parallel. More - /// calls to `acquire` will cause the calling thread to block. - /// - /// Note that the created `Client` is not automatically inherited into - /// spawned child processes from this program. Manual usage of the - /// `configure` function is required for a child process to have access to a - /// job server. - /// - /// # Examples - /// - /// ``` - /// use jobserver::Client; - /// - /// let client = Client::new(4).expect("failed to create jobserver"); - /// ``` - /// - /// # Errors - /// - /// Returns an error if any I/O error happens when attempting to create the - /// jobserver client. - pub fn new(limit: usize) -> io::Result { - Ok(Client { - inner: Arc::new(imp::Client::new(limit)?), - }) - } - - /// Attempts to connect to the jobserver specified in this process's - /// environment. - /// - /// When the a `make` executable calls a child process it will configure the - /// environment of the child to ensure that it has handles to the jobserver - /// it's passing down. This function will attempt to look for these details - /// and connect to the jobserver. - /// - /// Note that the created `Client` is not automatically inherited into - /// spawned child processes from this program. Manual usage of the - /// `configure` function is required for a child process to have access to a - /// job server. - /// - /// # Return value - /// - /// If a jobserver was found in the environment and it looks correct then - /// `Some` of the connected client will be returned. If no jobserver was - /// found then `None` will be returned. - /// - /// Note that on Unix the `Client` returned **takes ownership of the file - /// descriptors specified in the environment**. Jobservers on Unix are - /// implemented with `pipe` file descriptors, and they're inherited from - /// parent processes. This `Client` returned takes ownership of the file - /// descriptors for this process and will close the file descriptors after - /// this value is dropped. - /// - /// Additionally on Unix this function will configure the file descriptors - /// with `CLOEXEC` so they're not automatically inherited by spawned - /// children. - /// - /// # Safety - /// - /// This function is `unsafe` to call on Unix specifically as it - /// transitively requires usage of the `from_raw_fd` function, which is - /// itself unsafe in some circumstances. - /// - /// It's recommended to call this function very early in the lifetime of a - /// program before any other file descriptors are opened. That way you can - /// make sure to take ownership properly of the file descriptors passed - /// down, if any. - /// - /// It's generally unsafe to call this function twice in a program if the - /// previous invocation returned `Some`. - /// - /// Note, though, that on Windows it should be safe to call this function - /// any number of times. - pub unsafe fn from_env() -> Option { - let var = match env::var("CARGO_MAKEFLAGS") - .or_else(|_| env::var("MAKEFLAGS")) - .or_else(|_| env::var("MFLAGS")) - { - Ok(s) => s, - Err(_) => return None, - }; - let mut arg = "--jobserver-fds="; - let pos = match var.find(arg) { - Some(i) => i, - None => { - arg = "--jobserver-auth="; - match var.find(arg) { - Some(i) => i, - None => return None, - } - } - }; - - let s = var[pos + arg.len()..].split(' ').next().unwrap(); - imp::Client::open(s).map(|c| Client { inner: Arc::new(c) }) - } - - /// Acquires a token from this jobserver client. - /// - /// This function will block the calling thread until a new token can be - /// acquired from the jobserver. - /// - /// # Return value - /// - /// On successful acquisition of a token an instance of `Acquired` is - /// returned. This structure, when dropped, will release the token back to - /// the jobserver. It's recommended to avoid leaking this value. - /// - /// # Errors - /// - /// If an I/O error happens while acquiring a token then this function will - /// return immediately with the error. If an error is returned then a token - /// was not acquired. - pub fn acquire(&self) -> io::Result { - let data = self.inner.acquire()?; - Ok(Acquired { - client: self.inner.clone(), - data, - disabled: false, - }) - } - - /// Returns amount of tokens in the read-side pipe. - /// - /// # Return value - /// - /// Number of bytes available to be read from the jobserver pipe - /// - /// # Errors - /// - /// Underlying errors from the ioctl will be passed up. - pub fn available(&self) -> io::Result { - self.inner.available() - } - - /// Configures a child process to have access to this client's jobserver as - /// well. - /// - /// This function is required to be called to ensure that a jobserver is - /// properly inherited to a child process. If this function is *not* called - /// then this `Client` will not be accessible in the child process. In other - /// words, if not called, then `Client::from_env` will return `None` in the - /// child process (or the equivalent of `Child::from_env` that `make` uses). - /// - /// ## Platform-specific behavior - /// - /// On Unix and Windows this will clobber the `CARGO_MAKEFLAGS` environment - /// variables for the child process, and on Unix this will also allow the - /// two file descriptors for this client to be inherited to the child. - /// - /// On platforms other than Unix and Windows this panics. - pub fn configure(&self, cmd: &mut Command) { - cmd.env("CARGO_MAKEFLAGS", &self.mflags_env()); - self.inner.configure(cmd); - } - - /// Configures a child process to have access to this client's jobserver as - /// well. - /// - /// This function is required to be called to ensure that a jobserver is - /// properly inherited to a child process. If this function is *not* called - /// then this `Client` will not be accessible in the child process. In other - /// words, if not called, then `Client::from_env` will return `None` in the - /// child process (or the equivalent of `Child::from_env` that `make` uses). - /// - /// ## Platform-specific behavior - /// - /// On Unix and Windows this will clobber the `CARGO_MAKEFLAGS`, - /// `MAKEFLAGS` and `MFLAGS` environment variables for the child process, - /// and on Unix this will also allow the two file descriptors for - /// this client to be inherited to the child. - /// - /// On platforms other than Unix and Windows this panics. - pub fn configure_make(&self, cmd: &mut Command) { - let value = self.mflags_env(); - cmd.env("CARGO_MAKEFLAGS", &value); - cmd.env("MAKEFLAGS", &value); - cmd.env("MFLAGS", &value); - self.inner.configure(cmd); - } - - fn mflags_env(&self) -> String { - let arg = self.inner.string_arg(); - // Older implementations of make use `--jobserver-fds` and newer - // implementations use `--jobserver-auth`, pass both to try to catch - // both implementations. - format!("-j --jobserver-fds={0} --jobserver-auth={0}", arg) - } - - /// Converts this `Client` into a helper thread to deal with a blocking - /// `acquire` function a little more easily. - /// - /// The fact that the `acquire` function on `Client` blocks isn't always - /// the easiest to work with. Typically you're using a jobserver to - /// manage running other events in parallel! This means that you need to - /// either (a) wait for an existing job to finish or (b) wait for a - /// new token to become available. - /// - /// Unfortunately the blocking in `acquire` happens at the implementation - /// layer of jobservers. On Unix this requires a blocking call to `read` - /// and on Windows this requires one of the `WaitFor*` functions. Both - /// of these situations aren't the easiest to deal with: - /// - /// * On Unix there's basically only one way to wake up a `read` early, and - /// that's through a signal. This is what the `make` implementation - /// itself uses, relying on `SIGCHLD` to wake up a blocking acquisition - /// of a new job token. Unfortunately nonblocking I/O is not an option - /// here, so it means that "waiting for one of two events" means that - /// the latter event must generate a signal! This is not always the case - /// on unix for all jobservers. - /// - /// * On Windows you'd have to basically use the `WaitForMultipleObjects` - /// which means that you've got to canonicalize all your event sources - /// into a `HANDLE` which also isn't the easiest thing to do - /// unfortunately. - /// - /// This function essentially attempts to ease these limitations by - /// converting this `Client` into a helper thread spawned into this - /// process. The application can then request that the helper thread - /// acquires tokens and the provided closure will be invoked for each token - /// acquired. - /// - /// The intention is that this function can be used to translate the event - /// of a token acquisition into an arbitrary user-defined event. - /// - /// # Arguments - /// - /// This function will consume the `Client` provided to be transferred to - /// the helper thread that is spawned. Additionally a closure `f` is - /// provided to be invoked whenever a token is acquired. - /// - /// This closure is only invoked after calls to - /// `HelperThread::request_token` have been made and a token itself has - /// been acquired. If an error happens while acquiring the token then - /// an error will be yielded to the closure as well. - /// - /// # Return Value - /// - /// This function will return an instance of the `HelperThread` structure - /// which is used to manage the helper thread associated with this client. - /// Through the `HelperThread` you'll request that tokens are acquired. - /// When acquired, the closure provided here is invoked. - /// - /// When the `HelperThread` structure is returned it will be gracefully - /// torn down, and the calling thread will be blocked until the thread is - /// torn down (which should be prompt). - /// - /// # Errors - /// - /// This function may fail due to creation of the helper thread or - /// auxiliary I/O objects to manage the helper thread. In any of these - /// situations the error is propagated upwards. - /// - /// # Platform-specific behavior - /// - /// On Windows this function behaves pretty normally as expected, but on - /// Unix the implementation is... a little heinous. As mentioned above - /// we're forced into blocking I/O for token acquisition, namely a blocking - /// call to `read`. We must be able to unblock this, however, to tear down - /// the helper thread gracefully! - /// - /// Essentially what happens is that we'll send a signal to the helper - /// thread spawned and rely on `EINTR` being returned to wake up the helper - /// thread. This involves installing a global `SIGUSR1` handler that does - /// nothing along with sending signals to that thread. This may cause - /// odd behavior in some applications, so it's recommended to review and - /// test thoroughly before using this. - pub fn into_helper_thread(self, f: F) -> io::Result - where - F: FnMut(io::Result) + Send + 'static, - { - let state = Arc::new(HelperState::default()); - Ok(HelperThread { - inner: Some(imp::spawn_helper(self, state.clone(), Box::new(f))?), - state, - }) - } - - /// Blocks the current thread until a token is acquired. - /// - /// This is the same as `acquire`, except that it doesn't return an RAII - /// helper. If successful the process will need to guarantee that - /// `release_raw` is called in the future. - pub fn acquire_raw(&self) -> io::Result<()> { - self.inner.acquire()?; - Ok(()) - } - - /// Releases a jobserver token back to the original jobserver. - /// - /// This is intended to be paired with `acquire_raw` if it was called, but - /// in some situations it could also be called to relinquish a process's - /// implicit token temporarily which is then re-acquired later. - pub fn release_raw(&self) -> io::Result<()> { - self.inner.release(None)?; - Ok(()) - } -} - -impl Drop for Acquired { - fn drop(&mut self) { - if !self.disabled { - drop(self.client.release(Some(&self.data))); - } - } -} - -/// Structure returned from `Client::into_helper_thread` to manage the lifetime -/// of the helper thread returned, see those associated docs for more info. -#[derive(Debug)] -pub struct HelperThread { - inner: Option, - state: Arc, -} - -impl HelperThread { - /// Request that the helper thread acquires a token, eventually calling the - /// original closure with a token when it's available. - /// - /// For more information, see the docs on that function. - pub fn request_token(&self) { - // Indicate that there's one more request for a token and then wake up - // the helper thread if it's sleeping. - self.state.lock().requests += 1; - self.state.cvar.notify_one(); - } -} - -impl Drop for HelperThread { - fn drop(&mut self) { - // Flag that the producer half is done so the helper thread should exit - // quickly if it's waiting. Wake it up if it's actually waiting - self.state.lock().producer_done = true; - self.state.cvar.notify_one(); - - // ... and afterwards perform any thread cleanup logic - self.inner.take().unwrap().join(); - } -} - -impl HelperState { - fn lock(&self) -> MutexGuard<'_, HelperInner> { - self.lock.lock().unwrap_or_else(|e| e.into_inner()) - } - - /// Executes `f` for each request for a token, where `f` is expected to - /// block and then provide the original closure with a token once it's - /// acquired. - /// - /// This is an infinite loop until the helper thread is dropped, at which - /// point everything should get interrupted. - fn for_each_request(&self, mut f: impl FnMut(&HelperState)) { - let mut lock = self.lock(); - - // We only execute while we could receive requests, but as soon as - // that's `false` we're out of here. - while !lock.producer_done { - // If no one's requested a token then we wait for someone to - // request a token. - if lock.requests == 0 { - lock = self.cvar.wait(lock).unwrap_or_else(|e| e.into_inner()); - continue; - } - - // Consume the request for a token, and then actually acquire a - // token after unlocking our lock (not that acquisition happens in - // `f`). This ensures that we don't actually hold the lock if we - // wait for a long time for a token. - lock.requests -= 1; - drop(lock); - f(self); - lock = self.lock(); - } - lock.consumer_done = true; - self.cvar.notify_one(); - } - - fn producer_done(&self) -> bool { - self.lock().producer_done - } -} - -#[test] -fn no_helper_deadlock() { - let x = crate::Client::new(32).unwrap(); - let _y = x.clone(); - std::mem::drop(x.into_helper_thread(|_| {}).unwrap()); -} +//! An implementation of the GNU make jobserver. +//! +//! This crate is an implementation, in Rust, of the GNU `make` jobserver for +//! CLI tools that are interoperating with make or otherwise require some form +//! of parallelism limiting across process boundaries. This was originally +//! written for usage in Cargo to both (a) work when `cargo` is invoked from +//! `make` (using `make`'s jobserver) and (b) work when `cargo` invokes build +//! scripts, exporting a jobserver implementation for `make` processes to +//! transitively use. +//! +//! The jobserver implementation can be found in [detail online][docs] but +//! basically boils down to a cross-process semaphore. On Unix this is +//! implemented with the `pipe` syscall and read/write ends of a pipe and on +//! Windows this is implemented literally with IPC semaphores. Starting from +//! GNU `make` version 4.4, named pipe becomes the default way in communication +//! on Unix. This crate also supports that feature in the sense of inheriting +//! and forwarding the correct environment. +//! +//! The jobserver protocol in `make` also dictates when tokens are acquired to +//! run child work, and clients using this crate should take care to implement +//! such details to ensure correct interoperation with `make` itself. +//! +//! ## Examples +//! +//! Connect to a jobserver that was set up by `make` or a different process: +//! +//! ```no_run +//! use jobserver::Client; +//! +//! // See API documentation for why this is `unsafe` +//! let client = match unsafe { Client::from_env() } { +//! Some(client) => client, +//! None => panic!("client not configured"), +//! }; +//! ``` +//! +//! Acquire and release token from a jobserver: +//! +//! ```no_run +//! use jobserver::Client; +//! +//! let client = unsafe { Client::from_env().unwrap() }; +//! let token = client.acquire().unwrap(); // blocks until it is available +//! drop(token); // releases the token when the work is done +//! ``` +//! +//! Create a new jobserver and configure a child process to have access: +//! +//! ``` +//! use std::process::Command; +//! use jobserver::Client; +//! +//! let client = Client::new(4).expect("failed to create jobserver"); +//! let mut cmd = Command::new("make"); +//! client.configure(&mut cmd); +//! ``` +//! +//! ## Caveats +//! +//! This crate makes no attempt to release tokens back to a jobserver on +//! abnormal exit of a process. If a process which acquires a token is killed +//! with ctrl-c or some similar signal then tokens will not be released and the +//! jobserver may be in a corrupt state. +//! +//! Note that this is typically ok as ctrl-c means that an entire build process +//! is being torn down, but it's worth being aware of at least! +//! +//! ## Windows caveats +//! +//! There appear to be two implementations of `make` on Windows. On MSYS2 one +//! typically comes as `mingw32-make` and the other as `make` itself. I'm not +//! personally too familiar with what's going on here, but for jobserver-related +//! information the `mingw32-make` implementation uses Windows semaphores +//! whereas the `make` program does not. The `make` program appears to use file +//! descriptors and I'm not really sure how it works, so this crate is not +//! compatible with `make` on Windows. It is, however, compatible with +//! `mingw32-make`. +//! +//! [docs]: http://make.mad-scientist.net/papers/jobserver-implementation/ + +#![deny(missing_docs, missing_debug_implementations)] +#![doc(html_root_url = "https://docs.rs/jobserver/0.1")] + +use std::env; +use std::ffi::OsString; +use std::io; +use std::process::Command; +use std::sync::{Arc, Condvar, Mutex, MutexGuard}; + +mod error; +#[cfg(unix)] +#[path = "unix.rs"] +mod imp; +#[cfg(windows)] +#[path = "windows.rs"] +mod imp; +#[cfg(not(any(unix, windows)))] +#[path = "wasm.rs"] +mod imp; + +/// A client of a jobserver +/// +/// This structure is the main type exposed by this library, and is where +/// interaction to a jobserver is configured through. Clients are either created +/// from scratch in which case the internal semphore is initialied on the spot, +/// or a client is created from the environment to connect to a jobserver +/// already created. +/// +/// Some usage examples can be found in the crate documentation for using a +/// client. +/// +/// Note that a `Client` implements the `Clone` trait, and all instances of a +/// `Client` refer to the same jobserver instance. +#[derive(Clone, Debug)] +pub struct Client { + inner: Arc, +} + +/// An acquired token from a jobserver. +/// +/// This token will be released back to the jobserver when it is dropped and +/// otherwise represents the ability to spawn off another thread of work. +#[derive(Debug)] +pub struct Acquired { + client: Arc, + data: imp::Acquired, + disabled: bool, +} + +impl Acquired { + /// This drops the `Acquired` token without releasing the associated token. + /// + /// This is not generally useful, but can be helpful if you do not have the + /// ability to store an Acquired token but need to not yet release it. + /// + /// You'll typically want to follow this up with a call to `release_raw` or + /// similar to actually release the token later on. + pub fn drop_without_releasing(mut self) { + self.disabled = true; + } +} + +#[derive(Default, Debug)] +struct HelperState { + lock: Mutex, + cvar: Condvar, +} + +#[derive(Default, Debug)] +struct HelperInner { + requests: usize, + producer_done: bool, + consumer_done: bool, +} + +use error::FromEnvErrorInner; +pub use error::{FromEnvError, FromEnvErrorKind}; + +/// Return type for `from_env_ext` function. +#[derive(Debug)] +pub struct FromEnv { + /// Result of trying to get jobserver client from env. + pub client: Result, + /// Name and value of the environment variable. + /// `None` if no relevant environment variable is found. + pub var: Option<(&'static str, OsString)>, +} + +impl FromEnv { + fn new_ok(client: Client, var_name: &'static str, var_value: OsString) -> FromEnv { + FromEnv { + client: Ok(client), + var: Some((var_name, var_value)), + } + } + fn new_err(kind: FromEnvErrorInner, var_name: &'static str, var_value: OsString) -> FromEnv { + FromEnv { + client: Err(FromEnvError { inner: kind }), + var: Some((var_name, var_value)), + } + } +} + +impl Client { + /// Creates a new jobserver initialized with the given parallelism limit. + /// + /// A client to the jobserver created will be returned. This client will + /// allow at most `limit` tokens to be acquired from it in parallel. More + /// calls to `acquire` will cause the calling thread to block. + /// + /// Note that the created `Client` is not automatically inherited into + /// spawned child processes from this program. Manual usage of the + /// `configure` function is required for a child process to have access to a + /// job server. + /// + /// # Examples + /// + /// ``` + /// use jobserver::Client; + /// + /// let client = Client::new(4).expect("failed to create jobserver"); + /// ``` + /// + /// # Errors + /// + /// Returns an error if any I/O error happens when attempting to create the + /// jobserver client. + pub fn new(limit: usize) -> io::Result { + Ok(Client { + inner: Arc::new(imp::Client::new(limit)?), + }) + } + + /// Attempts to connect to the jobserver specified in this process's + /// environment. + /// + /// When the a `make` executable calls a child process it will configure the + /// environment of the child to ensure that it has handles to the jobserver + /// it's passing down. This function will attempt to look for these details + /// and connect to the jobserver. + /// + /// Note that the created `Client` is not automatically inherited into + /// spawned child processes from this program. Manual usage of the + /// `configure` function is required for a child process to have access to a + /// job server. + /// + /// # Return value + /// + /// `FromEnv` contains result and relevant environment variable. + /// If a jobserver was found in the environment and it looks correct then + /// result with the connected client will be returned. In other cases + /// result will contain `Err(FromEnvErr)`. + /// + /// Note that on Unix the `Client` returned **takes ownership of the file + /// descriptors specified in the environment**. Jobservers on Unix are + /// implemented with `pipe` file descriptors, and they're inherited from + /// parent processes. This `Client` returned takes ownership of the file + /// descriptors for this process and will close the file descriptors after + /// this value is dropped. + /// + /// Additionally on Unix this function will configure the file descriptors + /// with `CLOEXEC` so they're not automatically inherited by spawned + /// children. + /// + /// On unix if `check_pipe` enabled this function will check if provided + /// files are actually pipes. + /// + /// # Safety + /// + /// This function is `unsafe` to call on Unix specifically as it + /// transitively requires usage of the `from_raw_fd` function, which is + /// itself unsafe in some circumstances. + /// + /// It's recommended to call this function very early in the lifetime of a + /// program before any other file descriptors are opened. That way you can + /// make sure to take ownership properly of the file descriptors passed + /// down, if any. + /// + /// It's generally unsafe to call this function twice in a program if the + /// previous invocation returned `Some`. + /// + /// Note, though, that on Windows it should be safe to call this function + /// any number of times. + pub unsafe fn from_env_ext(check_pipe: bool) -> FromEnv { + let (env, var_os) = match ["CARGO_MAKEFLAGS", "MAKEFLAGS", "MFLAGS"] + .iter() + .map(|&env| env::var_os(env).map(|var| (env, var))) + .find_map(|p| p) + { + Some((env, var_os)) => (env, var_os), + None => return FromEnv::new_err(FromEnvErrorInner::NoEnvVar, "", Default::default()), + }; + + let var = match var_os.to_str() { + Some(var) => var, + None => { + let err = FromEnvErrorInner::CannotParse("not valid UTF-8".to_string()); + return FromEnv::new_err(err, env, var_os); + } + }; + + let (arg, pos) = match ["--jobserver-fds=", "--jobserver-auth="] + .iter() + .map(|&arg| var.find(arg).map(|pos| (arg, pos))) + .find_map(|pos| pos) + { + Some((arg, pos)) => (arg, pos), + None => return FromEnv::new_err(FromEnvErrorInner::NoJobserver, env, var_os), + }; + + let s = var[pos + arg.len()..].split(' ').next().unwrap(); + match imp::Client::open(s, check_pipe) { + Ok(c) => FromEnv::new_ok(Client { inner: Arc::new(c) }, env, var_os), + Err(err) => FromEnv::new_err(err, env, var_os), + } + } + + /// Attempts to connect to the jobserver specified in this process's + /// environment. + /// + /// Wraps `from_env_ext` and discards error details. + pub unsafe fn from_env() -> Option { + Self::from_env_ext(false).client.ok() + } + + /// Acquires a token from this jobserver client. + /// + /// This function will block the calling thread until a new token can be + /// acquired from the jobserver. + /// + /// # Return value + /// + /// On successful acquisition of a token an instance of `Acquired` is + /// returned. This structure, when dropped, will release the token back to + /// the jobserver. It's recommended to avoid leaking this value. + /// + /// # Errors + /// + /// If an I/O error happens while acquiring a token then this function will + /// return immediately with the error. If an error is returned then a token + /// was not acquired. + pub fn acquire(&self) -> io::Result { + let data = self.inner.acquire()?; + Ok(Acquired { + client: self.inner.clone(), + data, + disabled: false, + }) + } + + /// Returns amount of tokens in the read-side pipe. + /// + /// # Return value + /// + /// Number of bytes available to be read from the jobserver pipe + /// + /// # Errors + /// + /// Underlying errors from the ioctl will be passed up. + pub fn available(&self) -> io::Result { + self.inner.available() + } + + /// Configures a child process to have access to this client's jobserver as + /// well. + /// + /// This function is required to be called to ensure that a jobserver is + /// properly inherited to a child process. If this function is *not* called + /// then this `Client` will not be accessible in the child process. In other + /// words, if not called, then `Client::from_env` will return `None` in the + /// child process (or the equivalent of `Child::from_env` that `make` uses). + /// + /// ## Platform-specific behavior + /// + /// On Unix and Windows this will clobber the `CARGO_MAKEFLAGS` environment + /// variables for the child process, and on Unix this will also allow the + /// two file descriptors for this client to be inherited to the child. + /// + /// On platforms other than Unix and Windows this panics. + pub fn configure(&self, cmd: &mut Command) { + cmd.env("CARGO_MAKEFLAGS", &self.mflags_env()); + self.inner.configure(cmd); + } + + /// Configures a child process to have access to this client's jobserver as + /// well. + /// + /// This function is required to be called to ensure that a jobserver is + /// properly inherited to a child process. If this function is *not* called + /// then this `Client` will not be accessible in the child process. In other + /// words, if not called, then `Client::from_env` will return `None` in the + /// child process (or the equivalent of `Child::from_env` that `make` uses). + /// + /// ## Platform-specific behavior + /// + /// On Unix and Windows this will clobber the `CARGO_MAKEFLAGS`, + /// `MAKEFLAGS` and `MFLAGS` environment variables for the child process, + /// and on Unix this will also allow the two file descriptors for + /// this client to be inherited to the child. + /// + /// On platforms other than Unix and Windows this panics. + pub fn configure_make(&self, cmd: &mut Command) { + let value = self.mflags_env(); + cmd.env("CARGO_MAKEFLAGS", &value); + cmd.env("MAKEFLAGS", &value); + cmd.env("MFLAGS", &value); + self.inner.configure(cmd); + } + + fn mflags_env(&self) -> String { + let arg = self.inner.string_arg(); + // Older implementations of make use `--jobserver-fds` and newer + // implementations use `--jobserver-auth`, pass both to try to catch + // both implementations. + format!("-j --jobserver-fds={0} --jobserver-auth={0}", arg) + } + + /// Converts this `Client` into a helper thread to deal with a blocking + /// `acquire` function a little more easily. + /// + /// The fact that the `acquire` function on `Client` blocks isn't always + /// the easiest to work with. Typically you're using a jobserver to + /// manage running other events in parallel! This means that you need to + /// either (a) wait for an existing job to finish or (b) wait for a + /// new token to become available. + /// + /// Unfortunately the blocking in `acquire` happens at the implementation + /// layer of jobservers. On Unix this requires a blocking call to `read` + /// and on Windows this requires one of the `WaitFor*` functions. Both + /// of these situations aren't the easiest to deal with: + /// + /// * On Unix there's basically only one way to wake up a `read` early, and + /// that's through a signal. This is what the `make` implementation + /// itself uses, relying on `SIGCHLD` to wake up a blocking acquisition + /// of a new job token. Unfortunately nonblocking I/O is not an option + /// here, so it means that "waiting for one of two events" means that + /// the latter event must generate a signal! This is not always the case + /// on unix for all jobservers. + /// + /// * On Windows you'd have to basically use the `WaitForMultipleObjects` + /// which means that you've got to canonicalize all your event sources + /// into a `HANDLE` which also isn't the easiest thing to do + /// unfortunately. + /// + /// This function essentially attempts to ease these limitations by + /// converting this `Client` into a helper thread spawned into this + /// process. The application can then request that the helper thread + /// acquires tokens and the provided closure will be invoked for each token + /// acquired. + /// + /// The intention is that this function can be used to translate the event + /// of a token acquisition into an arbitrary user-defined event. + /// + /// # Arguments + /// + /// This function will consume the `Client` provided to be transferred to + /// the helper thread that is spawned. Additionally a closure `f` is + /// provided to be invoked whenever a token is acquired. + /// + /// This closure is only invoked after calls to + /// `HelperThread::request_token` have been made and a token itself has + /// been acquired. If an error happens while acquiring the token then + /// an error will be yielded to the closure as well. + /// + /// # Return Value + /// + /// This function will return an instance of the `HelperThread` structure + /// which is used to manage the helper thread associated with this client. + /// Through the `HelperThread` you'll request that tokens are acquired. + /// When acquired, the closure provided here is invoked. + /// + /// When the `HelperThread` structure is returned it will be gracefully + /// torn down, and the calling thread will be blocked until the thread is + /// torn down (which should be prompt). + /// + /// # Errors + /// + /// This function may fail due to creation of the helper thread or + /// auxiliary I/O objects to manage the helper thread. In any of these + /// situations the error is propagated upwards. + /// + /// # Platform-specific behavior + /// + /// On Windows this function behaves pretty normally as expected, but on + /// Unix the implementation is... a little heinous. As mentioned above + /// we're forced into blocking I/O for token acquisition, namely a blocking + /// call to `read`. We must be able to unblock this, however, to tear down + /// the helper thread gracefully! + /// + /// Essentially what happens is that we'll send a signal to the helper + /// thread spawned and rely on `EINTR` being returned to wake up the helper + /// thread. This involves installing a global `SIGUSR1` handler that does + /// nothing along with sending signals to that thread. This may cause + /// odd behavior in some applications, so it's recommended to review and + /// test thoroughly before using this. + pub fn into_helper_thread(self, f: F) -> io::Result + where + F: FnMut(io::Result) + Send + 'static, + { + let state = Arc::new(HelperState::default()); + Ok(HelperThread { + inner: Some(imp::spawn_helper(self, state.clone(), Box::new(f))?), + state, + }) + } + + /// Blocks the current thread until a token is acquired. + /// + /// This is the same as `acquire`, except that it doesn't return an RAII + /// helper. If successful the process will need to guarantee that + /// `release_raw` is called in the future. + pub fn acquire_raw(&self) -> io::Result<()> { + self.inner.acquire()?; + Ok(()) + } + + /// Releases a jobserver token back to the original jobserver. + /// + /// This is intended to be paired with `acquire_raw` if it was called, but + /// in some situations it could also be called to relinquish a process's + /// implicit token temporarily which is then re-acquired later. + pub fn release_raw(&self) -> io::Result<()> { + self.inner.release(None)?; + Ok(()) + } +} + +impl Drop for Acquired { + fn drop(&mut self) { + if !self.disabled { + drop(self.client.release(Some(&self.data))); + } + } +} + +/// Structure returned from `Client::into_helper_thread` to manage the lifetime +/// of the helper thread returned, see those associated docs for more info. +#[derive(Debug)] +pub struct HelperThread { + inner: Option, + state: Arc, +} + +impl HelperThread { + /// Request that the helper thread acquires a token, eventually calling the + /// original closure with a token when it's available. + /// + /// For more information, see the docs on that function. + pub fn request_token(&self) { + // Indicate that there's one more request for a token and then wake up + // the helper thread if it's sleeping. + self.state.lock().requests += 1; + self.state.cvar.notify_one(); + } +} + +impl Drop for HelperThread { + fn drop(&mut self) { + // Flag that the producer half is done so the helper thread should exit + // quickly if it's waiting. Wake it up if it's actually waiting + self.state.lock().producer_done = true; + self.state.cvar.notify_one(); + + // ... and afterwards perform any thread cleanup logic + self.inner.take().unwrap().join(); + } +} + +impl HelperState { + fn lock(&self) -> MutexGuard<'_, HelperInner> { + self.lock.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// Executes `f` for each request for a token, where `f` is expected to + /// block and then provide the original closure with a token once it's + /// acquired. + /// + /// This is an infinite loop until the helper thread is dropped, at which + /// point everything should get interrupted. + fn for_each_request(&self, mut f: impl FnMut(&HelperState)) { + let mut lock = self.lock(); + + // We only execute while we could receive requests, but as soon as + // that's `false` we're out of here. + while !lock.producer_done { + // If no one's requested a token then we wait for someone to + // request a token. + if lock.requests == 0 { + lock = self.cvar.wait(lock).unwrap_or_else(|e| e.into_inner()); + continue; + } + + // Consume the request for a token, and then actually acquire a + // token after unlocking our lock (not that acquisition happens in + // `f`). This ensures that we don't actually hold the lock if we + // wait for a long time for a token. + lock.requests -= 1; + drop(lock); + f(self); + lock = self.lock(); + } + lock.consumer_done = true; + self.cvar.notify_one(); + } + + fn producer_done(&self) -> bool { + self.lock().producer_done + } +} + +#[test] +fn no_helper_deadlock() { + let x = crate::Client::new(32).unwrap(); + let _y = x.clone(); + std::mem::drop(x.into_helper_thread(|_| {}).unwrap()); +} diff --git a/vendor/jobserver/src/unix.rs b/vendor/jobserver/src/unix.rs index e4b143505..b2312b08f 100644 --- a/vendor/jobserver/src/unix.rs +++ b/vendor/jobserver/src/unix.rs @@ -1,431 +1,477 @@ -use libc::c_int; - -use std::fs::{File, OpenOptions}; -use std::io::{self, Read, Write}; -use std::mem; -use std::mem::MaybeUninit; -use std::os::unix::prelude::*; -use std::path::{Path, PathBuf}; -use std::process::Command; -use std::ptr; -use std::sync::{Arc, Once}; -use std::thread::{self, Builder, JoinHandle}; -use std::time::Duration; - -#[derive(Debug)] -pub enum Client { - /// `--jobserver-auth=R,W` - Pipe { read: File, write: File }, - /// `--jobserver-auth=fifo:PATH` - Fifo { file: File, path: PathBuf }, -} - -#[derive(Debug)] -pub struct Acquired { - byte: u8, -} - -impl Client { - pub fn new(mut limit: usize) -> io::Result { - let client = unsafe { Client::mk()? }; - - // I don't think the character written here matters, but I could be - // wrong! - const BUFFER: [u8; 128] = [b'|'; 128]; - - let mut write = client.write(); - - set_nonblocking(write.as_raw_fd(), true)?; - - while limit > 0 { - let n = limit.min(BUFFER.len()); - - write.write_all(&BUFFER[..n])?; - limit -= n; - } - - set_nonblocking(write.as_raw_fd(), false)?; - - Ok(client) - } - - unsafe fn mk() -> io::Result { - let mut pipes = [0; 2]; - - // Attempt atomically-create-with-cloexec if we can on Linux, - // detected by using the `syscall` function in `libc` to try to work - // with as many kernels/glibc implementations as possible. - #[cfg(target_os = "linux")] - { - use std::sync::atomic::{AtomicBool, Ordering}; - - static PIPE2_AVAILABLE: AtomicBool = AtomicBool::new(true); - if PIPE2_AVAILABLE.load(Ordering::SeqCst) { - match libc::syscall(libc::SYS_pipe2, pipes.as_mut_ptr(), libc::O_CLOEXEC) { - -1 => { - let err = io::Error::last_os_error(); - if err.raw_os_error() == Some(libc::ENOSYS) { - PIPE2_AVAILABLE.store(false, Ordering::SeqCst); - } else { - return Err(err); - } - } - _ => return Ok(Client::from_fds(pipes[0], pipes[1])), - } - } - } - - cvt(libc::pipe(pipes.as_mut_ptr()))?; - drop(set_cloexec(pipes[0], true)); - drop(set_cloexec(pipes[1], true)); - Ok(Client::from_fds(pipes[0], pipes[1])) - } - - pub unsafe fn open(s: &str) -> Option { - Client::from_fifo(s).or_else(|| Client::from_pipe(s)) - } - - /// `--jobserver-auth=fifo:PATH` - fn from_fifo(s: &str) -> Option { - let mut parts = s.splitn(2, ':'); - if parts.next().unwrap() != "fifo" { - return None; - } - let path = match parts.next() { - Some(p) => Path::new(p), - None => return None, - }; - let file = match OpenOptions::new().read(true).write(true).open(path) { - Ok(f) => f, - Err(_) => return None, - }; - Some(Client::Fifo { - file, - path: path.into(), - }) - } - - /// `--jobserver-auth=R,W` - unsafe fn from_pipe(s: &str) -> Option { - let mut parts = s.splitn(2, ','); - let read = parts.next().unwrap(); - let write = match parts.next() { - Some(s) => s, - None => return None, - }; - - let read = match read.parse() { - Ok(n) => n, - Err(_) => return None, - }; - let write = match write.parse() { - Ok(n) => n, - Err(_) => return None, - }; - - // Ok so we've got two integers that look like file descriptors, but - // for extra sanity checking let's see if they actually look like - // instances of a pipe before we return the client. - // - // If we're called from `make` *without* the leading + on our rule - // then we'll have `MAKEFLAGS` env vars but won't actually have - // access to the file descriptors. - if is_valid_fd(read) && is_valid_fd(write) { - drop(set_cloexec(read, true)); - drop(set_cloexec(write, true)); - Some(Client::from_fds(read, write)) - } else { - None - } - } - - unsafe fn from_fds(read: c_int, write: c_int) -> Client { - Client::Pipe { - read: File::from_raw_fd(read), - write: File::from_raw_fd(write), - } - } - - /// Gets the read end of our jobserver client. - fn read(&self) -> &File { - match self { - Client::Pipe { read, .. } => read, - Client::Fifo { file, .. } => file, - } - } - - /// Gets the write end of our jobserver client. - fn write(&self) -> &File { - match self { - Client::Pipe { write, .. } => write, - Client::Fifo { file, .. } => file, - } - } - - pub fn acquire(&self) -> io::Result { - // Ignore interrupts and keep trying if that happens - loop { - if let Some(token) = self.acquire_allow_interrupts()? { - return Ok(token); - } - } - } - - /// Block waiting for a token, returning `None` if we're interrupted with - /// EINTR. - fn acquire_allow_interrupts(&self) -> io::Result> { - // We don't actually know if the file descriptor here is set in - // blocking or nonblocking mode. AFAIK all released versions of - // `make` use blocking fds for the jobserver, but the unreleased - // version of `make` doesn't. In the unreleased version jobserver - // fds are set to nonblocking and combined with `pselect` - // internally. - // - // Here we try to be compatible with both strategies. We optimistically - // try to read from the file descriptor which then may block, return - // a token or indicate that polling is needed. - // Blocking reads (if possible) allows the kernel to be more selective - // about which readers to wake up when a token is written to the pipe. - // - // We use `poll` here to block this thread waiting for read - // readiness, and then afterwards we perform the `read` itself. If - // the `read` returns that it would block then we start over and try - // again. - // - // Also note that we explicitly don't handle EINTR here. That's used - // to shut us down, so we otherwise punt all errors upwards. - unsafe { - let mut fd: libc::pollfd = mem::zeroed(); - let mut read = self.read(); - fd.fd = read.as_raw_fd(); - fd.events = libc::POLLIN; - loop { - let mut buf = [0]; - match read.read(&mut buf) { - Ok(1) => return Ok(Some(Acquired { byte: buf[0] })), - Ok(_) => { - return Err(io::Error::new( - io::ErrorKind::Other, - "early EOF on jobserver pipe", - )) - } - Err(e) => match e.kind() { - io::ErrorKind::WouldBlock => { /* fall through to polling */ } - io::ErrorKind::Interrupted => return Ok(None), - _ => return Err(e), - }, - } - - loop { - fd.revents = 0; - if libc::poll(&mut fd, 1, -1) == -1 { - let e = io::Error::last_os_error(); - return match e.kind() { - io::ErrorKind::Interrupted => Ok(None), - _ => Err(e), - }; - } - if fd.revents != 0 { - break; - } - } - } - } - } - - pub fn release(&self, data: Option<&Acquired>) -> io::Result<()> { - // Note that the fd may be nonblocking but we're going to go ahead - // and assume that the writes here are always nonblocking (we can - // always quickly release a token). If that turns out to not be the - // case we'll get an error anyway! - let byte = data.map(|d| d.byte).unwrap_or(b'+'); - match self.write().write(&[byte])? { - 1 => Ok(()), - _ => Err(io::Error::new( - io::ErrorKind::Other, - "failed to write token back to jobserver", - )), - } - } - - pub fn string_arg(&self) -> String { - match self { - Client::Pipe { read, write } => format!("{},{}", read.as_raw_fd(), write.as_raw_fd()), - Client::Fifo { path, .. } => format!("fifo:{}", path.to_str().unwrap()), - } - } - - pub fn available(&self) -> io::Result { - let mut len = MaybeUninit::::uninit(); - cvt(unsafe { libc::ioctl(self.read().as_raw_fd(), libc::FIONREAD, len.as_mut_ptr()) })?; - Ok(unsafe { len.assume_init() } as usize) - } - - pub fn configure(&self, cmd: &mut Command) { - match self { - // We `File::open`ed it when inheriting from environment, - // so no need to set cloexec for fifo. - Client::Fifo { .. } => return, - Client::Pipe { .. } => {} - }; - // Here we basically just want to say that in the child process - // we'll configure the read/write file descriptors to *not* be - // cloexec, so they're inherited across the exec and specified as - // integers through `string_arg` above. - let read = self.read().as_raw_fd(); - let write = self.write().as_raw_fd(); - unsafe { - cmd.pre_exec(move || { - set_cloexec(read, false)?; - set_cloexec(write, false)?; - Ok(()) - }); - } - } -} - -#[derive(Debug)] -pub struct Helper { - thread: JoinHandle<()>, - state: Arc, -} - -pub(crate) fn spawn_helper( - client: crate::Client, - state: Arc, - mut f: Box) + Send>, -) -> io::Result { - static USR1_INIT: Once = Once::new(); - let mut err = None; - USR1_INIT.call_once(|| unsafe { - let mut new: libc::sigaction = mem::zeroed(); - #[cfg(target_os = "aix")] - { - new.sa_union.__su_sigaction = sigusr1_handler; - } - #[cfg(not(target_os = "aix"))] - { - new.sa_sigaction = sigusr1_handler as usize; - } - new.sa_flags = libc::SA_SIGINFO as _; - if libc::sigaction(libc::SIGUSR1, &new, ptr::null_mut()) != 0 { - err = Some(io::Error::last_os_error()); - } - }); - - if let Some(e) = err.take() { - return Err(e); - } - - let state2 = state.clone(); - let thread = Builder::new().spawn(move || { - state2.for_each_request(|helper| loop { - match client.inner.acquire_allow_interrupts() { - Ok(Some(data)) => { - break f(Ok(crate::Acquired { - client: client.inner.clone(), - data, - disabled: false, - })) - } - Err(e) => break f(Err(e)), - Ok(None) if helper.producer_done() => break, - Ok(None) => {} - } - }); - })?; - - Ok(Helper { thread, state }) -} - -impl Helper { - pub fn join(self) { - let dur = Duration::from_millis(10); - let mut state = self.state.lock(); - debug_assert!(state.producer_done); - - // We need to join our helper thread, and it could be blocked in one - // of two locations. First is the wait for a request, but the - // initial drop of `HelperState` will take care of that. Otherwise - // it may be blocked in `client.acquire()`. We actually have no way - // of interrupting that, so resort to `pthread_kill` as a fallback. - // This signal should interrupt any blocking `read` call with - // `io::ErrorKind::Interrupt` and cause the thread to cleanly exit. - // - // Note that we don't do this forever though since there's a chance - // of bugs, so only do this opportunistically to make a best effort - // at clearing ourselves up. - for _ in 0..100 { - if state.consumer_done { - break; - } - unsafe { - // Ignore the return value here of `pthread_kill`, - // apparently on OSX if you kill a dead thread it will - // return an error, but on other platforms it may not. In - // that sense we don't actually know if this will succeed or - // not! - libc::pthread_kill(self.thread.as_pthread_t() as _, libc::SIGUSR1); - } - state = self - .state - .cvar - .wait_timeout(state, dur) - .unwrap_or_else(|e| e.into_inner()) - .0; - thread::yield_now(); // we really want the other thread to run - } - - // If we managed to actually see the consumer get done, then we can - // definitely wait for the thread. Otherwise it's... off in the ether - // I guess? - if state.consumer_done { - drop(self.thread.join()); - } - } -} - -fn is_valid_fd(fd: c_int) -> bool { - unsafe { libc::fcntl(fd, libc::F_GETFD) != -1 } -} - -fn set_cloexec(fd: c_int, set: bool) -> io::Result<()> { - unsafe { - let previous = cvt(libc::fcntl(fd, libc::F_GETFD))?; - let new = if set { - previous | libc::FD_CLOEXEC - } else { - previous & !libc::FD_CLOEXEC - }; - if new != previous { - cvt(libc::fcntl(fd, libc::F_SETFD, new))?; - } - Ok(()) - } -} - -fn set_nonblocking(fd: c_int, set: bool) -> io::Result<()> { - let status_flag = if set { libc::O_NONBLOCK } else { 0 }; - - unsafe { - cvt(libc::fcntl(fd, libc::F_SETFL, status_flag))?; - } - - Ok(()) -} - -fn cvt(t: c_int) -> io::Result { - if t == -1 { - Err(io::Error::last_os_error()) - } else { - Ok(t) - } -} - -extern "C" fn sigusr1_handler( - _signum: c_int, - _info: *mut libc::siginfo_t, - _ptr: *mut libc::c_void, -) { - // nothing to do -} +use libc::c_int; + +use crate::FromEnvErrorInner; +use std::fs::{File, OpenOptions}; +use std::io::{self, Read, Write}; +use std::mem; +use std::mem::MaybeUninit; +use std::os::unix::prelude::*; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::ptr; +use std::sync::{Arc, Once}; +use std::thread::{self, Builder, JoinHandle}; +use std::time::Duration; + +#[derive(Debug)] +pub enum Client { + /// `--jobserver-auth=R,W` + Pipe { read: File, write: File }, + /// `--jobserver-auth=fifo:PATH` + Fifo { file: File, path: PathBuf }, +} + +#[derive(Debug)] +pub struct Acquired { + byte: u8, +} + +impl Client { + pub fn new(mut limit: usize) -> io::Result { + let client = unsafe { Client::mk()? }; + + // I don't think the character written here matters, but I could be + // wrong! + const BUFFER: [u8; 128] = [b'|'; 128]; + + let mut write = client.write(); + + set_nonblocking(write.as_raw_fd(), true)?; + + while limit > 0 { + let n = limit.min(BUFFER.len()); + + write.write_all(&BUFFER[..n])?; + limit -= n; + } + + set_nonblocking(write.as_raw_fd(), false)?; + + Ok(client) + } + + unsafe fn mk() -> io::Result { + let mut pipes = [0; 2]; + + // Attempt atomically-create-with-cloexec if we can on Linux, + // detected by using the `syscall` function in `libc` to try to work + // with as many kernels/glibc implementations as possible. + #[cfg(target_os = "linux")] + { + use std::sync::atomic::{AtomicBool, Ordering}; + + static PIPE2_AVAILABLE: AtomicBool = AtomicBool::new(true); + if PIPE2_AVAILABLE.load(Ordering::SeqCst) { + match libc::syscall(libc::SYS_pipe2, pipes.as_mut_ptr(), libc::O_CLOEXEC) { + -1 => { + let err = io::Error::last_os_error(); + if err.raw_os_error() == Some(libc::ENOSYS) { + PIPE2_AVAILABLE.store(false, Ordering::SeqCst); + } else { + return Err(err); + } + } + _ => return Ok(Client::from_fds(pipes[0], pipes[1])), + } + } + } + + cvt(libc::pipe(pipes.as_mut_ptr()))?; + drop(set_cloexec(pipes[0], true)); + drop(set_cloexec(pipes[1], true)); + Ok(Client::from_fds(pipes[0], pipes[1])) + } + + pub(crate) unsafe fn open(s: &str, check_pipe: bool) -> Result { + if let Some(client) = Self::from_fifo(s)? { + return Ok(client); + } + if let Some(client) = Self::from_pipe(s, check_pipe)? { + return Ok(client); + } + Err(FromEnvErrorInner::CannotParse(format!( + "expected `fifo:PATH` or `R,W`, found `{s}`" + ))) + } + + /// `--jobserver-auth=fifo:PATH` + fn from_fifo(s: &str) -> Result, FromEnvErrorInner> { + let mut parts = s.splitn(2, ':'); + if parts.next().unwrap() != "fifo" { + return Ok(None); + } + let path_str = parts.next().ok_or_else(|| { + FromEnvErrorInner::CannotParse("expected a path after `fifo:`".to_string()) + })?; + let path = Path::new(path_str); + let file = OpenOptions::new() + .read(true) + .write(true) + .open(path) + .map_err(|err| FromEnvErrorInner::CannotOpenPath(path_str.to_string(), err))?; + Ok(Some(Client::Fifo { + file, + path: path.into(), + })) + } + + /// `--jobserver-auth=R,W` + unsafe fn from_pipe(s: &str, check_pipe: bool) -> Result, FromEnvErrorInner> { + let mut parts = s.splitn(2, ','); + let read = parts.next().unwrap(); + let write = match parts.next() { + Some(w) => w, + None => return Ok(None), + }; + let read = read + .parse() + .map_err(|e| FromEnvErrorInner::CannotParse(format!("cannot parse `read` fd: {e}")))?; + let write = write + .parse() + .map_err(|e| FromEnvErrorInner::CannotParse(format!("cannot parse `write` fd: {e}")))?; + + // Ok so we've got two integers that look like file descriptors, but + // for extra sanity checking let's see if they actually look like + // valid files and instances of a pipe if feature enabled before we + // return the client. + // + // If we're called from `make` *without* the leading + on our rule + // then we'll have `MAKEFLAGS` env vars but won't actually have + // access to the file descriptors. + // + // `NotAPipe` is a worse error, return it if it's reported for any of the two fds. + match (fd_check(read, check_pipe), fd_check(write, check_pipe)) { + (read_err @ Err(FromEnvErrorInner::NotAPipe(..)), _) => read_err?, + (_, write_err @ Err(FromEnvErrorInner::NotAPipe(..))) => write_err?, + (read_err, write_err) => { + read_err?; + write_err?; + } + } + + drop(set_cloexec(read, true)); + drop(set_cloexec(write, true)); + Ok(Some(Client::from_fds(read, write))) + } + + unsafe fn from_fds(read: c_int, write: c_int) -> Client { + Client::Pipe { + read: File::from_raw_fd(read), + write: File::from_raw_fd(write), + } + } + + /// Gets the read end of our jobserver client. + fn read(&self) -> &File { + match self { + Client::Pipe { read, .. } => read, + Client::Fifo { file, .. } => file, + } + } + + /// Gets the write end of our jobserver client. + fn write(&self) -> &File { + match self { + Client::Pipe { write, .. } => write, + Client::Fifo { file, .. } => file, + } + } + + pub fn acquire(&self) -> io::Result { + // Ignore interrupts and keep trying if that happens + loop { + if let Some(token) = self.acquire_allow_interrupts()? { + return Ok(token); + } + } + } + + /// Block waiting for a token, returning `None` if we're interrupted with + /// EINTR. + fn acquire_allow_interrupts(&self) -> io::Result> { + // We don't actually know if the file descriptor here is set in + // blocking or nonblocking mode. AFAIK all released versions of + // `make` use blocking fds for the jobserver, but the unreleased + // version of `make` doesn't. In the unreleased version jobserver + // fds are set to nonblocking and combined with `pselect` + // internally. + // + // Here we try to be compatible with both strategies. We optimistically + // try to read from the file descriptor which then may block, return + // a token or indicate that polling is needed. + // Blocking reads (if possible) allows the kernel to be more selective + // about which readers to wake up when a token is written to the pipe. + // + // We use `poll` here to block this thread waiting for read + // readiness, and then afterwards we perform the `read` itself. If + // the `read` returns that it would block then we start over and try + // again. + // + // Also note that we explicitly don't handle EINTR here. That's used + // to shut us down, so we otherwise punt all errors upwards. + unsafe { + let mut fd: libc::pollfd = mem::zeroed(); + let mut read = self.read(); + fd.fd = read.as_raw_fd(); + fd.events = libc::POLLIN; + loop { + let mut buf = [0]; + match read.read(&mut buf) { + Ok(1) => return Ok(Some(Acquired { byte: buf[0] })), + Ok(_) => { + return Err(io::Error::new( + io::ErrorKind::Other, + "early EOF on jobserver pipe", + )); + } + Err(e) => match e.kind() { + io::ErrorKind::WouldBlock => { /* fall through to polling */ } + io::ErrorKind::Interrupted => return Ok(None), + _ => return Err(e), + }, + } + + loop { + fd.revents = 0; + if libc::poll(&mut fd, 1, -1) == -1 { + let e = io::Error::last_os_error(); + return match e.kind() { + io::ErrorKind::Interrupted => Ok(None), + _ => Err(e), + }; + } + if fd.revents != 0 { + break; + } + } + } + } + } + + pub fn release(&self, data: Option<&Acquired>) -> io::Result<()> { + // Note that the fd may be nonblocking but we're going to go ahead + // and assume that the writes here are always nonblocking (we can + // always quickly release a token). If that turns out to not be the + // case we'll get an error anyway! + let byte = data.map(|d| d.byte).unwrap_or(b'+'); + match self.write().write(&[byte])? { + 1 => Ok(()), + _ => Err(io::Error::new( + io::ErrorKind::Other, + "failed to write token back to jobserver", + )), + } + } + + pub fn string_arg(&self) -> String { + match self { + Client::Pipe { read, write } => format!("{},{}", read.as_raw_fd(), write.as_raw_fd()), + Client::Fifo { path, .. } => format!("fifo:{}", path.to_str().unwrap()), + } + } + + pub fn available(&self) -> io::Result { + let mut len = MaybeUninit::::uninit(); + cvt(unsafe { libc::ioctl(self.read().as_raw_fd(), libc::FIONREAD, len.as_mut_ptr()) })?; + Ok(unsafe { len.assume_init() } as usize) + } + + pub fn configure(&self, cmd: &mut Command) { + match self { + // We `File::open`ed it when inheriting from environment, + // so no need to set cloexec for fifo. + Client::Fifo { .. } => return, + Client::Pipe { .. } => {} + }; + // Here we basically just want to say that in the child process + // we'll configure the read/write file descriptors to *not* be + // cloexec, so they're inherited across the exec and specified as + // integers through `string_arg` above. + let read = self.read().as_raw_fd(); + let write = self.write().as_raw_fd(); + unsafe { + cmd.pre_exec(move || { + set_cloexec(read, false)?; + set_cloexec(write, false)?; + Ok(()) + }); + } + } +} + +#[derive(Debug)] +pub struct Helper { + thread: JoinHandle<()>, + state: Arc, +} + +pub(crate) fn spawn_helper( + client: crate::Client, + state: Arc, + mut f: Box) + Send>, +) -> io::Result { + static USR1_INIT: Once = Once::new(); + let mut err = None; + USR1_INIT.call_once(|| unsafe { + let mut new: libc::sigaction = mem::zeroed(); + #[cfg(target_os = "aix")] + { + new.sa_union.__su_sigaction = sigusr1_handler; + } + #[cfg(not(target_os = "aix"))] + { + new.sa_sigaction = sigusr1_handler as usize; + } + new.sa_flags = libc::SA_SIGINFO as _; + if libc::sigaction(libc::SIGUSR1, &new, ptr::null_mut()) != 0 { + err = Some(io::Error::last_os_error()); + } + }); + + if let Some(e) = err.take() { + return Err(e); + } + + let state2 = state.clone(); + let thread = Builder::new().spawn(move || { + state2.for_each_request(|helper| loop { + match client.inner.acquire_allow_interrupts() { + Ok(Some(data)) => { + break f(Ok(crate::Acquired { + client: client.inner.clone(), + data, + disabled: false, + })); + } + Err(e) => break f(Err(e)), + Ok(None) if helper.producer_done() => break, + Ok(None) => {} + } + }); + })?; + + Ok(Helper { thread, state }) +} + +impl Helper { + pub fn join(self) { + let dur = Duration::from_millis(10); + let mut state = self.state.lock(); + debug_assert!(state.producer_done); + + // We need to join our helper thread, and it could be blocked in one + // of two locations. First is the wait for a request, but the + // initial drop of `HelperState` will take care of that. Otherwise + // it may be blocked in `client.acquire()`. We actually have no way + // of interrupting that, so resort to `pthread_kill` as a fallback. + // This signal should interrupt any blocking `read` call with + // `io::ErrorKind::Interrupt` and cause the thread to cleanly exit. + // + // Note that we don't do this forever though since there's a chance + // of bugs, so only do this opportunistically to make a best effort + // at clearing ourselves up. + for _ in 0..100 { + if state.consumer_done { + break; + } + unsafe { + // Ignore the return value here of `pthread_kill`, + // apparently on OSX if you kill a dead thread it will + // return an error, but on other platforms it may not. In + // that sense we don't actually know if this will succeed or + // not! + libc::pthread_kill(self.thread.as_pthread_t() as _, libc::SIGUSR1); + } + state = self + .state + .cvar + .wait_timeout(state, dur) + .unwrap_or_else(|e| e.into_inner()) + .0; + thread::yield_now(); // we really want the other thread to run + } + + // If we managed to actually see the consumer get done, then we can + // definitely wait for the thread. Otherwise it's... off in the ether + // I guess? + if state.consumer_done { + drop(self.thread.join()); + } + } +} + +unsafe fn fcntl_check(fd: c_int) -> Result<(), FromEnvErrorInner> { + match libc::fcntl(fd, libc::F_GETFD) { + -1 => Err(FromEnvErrorInner::CannotOpenFd( + fd, + io::Error::last_os_error(), + )), + _ => Ok(()), + } +} + +unsafe fn fd_check(fd: c_int, check_pipe: bool) -> Result<(), FromEnvErrorInner> { + if check_pipe { + let mut stat = mem::zeroed(); + if libc::fstat(fd, &mut stat) == -1 { + let last_os_error = io::Error::last_os_error(); + fcntl_check(fd)?; + Err(FromEnvErrorInner::NotAPipe(fd, Some(last_os_error))) + } else { + // On android arm and i686 mode_t is u16 and st_mode is u32, + // this generates a type mismatch when S_IFIFO (declared as mode_t) + // is used in operations with st_mode, so we use this workaround + // to get the value of S_IFIFO with the same type of st_mode. + #[allow(unused_assignments)] + let mut s_ififo = stat.st_mode; + s_ififo = libc::S_IFIFO as _; + if stat.st_mode & s_ififo == s_ififo { + return Ok(()); + } + Err(FromEnvErrorInner::NotAPipe(fd, None)) + } + } else { + fcntl_check(fd) + } +} + +fn set_cloexec(fd: c_int, set: bool) -> io::Result<()> { + unsafe { + let previous = cvt(libc::fcntl(fd, libc::F_GETFD))?; + let new = if set { + previous | libc::FD_CLOEXEC + } else { + previous & !libc::FD_CLOEXEC + }; + if new != previous { + cvt(libc::fcntl(fd, libc::F_SETFD, new))?; + } + Ok(()) + } +} + +fn set_nonblocking(fd: c_int, set: bool) -> io::Result<()> { + let status_flag = if set { libc::O_NONBLOCK } else { 0 }; + + unsafe { + cvt(libc::fcntl(fd, libc::F_SETFL, status_flag))?; + } + + Ok(()) +} + +fn cvt(t: c_int) -> io::Result { + if t == -1 { + Err(io::Error::last_os_error()) + } else { + Ok(t) + } +} + +extern "C" fn sigusr1_handler( + _signum: c_int, + _info: *mut libc::siginfo_t, + _ptr: *mut libc::c_void, +) { + // nothing to do +} diff --git a/vendor/jobserver/src/wasm.rs b/vendor/jobserver/src/wasm.rs index 3793bd67c..3dda675fd 100644 --- a/vendor/jobserver/src/wasm.rs +++ b/vendor/jobserver/src/wasm.rs @@ -1,95 +1,96 @@ -use std::io; -use std::process::Command; -use std::sync::{Arc, Condvar, Mutex}; -use std::thread::{Builder, JoinHandle}; - -#[derive(Debug)] -pub struct Client { - inner: Arc, -} - -#[derive(Debug)] -struct Inner { - count: Mutex, - cvar: Condvar, -} - -#[derive(Debug)] -pub struct Acquired(()); - -impl Client { - pub fn new(limit: usize) -> io::Result { - Ok(Client { - inner: Arc::new(Inner { - count: Mutex::new(limit), - cvar: Condvar::new(), - }), - }) - } - - pub unsafe fn open(_s: &str) -> Option { - None - } - - pub fn acquire(&self) -> io::Result { - let mut lock = self.inner.count.lock().unwrap_or_else(|e| e.into_inner()); - while *lock == 0 { - lock = self - .inner - .cvar - .wait(lock) - .unwrap_or_else(|e| e.into_inner()); - } - *lock -= 1; - Ok(Acquired(())) - } - - pub fn release(&self, _data: Option<&Acquired>) -> io::Result<()> { - let mut lock = self.inner.count.lock().unwrap_or_else(|e| e.into_inner()); - *lock += 1; - drop(lock); - self.inner.cvar.notify_one(); - Ok(()) - } - - pub fn string_arg(&self) -> String { - panic!( - "On this platform there is no cross process jobserver support, - so Client::configure is not supported." - ); - } - - pub fn available(&self) -> io::Result { - let lock = self.inner.count.lock().unwrap_or_else(|e| e.into_inner()); - Ok(*lock) - } - - pub fn configure(&self, _cmd: &mut Command) { - unreachable!(); - } -} - -#[derive(Debug)] -pub struct Helper { - thread: JoinHandle<()>, -} - -pub(crate) fn spawn_helper( - client: crate::Client, - state: Arc, - mut f: Box) + Send>, -) -> io::Result { - let thread = Builder::new().spawn(move || { - state.for_each_request(|_| f(client.acquire())); - })?; - - Ok(Helper { thread: thread }) -} - -impl Helper { - pub fn join(self) { - // TODO: this is not correct if the thread is blocked in - // `client.acquire()`. - drop(self.thread.join()); - } -} +use crate::FromEnvErrorInner; +use std::io; +use std::process::Command; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread::{Builder, JoinHandle}; + +#[derive(Debug)] +pub struct Client { + inner: Arc, +} + +#[derive(Debug)] +struct Inner { + count: Mutex, + cvar: Condvar, +} + +#[derive(Debug)] +pub struct Acquired(()); + +impl Client { + pub fn new(limit: usize) -> io::Result { + Ok(Client { + inner: Arc::new(Inner { + count: Mutex::new(limit), + cvar: Condvar::new(), + }), + }) + } + + pub(crate) unsafe fn open(_s: &str, _check_pipe: bool) -> Result { + Err(FromEnvErrorInner::Unsupported) + } + + pub fn acquire(&self) -> io::Result { + let mut lock = self.inner.count.lock().unwrap_or_else(|e| e.into_inner()); + while *lock == 0 { + lock = self + .inner + .cvar + .wait(lock) + .unwrap_or_else(|e| e.into_inner()); + } + *lock -= 1; + Ok(Acquired(())) + } + + pub fn release(&self, _data: Option<&Acquired>) -> io::Result<()> { + let mut lock = self.inner.count.lock().unwrap_or_else(|e| e.into_inner()); + *lock += 1; + drop(lock); + self.inner.cvar.notify_one(); + Ok(()) + } + + pub fn string_arg(&self) -> String { + panic!( + "On this platform there is no cross process jobserver support, + so Client::configure is not supported." + ); + } + + pub fn available(&self) -> io::Result { + let lock = self.inner.count.lock().unwrap_or_else(|e| e.into_inner()); + Ok(*lock) + } + + pub fn configure(&self, _cmd: &mut Command) { + unreachable!(); + } +} + +#[derive(Debug)] +pub struct Helper { + thread: JoinHandle<()>, +} + +pub(crate) fn spawn_helper( + client: crate::Client, + state: Arc, + mut f: Box) + Send>, +) -> io::Result { + let thread = Builder::new().spawn(move || { + state.for_each_request(|_| f(client.acquire())); + })?; + + Ok(Helper { thread: thread }) +} + +impl Helper { + pub fn join(self) { + // TODO: this is not correct if the thread is blocked in + // `client.acquire()`. + drop(self.thread.join()); + } +} diff --git a/vendor/jobserver/src/windows.rs b/vendor/jobserver/src/windows.rs index 6791efea4..bff89c1b0 100644 --- a/vendor/jobserver/src/windows.rs +++ b/vendor/jobserver/src/windows.rs @@ -1,266 +1,270 @@ -use std::ffi::CString; -use std::io; -use std::process::Command; -use std::ptr; -use std::sync::Arc; -use std::thread::{Builder, JoinHandle}; - -#[derive(Debug)] -pub struct Client { - sem: Handle, - name: String, -} - -#[derive(Debug)] -pub struct Acquired; - -type BOOL = i32; -type DWORD = u32; -type HANDLE = *mut u8; -type LONG = i32; - -const ERROR_ALREADY_EXISTS: DWORD = 183; -const FALSE: BOOL = 0; -const INFINITE: DWORD = 0xffffffff; -const SEMAPHORE_MODIFY_STATE: DWORD = 0x2; -const SYNCHRONIZE: DWORD = 0x00100000; -const TRUE: BOOL = 1; -const WAIT_OBJECT_0: DWORD = 0; - -extern "system" { - fn CloseHandle(handle: HANDLE) -> BOOL; - fn SetEvent(hEvent: HANDLE) -> BOOL; - fn WaitForMultipleObjects( - ncount: DWORD, - lpHandles: *const HANDLE, - bWaitAll: BOOL, - dwMilliseconds: DWORD, - ) -> DWORD; - fn CreateEventA( - lpEventAttributes: *mut u8, - bManualReset: BOOL, - bInitialState: BOOL, - lpName: *const i8, - ) -> HANDLE; - fn ReleaseSemaphore( - hSemaphore: HANDLE, - lReleaseCount: LONG, - lpPreviousCount: *mut LONG, - ) -> BOOL; - fn CreateSemaphoreA( - lpEventAttributes: *mut u8, - lInitialCount: LONG, - lMaximumCount: LONG, - lpName: *const i8, - ) -> HANDLE; - fn OpenSemaphoreA(dwDesiredAccess: DWORD, bInheritHandle: BOOL, lpName: *const i8) -> HANDLE; - fn WaitForSingleObject(hHandle: HANDLE, dwMilliseconds: DWORD) -> DWORD; - #[link_name = "SystemFunction036"] - fn RtlGenRandom(RandomBuffer: *mut u8, RandomBufferLength: u32) -> u8; -} - -// Note that we ideally would use the `getrandom` crate, but unfortunately -// that causes build issues when this crate is used in rust-lang/rust (see -// rust-lang/rust#65014 for more information). As a result we just inline -// the pretty simple Windows-specific implementation of generating -// randomness. -fn getrandom(dest: &mut [u8]) -> io::Result<()> { - // Prevent overflow of u32 - for chunk in dest.chunks_mut(u32::max_value() as usize) { - let ret = unsafe { RtlGenRandom(chunk.as_mut_ptr(), chunk.len() as u32) }; - if ret == 0 { - return Err(io::Error::new( - io::ErrorKind::Other, - "failed to generate random bytes", - )); - } - } - Ok(()) -} - -impl Client { - pub fn new(limit: usize) -> io::Result { - // Try a bunch of random semaphore names until we get a unique one, - // but don't try for too long. - // - // Note that `limit == 0` is a valid argument above but Windows - // won't let us create a semaphore with 0 slots available to it. Get - // `limit == 0` working by creating a semaphore instead with one - // slot and then immediately acquire it (without ever releaseing it - // back). - for _ in 0..100 { - let mut bytes = [0; 4]; - getrandom(&mut bytes)?; - let mut name = format!("__rust_jobserver_semaphore_{}\0", u32::from_ne_bytes(bytes)); - unsafe { - let create_limit = if limit == 0 { 1 } else { limit }; - let r = CreateSemaphoreA( - ptr::null_mut(), - create_limit as LONG, - create_limit as LONG, - name.as_ptr() as *const _, - ); - if r.is_null() { - return Err(io::Error::last_os_error()); - } - let handle = Handle(r); - - let err = io::Error::last_os_error(); - if err.raw_os_error() == Some(ERROR_ALREADY_EXISTS as i32) { - continue; - } - name.pop(); // chop off the trailing nul - let client = Client { - sem: handle, - name: name, - }; - if create_limit != limit { - client.acquire()?; - } - return Ok(client); - } - } - - Err(io::Error::new( - io::ErrorKind::Other, - "failed to find a unique name for a semaphore", - )) - } - - pub unsafe fn open(s: &str) -> Option { - let name = match CString::new(s) { - Ok(s) => s, - Err(_) => return None, - }; - - let sem = OpenSemaphoreA(SYNCHRONIZE | SEMAPHORE_MODIFY_STATE, FALSE, name.as_ptr()); - if sem.is_null() { - None - } else { - Some(Client { - sem: Handle(sem), - name: s.to_string(), - }) - } - } - - pub fn acquire(&self) -> io::Result { - unsafe { - let r = WaitForSingleObject(self.sem.0, INFINITE); - if r == WAIT_OBJECT_0 { - Ok(Acquired) - } else { - Err(io::Error::last_os_error()) - } - } - } - - pub fn release(&self, _data: Option<&Acquired>) -> io::Result<()> { - unsafe { - let r = ReleaseSemaphore(self.sem.0, 1, ptr::null_mut()); - if r != 0 { - Ok(()) - } else { - Err(io::Error::last_os_error()) - } - } - } - - pub fn string_arg(&self) -> String { - self.name.clone() - } - - pub fn available(&self) -> io::Result { - // Can't read value of a semaphore on Windows, so - // try to acquire without sleeping, since we can find out the - // old value on release. If acquisiton fails, then available is 0. - unsafe { - let r = WaitForSingleObject(self.sem.0, 0); - if r != WAIT_OBJECT_0 { - Ok(0) - } else { - let mut prev: LONG = 0; - let r = ReleaseSemaphore(self.sem.0, 1, &mut prev); - if r != 0 { - Ok(prev as usize + 1) - } else { - Err(io::Error::last_os_error()) - } - } - } - } - - pub fn configure(&self, _cmd: &mut Command) { - // nothing to do here, we gave the name of our semaphore to the - // child above - } -} - -#[derive(Debug)] -struct Handle(HANDLE); -// HANDLE is a raw ptr, but we're send/sync -unsafe impl Sync for Handle {} -unsafe impl Send for Handle {} - -impl Drop for Handle { - fn drop(&mut self) { - unsafe { - CloseHandle(self.0); - } - } -} - -#[derive(Debug)] -pub struct Helper { - event: Arc, - thread: JoinHandle<()>, -} - -pub(crate) fn spawn_helper( - client: crate::Client, - state: Arc, - mut f: Box) + Send>, -) -> io::Result { - let event = unsafe { - let r = CreateEventA(ptr::null_mut(), TRUE, FALSE, ptr::null()); - if r.is_null() { - return Err(io::Error::last_os_error()); - } else { - Handle(r) - } - }; - let event = Arc::new(event); - let event2 = event.clone(); - let thread = Builder::new().spawn(move || { - let objects = [event2.0, client.inner.sem.0]; - state.for_each_request(|_| { - const WAIT_OBJECT_1: u32 = WAIT_OBJECT_0 + 1; - match unsafe { WaitForMultipleObjects(2, objects.as_ptr(), FALSE, INFINITE) } { - WAIT_OBJECT_0 => return, - WAIT_OBJECT_1 => f(Ok(crate::Acquired { - client: client.inner.clone(), - data: Acquired, - disabled: false, - })), - _ => f(Err(io::Error::last_os_error())), - } - }); - })?; - Ok(Helper { thread, event }) -} - -impl Helper { - pub fn join(self) { - // Unlike unix this logic is much easier. If our thread was blocked - // in waiting for requests it should already be woken up and - // exiting. Otherwise it's waiting for a token, so we wake it up - // with a different event that it's also waiting on here. After - // these two we should be guaranteed the thread is on its way out, - // so we can safely `join`. - let r = unsafe { SetEvent(self.event.0) }; - if r == 0 { - panic!("failed to set event: {}", io::Error::last_os_error()); - } - drop(self.thread.join()); - } -} +use crate::FromEnvErrorInner; +use std::ffi::CString; +use std::io; +use std::process::Command; +use std::ptr; +use std::sync::Arc; +use std::thread::{Builder, JoinHandle}; + +#[derive(Debug)] +pub struct Client { + sem: Handle, + name: String, +} + +#[derive(Debug)] +pub struct Acquired; + +type BOOL = i32; +type DWORD = u32; +type HANDLE = *mut u8; +type LONG = i32; + +const ERROR_ALREADY_EXISTS: DWORD = 183; +const FALSE: BOOL = 0; +const INFINITE: DWORD = 0xffffffff; +const SEMAPHORE_MODIFY_STATE: DWORD = 0x2; +const SYNCHRONIZE: DWORD = 0x00100000; +const TRUE: BOOL = 1; +const WAIT_OBJECT_0: DWORD = 0; + +extern "system" { + fn CloseHandle(handle: HANDLE) -> BOOL; + fn SetEvent(hEvent: HANDLE) -> BOOL; + fn WaitForMultipleObjects( + ncount: DWORD, + lpHandles: *const HANDLE, + bWaitAll: BOOL, + dwMilliseconds: DWORD, + ) -> DWORD; + fn CreateEventA( + lpEventAttributes: *mut u8, + bManualReset: BOOL, + bInitialState: BOOL, + lpName: *const i8, + ) -> HANDLE; + fn ReleaseSemaphore( + hSemaphore: HANDLE, + lReleaseCount: LONG, + lpPreviousCount: *mut LONG, + ) -> BOOL; + fn CreateSemaphoreA( + lpEventAttributes: *mut u8, + lInitialCount: LONG, + lMaximumCount: LONG, + lpName: *const i8, + ) -> HANDLE; + fn OpenSemaphoreA(dwDesiredAccess: DWORD, bInheritHandle: BOOL, lpName: *const i8) -> HANDLE; + fn WaitForSingleObject(hHandle: HANDLE, dwMilliseconds: DWORD) -> DWORD; + #[link_name = "SystemFunction036"] + fn RtlGenRandom(RandomBuffer: *mut u8, RandomBufferLength: u32) -> u8; +} + +// Note that we ideally would use the `getrandom` crate, but unfortunately +// that causes build issues when this crate is used in rust-lang/rust (see +// rust-lang/rust#65014 for more information). As a result we just inline +// the pretty simple Windows-specific implementation of generating +// randomness. +fn getrandom(dest: &mut [u8]) -> io::Result<()> { + // Prevent overflow of u32 + for chunk in dest.chunks_mut(u32::max_value() as usize) { + let ret = unsafe { RtlGenRandom(chunk.as_mut_ptr(), chunk.len() as u32) }; + if ret == 0 { + return Err(io::Error::new( + io::ErrorKind::Other, + "failed to generate random bytes", + )); + } + } + Ok(()) +} + +impl Client { + pub fn new(limit: usize) -> io::Result { + // Try a bunch of random semaphore names until we get a unique one, + // but don't try for too long. + // + // Note that `limit == 0` is a valid argument above but Windows + // won't let us create a semaphore with 0 slots available to it. Get + // `limit == 0` working by creating a semaphore instead with one + // slot and then immediately acquire it (without ever releaseing it + // back). + for _ in 0..100 { + let mut bytes = [0; 4]; + getrandom(&mut bytes)?; + let mut name = format!("__rust_jobserver_semaphore_{}\0", u32::from_ne_bytes(bytes)); + unsafe { + let create_limit = if limit == 0 { 1 } else { limit }; + let r = CreateSemaphoreA( + ptr::null_mut(), + create_limit as LONG, + create_limit as LONG, + name.as_ptr() as *const _, + ); + if r.is_null() { + return Err(io::Error::last_os_error()); + } + let handle = Handle(r); + + let err = io::Error::last_os_error(); + if err.raw_os_error() == Some(ERROR_ALREADY_EXISTS as i32) { + continue; + } + name.pop(); // chop off the trailing nul + let client = Client { + sem: handle, + name: name, + }; + if create_limit != limit { + client.acquire()?; + } + return Ok(client); + } + } + + Err(io::Error::new( + io::ErrorKind::Other, + "failed to find a unique name for a semaphore", + )) + } + + pub(crate) unsafe fn open(s: &str, _check_pipe: bool) -> Result { + let name = match CString::new(s) { + Ok(s) => s, + Err(e) => return Err(FromEnvErrorInner::CannotParse(e.to_string())), + }; + + let sem = OpenSemaphoreA(SYNCHRONIZE | SEMAPHORE_MODIFY_STATE, FALSE, name.as_ptr()); + if sem.is_null() { + Err(FromEnvErrorInner::CannotOpenPath( + s.to_string(), + io::Error::last_os_error(), + )) + } else { + Ok(Client { + sem: Handle(sem), + name: s.to_string(), + }) + } + } + + pub fn acquire(&self) -> io::Result { + unsafe { + let r = WaitForSingleObject(self.sem.0, INFINITE); + if r == WAIT_OBJECT_0 { + Ok(Acquired) + } else { + Err(io::Error::last_os_error()) + } + } + } + + pub fn release(&self, _data: Option<&Acquired>) -> io::Result<()> { + unsafe { + let r = ReleaseSemaphore(self.sem.0, 1, ptr::null_mut()); + if r != 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } + } + } + + pub fn string_arg(&self) -> String { + self.name.clone() + } + + pub fn available(&self) -> io::Result { + // Can't read value of a semaphore on Windows, so + // try to acquire without sleeping, since we can find out the + // old value on release. If acquisiton fails, then available is 0. + unsafe { + let r = WaitForSingleObject(self.sem.0, 0); + if r != WAIT_OBJECT_0 { + Ok(0) + } else { + let mut prev: LONG = 0; + let r = ReleaseSemaphore(self.sem.0, 1, &mut prev); + if r != 0 { + Ok(prev as usize + 1) + } else { + Err(io::Error::last_os_error()) + } + } + } + } + + pub fn configure(&self, _cmd: &mut Command) { + // nothing to do here, we gave the name of our semaphore to the + // child above + } +} + +#[derive(Debug)] +struct Handle(HANDLE); +// HANDLE is a raw ptr, but we're send/sync +unsafe impl Sync for Handle {} +unsafe impl Send for Handle {} + +impl Drop for Handle { + fn drop(&mut self) { + unsafe { + CloseHandle(self.0); + } + } +} + +#[derive(Debug)] +pub struct Helper { + event: Arc, + thread: JoinHandle<()>, +} + +pub(crate) fn spawn_helper( + client: crate::Client, + state: Arc, + mut f: Box) + Send>, +) -> io::Result { + let event = unsafe { + let r = CreateEventA(ptr::null_mut(), TRUE, FALSE, ptr::null()); + if r.is_null() { + return Err(io::Error::last_os_error()); + } else { + Handle(r) + } + }; + let event = Arc::new(event); + let event2 = event.clone(); + let thread = Builder::new().spawn(move || { + let objects = [event2.0, client.inner.sem.0]; + state.for_each_request(|_| { + const WAIT_OBJECT_1: u32 = WAIT_OBJECT_0 + 1; + match unsafe { WaitForMultipleObjects(2, objects.as_ptr(), FALSE, INFINITE) } { + WAIT_OBJECT_0 => return, + WAIT_OBJECT_1 => f(Ok(crate::Acquired { + client: client.inner.clone(), + data: Acquired, + disabled: false, + })), + _ => f(Err(io::Error::last_os_error())), + } + }); + })?; + Ok(Helper { thread, event }) +} + +impl Helper { + pub fn join(self) { + // Unlike unix this logic is much easier. If our thread was blocked + // in waiting for requests it should already be woken up and + // exiting. Otherwise it's waiting for a token, so we wake it up + // with a different event that it's also waiting on here. After + // these two we should be guaranteed the thread is on its way out, + // so we can safely `join`. + let r = unsafe { SetEvent(self.event.0) }; + if r == 0 { + panic!("failed to set event: {}", io::Error::last_os_error()); + } + drop(self.thread.join()); + } +} -- cgit v1.2.3