1//! An implementation of the GNU make jobserver.
2//!
3//! This crate is an implementation, in Rust, of the GNU `make` jobserver for
4//! CLI tools that are interoperating with make or otherwise require some form
5//! of parallelism limiting across process boundaries. This was originally
6//! written for usage in Cargo to both (a) work when `cargo` is invoked from
7//! `make` (using `make`'s jobserver) and (b) work when `cargo` invokes build
8//! scripts, exporting a jobserver implementation for `make` processes to
9//! transitively use.
10//!
11//! The jobserver implementation can be found in [detail online][docs] but
12//! basically boils down to a cross-process semaphore. On Unix this is
13//! implemented with the `pipe` syscall and read/write ends of a pipe and on
14//! Windows this is implemented literally with IPC semaphores. Starting from
15//! GNU `make` version 4.4, named pipe becomes the default way in communication
16//! on Unix. This crate also supports that feature in the sense of inheriting
17//! and forwarding the correct environment.
18//!
19//! The jobserver protocol in `make` also dictates when tokens are acquired to
20//! run child work, and clients using this crate should take care to implement
21//! such details to ensure correct interoperation with `make` itself.
22//!
23//! ## Examples
24//!
25//! Connect to a jobserver that was set up by `make` or a different process:
26//!
27//! ```no_run
28//! use jobserver::Client;
29//!
30//! // See API documentation for why this is `unsafe`
31//! let client = match unsafe { Client::from_env() } {
32//! Some(client) => client,
33//! None => panic!("client not configured"),
34//! };
35//! ```
36//!
37//! Acquire and release token from a jobserver:
38//!
39//! ```no_run
40//! use jobserver::Client;
41//!
42//! let client = unsafe { Client::from_env().unwrap() };
43//! let token = client.acquire().unwrap(); // blocks until it is available
44//! drop(token); // releases the token when the work is done
45//! ```
46//!
47//! Create a new jobserver and configure a child process to have access:
48//!
49//! ```
50//! use std::process::Command;
51//! use jobserver::Client;
52//!
53//! let client = Client::new(4).expect("failed to create jobserver");
54//! let mut cmd = Command::new("make");
55//! client.configure(&mut cmd);
56//! ```
57//!
58//! ## Caveats
59//!
60//! This crate makes no attempt to release tokens back to a jobserver on
61//! abnormal exit of a process. If a process which acquires a token is killed
62//! with ctrl-c or some similar signal then tokens will not be released and the
63//! jobserver may be in a corrupt state.
64//!
65//! Note that this is typically ok as ctrl-c means that an entire build process
66//! is being torn down, but it's worth being aware of at least!
67//!
68//! ## Windows caveats
69//!
70//! There appear to be two implementations of `make` on Windows. On MSYS2 one
71//! typically comes as `mingw32-make` and the other as `make` itself. I'm not
72//! personally too familiar with what's going on here, but for jobserver-related
73//! information the `mingw32-make` implementation uses Windows semaphores
74//! whereas the `make` program does not. The `make` program appears to use file
75//! descriptors and I'm not really sure how it works, so this crate is not
76//! compatible with `make` on Windows. It is, however, compatible with
77//! `mingw32-make`.
78//!
79//! [docs]: http://make.mad-scientist.net/papers/jobserver-implementation/
80
81#![deny(missing_docs, missing_debug_implementations)]
82#![doc(html_root_url = "https://docs.rs/jobserver/0.1")]
83
84use std::env;
85use std::ffi::OsString;
86use std::io;
87use std::process::Command;
88use std::sync::{Arc, Condvar, Mutex, MutexGuard};
89
90mod error;
91#[cfg(unix)]
92#[path = "unix.rs"]
93mod imp;
94#[cfg(windows)]
95#[path = "windows.rs"]
96mod imp;
97#[cfg(not(any(unix, windows)))]
98#[path = "wasm.rs"]
99mod imp;
100
101/// A client of a jobserver
102///
103/// This structure is the main type exposed by this library, and is where
104/// interaction to a jobserver is configured through. Clients are either created
105/// from scratch in which case the internal semphore is initialied on the spot,
106/// or a client is created from the environment to connect to a jobserver
107/// already created.
108///
109/// Some usage examples can be found in the crate documentation for using a
110/// client.
111///
112/// Note that a `Client` implements the `Clone` trait, and all instances of a
113/// `Client` refer to the same jobserver instance.
114#[derive(Clone, Debug)]
115pub struct Client {
116 inner: Arc<imp::Client>,
117}
118
119/// An acquired token from a jobserver.
120///
121/// This token will be released back to the jobserver when it is dropped and
122/// otherwise represents the ability to spawn off another thread of work.
123#[derive(Debug)]
124pub struct Acquired {
125 client: Arc<imp::Client>,
126 data: imp::Acquired,
127 disabled: bool,
128}
129
130impl Acquired {
131 /// This drops the `Acquired` token without releasing the associated token.
132 ///
133 /// This is not generally useful, but can be helpful if you do not have the
134 /// ability to store an Acquired token but need to not yet release it.
135 ///
136 /// You'll typically want to follow this up with a call to `release_raw` or
137 /// similar to actually release the token later on.
138 pub fn drop_without_releasing(mut self) {
139 self.disabled = true;
140 }
141}
142
143#[derive(Default, Debug)]
144struct HelperState {
145 lock: Mutex<HelperInner>,
146 cvar: Condvar,
147}
148
149#[derive(Default, Debug)]
150struct HelperInner {
151 requests: usize,
152 producer_done: bool,
153 consumer_done: bool,
154}
155
156use error::FromEnvErrorInner;
157pub use error::{FromEnvError, FromEnvErrorKind};
158
159/// Return type for `from_env_ext` function.
160#[derive(Debug)]
161pub struct FromEnv {
162 /// Result of trying to get jobserver client from env.
163 pub client: Result<Client, FromEnvError>,
164 /// Name and value of the environment variable.
165 /// `None` if no relevant environment variable is found.
166 pub var: Option<(&'static str, OsString)>,
167}
168
169impl FromEnv {
170 fn new_ok(client: Client, var_name: &'static str, var_value: OsString) -> FromEnv {
171 FromEnv {
172 client: Ok(client),
173 var: Some((var_name, var_value)),
174 }
175 }
176 fn new_err(kind: FromEnvErrorInner, var_name: &'static str, var_value: OsString) -> FromEnv {
177 FromEnv {
178 client: Err(FromEnvError { inner: kind }),
179 var: Some((var_name, var_value)),
180 }
181 }
182}
183
184impl Client {
185 /// Creates a new jobserver initialized with the given parallelism limit.
186 ///
187 /// A client to the jobserver created will be returned. This client will
188 /// allow at most `limit` tokens to be acquired from it in parallel. More
189 /// calls to `acquire` will cause the calling thread to block.
190 ///
191 /// Note that the created `Client` is not automatically inherited into
192 /// spawned child processes from this program. Manual usage of the
193 /// `configure` function is required for a child process to have access to a
194 /// job server.
195 ///
196 /// # Examples
197 ///
198 /// ```
199 /// use jobserver::Client;
200 ///
201 /// let client = Client::new(4).expect("failed to create jobserver");
202 /// ```
203 ///
204 /// # Errors
205 ///
206 /// Returns an error if any I/O error happens when attempting to create the
207 /// jobserver client.
208 pub fn new(limit: usize) -> io::Result<Client> {
209 Ok(Client {
210 inner: Arc::new(imp::Client::new(limit)?),
211 })
212 }
213
214 /// Attempts to connect to the jobserver specified in this process's
215 /// environment.
216 ///
217 /// When the a `make` executable calls a child process it will configure the
218 /// environment of the child to ensure that it has handles to the jobserver
219 /// it's passing down. This function will attempt to look for these details
220 /// and connect to the jobserver.
221 ///
222 /// Note that the created `Client` is not automatically inherited into
223 /// spawned child processes from this program. Manual usage of the
224 /// `configure` function is required for a child process to have access to a
225 /// job server.
226 ///
227 /// # Return value
228 ///
229 /// `FromEnv` contains result and relevant environment variable.
230 /// If a jobserver was found in the environment and it looks correct then
231 /// result with the connected client will be returned. In other cases
232 /// result will contain `Err(FromEnvErr)`.
233 ///
234 /// Note that on Unix the `Client` returned **takes ownership of the file
235 /// descriptors specified in the environment**. Jobservers on Unix are
236 /// implemented with `pipe` file descriptors, and they're inherited from
237 /// parent processes. This `Client` returned takes ownership of the file
238 /// descriptors for this process and will close the file descriptors after
239 /// this value is dropped.
240 ///
241 /// Additionally on Unix this function will configure the file descriptors
242 /// with `CLOEXEC` so they're not automatically inherited by spawned
243 /// children.
244 ///
245 /// On unix if `check_pipe` enabled this function will check if provided
246 /// files are actually pipes.
247 ///
248 /// # Safety
249 ///
250 /// This function is `unsafe` to call on Unix specifically as it
251 /// transitively requires usage of the `from_raw_fd` function, which is
252 /// itself unsafe in some circumstances.
253 ///
254 /// It's recommended to call this function very early in the lifetime of a
255 /// program before any other file descriptors are opened. That way you can
256 /// make sure to take ownership properly of the file descriptors passed
257 /// down, if any.
258 ///
259 /// It's generally unsafe to call this function twice in a program if the
260 /// previous invocation returned `Some`.
261 ///
262 /// Note, though, that on Windows it should be safe to call this function
263 /// any number of times.
264 pub unsafe fn from_env_ext(check_pipe: bool) -> FromEnv {
265 let (env, var_os) = match ["CARGO_MAKEFLAGS", "MAKEFLAGS", "MFLAGS"]
266 .iter()
267 .map(|&env| env::var_os(env).map(|var| (env, var)))
268 .find_map(|p| p)
269 {
270 Some((env, var_os)) => (env, var_os),
271 None => return FromEnv::new_err(FromEnvErrorInner::NoEnvVar, "", Default::default()),
272 };
273
274 let var = match var_os.to_str() {
275 Some(var) => var,
276 None => {
277 let err = FromEnvErrorInner::CannotParse("not valid UTF-8".to_string());
278 return FromEnv::new_err(err, env, var_os);
279 }
280 };
281
282 let s = match find_jobserver_auth(var) {
283 Some(s) => s,
284 None => return FromEnv::new_err(FromEnvErrorInner::NoJobserver, env, var_os),
285 };
286 match imp::Client::open(s, check_pipe) {
287 Ok(c) => FromEnv::new_ok(Client { inner: Arc::new(c) }, env, var_os),
288 Err(err) => FromEnv::new_err(err, env, var_os),
289 }
290 }
291
292 /// Attempts to connect to the jobserver specified in this process's
293 /// environment.
294 ///
295 /// Wraps `from_env_ext` and discards error details.
296 pub unsafe fn from_env() -> Option<Client> {
297 Self::from_env_ext(false).client.ok()
298 }
299
300 /// Acquires a token from this jobserver client.
301 ///
302 /// This function will block the calling thread until a new token can be
303 /// acquired from the jobserver.
304 ///
305 /// # Return value
306 ///
307 /// On successful acquisition of a token an instance of `Acquired` is
308 /// returned. This structure, when dropped, will release the token back to
309 /// the jobserver. It's recommended to avoid leaking this value.
310 ///
311 /// # Errors
312 ///
313 /// If an I/O error happens while acquiring a token then this function will
314 /// return immediately with the error. If an error is returned then a token
315 /// was not acquired.
316 pub fn acquire(&self) -> io::Result<Acquired> {
317 let data = self.inner.acquire()?;
318 Ok(Acquired {
319 client: self.inner.clone(),
320 data,
321 disabled: false,
322 })
323 }
324
325 /// Returns amount of tokens in the read-side pipe.
326 ///
327 /// # Return value
328 ///
329 /// Number of bytes available to be read from the jobserver pipe
330 ///
331 /// # Errors
332 ///
333 /// Underlying errors from the ioctl will be passed up.
334 pub fn available(&self) -> io::Result<usize> {
335 self.inner.available()
336 }
337
338 /// Configures a child process to have access to this client's jobserver as
339 /// well.
340 ///
341 /// This function is required to be called to ensure that a jobserver is
342 /// properly inherited to a child process. If this function is *not* called
343 /// then this `Client` will not be accessible in the child process. In other
344 /// words, if not called, then `Client::from_env` will return `None` in the
345 /// child process (or the equivalent of `Child::from_env` that `make` uses).
346 ///
347 /// ## Platform-specific behavior
348 ///
349 /// On Unix and Windows this will clobber the `CARGO_MAKEFLAGS` environment
350 /// variables for the child process, and on Unix this will also allow the
351 /// two file descriptors for this client to be inherited to the child.
352 ///
353 /// On platforms other than Unix and Windows this panics.
354 pub fn configure(&self, cmd: &mut Command) {
355 cmd.env("CARGO_MAKEFLAGS", &self.mflags_env());
356 self.inner.configure(cmd);
357 }
358
359 /// Configures a child process to have access to this client's jobserver as
360 /// well.
361 ///
362 /// This function is required to be called to ensure that a jobserver is
363 /// properly inherited to a child process. If this function is *not* called
364 /// then this `Client` will not be accessible in the child process. In other
365 /// words, if not called, then `Client::from_env` will return `None` in the
366 /// child process (or the equivalent of `Child::from_env` that `make` uses).
367 ///
368 /// ## Platform-specific behavior
369 ///
370 /// On Unix and Windows this will clobber the `CARGO_MAKEFLAGS`,
371 /// `MAKEFLAGS` and `MFLAGS` environment variables for the child process,
372 /// and on Unix this will also allow the two file descriptors for
373 /// this client to be inherited to the child.
374 ///
375 /// On platforms other than Unix and Windows this panics.
376 pub fn configure_make(&self, cmd: &mut Command) {
377 let value = self.mflags_env();
378 cmd.env("CARGO_MAKEFLAGS", &value);
379 cmd.env("MAKEFLAGS", &value);
380 cmd.env("MFLAGS", &value);
381 self.inner.configure(cmd);
382 }
383
384 fn mflags_env(&self) -> String {
385 let arg = self.inner.string_arg();
386 // Older implementations of make use `--jobserver-fds` and newer
387 // implementations use `--jobserver-auth`, pass both to try to catch
388 // both implementations.
389 format!("-j --jobserver-fds={0} --jobserver-auth={0}", arg)
390 }
391
392 /// Converts this `Client` into a helper thread to deal with a blocking
393 /// `acquire` function a little more easily.
394 ///
395 /// The fact that the `acquire` function on `Client` blocks isn't always
396 /// the easiest to work with. Typically you're using a jobserver to
397 /// manage running other events in parallel! This means that you need to
398 /// either (a) wait for an existing job to finish or (b) wait for a
399 /// new token to become available.
400 ///
401 /// Unfortunately the blocking in `acquire` happens at the implementation
402 /// layer of jobservers. On Unix this requires a blocking call to `read`
403 /// and on Windows this requires one of the `WaitFor*` functions. Both
404 /// of these situations aren't the easiest to deal with:
405 ///
406 /// * On Unix there's basically only one way to wake up a `read` early, and
407 /// that's through a signal. This is what the `make` implementation
408 /// itself uses, relying on `SIGCHLD` to wake up a blocking acquisition
409 /// of a new job token. Unfortunately nonblocking I/O is not an option
410 /// here, so it means that "waiting for one of two events" means that
411 /// the latter event must generate a signal! This is not always the case
412 /// on unix for all jobservers.
413 ///
414 /// * On Windows you'd have to basically use the `WaitForMultipleObjects`
415 /// which means that you've got to canonicalize all your event sources
416 /// into a `HANDLE` which also isn't the easiest thing to do
417 /// unfortunately.
418 ///
419 /// This function essentially attempts to ease these limitations by
420 /// converting this `Client` into a helper thread spawned into this
421 /// process. The application can then request that the helper thread
422 /// acquires tokens and the provided closure will be invoked for each token
423 /// acquired.
424 ///
425 /// The intention is that this function can be used to translate the event
426 /// of a token acquisition into an arbitrary user-defined event.
427 ///
428 /// # Arguments
429 ///
430 /// This function will consume the `Client` provided to be transferred to
431 /// the helper thread that is spawned. Additionally a closure `f` is
432 /// provided to be invoked whenever a token is acquired.
433 ///
434 /// This closure is only invoked after calls to
435 /// `HelperThread::request_token` have been made and a token itself has
436 /// been acquired. If an error happens while acquiring the token then
437 /// an error will be yielded to the closure as well.
438 ///
439 /// # Return Value
440 ///
441 /// This function will return an instance of the `HelperThread` structure
442 /// which is used to manage the helper thread associated with this client.
443 /// Through the `HelperThread` you'll request that tokens are acquired.
444 /// When acquired, the closure provided here is invoked.
445 ///
446 /// When the `HelperThread` structure is returned it will be gracefully
447 /// torn down, and the calling thread will be blocked until the thread is
448 /// torn down (which should be prompt).
449 ///
450 /// # Errors
451 ///
452 /// This function may fail due to creation of the helper thread or
453 /// auxiliary I/O objects to manage the helper thread. In any of these
454 /// situations the error is propagated upwards.
455 ///
456 /// # Platform-specific behavior
457 ///
458 /// On Windows this function behaves pretty normally as expected, but on
459 /// Unix the implementation is... a little heinous. As mentioned above
460 /// we're forced into blocking I/O for token acquisition, namely a blocking
461 /// call to `read`. We must be able to unblock this, however, to tear down
462 /// the helper thread gracefully!
463 ///
464 /// Essentially what happens is that we'll send a signal to the helper
465 /// thread spawned and rely on `EINTR` being returned to wake up the helper
466 /// thread. This involves installing a global `SIGUSR1` handler that does
467 /// nothing along with sending signals to that thread. This may cause
468 /// odd behavior in some applications, so it's recommended to review and
469 /// test thoroughly before using this.
470 pub fn into_helper_thread<F>(self, f: F) -> io::Result<HelperThread>
471 where
472 F: FnMut(io::Result<Acquired>) + Send + 'static,
473 {
474 let state = Arc::new(HelperState::default());
475 Ok(HelperThread {
476 inner: Some(imp::spawn_helper(self, state.clone(), Box::new(f))?),
477 state,
478 })
479 }
480
481 /// Blocks the current thread until a token is acquired.
482 ///
483 /// This is the same as `acquire`, except that it doesn't return an RAII
484 /// helper. If successful the process will need to guarantee that
485 /// `release_raw` is called in the future.
486 pub fn acquire_raw(&self) -> io::Result<()> {
487 self.inner.acquire()?;
488 Ok(())
489 }
490
491 /// Releases a jobserver token back to the original jobserver.
492 ///
493 /// This is intended to be paired with `acquire_raw` if it was called, but
494 /// in some situations it could also be called to relinquish a process's
495 /// implicit token temporarily which is then re-acquired later.
496 pub fn release_raw(&self) -> io::Result<()> {
497 self.inner.release(None)?;
498 Ok(())
499 }
500}
501
502impl Drop for Acquired {
503 fn drop(&mut self) {
504 if !self.disabled {
505 drop(self.client.release(data:Some(&self.data)));
506 }
507 }
508}
509
510/// Structure returned from `Client::into_helper_thread` to manage the lifetime
511/// of the helper thread returned, see those associated docs for more info.
512#[derive(Debug)]
513pub struct HelperThread {
514 inner: Option<imp::Helper>,
515 state: Arc<HelperState>,
516}
517
518impl HelperThread {
519 /// Request that the helper thread acquires a token, eventually calling the
520 /// original closure with a token when it's available.
521 ///
522 /// For more information, see the docs on that function.
523 pub fn request_token(&self) {
524 // Indicate that there's one more request for a token and then wake up
525 // the helper thread if it's sleeping.
526 self.state.lock().requests += 1;
527 self.state.cvar.notify_one();
528 }
529}
530
531impl Drop for HelperThread {
532 fn drop(&mut self) {
533 // Flag that the producer half is done so the helper thread should exit
534 // quickly if it's waiting. Wake it up if it's actually waiting
535 self.state.lock().producer_done = true;
536 self.state.cvar.notify_one();
537
538 // ... and afterwards perform any thread cleanup logic
539 self.inner.take().unwrap().join();
540 }
541}
542
543impl HelperState {
544 fn lock(&self) -> MutexGuard<'_, HelperInner> {
545 self.lock.lock().unwrap_or_else(|e| e.into_inner())
546 }
547
548 /// Executes `f` for each request for a token, where `f` is expected to
549 /// block and then provide the original closure with a token once it's
550 /// acquired.
551 ///
552 /// This is an infinite loop until the helper thread is dropped, at which
553 /// point everything should get interrupted.
554 fn for_each_request(&self, mut f: impl FnMut(&HelperState)) {
555 let mut lock = self.lock();
556
557 // We only execute while we could receive requests, but as soon as
558 // that's `false` we're out of here.
559 while !lock.producer_done {
560 // If no one's requested a token then we wait for someone to
561 // request a token.
562 if lock.requests == 0 {
563 lock = self.cvar.wait(lock).unwrap_or_else(|e| e.into_inner());
564 continue;
565 }
566
567 // Consume the request for a token, and then actually acquire a
568 // token after unlocking our lock (not that acquisition happens in
569 // `f`). This ensures that we don't actually hold the lock if we
570 // wait for a long time for a token.
571 lock.requests -= 1;
572 drop(lock);
573 f(self);
574 lock = self.lock();
575 }
576 lock.consumer_done = true;
577 self.cvar.notify_one();
578 }
579
580 fn producer_done(&self) -> bool {
581 self.lock().producer_done
582 }
583}
584
585/// Finds and returns the value of `--jobserver-auth=<VALUE>` in the given
586/// environment variable.
587///
588/// Precedence rules:
589///
590/// * The last instance wins [^1].
591/// * `--jobserver-fds=` as a fallback when no `--jobserver-auth=` is present [^2].
592///
593/// [^1]: See ["GNU `make` manual: Sharing Job Slots with GNU `make`"](https://www.gnu.org/software/make/manual/make.html#Job-Slots)
594/// _"Be aware that the `MAKEFLAGS` variable may contain multiple instances of
595/// the `--jobserver-auth=` option. Only the last instance is relevant."_
596///
597/// [^2]: Refer to [the release announcement](https://git.savannah.gnu.org/cgit/make.git/tree/NEWS?h=4.2#n31)
598/// of GNU Make 4.2, which states that `--jobserver-fds` was initially an
599/// internal-only flag and was later renamed to `--jobserver-auth`.
600fn find_jobserver_auth(var: &str) -> Option<&str> {
601 ["--jobserver-auth=", "--jobserver-fds="]
602 .iter()
603 .find_map(|&arg: &str| var.rsplit_once(delimiter:arg).map(|(_, s: &str)| s))
604 .and_then(|s: &str| s.split(' ').next())
605}
606
607#[test]
608fn no_helper_deadlock() {
609 let x: Client = crate::Client::new(limit:32).unwrap();
610 let _y: Client = x.clone();
611 std::mem::drop(x.into_helper_thread(|_| {}).unwrap());
612}
613
614#[test]
615fn test_find_jobserver_auth() {
616 let cases = [
617 ("", None),
618 ("-j2", None),
619 ("-j2 --jobserver-auth=3,4", Some("3,4")),
620 ("--jobserver-auth=3,4 -j2", Some("3,4")),
621 ("--jobserver-auth=3,4", Some("3,4")),
622 ("--jobserver-auth=fifo:/myfifo", Some("fifo:/myfifo")),
623 ("--jobserver-auth=", Some("")),
624 ("--jobserver-auth", None),
625 ("--jobserver-fds=3,4", Some("3,4")),
626 ("--jobserver-fds=fifo:/myfifo", Some("fifo:/myfifo")),
627 ("--jobserver-fds=", Some("")),
628 ("--jobserver-fds", None),
629 (
630 "--jobserver-auth=auth-a --jobserver-auth=auth-b",
631 Some("auth-b"),
632 ),
633 (
634 "--jobserver-auth=auth-b --jobserver-auth=auth-a",
635 Some("auth-a"),
636 ),
637 ("--jobserver-fds=fds-a --jobserver-fds=fds-b", Some("fds-b")),
638 ("--jobserver-fds=fds-b --jobserver-fds=fds-a", Some("fds-a")),
639 (
640 "--jobserver-auth=auth-a --jobserver-fds=fds-a --jobserver-auth=auth-b",
641 Some("auth-b"),
642 ),
643 (
644 "--jobserver-fds=fds-a --jobserver-auth=auth-a --jobserver-fds=fds-b",
645 Some("auth-a"),
646 ),
647 ];
648 for (var, expected) in cases {
649 let actual = find_jobserver_auth(var);
650 assert_eq!(
651 actual, expected,
652 "expect {expected:?}, got {actual:?}, input `{var:?}`"
653 );
654 }
655}
656