std/sys/sync/futex/
unix.rs

1use crate::sync::atomic::Atomic;
2use crate::time::Duration;
3
4/// An atomic for use as a futex that is at least 32-bits but may be larger
5pub type Futex = Atomic<Primitive>;
6/// Must be the underlying type of Futex
7pub type Primitive = u32;
8
9/// An atomic for use as a futex that is at least 8-bits but may be larger.
10pub type SmallFutex = Atomic<SmallPrimitive>;
11/// Must be the underlying type of SmallFutex
12pub type SmallPrimitive = u32;
13
14/// Waits for a `futex_wake` operation to wake us.
15///
16/// Returns directly if the futex doesn't hold the expected value.
17///
18/// Returns false on timeout, and true in all other cases.
19#[cfg(any(target_os = "linux", target_os = "android", target_os = "freebsd"))]
20pub fn futex_wait(futex: &Atomic<u32>, expected: u32, timeout: Option<Duration>) -> bool {
21    use crate::ptr::null;
22    use crate::sync::atomic::Ordering::Relaxed;
23    use crate::sys::pal::time::Timespec;
24
25    // Calculate the timeout as an absolute timespec.
26    //
27    // Overflows are rounded up to an infinite timeout (None).
28    let timespec = timeout
29        .and_then(|d| Timespec::now(libc::CLOCK_MONOTONIC).checked_add_duration(&d))
30        .and_then(|t| t.to_timespec());
31
32    loop {
33        // No need to wait if the value already changed.
34        if futex.load(Relaxed) != expected {
35            return true;
36        }
37
38        let r = unsafe {
39            cfg_select! {
40                target_os = "freebsd" => {
41                    // FreeBSD doesn't have futex(), but it has
42                    // _umtx_op(UMTX_OP_WAIT_UINT_PRIVATE), which is nearly
43                    // identical. It supports absolute timeouts through a flag
44                    // in the _umtx_time struct.
45                    let umtx_timeout = timespec.map(|t| libc::_umtx_time {
46                        _timeout: t,
47                        _flags: libc::UMTX_ABSTIME,
48                        _clockid: libc::CLOCK_MONOTONIC as u32,
49                    });
50                    let umtx_timeout_ptr = umtx_timeout.as_ref().map_or(null(), |t| t as *const _);
51                    let umtx_timeout_size = umtx_timeout.as_ref().map_or(0, |t| size_of_val(t));
52                    libc::_umtx_op(
53                        futex as *const Atomic<u32> as *mut _,
54                        libc::UMTX_OP_WAIT_UINT_PRIVATE,
55                        expected as libc::c_ulong,
56                        crate::ptr::without_provenance_mut(umtx_timeout_size),
57                        umtx_timeout_ptr as *mut _,
58                    )
59                }
60                any(target_os = "linux", target_os = "android") => {
61                    // Use FUTEX_WAIT_BITSET rather than FUTEX_WAIT to be able to give an
62                    // absolute time rather than a relative time.
63                    libc::syscall(
64                        libc::SYS_futex,
65                        futex as *const Atomic<u32>,
66                        libc::FUTEX_WAIT_BITSET | libc::FUTEX_PRIVATE_FLAG,
67                        expected,
68                        timespec.as_ref().map_or(null(), |t| t as *const libc::timespec),
69                        null::<u32>(), // This argument is unused for FUTEX_WAIT_BITSET.
70                        !0u32, // A full bitmask, to make it behave like a regular FUTEX_WAIT.
71                    )
72                }
73                _ => {
74                    compile_error!("unknown target_os");
75                }
76            }
77        };
78
79        match (r < 0).then(crate::sys::io::errno) {
80            Some(libc::ETIMEDOUT) => return false,
81            Some(libc::EINTR) => continue,
82            _ => return true,
83        }
84    }
85}
86
87/// Wakes up one thread that's blocked on `futex_wait` on this futex.
88///
89/// Returns true if this actually woke up such a thread,
90/// or false if no thread was waiting on this futex.
91///
92/// On some platforms, this always returns false.
93#[cfg(any(target_os = "linux", target_os = "android"))]
94pub fn futex_wake(futex: &Atomic<u32>) -> bool {
95    let ptr = futex as *const Atomic<u32>;
96    let op = libc::FUTEX_WAKE | libc::FUTEX_PRIVATE_FLAG;
97    unsafe { libc::syscall(libc::SYS_futex, ptr, op, 1) > 0 }
98}
99
100/// Wakes up all threads that are waiting on `futex_wait` on this futex.
101#[cfg(any(target_os = "linux", target_os = "android"))]
102pub fn futex_wake_all(futex: &Atomic<u32>) {
103    let ptr = futex as *const Atomic<u32>;
104    let op = libc::FUTEX_WAKE | libc::FUTEX_PRIVATE_FLAG;
105    unsafe {
106        libc::syscall(libc::SYS_futex, ptr, op, i32::MAX);
107    }
108}
109
110// FreeBSD doesn't tell us how many threads are woken up, so this always returns false.
111#[cfg(target_os = "freebsd")]
112pub fn futex_wake(futex: &Atomic<u32>) -> bool {
113    use crate::ptr::null_mut;
114    unsafe {
115        libc::_umtx_op(
116            futex as *const Atomic<u32> as *mut _,
117            libc::UMTX_OP_WAKE_PRIVATE,
118            1,
119            null_mut(),
120            null_mut(),
121        )
122    };
123    false
124}
125
126#[cfg(target_os = "freebsd")]
127pub fn futex_wake_all(futex: &Atomic<u32>) {
128    use crate::ptr::null_mut;
129    unsafe {
130        libc::_umtx_op(
131            futex as *const Atomic<u32> as *mut _,
132            libc::UMTX_OP_WAKE_PRIVATE,
133            i32::MAX as libc::c_ulong,
134            null_mut(),
135            null_mut(),
136        )
137    };
138}
139
140#[cfg(target_os = "openbsd")]
141pub fn futex_wait(futex: &Atomic<u32>, expected: u32, timeout: Option<Duration>) -> bool {
142    use crate::ptr::{null, null_mut};
143    use crate::sys::pal::time::Timespec;
144
145    // Overflows are rounded up to an infinite timeout (None).
146    let timespec = timeout
147        .and_then(|d| Timespec::zero().checked_add_duration(&d))
148        .and_then(|t| t.to_timespec());
149
150    let r = unsafe {
151        libc::futex(
152            futex as *const Atomic<u32> as *mut u32,
153            libc::FUTEX_WAIT,
154            expected as i32,
155            timespec.as_ref().map_or(null(), |t| t as *const libc::timespec),
156            null_mut(),
157        )
158    };
159
160    r == 0 || crate::sys::io::errno() != libc::ETIMEDOUT
161}
162
163#[cfg(target_os = "openbsd")]
164pub fn futex_wake(futex: &Atomic<u32>) -> bool {
165    use crate::ptr::{null, null_mut};
166    unsafe {
167        libc::futex(
168            futex as *const Atomic<u32> as *mut u32,
169            libc::FUTEX_WAKE,
170            1,
171            null(),
172            null_mut(),
173        ) > 0
174    }
175}
176
177#[cfg(target_os = "openbsd")]
178pub fn futex_wake_all(futex: &Atomic<u32>) {
179    use crate::ptr::{null, null_mut};
180    unsafe {
181        libc::futex(
182            futex as *const Atomic<u32> as *mut u32,
183            libc::FUTEX_WAKE,
184            i32::MAX,
185            null(),
186            null_mut(),
187        );
188    }
189}
190
191#[cfg(target_os = "dragonfly")]
192pub fn futex_wait(futex: &Atomic<u32>, expected: u32, timeout: Option<Duration>) -> bool {
193    // A timeout of 0 means infinite.
194    // We round smaller timeouts up to 1 millisecond.
195    // Overflows are rounded up to an infinite timeout.
196    let timeout_ms =
197        timeout.and_then(|d| Some(i32::try_from(d.as_millis()).ok()?.max(1))).unwrap_or(0);
198
199    let r = unsafe {
200        libc::umtx_sleep(futex as *const Atomic<u32> as *const i32, expected as i32, timeout_ms)
201    };
202
203    r == 0 || crate::sys::io::errno() != libc::ETIMEDOUT
204}
205
206// DragonflyBSD doesn't tell us how many threads are woken up, so this always returns false.
207#[cfg(target_os = "dragonfly")]
208pub fn futex_wake(futex: &Atomic<u32>) -> bool {
209    unsafe { libc::umtx_wakeup(futex as *const Atomic<u32> as *const i32, 1) };
210    false
211}
212
213#[cfg(target_os = "dragonfly")]
214pub fn futex_wake_all(futex: &Atomic<u32>) {
215    unsafe { libc::umtx_wakeup(futex as *const Atomic<u32> as *const i32, i32::MAX) };
216}
217
218#[cfg(target_os = "emscripten")]
219unsafe extern "C" {
220    fn emscripten_futex_wake(addr: *const Atomic<u32>, count: libc::c_int) -> libc::c_int;
221    fn emscripten_futex_wait(
222        addr: *const Atomic<u32>,
223        val: libc::c_uint,
224        max_wait_ms: libc::c_double,
225    ) -> libc::c_int;
226}
227
228#[cfg(target_os = "emscripten")]
229pub fn futex_wait(futex: &Atomic<u32>, expected: u32, timeout: Option<Duration>) -> bool {
230    unsafe {
231        emscripten_futex_wait(
232            futex,
233            expected,
234            timeout.map_or(f64::INFINITY, |d| d.as_secs_f64() * 1000.0),
235        ) != -libc::ETIMEDOUT
236    }
237}
238
239#[cfg(target_os = "emscripten")]
240pub fn futex_wake(futex: &Atomic<u32>) -> bool {
241    unsafe { emscripten_futex_wake(futex, 1) > 0 }
242}
243
244#[cfg(target_os = "emscripten")]
245pub fn futex_wake_all(futex: &Atomic<u32>) {
246    unsafe { emscripten_futex_wake(futex, i32::MAX) };
247}
248
249#[cfg(target_os = "fuchsia")]
250pub fn futex_wait(futex: &Atomic<u32>, expected: u32, timeout: Option<Duration>) -> bool {
251    use crate::sys::pal::fuchsia::*;
252
253    // Sleep forever if the timeout is longer than fits in a i64.
254    let deadline = timeout
255        .and_then(|d| i64::try_from(d.as_nanos()).ok()?.checked_add(zx_clock_get_monotonic()))
256        .unwrap_or(ZX_TIME_INFINITE);
257
258    unsafe {
259        zx_futex_wait(futex, zx_futex_t::new(expected), ZX_HANDLE_INVALID, deadline)
260            != ZX_ERR_TIMED_OUT
261    }
262}
263
264// Fuchsia doesn't tell us how many threads are woken up, so this always returns false.
265#[cfg(target_os = "fuchsia")]
266pub fn futex_wake(futex: &Atomic<u32>) -> bool {
267    unsafe { crate::sys::pal::fuchsia::zx_futex_wake(futex, 1) };
268    false
269}
270
271#[cfg(target_os = "fuchsia")]
272pub fn futex_wake_all(futex: &Atomic<u32>) {
273    unsafe { crate::sys::pal::fuchsia::zx_futex_wake(futex, u32::MAX) };
274}
std/sys/sync/futex/unix.rs

std/sys/sync/futex/
unix.rs
