Skip to content

Commit 32aa8e9

Browse files
JonathanWoollett-LightJonathanWoollett-Light
committed
feat: Futex
1 parent ed0e859 commit 32aa8e9

File tree

3 files changed

+387
-0
lines changed

3 files changed

+387
-0
lines changed

‎CHANGELOG.md

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,9 @@ This project adheres to [Semantic Versioning](https://semver.org/).
55

66
## [Unreleased] - ReleaseDate
77
### Added
8+
9+
- Added futex interface.
10+
([#1907](https://github.com/nix-rust/nix/pull/1907))
811
- Add `PF_ROUTE` to `SockType` on macOS, iOS, all of the BSDs, Fuchsia, Haiku, Illumos.
912
([#1867](https://github.com/nix-rust/nix/pull/1867))
1013
- Added `nix::ucontext` module on `aarch64-unknown-linux-gnu`.

‎src/sys/futex.rs

Lines changed: 380 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,380 @@
1+
use crate::{Errno, Result};
2+
use libc::{syscall, SYS_futex};
3+
use std::cell::UnsafeCell;
4+
use std::convert::TryFrom;
5+
use std::os::unix::io::{FromRawFd, OwnedFd};
6+
use std::time::Duration;
7+
8+
fn timespec(duration: Duration) -> libc::timespec {
9+
let tv_sec = duration.as_secs().try_into().unwrap();
10+
let tv_nsec = duration.subsec_nanos().try_into().unwrap();
11+
libc::timespec { tv_sec, tv_nsec }
12+
}
13+
14+
fn unwrap_or_null<T>(option: Option<&T>) -> *const T {
15+
match option {
16+
Some(t) => t,
17+
None => std::ptr::null(),
18+
}
19+
}
20+
21+
/// Fast user-space locking.
22+
///
23+
/// By default we presume the futex is not process-private, that is, it is used across processes. If
24+
/// you know it is process-private you can set `PRIVATE` to `true` which allows some additional
25+
/// optimizations.
26+
/// ```
27+
/// # use nix::{
28+
/// # sys::{futex::Futex, mman::{mmap, MapFlags, ProtFlags}},
29+
/// # errno::Errno,
30+
/// # unistd::{fork,ForkResult},
31+
/// # };
32+
/// # use std::{
33+
/// # time::{Instant, Duration},
34+
/// # num::NonZeroUsize,
35+
/// # mem::{ManuallyDrop, size_of},
36+
/// # os::unix::io::OwnedFd,
37+
/// # sync::Arc,
38+
/// # thread::{spawn, sleep},
39+
/// # };
40+
/// const TIMEOUT: Duration = Duration::from_millis(500);
41+
/// const DELTA: Duration = Duration::from_millis(100);
42+
/// # fn main() -> nix::Result<()> {
43+
/// let futex: Futex = Futex::new(0);
44+
///
45+
/// // If the value of the futex is 0, wait for wake. Since the value is 0 and no wake occurs,
46+
/// // we expect the timeout will pass.
47+
///
48+
/// let instant = Instant::now();
49+
/// assert_eq!(futex.wait(0, Some(TIMEOUT)),Err(Errno::ETIMEDOUT));
50+
/// assert!(instant.elapsed() > TIMEOUT);
51+
///
52+
/// // If the value of the futex is 1, wait for wake. Since the value is 0, not 1, this will
53+
/// // return immediately.
54+
///
55+
/// let instant = Instant::now();
56+
/// assert_eq!(futex.wait(1, Some(TIMEOUT)),Err(Errno::EAGAIN));
57+
/// assert!(instant.elapsed() < DELTA);
58+
///
59+
/// // Test across threads
60+
/// // -------------------------------------------------------------------------
61+
///
62+
/// let futex = Arc::new(futex);
63+
/// let futex_clone = futex.clone();
64+
/// let instant = Instant::now();
65+
/// spawn(move || {
66+
/// sleep(TIMEOUT);
67+
/// assert_eq!(futex_clone.wake(1),Ok(1));
68+
/// });
69+
/// assert_eq!(futex.wait(0, Some(2 * TIMEOUT)), Ok(()));
70+
/// assert!(instant.elapsed() > TIMEOUT && instant.elapsed() < TIMEOUT + DELTA);
71+
///
72+
/// // Test across processes
73+
/// // -------------------------------------------------------------------------
74+
///
75+
/// let shared_memory = unsafe { mmap::<OwnedFd>(
76+
/// None,
77+
/// NonZeroUsize::new_unchecked(size_of::<Futex<false>>()),
78+
/// ProtFlags::PROT_WRITE | ProtFlags::PROT_READ,
79+
/// MapFlags::MAP_SHARED | MapFlags::MAP_ANONYMOUS,
80+
/// None,
81+
/// 0
82+
/// )? };
83+
/// let futex_ptr = shared_memory.cast::<Futex<false>>();
84+
/// let futex = unsafe { &*futex_ptr };
85+
/// match unsafe { fork()? } {
86+
/// ForkResult::Parent { child } => {
87+
/// sleep(TIMEOUT);
88+
/// assert_eq!(futex.wake(1),Ok(1));
89+
/// // Wait for child process to exit
90+
/// unsafe {
91+
/// assert_eq!(libc::waitpid(child.as_raw(), std::ptr::null_mut(), 0), child.as_raw());
92+
/// }
93+
/// },
94+
/// ForkResult::Child => {
95+
/// let now = Instant::now();
96+
/// assert_eq!(futex.wait(0, Some(2 * TIMEOUT)),Ok(()));
97+
/// assert!(now.elapsed() > TIMEOUT && now.elapsed() < TIMEOUT + DELTA);
98+
/// }
99+
/// }
100+
/// # Ok(())
101+
/// # }
102+
/// ```
103+
#[derive(Debug)]
104+
pub struct Futex<const PRIVATE: bool = false>(UnsafeCell<u32>);
105+
106+
impl<const PRIVATE: bool> Futex<PRIVATE> {
107+
const MASK: i32 = if PRIVATE { libc::FUTEX_PRIVATE_FLAG } else { 0 };
108+
109+
/// Constructs new futex with a given `val`.
110+
pub fn new(val: u32) -> Self {
111+
Self(UnsafeCell::new(val))
112+
}
113+
114+
/// If the value of the futex:
115+
/// - `== val`, the thread sleeps waiting for a [`Futex::wake`] call, in this case this thread
116+
/// is considered a waiter on this futex.
117+
/// - `!= val`, then `Err` with [`Errno::EAGAIN`] is immediately returned.
118+
///
119+
/// If the timeout is:
120+
/// - `Some(_)` it specifies a timeout for the wait.
121+
/// - `None` it will block indefinitely.
122+
///
123+
/// Wraps [`libc::FUTEX_WAIT`].
124+
pub fn wait(&self, val: u32, timeout: Option<Duration>) -> Result<()> {
125+
let timespec = timeout.map(timespec);
126+
let timespec_ptr = unwrap_or_null(timespec.as_ref());
127+
128+
let res = unsafe {
129+
syscall(
130+
SYS_futex,
131+
self.0.get(),
132+
Self::MASK | libc::FUTEX_WAIT,
133+
val,
134+
timespec_ptr,
135+
)
136+
};
137+
Errno::result(res).map(drop)
138+
}
139+
140+
/// Wakes at most `val` waiters.
141+
///
142+
/// - `val == 1` wakes a single waiter.
143+
/// - `val == u32::MAX` wakes all waiters.
144+
///
145+
/// No guarantee is provided about which waiters are awoken. A waiter with a higher scheduling
146+
/// priority is not guaranteed to be awoken in preference to a waiter with a lower priority.
147+
///
148+
/// Wraps [`libc::FUTEX_WAKE`].
149+
pub fn wake(&self, val: u32) -> Result<u32> {
150+
let res = unsafe {
151+
syscall(SYS_futex, self.0.get(), Self::MASK | libc::FUTEX_WAKE, val)
152+
};
153+
Errno::result(res).map(|x| u32::try_from(x).unwrap())
154+
}
155+
156+
/// Creates a file descriptor associated with the futex.
157+
///
158+
/// When [`Futex::wake`] is performed on the futex this file indicates being readable with
159+
/// `select`, `poll` and `epoll`.
160+
///
161+
/// The file descriptor can be used to obtain asynchronous notifications: if val is nonzero,
162+
/// then, when another process or thread executes a FUTEX_WAKE, the caller will receive the
163+
/// signal number that was passed in val.
164+
///
165+
/// **Because it was inherently racy, this is unsupported from Linux 2.6.26 onward.**
166+
///
167+
/// Wraps [`libc::FUTEX_FD`].
168+
pub fn fd(&self, val: u32) -> Result<OwnedFd> {
169+
let res = unsafe {
170+
syscall(SYS_futex, self.0.get(), Self::MASK | libc::FUTEX_WAKE, val)
171+
};
172+
173+
// On a 32 bit arch `x` will be an `i32` and will trigger this lint.
174+
#[allow(clippy::useless_conversion)]
175+
Errno::result(res)
176+
.map(|x| unsafe { OwnedFd::from_raw_fd(i32::try_from(x).unwrap()) })
177+
}
178+
179+
/// [`Futex::cmp_requeue`] without the check being made using `val3`.
180+
///
181+
/// Wraps [`libc::FUTEX_REQUEUE`].
182+
pub fn requeue(&self, val: u32, val2: u32, uaddr2: &Self) -> Result<u32> {
183+
let res = unsafe {
184+
syscall(
185+
SYS_futex,
186+
self.0.get(),
187+
Self::MASK | libc::FUTEX_CMP_REQUEUE,
188+
val,
189+
val2,
190+
&uaddr2.0,
191+
)
192+
};
193+
Errno::result(res).map(|x| u32::try_from(x).unwrap())
194+
}
195+
196+
/// Wakes `val` waiters, moving remaining (up to `val2`) waiters to `uaddr2`.
197+
///
198+
/// If the value of this futex `== val3` returns `Err` with [`Errno::EAGAIN`].
199+
///
200+
/// Typical values to specify for `val` are `0` or `1` (Specifying `u32::MAX` makes the
201+
/// [`Futex::cmp_requeue`] equivalent to [`Futex::wake`]).
202+
///
203+
/// Typical values to specify for `val2` are `1` or `u32::MAX` (Specifying `0` makes
204+
/// [`Futex::cmp_requeue`] equivalent to [`Futex::wait`]).
205+
///
206+
/// Wraps [`libc::FUTEX_CMP_REQUEUE`].
207+
pub fn cmp_requeue(
208+
&self,
209+
val: u32,
210+
val2: u32,
211+
uaddr2: &Self,
212+
val3: u32,
213+
) -> Result<u32> {
214+
let res = unsafe {
215+
syscall(
216+
SYS_futex,
217+
self.0.get(),
218+
Self::MASK | libc::FUTEX_CMP_REQUEUE,
219+
val,
220+
val2,
221+
&uaddr2.0,
222+
val3,
223+
)
224+
};
225+
Errno::result(res).map(|x| u32::try_from(x).unwrap())
226+
}
227+
228+
/// Wraps [`libc::FUTEX_WAKE_OP`].
229+
pub fn wake_op(
230+
&self,
231+
val: u32,
232+
val2: u32,
233+
uaddr2: &Self,
234+
val3: u32,
235+
) -> Result<u32> {
236+
let res = unsafe {
237+
syscall(
238+
SYS_futex,
239+
self.0.get(),
240+
Self::MASK | libc::FUTEX_WAKE_OP,
241+
val,
242+
val2,
243+
&uaddr2.0,
244+
val3,
245+
)
246+
};
247+
Errno::result(res).map(|x| u32::try_from(x).unwrap())
248+
}
249+
250+
/// Wraps [`libc::FUTEX_WAIT_BITSET`].
251+
pub fn wait_bitset(
252+
&self,
253+
val: u32,
254+
timeout: Option<Duration>,
255+
val3: u32,
256+
) -> Result<()> {
257+
let timespec = timeout.map(timespec);
258+
let timespec_ptr = unwrap_or_null(timespec.as_ref());
259+
260+
let res = unsafe {
261+
syscall(
262+
SYS_futex,
263+
self.0.get(),
264+
Self::MASK | libc::FUTEX_WAIT_BITSET,
265+
val,
266+
timespec_ptr,
267+
val3,
268+
)
269+
};
270+
Errno::result(res).map(drop)
271+
}
272+
273+
/// Wraps [`libc::FUTEX_WAKE_BITSET`].
274+
pub fn wake_bitset(&self, val: u32, val3: u32) -> Result<u32> {
275+
let res = unsafe {
276+
syscall(SYS_futex, self.0.get(), libc::FUTEX_WAKE_BITSET, val, val3)
277+
};
278+
Errno::result(res).map(|x| u32::try_from(x).unwrap())
279+
}
280+
281+
/// Wraps [`libc::FUTEX_LOCK_PI`].
282+
pub fn lock_pi(&self, timeout: Option<Duration>) -> Result<()> {
283+
let timespec = timeout.map(timespec);
284+
let timespec_ptr = unwrap_or_null(timespec.as_ref());
285+
286+
let res = unsafe {
287+
syscall(
288+
SYS_futex,
289+
self.0.get(),
290+
Self::MASK | libc::FUTEX_LOCK_PI,
291+
timespec_ptr,
292+
)
293+
};
294+
Errno::result(res).map(drop)
295+
}
296+
297+
/// Wraps [`libc::FUTEX_LOCK_PI2`].
298+
#[cfg(target_os = "linux")]
299+
pub fn lock_pi2(&self, timeout: Option<Duration>) -> Result<()> {
300+
let timespec = timeout.map(timespec);
301+
let timespec_ptr = unwrap_or_null(timespec.as_ref());
302+
303+
let res = unsafe {
304+
syscall(
305+
SYS_futex,
306+
self.0.get(),
307+
Self::MASK | libc::FUTEX_LOCK_PI2,
308+
timespec_ptr,
309+
)
310+
};
311+
Errno::result(res).map(drop)
312+
}
313+
314+
/// Wraps [`libc::FUTEX_TRYLOCK_PI`].
315+
pub fn trylock_pi(&self) -> Result<()> {
316+
let res = unsafe {
317+
syscall(
318+
SYS_futex,
319+
self.0.get(),
320+
Self::MASK | libc::FUTEX_TRYLOCK_PI,
321+
)
322+
};
323+
Errno::result(res).map(drop)
324+
}
325+
326+
/// `libc::FUTEX_UNLOCK_PI`
327+
pub fn unlock_pi(&self) -> Result<()> {
328+
let res = unsafe {
329+
syscall(SYS_futex, self.0.get(), Self::MASK | libc::FUTEX_UNLOCK_PI)
330+
};
331+
Errno::result(res).map(drop)
332+
}
333+
334+
/// Wraps [`libc::FUTEX_CMP_REQUEUE_PI`].
335+
pub fn cmp_requeue_pi(
336+
&self,
337+
val: u32,
338+
val2: u32,
339+
uaddr2: &Self,
340+
val3: u32,
341+
) -> Result<u32> {
342+
let res = unsafe {
343+
syscall(
344+
SYS_futex,
345+
self.0.get(),
346+
Self::MASK | libc::FUTEX_CMP_REQUEUE_PI,
347+
val,
348+
val2,
349+
&uaddr2.0,
350+
val3,
351+
)
352+
};
353+
Errno::result(res).map(|x| u32::try_from(x).unwrap())
354+
}
355+
356+
/// Wraps [`libc::FUTEX_WAIT_REQUEUE_PI`].
357+
pub fn wait_requeue_pi(
358+
&self,
359+
val: u32,
360+
timeout: Option<Duration>,
361+
uaddr2: &Self,
362+
) -> Result<()> {
363+
let timespec = timeout.map(timespec);
364+
let timespec_ptr = unwrap_or_null(timespec.as_ref());
365+
366+
let res = unsafe {
367+
syscall(
368+
SYS_futex,
369+
self.0.get(),
370+
Self::MASK | libc::FUTEX_WAIT_REQUEUE_PI,
371+
val,
372+
timespec_ptr,
373+
&uaddr2.0,
374+
)
375+
};
376+
Errno::result(res).map(drop)
377+
}
378+
}
379+
380+
unsafe impl Sync for Futex {}

‎src/sys/mod.rs

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -226,3 +226,7 @@ feature! {
226226
#![feature = "time"]
227227
pub mod timer;
228228
}
229+
230+
/// Fast user-space locking.
231+
#[cfg(any(target_os = "android", target_os = "linux"))]
232+
pub mod futex;

0 commit comments

Comments
 (0)