diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 19:33:14 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 19:33:14 +0000 |
commit | 36d22d82aa202bb199967e9512281e9a53db42c9 (patch) | |
tree | 105e8c98ddea1c1e4784a60a5a6410fa416be2de /third_party/rust/jobserver | |
parent | Initial commit. (diff) | |
download | firefox-esr-36d22d82aa202bb199967e9512281e9a53db42c9.tar.xz firefox-esr-36d22d82aa202bb199967e9512281e9a53db42c9.zip |
Adding upstream version 115.7.0esr.upstream/115.7.0esr
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'third_party/rust/jobserver')
-rw-r--r-- | third_party/rust/jobserver/.cargo-checksum.json | 1 | ||||
-rw-r--r-- | third_party/rust/jobserver/Cargo.toml | 65 | ||||
-rw-r--r-- | third_party/rust/jobserver/LICENSE-APACHE | 201 | ||||
-rw-r--r-- | third_party/rust/jobserver/LICENSE-MIT | 25 | ||||
-rw-r--r-- | third_party/rust/jobserver/README.md | 39 | ||||
-rw-r--r-- | third_party/rust/jobserver/src/lib.rs | 541 | ||||
-rw-r--r-- | third_party/rust/jobserver/src/unix.rs | 368 | ||||
-rw-r--r-- | third_party/rust/jobserver/src/wasm.rs | 95 | ||||
-rw-r--r-- | third_party/rust/jobserver/src/windows.rs | 266 | ||||
-rw-r--r-- | third_party/rust/jobserver/tests/client-of-myself.rs | 59 | ||||
-rw-r--r-- | third_party/rust/jobserver/tests/client.rs | 198 | ||||
-rw-r--r-- | third_party/rust/jobserver/tests/helper.rs | 77 | ||||
-rw-r--r-- | third_party/rust/jobserver/tests/make-as-a-client.rs | 81 | ||||
-rw-r--r-- | third_party/rust/jobserver/tests/server.rs | 181 |
14 files changed, 2197 insertions, 0 deletions
diff --git a/third_party/rust/jobserver/.cargo-checksum.json b/third_party/rust/jobserver/.cargo-checksum.json new file mode 100644 index 0000000000..7ee2166816 --- /dev/null +++ b/third_party/rust/jobserver/.cargo-checksum.json @@ -0,0 +1 @@ +{"files":{"Cargo.toml":"8bdb529d87cf3cd9eed1909a2f32216ea34caa2f30ac552f57cb29a1bdad5568","LICENSE-APACHE":"a60eea817514531668d7e00765731449fe14d059d3249e0bc93b36de45f759f2","LICENSE-MIT":"378f5840b258e2779c39418f3f2d7b2ba96f1c7917dd6be0713f88305dbda397","README.md":"07d8d79f8f6b6a94321fe8db78d26ed409de47cee49290947bd6bbfa29d05e9c","src/lib.rs":"4e841d58f95ea5f5cd4f975d0ddefaeef4dc364717cda16d6f0b908850b31ab8","src/unix.rs":"68b089c8c96ee0f8ac867db66520f0bd05cb6ed818f9f3c06aa1a079c9379aec","src/wasm.rs":"65d3d8ed45972b4459581505906481d32a50d2f7514cd7ff2a595fceeaa672f0","src/windows.rs":"8e0fa3ab29757d809d4fa03c8101870435ce8c4ceaebe491df3144d62fe0aaaf","tests/client-of-myself.rs":"ca09bf398f69df4bac1730999e954dbbc3faf3c6512678c136e0938e7e9cd0ab","tests/client.rs":"d4745cdd650c86d19bc81f6c9b35df498996deffb86ae6412ad040af96a19183","tests/helper.rs":"c0e6c00eaf849295d8ec23e374690b6645c0f7d993e91abf7ad53ac960f71762","tests/make-as-a-client.rs":"8be1f3fef1e9e65c7904dbaa04364bf0f44e9deab84a2a247a5a94b5cf0df9bc","tests/server.rs":"da15bf12e1df1883f660892b996c9e0d92485aace3f7b50ee70c4a8e6deae8da"},"package":"068b1ee6743e4d11fb9c6a1e6064b3693a1b600e7f5f5988047d98b3dc9fb90b"}
\ No newline at end of file diff --git a/third_party/rust/jobserver/Cargo.toml b/third_party/rust/jobserver/Cargo.toml new file mode 100644 index 0000000000..c1d0481e97 --- /dev/null +++ b/third_party/rust/jobserver/Cargo.toml @@ -0,0 +1,65 @@ +# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO +# +# When uploading crates to the registry Cargo will automatically +# "normalize" Cargo.toml files for maximal compatibility +# with all versions of Cargo and also rewrite `path` dependencies +# to registry (e.g., crates.io) dependencies. +# +# If you are reading this file be aware that the original Cargo.toml +# will likely look very different (and much more reasonable). +# See Cargo.toml.orig for the original contents. + +[package] +edition = "2018" +name = "jobserver" +version = "0.1.25" +authors = ["Alex Crichton <alex@alexcrichton.com>"] +description = """ +An implementation of the GNU make jobserver for Rust +""" +homepage = "https://github.com/alexcrichton/jobserver-rs" +documentation = "https://docs.rs/jobserver" +readme = "README.md" +license = "MIT/Apache-2.0" +repository = "https://github.com/alexcrichton/jobserver-rs" + +[[test]] +name = "client" +path = "tests/client.rs" +harness = false + +[[test]] +name = "server" +path = "tests/server.rs" + +[[test]] +name = "client-of-myself" +path = "tests/client-of-myself.rs" +harness = false + +[[test]] +name = "make-as-a-client" +path = "tests/make-as-a-client.rs" +harness = false + +[[test]] +name = "helper" +path = "tests/helper.rs" + +[dev-dependencies.futures] +version = "0.1" + +[dev-dependencies.num_cpus] +version = "1.0" + +[dev-dependencies.tempfile] +version = "3" + +[dev-dependencies.tokio-core] +version = "0.1" + +[dev-dependencies.tokio-process] +version = "0.2" + +[target."cfg(unix)".dependencies.libc] +version = "0.2.50" diff --git a/third_party/rust/jobserver/LICENSE-APACHE b/third_party/rust/jobserver/LICENSE-APACHE new file mode 100644 index 0000000000..16fe87b06e --- /dev/null +++ b/third_party/rust/jobserver/LICENSE-APACHE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + +TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + +1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + +2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + +3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + +4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + +5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + +6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + +7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + +8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + +9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + +END OF TERMS AND CONDITIONS + +APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + +Copyright [yyyy] [name of copyright owner] + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. diff --git a/third_party/rust/jobserver/LICENSE-MIT b/third_party/rust/jobserver/LICENSE-MIT new file mode 100644 index 0000000000..39e0ed6602 --- /dev/null +++ b/third_party/rust/jobserver/LICENSE-MIT @@ -0,0 +1,25 @@ +Copyright (c) 2014 Alex Crichton + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. diff --git a/third_party/rust/jobserver/README.md b/third_party/rust/jobserver/README.md new file mode 100644 index 0000000000..7b06c3eb0a --- /dev/null +++ b/third_party/rust/jobserver/README.md @@ -0,0 +1,39 @@ +# jobserver-rs + +An implementation of the GNU make jobserver for Rust + +[![Crates.io](https://img.shields.io/crates/v/jobserver.svg?maxAge=2592000)](https://crates.io/crates/jobserver) + +[Documentation](https://docs.rs/jobserver) + +## Usage + +First, add this to your `Cargo.toml`: + +```toml +[dependencies] +jobserver = "0.1" +``` + +Next, add this to your crate: + +```rust +extern crate jobserver; +``` + +# License + +This project is licensed under either of + + * Apache License, Version 2.0, ([LICENSE-APACHE](LICENSE-APACHE) or + http://www.apache.org/licenses/LICENSE-2.0) + * MIT license ([LICENSE-MIT](LICENSE-MIT) or + http://opensource.org/licenses/MIT) + +at your option. + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in jobserver-rs by you, as defined in the Apache-2.0 license, shall be +dual licensed as above, without any additional terms or conditions. diff --git a/third_party/rust/jobserver/src/lib.rs b/third_party/rust/jobserver/src/lib.rs new file mode 100644 index 0000000000..6d07884b18 --- /dev/null +++ b/third_party/rust/jobserver/src/lib.rs @@ -0,0 +1,541 @@ +//! 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. +//! +//! 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<imp::Client>, +} + +/// 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<imp::Client>, + 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<HelperInner>, + 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<Client> { + 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<Client> { + 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<Acquired> { + 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<usize> { + 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<F>(self, f: F) -> io::Result<HelperThread> + where + F: FnMut(io::Result<Acquired>) + 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<imp::Helper>, + state: Arc<HelperState>, +} + +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/third_party/rust/jobserver/src/unix.rs b/third_party/rust/jobserver/src/unix.rs new file mode 100644 index 0000000000..2474deefc5 --- /dev/null +++ b/third_party/rust/jobserver/src/unix.rs @@ -0,0 +1,368 @@ +use libc::c_int; + +use std::fs::File; +use std::io::{self, Read, Write}; +use std::mem; +use std::mem::MaybeUninit; +use std::os::unix::prelude::*; +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 struct Client { + read: File, + write: File, +} + +#[derive(Debug)] +pub struct Acquired { + byte: u8, +} + +impl Client { + pub fn new(mut limit: usize) -> io::Result<Client> { + 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]; + + set_nonblocking(client.write.as_raw_fd(), true)?; + + while limit > 0 { + let n = limit.min(BUFFER.len()); + + (&client.write).write_all(&BUFFER[..n])?; + limit -= n; + } + + set_nonblocking(client.write.as_raw_fd(), false)?; + + Ok(client) + } + + unsafe fn mk() -> io::Result<Client> { + 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> { + 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 { + read: File::from_raw_fd(read), + write: File::from_raw_fd(write), + } + } + + pub fn acquire(&self) -> io::Result<Acquired> { + // 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<Option<Acquired>> { + // 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(); + fd.fd = self.read.as_raw_fd(); + fd.events = libc::POLLIN; + loop { + let mut buf = [0]; + match (&self.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 { + format!("{},{}", self.read.as_raw_fd(), self.write.as_raw_fd()) + } + + pub fn available(&self) -> io::Result<usize> { + let mut len = MaybeUninit::<c_int>::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) { + // 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<super::HelperState>, +} + +pub(crate) fn spawn_helper( + client: crate::Client, + state: Arc<super::HelperState>, + mut f: Box<dyn FnMut(io::Result<crate::Acquired>) + Send>, +) -> io::Result<Helper> { + static USR1_INIT: Once = Once::new(); + let mut err = None; + USR1_INIT.call_once(|| unsafe { + let mut new: libc::sigaction = mem::zeroed(); + 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<c_int> { + 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/third_party/rust/jobserver/src/wasm.rs b/third_party/rust/jobserver/src/wasm.rs new file mode 100644 index 0000000000..3793bd67cc --- /dev/null +++ b/third_party/rust/jobserver/src/wasm.rs @@ -0,0 +1,95 @@ +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<Inner>, +} + +#[derive(Debug)] +struct Inner { + count: Mutex<usize>, + cvar: Condvar, +} + +#[derive(Debug)] +pub struct Acquired(()); + +impl Client { + pub fn new(limit: usize) -> io::Result<Client> { + Ok(Client { + inner: Arc::new(Inner { + count: Mutex::new(limit), + cvar: Condvar::new(), + }), + }) + } + + pub unsafe fn open(_s: &str) -> Option<Client> { + None + } + + pub fn acquire(&self) -> io::Result<Acquired> { + 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<usize> { + 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<super::HelperState>, + mut f: Box<dyn FnMut(io::Result<crate::Acquired>) + Send>, +) -> io::Result<Helper> { + 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/third_party/rust/jobserver/src/windows.rs b/third_party/rust/jobserver/src/windows.rs new file mode 100644 index 0000000000..6791efea4f --- /dev/null +++ b/third_party/rust/jobserver/src/windows.rs @@ -0,0 +1,266 @@ +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<Client> { + // 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<Client> { + 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<Acquired> { + 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<usize> { + // 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<Handle>, + thread: JoinHandle<()>, +} + +pub(crate) fn spawn_helper( + client: crate::Client, + state: Arc<super::HelperState>, + mut f: Box<dyn FnMut(io::Result<crate::Acquired>) + Send>, +) -> io::Result<Helper> { + 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()); + } +} diff --git a/third_party/rust/jobserver/tests/client-of-myself.rs b/third_party/rust/jobserver/tests/client-of-myself.rs new file mode 100644 index 0000000000..45d57b0b58 --- /dev/null +++ b/third_party/rust/jobserver/tests/client-of-myself.rs @@ -0,0 +1,59 @@ +use std::env; +use std::io::prelude::*; +use std::io::BufReader; +use std::process::{Command, Stdio}; +use std::sync::mpsc; +use std::thread; + +use jobserver::Client; + +macro_rules! t { + ($e:expr) => { + match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {}", stringify!($e), e), + } + }; +} + +fn main() { + if env::var("I_AM_THE_CLIENT").is_ok() { + client(); + } else { + server(); + } +} + +fn server() { + let me = t!(env::current_exe()); + let client = t!(Client::new(1)); + let mut cmd = Command::new(me); + cmd.env("I_AM_THE_CLIENT", "1").stdout(Stdio::piped()); + client.configure(&mut cmd); + let acq = client.acquire().unwrap(); + let mut child = t!(cmd.spawn()); + let stdout = child.stdout.take().unwrap(); + let (tx, rx) = mpsc::channel(); + let t = thread::spawn(move || { + for line in BufReader::new(stdout).lines() { + tx.send(t!(line)).unwrap(); + } + }); + + for _ in 0..100 { + assert!(rx.try_recv().is_err()); + } + + drop(acq); + assert_eq!(rx.recv().unwrap(), "hello!"); + t.join().unwrap(); + assert!(rx.recv().is_err()); + client.acquire().unwrap(); +} + +fn client() { + let client = unsafe { Client::from_env().unwrap() }; + let acq = client.acquire().unwrap(); + println!("hello!"); + drop(acq); +} diff --git a/third_party/rust/jobserver/tests/client.rs b/third_party/rust/jobserver/tests/client.rs new file mode 100644 index 0000000000..2516b8ccf7 --- /dev/null +++ b/third_party/rust/jobserver/tests/client.rs @@ -0,0 +1,198 @@ +use std::env; +use std::fs::File; +use std::io::Write; +use std::process::Command; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc; +use std::sync::Arc; +use std::thread; + +use futures::future::{self, Future}; +use futures::stream::{self, Stream}; +use jobserver::Client; +use tokio_core::reactor::Core; +use tokio_process::CommandExt; + +macro_rules! t { + ($e:expr) => { + match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {}", stringify!($e), e), + } + }; +} + +struct Test { + name: &'static str, + f: &'static dyn Fn(), + make_args: &'static [&'static str], + rule: &'static dyn Fn(&str) -> String, +} + +const TESTS: &[Test] = &[ + Test { + name: "no j args", + make_args: &[], + rule: &|me| me.to_string(), + f: &|| { + assert!(unsafe { Client::from_env().is_none() }); + }, + }, + Test { + name: "no j args with plus", + make_args: &[], + rule: &|me| format!("+{}", me), + f: &|| { + assert!(unsafe { Client::from_env().is_none() }); + }, + }, + Test { + name: "j args with plus", + make_args: &["-j2"], + rule: &|me| format!("+{}", me), + f: &|| { + assert!(unsafe { Client::from_env().is_some() }); + }, + }, + Test { + name: "acquire", + make_args: &["-j2"], + rule: &|me| format!("+{}", me), + f: &|| { + let c = unsafe { Client::from_env().unwrap() }; + drop(c.acquire().unwrap()); + drop(c.acquire().unwrap()); + }, + }, + Test { + name: "acquire3", + make_args: &["-j3"], + rule: &|me| format!("+{}", me), + f: &|| { + let c = unsafe { Client::from_env().unwrap() }; + let a = c.acquire().unwrap(); + let b = c.acquire().unwrap(); + drop((a, b)); + }, + }, + Test { + name: "acquire blocks", + make_args: &["-j2"], + rule: &|me| format!("+{}", me), + f: &|| { + let c = unsafe { Client::from_env().unwrap() }; + let a = c.acquire().unwrap(); + let hit = Arc::new(AtomicBool::new(false)); + let hit2 = hit.clone(); + let (tx, rx) = mpsc::channel(); + let t = thread::spawn(move || { + tx.send(()).unwrap(); + let _b = c.acquire().unwrap(); + hit2.store(true, Ordering::SeqCst); + }); + rx.recv().unwrap(); + assert!(!hit.load(Ordering::SeqCst)); + drop(a); + t.join().unwrap(); + assert!(hit.load(Ordering::SeqCst)); + }, + }, + Test { + name: "acquire_raw", + make_args: &["-j2"], + rule: &|me| format!("+{}", me), + f: &|| { + let c = unsafe { Client::from_env().unwrap() }; + c.acquire_raw().unwrap(); + c.release_raw().unwrap(); + }, + }, +]; + +fn main() { + if let Ok(test) = env::var("TEST_TO_RUN") { + return (TESTS.iter().find(|t| t.name == test).unwrap().f)(); + } + + let me = t!(env::current_exe()); + let me = me.to_str().unwrap(); + let filter = env::args().nth(1); + + let mut core = t!(Core::new()); + + let futures = TESTS + .iter() + .filter(|test| match filter { + Some(ref s) => test.name.contains(s), + None => true, + }) + .map(|test| { + let td = t!(tempfile::tempdir()); + let makefile = format!( + "\ +all: export TEST_TO_RUN={} +all: +\t{} +", + test.name, + (test.rule)(me) + ); + t!(t!(File::create(td.path().join("Makefile"))).write_all(makefile.as_bytes())); + let prog = env::var("MAKE").unwrap_or_else(|_| "make".to_string()); + let mut cmd = Command::new(prog); + cmd.args(test.make_args); + cmd.current_dir(td.path()); + future::lazy(move || { + cmd.output_async().map(move |e| { + drop(td); + (test, e) + }) + }) + }) + .collect::<Vec<_>>(); + + println!("\nrunning {} tests\n", futures.len()); + + let stream = stream::iter(futures.into_iter().map(Ok)).buffer_unordered(num_cpus::get()); + + let mut failures = Vec::new(); + t!(core.run(stream.for_each(|(test, output)| { + if output.status.success() { + println!("test {} ... ok", test.name); + } else { + println!("test {} ... FAIL", test.name); + failures.push((test, output)); + } + Ok(()) + }))); + + if failures.is_empty() { + println!("\ntest result: ok\n"); + return; + } + + println!("\n----------- failures"); + + for (test, output) in failures { + println!("test {}", test.name); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + + println!("\texit status: {}", output.status); + if !stdout.is_empty() { + println!("\tstdout ==="); + for line in stdout.lines() { + println!("\t\t{}", line); + } + } + + if !stderr.is_empty() { + println!("\tstderr ==="); + for line in stderr.lines() { + println!("\t\t{}", line); + } + } + } + + std::process::exit(4); +} diff --git a/third_party/rust/jobserver/tests/helper.rs b/third_party/rust/jobserver/tests/helper.rs new file mode 100644 index 0000000000..0b3ba88a70 --- /dev/null +++ b/third_party/rust/jobserver/tests/helper.rs @@ -0,0 +1,77 @@ +use jobserver::Client; +use std::sync::atomic::*; +use std::sync::mpsc; +use std::sync::*; + +macro_rules! t { + ($e:expr) => { + match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {}", stringify!($e), e), + } + }; +} + +#[test] +fn helper_smoke() { + let client = t!(Client::new(1)); + drop(client.clone().into_helper_thread(|_| ()).unwrap()); + drop(client.clone().into_helper_thread(|_| ()).unwrap()); + drop(client.clone().into_helper_thread(|_| ()).unwrap()); + drop(client.clone().into_helper_thread(|_| ()).unwrap()); + drop(client.clone().into_helper_thread(|_| ()).unwrap()); + drop(client.into_helper_thread(|_| ()).unwrap()); +} + +#[test] +fn acquire() { + let (tx, rx) = mpsc::channel(); + let client = t!(Client::new(1)); + let helper = client + .into_helper_thread(move |a| drop(tx.send(a))) + .unwrap(); + assert!(rx.try_recv().is_err()); + helper.request_token(); + rx.recv().unwrap().unwrap(); + helper.request_token(); + rx.recv().unwrap().unwrap(); + + helper.request_token(); + helper.request_token(); + rx.recv().unwrap().unwrap(); + rx.recv().unwrap().unwrap(); + + helper.request_token(); + helper.request_token(); + drop(helper); +} + +#[test] +fn prompt_shutdown() { + for _ in 0..100 { + let client = jobserver::Client::new(4).unwrap(); + let count = Arc::new(AtomicU32::new(0)); + let count2 = count.clone(); + let tokens = Arc::new(Mutex::new(Vec::new())); + let helper = client + .into_helper_thread(move |token| { + tokens.lock().unwrap().push(token); + count2.fetch_add(1, Ordering::SeqCst); + }) + .unwrap(); + + // Request more tokens than what are available. + for _ in 0..5 { + helper.request_token(); + } + // Wait for at least some of the requests to finish. + while count.load(Ordering::SeqCst) < 3 { + std::thread::yield_now(); + } + // Drop helper + let t = std::time::Instant::now(); + drop(helper); + let d = t.elapsed(); + assert!(d.as_secs_f64() < 0.5); + } +} diff --git a/third_party/rust/jobserver/tests/make-as-a-client.rs b/third_party/rust/jobserver/tests/make-as-a-client.rs new file mode 100644 index 0000000000..4faac5b880 --- /dev/null +++ b/third_party/rust/jobserver/tests/make-as-a-client.rs @@ -0,0 +1,81 @@ +use std::env; +use std::fs::File; +use std::io::prelude::*; +use std::net::{TcpListener, TcpStream}; +use std::process::Command; + +use jobserver::Client; + +macro_rules! t { + ($e:expr) => { + match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {}", stringify!($e), e), + } + }; +} + +fn main() { + if env::var("_DO_THE_TEST").is_ok() { + std::process::exit( + Command::new(env::var_os("MAKE").unwrap()) + .env("MAKEFLAGS", env::var_os("CARGO_MAKEFLAGS").unwrap()) + .env_remove("_DO_THE_TEST") + .args(&env::args_os().skip(1).collect::<Vec<_>>()) + .status() + .unwrap() + .code() + .unwrap_or(1), + ); + } + + if let Ok(s) = env::var("TEST_ADDR") { + let mut contents = Vec::new(); + t!(t!(TcpStream::connect(&s)).read_to_end(&mut contents)); + return; + } + + let c = t!(Client::new(1)); + let td = tempfile::tempdir().unwrap(); + + let prog = env::var("MAKE").unwrap_or_else(|_| "make".to_string()); + + let me = t!(env::current_exe()); + let me = me.to_str().unwrap(); + + let mut cmd = Command::new(&me); + cmd.current_dir(td.path()); + cmd.env("MAKE", prog); + cmd.env("_DO_THE_TEST", "1"); + + t!(t!(File::create(td.path().join("Makefile"))).write_all( + format!( + "\ +all: foo bar +foo: +\t{0} +bar: +\t{0} +", + me + ) + .as_bytes() + )); + + // We're leaking one extra token to `make` sort of violating the makefile + // jobserver protocol. It has the desired effect though. + c.configure(&mut cmd); + + let listener = t!(TcpListener::bind("127.0.0.1:0")); + let addr = t!(listener.local_addr()); + cmd.env("TEST_ADDR", addr.to_string()); + let mut child = t!(cmd.spawn()); + + // We should get both connections as the two programs should be run + // concurrently. + let a = t!(listener.accept()); + let b = t!(listener.accept()); + drop((a, b)); + + assert!(t!(child.wait()).success()); +} diff --git a/third_party/rust/jobserver/tests/server.rs b/third_party/rust/jobserver/tests/server.rs new file mode 100644 index 0000000000..70ea218fc7 --- /dev/null +++ b/third_party/rust/jobserver/tests/server.rs @@ -0,0 +1,181 @@ +use std::env; +use std::fs::File; +use std::io::prelude::*; +use std::process::Command; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc; +use std::sync::Arc; +use std::thread; + +use jobserver::Client; + +macro_rules! t { + ($e:expr) => { + match $e { + Ok(e) => e, + Err(e) => panic!("{} failed with {}", stringify!($e), e), + } + }; +} + +#[test] +fn server_smoke() { + let c = t!(Client::new(1)); + drop(c.acquire().unwrap()); + drop(c.acquire().unwrap()); +} + +#[test] +fn server_multiple() { + let c = t!(Client::new(2)); + let a = c.acquire().unwrap(); + let b = c.acquire().unwrap(); + drop((a, b)); +} + +#[test] +fn server_available() { + let c = t!(Client::new(10)); + assert_eq!(c.available().unwrap(), 10); + let a = c.acquire().unwrap(); + assert_eq!(c.available().unwrap(), 9); + drop(a); + assert_eq!(c.available().unwrap(), 10); +} + +#[test] +fn server_none_available() { + let c = t!(Client::new(2)); + assert_eq!(c.available().unwrap(), 2); + let a = c.acquire().unwrap(); + assert_eq!(c.available().unwrap(), 1); + let b = c.acquire().unwrap(); + assert_eq!(c.available().unwrap(), 0); + drop(a); + assert_eq!(c.available().unwrap(), 1); + drop(b); + assert_eq!(c.available().unwrap(), 2); +} + +#[test] +fn server_blocks() { + let c = t!(Client::new(1)); + let a = c.acquire().unwrap(); + let hit = Arc::new(AtomicBool::new(false)); + let hit2 = hit.clone(); + let (tx, rx) = mpsc::channel(); + let t = thread::spawn(move || { + tx.send(()).unwrap(); + let _b = c.acquire().unwrap(); + hit2.store(true, Ordering::SeqCst); + }); + rx.recv().unwrap(); + assert!(!hit.load(Ordering::SeqCst)); + drop(a); + t.join().unwrap(); + assert!(hit.load(Ordering::SeqCst)); +} + +#[test] +fn make_as_a_single_thread_client() { + let c = t!(Client::new(1)); + let td = tempfile::tempdir().unwrap(); + + let prog = env::var("MAKE").unwrap_or_else(|_| "make".to_string()); + let mut cmd = Command::new(prog); + cmd.current_dir(td.path()); + + t!(t!(File::create(td.path().join("Makefile"))).write_all( + b" +all: foo bar +foo: +\techo foo +bar: +\techo bar +" + )); + + // The jobserver protocol means that the `make` process itself "runs with a + // token", so we acquire our one token to drain the jobserver, and this + // should mean that `make` itself never has a second token available to it. + let _a = c.acquire(); + c.configure(&mut cmd); + let output = t!(cmd.output()); + println!( + "\n\t=== stderr\n\t\t{}", + String::from_utf8_lossy(&output.stderr).replace("\n", "\n\t\t") + ); + println!( + "\t=== stdout\n\t\t{}", + String::from_utf8_lossy(&output.stdout).replace("\n", "\n\t\t") + ); + + assert!(output.status.success()); + assert!(output.stderr.is_empty()); + + let stdout = String::from_utf8_lossy(&output.stdout).replace("\r\n", "\n"); + let a = "\ +echo foo +foo +echo bar +bar +"; + let b = "\ +echo bar +bar +echo foo +foo +"; + + assert!(stdout == a || stdout == b); +} + +#[test] +fn make_as_a_multi_thread_client() { + let c = t!(Client::new(1)); + let td = tempfile::tempdir().unwrap(); + + let prog = env::var("MAKE").unwrap_or_else(|_| "make".to_string()); + let mut cmd = Command::new(prog); + cmd.current_dir(td.path()); + + t!(t!(File::create(td.path().join("Makefile"))).write_all( + b" +all: foo bar +foo: +\techo foo +bar: +\techo bar +" + )); + + // We're leaking one extra token to `make` sort of violating the makefile + // jobserver protocol. It has the desired effect though. + c.configure(&mut cmd); + let output = t!(cmd.output()); + println!( + "\n\t=== stderr\n\t\t{}", + String::from_utf8_lossy(&output.stderr).replace("\n", "\n\t\t") + ); + println!( + "\t=== stdout\n\t\t{}", + String::from_utf8_lossy(&output.stdout).replace("\n", "\n\t\t") + ); + + assert!(output.status.success()); +} + +#[test] +fn zero_client() { + let client = t!(Client::new(0)); + let (tx, rx) = mpsc::channel(); + let helper = client + .into_helper_thread(move |a| drop(tx.send(a))) + .unwrap(); + helper.request_token(); + helper.request_token(); + + for _ in 0..1000 { + assert!(rx.try_recv().is_err()); + } +} |