1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
//! # Address Lock
//! Blocking a thread on the value of a word, and waking
//! everything blocked on one
//!
//! #### Note
//! A wait only sleeps while the word still reads the value it
//! was given, so a wake landing between the caller's read and
//! its wait is never lost
use crate::RuntimeError;
use libc::c_void;
use std::{io::Error, mem, sync::atomic::AtomicU32, time::Duration};
/// Blocks while a word still reads `value`
///
/// ## Returns
/// `Ok` when the word may have changed and the caller should
/// look again, and an error only when the kernel refused in a
/// way that going round again won't fix
pub(crate) fn wait(address: *mut c_void, value: u32) -> Result<(), RuntimeError> {
let status = unsafe {
libc::os_sync_wait_on_address(
address,
value as u64, // Sleep only while it still reads this
mem::size_of::<u32>(), // A single word, whatever it holds
libc::OS_SYNC_WAIT_ON_ADDRESS_NONE, // Single process waiting
)
};
if status >= 0 {
return Ok(());
}
let error = Error::last_os_error().raw_os_error();
// A signal, a value that moved, or the kernel short of memory all
// mean look again
if matches!(
error,
Some(libc::EINTR | libc::EAGAIN | libc::ENOMEM | libc::EFAULT)
) {
return Ok(());
}
Err(RuntimeError::AddressLock)
}
/// Blocks while a word still reads `value`, for at most
/// `timeout`
///
/// ## Returns
/// `Ok(true)` when the word may have changed, `Ok(false)` when
/// the timeout ran out, and an error only when the kernel
/// refused in a way that going round again won't fix
///
/// A zero timeout returns `Ok(false)` without waiting
pub(crate) fn wait_until(
address: *mut c_void,
value: u32,
timeout: Duration,
) -> Result<bool, RuntimeError> {
let nanos = timeout.as_nanos();
if nanos == 0 {
return Ok(false);
}
let status = unsafe {
libc::os_sync_wait_on_address_with_timeout(
address,
value as u64, // Sleep only while it still reads this
mem::size_of::<u32>(), // A single word, whatever it holds
libc::OS_SYNC_WAIT_ON_ADDRESS_NONE, // Single process waiting
libc::OS_CLOCK_MACH_ABSOLUTE_TIME, // Monotonic, so setting the clock can't move it
// Saturated, so a huge timeout can't wrap to a tiny one
nanos.min(u64::MAX as u128) as u64,
)
};
if status >= 0 {
return Ok(true);
}
let error = Error::last_os_error().raw_os_error();
// The time ran out
if error == Some(libc::ETIMEDOUT) {
return Ok(false);
}
// A signal, a value that moved, or the kernel short of memory all
// mean look again
if matches!(
error,
Some(libc::EINTR | libc::EAGAIN | libc::ENOMEM | libc::EFAULT)
) {
return Ok(true);
}
Err(RuntimeError::AddressLock)
}
/// Wakes everything blocked on a word
pub(crate) fn wake(address: *mut c_void) {
let _ = unsafe {
libc::os_sync_wake_by_address_all(
address,
mem::size_of::<u32>(), // A single word, whatever it holds
libc::OS_SYNC_WAKE_BY_ADDRESS_NONE, // Single process waiting
)
};
}
/// The address of a word, in the shape the kernel wants it
#[inline(always)]
pub(crate) fn address(word: &AtomicU32) -> *mut c_void {
word as *const AtomicU32 as *mut c_void
}