Skip to content

Commit 0e2def5

Browse files
borsbors
committed
Auto merge of #2054 - m-ou-se:futex-wait-bitset, r=RalfJung
Add support for FUTEX_{WAIT,WAKE}_BITSET FUTEX_WAIT_BITSET and FUTEX_WAKE_BITSET are extensions of FUTEX_WAIT and FUTEX_WAKE that allow tagging each waiting thread with up to 32 'labels', and then only wake up threads that match certain labels. The non-bitset operations behave like their bitset was fully set (u32::MAX), meaning that they'll wait for anything, and wake up anything. The only other difference is that FUTEX_WAIT_BITSET uses an absolute timeout instead of an relative timeout like FUTEX_WAIT. Often, FUTEX_WAIT_BITSET is used not for its bitset functionality, but only for its absolute timeout functionality. It is then used with a bitset of u32::MAX. ~~This adds support for only that use case to Miri, as that's all `std` currently needs. Any other bitset is still unsupported.~~ Update: This adds full support for both these syscalls.
2 parents fb01df5 + 4fdda31 commit 0e2def5

File tree

4 files changed

+165
-19
lines changed

4 files changed

+165
-19
lines changed

‎rust-version

+1-1
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
f262ca12aac76152c4b46cefcf8300f0249a5eb2
1+
306ba8357fb36212b7d30efb9eb9e41659ac1445

‎src/shims/posix/linux/sync.rs

+68-12
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,9 @@ pub fn futex<'tcx>(
3636

3737
let futex_private = this.eval_libc_i32("FUTEX_PRIVATE_FLAG")?;
3838
let futex_wait = this.eval_libc_i32("FUTEX_WAIT")?;
39+
let futex_wait_bitset = this.eval_libc_i32("FUTEX_WAIT_BITSET")?;
3940
let futex_wake = this.eval_libc_i32("FUTEX_WAKE")?;
41+
let futex_wake_bitset = this.eval_libc_i32("FUTEX_WAKE_BITSET")?;
4042
let futex_realtime = this.eval_libc_i32("FUTEX_CLOCK_REALTIME")?;
4143

4244
// FUTEX_PRIVATE enables an optimization that stops it from working across processes.
@@ -45,12 +47,37 @@ pub fn futex<'tcx>(
4547
// FUTEX_WAIT: (int *addr, int op = FUTEX_WAIT, int val, const timespec *timeout)
4648
// Blocks the thread if *addr still equals val. Wakes up when FUTEX_WAKE is called on the same address,
4749
// or *timeout expires. `timeout == null` for an infinite timeout.
48-
op if op & !futex_realtime == futex_wait => {
49-
if args.len() < 5 {
50-
throw_ub_format!(
51-
"incorrect number of arguments for `futex` syscall with `op=FUTEX_WAIT`: got {}, expected at least 5",
52-
args.len()
53-
);
50+
//
51+
// FUTEX_WAIT_BITSET: (int *addr, int op = FUTEX_WAIT_BITSET, int val, const timespec *timeout, int *_ignored, unsigned int bitset)
52+
// This is identical to FUTEX_WAIT, except:
53+
// - The timeout is absolute rather than relative.
54+
// - You can specify the bitset to selecting what WAKE operations to respond to.
55+
op if op & !futex_realtime == futex_wait || op & !futex_realtime == futex_wait_bitset => {
56+
let wait_bitset = op & !futex_realtime == futex_wait_bitset;
57+
58+
let bitset = if wait_bitset {
59+
if args.len() != 7 {
60+
throw_ub_format!(
61+
"incorrect number of arguments for `futex` syscall with `op=FUTEX_WAIT_BITSET`: got {}, expected 7",
62+
args.len()
63+
);
64+
}
65+
this.read_scalar(&args[6])?.to_u32()?
66+
} else {
67+
if args.len() < 5 {
68+
throw_ub_format!(
69+
"incorrect number of arguments for `futex` syscall with `op=FUTEX_WAIT`: got {}, expected at least 5",
70+
args.len()
71+
);
72+
}
73+
u32::MAX
74+
};
75+
76+
if bitset == 0 {
77+
let einval = this.eval_libc("EINVAL")?;
78+
this.set_last_error(einval)?;
79+
this.write_scalar(Scalar::from_machine_isize(-1, this), dest)?;
80+
return Ok(());
5481
}
5582

5683
// `deref_operand` but not actually dereferencing the ptr yet (it might be NULL!).
@@ -70,10 +97,20 @@ pub fn futex<'tcx>(
7097
return Ok(());
7198
}
7299
};
73-
Some(if op & futex_realtime != 0 {
74-
Time::RealTime(SystemTime::now().checked_add(duration).unwrap())
100+
Some(if wait_bitset {
101+
// FUTEX_WAIT_BITSET uses an absolute timestamp.
102+
if op & futex_realtime != 0 {
103+
Time::RealTime(SystemTime::UNIX_EPOCH.checked_add(duration).unwrap())
104+
} else {
105+
Time::Monotonic(this.machine.time_anchor.checked_add(duration).unwrap())
106+
}
75107
} else {
76-
Time::Monotonic(Instant::now().checked_add(duration).unwrap())
108+
// FUTEX_WAIT uses a relative timestamp.
109+
if op & futex_realtime != 0 {
110+
Time::RealTime(SystemTime::now().checked_add(duration).unwrap())
111+
} else {
112+
Time::Monotonic(Instant::now().checked_add(duration).unwrap())
113+
}
77114
})
78115
};
79116
// Check the pointer for alignment and validity.
@@ -108,7 +145,7 @@ pub fn futex<'tcx>(
108145
if val == futex_val {
109146
// The value still matches, so we block the trait make it wait for FUTEX_WAKE.
110147
this.block_thread(thread);
111-
this.futex_wait(addr_scalar.to_machine_usize(this)?, thread);
148+
this.futex_wait(addr_scalar.to_machine_usize(this)?, thread, bitset);
112149
// Succesfully waking up from FUTEX_WAIT always returns zero.
113150
this.write_scalar(Scalar::from_machine_isize(0, this), dest)?;
114151
// Register a timeout callback if a timeout was specified.
@@ -140,10 +177,29 @@ pub fn futex<'tcx>(
140177
// Wakes at most `val` threads waiting on the futex at `addr`.
141178
// Returns the amount of threads woken up.
142179
// Does not access the futex value at *addr.
143-
op if op == futex_wake => {
180+
// FUTEX_WAKE_BITSET: (int *addr, int op = FUTEX_WAKE, int val, const timespect *_unused, int *_unused, unsigned int bitset)
181+
// Same as FUTEX_WAKE, but allows you to specify a bitset to select which threads to wake up.
182+
op if op == futex_wake || op == futex_wake_bitset => {
183+
let bitset = if op == futex_wake_bitset {
184+
if args.len() != 7 {
185+
throw_ub_format!(
186+
"incorrect number of arguments for `futex` syscall with `op=FUTEX_WAKE_BITSET`: got {}, expected 7",
187+
args.len()
188+
);
189+
}
190+
this.read_scalar(&args[6])?.to_u32()?
191+
} else {
192+
u32::MAX
193+
};
194+
if bitset == 0 {
195+
let einval = this.eval_libc("EINVAL")?;
196+
this.set_last_error(einval)?;
197+
this.write_scalar(Scalar::from_machine_isize(-1, this), dest)?;
198+
return Ok(());
199+
}
144200
let mut n = 0;
145201
for _ in 0..val {
146-
if let Some(thread) = this.futex_wake(addr_scalar.to_machine_usize(this)?) {
202+
if let Some(thread) = this.futex_wake(addr_scalar.to_machine_usize(this)?, bitset) {
147203
this.unblock_thread(thread);
148204
this.unregister_timeout_callback_if_exists(thread);
149205
n += 1;

‎src/sync.rs

+10-6
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,8 @@ struct Futex {
144144
struct FutexWaiter {
145145
/// The thread that is waiting on this futex.
146146
thread: ThreadId,
147+
/// The bitset used by FUTEX_*_BITSET, or u32::MAX for other operations.
148+
bitset: u32,
147149
}
148150

149151
/// The state of all synchronization variables.
@@ -486,15 +488,15 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriEvalContextExt<'mir, 'tcx
486488
this.machine.threads.sync.condvars[id].waiters.retain(|waiter| waiter.thread != thread);
487489
}
488490

489-
fn futex_wait(&mut self, addr: u64, thread: ThreadId) {
491+
fn futex_wait(&mut self, addr: u64, thread: ThreadId, bitset: u32) {
490492
let this = self.eval_context_mut();
491493
let futex = &mut this.machine.threads.sync.futexes.entry(addr).or_default();
492494
let waiters = &mut futex.waiters;
493495
assert!(waiters.iter().all(|waiter| waiter.thread != thread), "thread is already waiting");
494-
waiters.push_back(FutexWaiter { thread });
496+
waiters.push_back(FutexWaiter { thread, bitset });
495497
}
496498

497-
fn futex_wake(&mut self, addr: u64) -> Option<ThreadId> {
499+
fn futex_wake(&mut self, addr: u64, bitset: u32) -> Option<ThreadId> {
498500
let this = self.eval_context_mut();
499501
let current_thread = this.get_active_thread();
500502
let futex = &mut this.machine.threads.sync.futexes.get_mut(&addr)?;
@@ -504,13 +506,15 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriEvalContextExt<'mir, 'tcx
504506
if let Some(data_race) = data_race {
505507
data_race.validate_lock_release(&mut futex.data_race, current_thread);
506508
}
507-
let res = futex.waiters.pop_front().map(|waiter| {
509+
510+
// Wake up the first thread in the queue that matches any of the bits in the bitset.
511+
futex.waiters.iter().position(|w| w.bitset & bitset != 0).map(|i| {
512+
let waiter = futex.waiters.remove(i).unwrap();
508513
if let Some(data_race) = data_race {
509514
data_race.validate_lock_acquire(&futex.data_race, waiter.thread);
510515
}
511516
waiter.thread
512-
});
513-
res
517+
})
514518
}
515519

516520
fn futex_remove_waiter(&mut self, addr: u64, thread: ThreadId) {

‎tests/run-pass/concurrency/linux-futex.rs

+86
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
#![feature(rustc_private)]
88
extern crate libc;
99

10+
use std::mem::MaybeUninit;
1011
use std::ptr;
1112
use std::thread;
1213
use std::time::{Duration, Instant};
@@ -93,6 +94,42 @@ fn wait_timeout() {
9394
assert!((200..1000).contains(&start.elapsed().as_millis()));
9495
}
9596

97+
fn wait_absolute_timeout() {
98+
let start = Instant::now();
99+
100+
// Get the current monotonic timestamp as timespec.
101+
let mut timeout = unsafe {
102+
let mut now: MaybeUninit<libc::timespec> = MaybeUninit::uninit();
103+
assert_eq!(libc::clock_gettime(libc::CLOCK_MONOTONIC, now.as_mut_ptr()), 0);
104+
now.assume_init()
105+
};
106+
107+
// Add 200ms.
108+
timeout.tv_nsec += 200_000_000;
109+
if timeout.tv_nsec > 1_000_000_000 {
110+
timeout.tv_nsec -= 1_000_000_000;
111+
timeout.tv_sec += 1;
112+
}
113+
114+
let futex: i32 = 123;
115+
116+
// Wait for 200ms from now, with nobody waking us up early.
117+
unsafe {
118+
assert_eq!(libc::syscall(
119+
libc::SYS_futex,
120+
&futex as *const i32,
121+
libc::FUTEX_WAIT_BITSET,
122+
123,
123+
&timeout,
124+
0,
125+
u32::MAX,
126+
), -1);
127+
assert_eq!(*libc::__errno_location(), libc::ETIMEDOUT);
128+
}
129+
130+
assert!((200..1000).contains(&start.elapsed().as_millis()));
131+
}
132+
96133
fn wait_wake() {
97134
let start = Instant::now();
98135

@@ -123,10 +160,59 @@ fn wait_wake() {
123160
assert!((200..1000).contains(&start.elapsed().as_millis()));
124161
}
125162

163+
fn wait_wake_bitset() {
164+
let start = Instant::now();
165+
166+
static FUTEX: i32 = 0;
167+
168+
thread::spawn(move || {
169+
thread::sleep(Duration::from_millis(200));
170+
unsafe {
171+
assert_eq!(libc::syscall(
172+
libc::SYS_futex,
173+
&FUTEX as *const i32,
174+
libc::FUTEX_WAKE_BITSET,
175+
10, // Wake up at most 10 threads.
176+
0,
177+
0,
178+
0b1001, // bitset
179+
), 0); // Didn't match any thread.
180+
}
181+
thread::sleep(Duration::from_millis(200));
182+
unsafe {
183+
assert_eq!(libc::syscall(
184+
libc::SYS_futex,
185+
&FUTEX as *const i32,
186+
libc::FUTEX_WAKE_BITSET,
187+
10, // Wake up at most 10 threads.
188+
0,
189+
0,
190+
0b0110, // bitset
191+
), 1); // Woken up one thread.
192+
}
193+
});
194+
195+
unsafe {
196+
assert_eq!(libc::syscall(
197+
libc::SYS_futex,
198+
&FUTEX as *const i32,
199+
libc::FUTEX_WAIT_BITSET,
200+
0,
201+
ptr::null::<libc::timespec>(),
202+
0,
203+
0b0100, // bitset
204+
), 0);
205+
}
206+
207+
assert!((400..1000).contains(&start.elapsed().as_millis()));
208+
}
209+
126210
fn main() {
127211
wake_nobody();
128212
wake_dangling();
129213
wait_wrong_val();
130214
wait_timeout();
215+
wait_absolute_timeout();
131216
wait_wake();
217+
wait_wake_bitset();
132218
}

0 commit comments

Comments
 (0)