Review Linux futex implementation for weak memory

[cbeuw]

This appears to be the smoking gun of std::sync::mpsc matklad/once_cell#182 and crossbeam channel rust-lang/rust#55005 (comment) deadlocks

miri/src/shims/unix/linux/sync.rs

Lines 129 to 148 in 32a7580

	// Read an `i32` through the pointer, regardless of any wrapper types.
	// It's not uncommon for `addr` to be passed as another type than `*mut i32`, such as `*const AtomicI32`.
	// FIXME: this fails if `addr` is not a pointer type.
	// The atomic ordering for futex(https://man7.org/linux/man-pages/man2/futex.2.html):
	// "The load of the value of the futex word is an
	// atomic memory access (i.e., using atomic machine instructions
	// of the respective architecture). This load, the comparison
	// with the expected value, and starting to sleep are performed
	// atomically and totally ordered with respect to other futex
	// operations on the same futex word."
	// SeqCst is total order over all operations.
	// FIXME: check if this should be changed when weak memory orders are added.
	let futex_val = this
	.read_scalar_at_offset_atomic(
	&addr.into(),
	0,
	this.machine.layouts.i32,
	AtomicReadOp::SeqCst,
	)?
	.to_i32()?;

The deadlock can be seen running this

use std::{sync::mpsc::channel, thread};

fn test() {
    let (tx1, rx1) = channel();
    let (tx2, rx2) = channel();
    let t1 = thread::spawn(move || {
        tx1.send(()).unwrap();
        rx2.recv().unwrap();
    });

    rx1.recv().unwrap();

    tx2.send(()).unwrap();

    assert!(t1.join().is_ok());
}

pub fn main() {
    for i in 0.. {
        dbg!(i);
        test();
    }
}

[src/main.rs:20] i = 0
[src/main.rs:20] i = 1
[2022-06-10T22:00:00Z TRACE miri::concurrency::weak_memory] outdated value read at /home/cbeuw/.rustup/toolchains/miri/lib/rustlib/src/rust/library/std/src/sys/unix/futex.rs:61:21: 69:22 (#0)
error: deadlock: the evaluated program deadlocked
   --> /home/cbeuw/.rustup/toolchains/miri/lib/rustlib/src/rust/library/std/src/sys/unix/thread.rs:250:66
    |
250 |             let ret = libc::pthread_join(self.id, ptr::null_mut());
    |                                                                  ^ the evaluated program deadlocked
    |
    = note: inside `std::sys::unix::thread::Thread::join` at /home/cbeuw/.rustup/toolchains/miri/lib/rustlib/src/rust/library/std/src/sys/unix/thread.rs:250:66
    = note: inside `std::thread::JoinInner::<()>::join` at /home/cbeuw/.rustup/toolchains/miri/lib/rustlib/src/rust/library/std/src/thread/mod.rs:1352:9
    = note: inside `std::thread::JoinHandle::<()>::join` at /home/cbeuw/.rustup/toolchains/miri/lib/rustlib/src/rust/library/std/src/thread/mod.rs:1485:9
note: inside `test` at src/main.rs:15:10
   --> src/main.rs:15:10
    |
15  |     assert!(t1.join().is_ok());
    |             ^^^^^^^^^
note: inside `main` at src/main.rs:21:9
   --> src/main.rs:21:9
    |
21  |         test();
    |         ^^^^^^

note: some details are omitted, run with `MIRIFLAGS=-Zmiri-backtrace=full` for a verbose backtrace

error: aborting due to previous error

(That TRACE line was added locally. I'll make a separate PR to add weak memory logging)

[RalfJung]

I think that FIXME refers to the fact that we use SeqCst. Is that the right ordering or not? The futex documentation should specify how it interacts with other non-futex atomic accesses to the same location -- what kind of an access does it behave like?

[RalfJung]

This also seems suspicious to me:

miri/src/shims/unix/linux/sync.rs

Line 209 in 32a7580

this.unblock_thread(thread);

When waking up a thread, shouldn't we also tell that thread that it "synchronized with" the thread that triggered the wakeup? Or does futex not guarantee that?

The corresponding calls in sync.rs (the lock implementations) do do something with the data race model. It's a bit non-local unfortunately; maybe unblock_thread should take some kind of parameter (a vclock or so) so that it can do all that work itself so that we cannot possibly forget this?

[RalfJung]

Ah, there actually is code to inform data_race about the futex happens-before, it is just in a different file:

miri/src/sync.rs

Line 566 in 10d978c

data_race.validate_lock_release(&mut futex.data_race, current_thread);

[cbeuw]

What happened here is that an outdated value was read from addr, which happens to be equal to expected and the thread starts waiting even though someone else has already changed *addr and called futex_wake.

This exact issue was mentioned in the Linux kernel implementation:
https://github.com/torvalds/linux/blob/0885eacdc81f920c3e0554d5615e69a66504a28d/kernel/futex/waitwake.c#L27-L95

 * In futex wake up scenarios where no tasks are blocked on a futex, taking
 * the hb spinlock can be avoided and simply return. In order for this
 * optimization to work, ordering guarantees must exist so that the waiter
 * being added to the list is acknowledged when the list is concurrently being
 * checked by the waker, avoiding scenarios like the following:
 *
 * CPU 0                               CPU 1
 * val = *futex;
 * sys_futex(WAIT, futex, val);
 *   futex_wait(futex, val);
 *   uval = *futex;
 *                                     *futex = newval;            // cbeuw: They are assuming interleaving semantics here but with weak memory this assignment can be executed before CPU 0 even ran anything
 *                                     sys_futex(WAKE, futex);
 *                                       futex_wake(futex);
 *                                       if (queue_empty())
 *                                         return;
 *   if (uval == val)
 *      lock(hash_bucket(futex));
 *      queue();
 *     unlock(hash_bucket(futex));
 *     schedule();
 *
 * This would cause the waiter on CPU 0 to wait forever because it
 * missed the transition of the user space value from val to newval
 * and the waker did not find the waiter in the hash bucket queue.
 *
 * The correct serialization ensures that a waiter either observes
 * the changed user space value before blocking or is woken by a
 * concurrent waker:
 *
 * CPU 0                                 CPU 1
 * val = *futex;
 * sys_futex(WAIT, futex, val);
 *   futex_wait(futex, val);
 *
 *   waiters++; (a)
 *   smp_mb(); (A) <-- paired with -.
 *                                  |
 *   lock(hash_bucket(futex));      |
 *                                  |
 *   uval = *futex;                 |
 *                                  |        *futex = newval;
 *                                  |        sys_futex(WAKE, futex);
 *                                  |          futex_wake(futex);
 *                                  |
 *                                  `--------> smp_mb(); (B)
 *   if (uval == val)
 *     queue();
 *     unlock(hash_bucket(futex));
 *     schedule();                         if (waiters)
 *                                           lock(hash_bucket(futex));
 *   else                                    wake_waiters(futex);
 *     waiters--; (b)                        unlock(hash_bucket(futex));
 *
 * Where (A) orders the waiters increment and the futex value read through
 * atomic operations (see futex_hb_waiters_inc) and where (B) orders the write
 * to futex and the waiters read (see futex_hb_waiters_pending()).
 *
 * This yields the following case (where X:=waiters, Y:=futex):
 *
 *	X = Y = 0
 *
 *	w[X]=1		w[Y]=1
 *	MB		MB
 *	r[Y]=y		r[X]=x
 *
 * Which guarantees that x==0 && y==0 is impossible; which translates back into
 * the guarantee that we cannot both miss the futex variable change and the
 * enqueue.

But our futex_wake doesn't handle a concurrent wait call that hasn't added itself to the queue yet:

miri/src/sync.rs

Lines 550 to 577 in 31da2d9

	fn futex_wait(&mut self, addr: u64, thread: ThreadId, bitset: u32) {
	let this = self.eval_context_mut();
	let futex = &mut this.machine.threads.sync.futexes.entry(addr).or_default();
	let waiters = &mut futex.waiters;
	assert!(waiters.iter().all(|waiter| waiter.thread != thread), "thread is already waiting");
	waiters.push_back(FutexWaiter { thread, bitset });
	}
	fn futex_wake(&mut self, addr: u64, bitset: u32) -> Option<ThreadId> {
	let this = self.eval_context_mut();
	let current_thread = this.get_active_thread();
	let futex = &mut this.machine.threads.sync.futexes.get_mut(&addr)?;
	let data_race = &this.machine.data_race;
	// Each futex-wake happens-before the end of the futex wait
	if let Some(data_race) = data_race {
	data_race.validate_lock_release(&mut futex.data_race, current_thread);
	}
	// Wake up the first thread in the queue that matches any of the bits in the bitset.
	futex.waiters.iter().position(|w| w.bitset & bitset != 0).map(|i| {
	let waiter = futex.waiters.remove(i).unwrap();
	if let Some(data_race) = data_race {
	data_race.validate_lock_acquire(&futex.data_race, waiter.thread);
	}
	waiter.thread
	})
	}

We could use addr.fetch_add(0) to read and ensure "waiter observes the changed user space value before blocking" in this case, and this will work because Miri will never yield to a concurrent waker before our futex_wait call is enqueued and finished - but it's still incorrect and will deadlock again once we add pre-emptive scheduling (my recent attempt to add a very simple time-quanta based preemption failed due to deadlocks in tests. I didn't figure out why but I think this is it)

[RalfJung]

(FWIW the Miri scheduler is preemptive now, though it doesn't use time quanta -- it uses a simple "probability to yield at the end of each basic block".)

So is this a bug in how stdlib uses the futex API, or in how Miri implements it?

[cbeuw]

but it's still incorrect and will deadlock again once we add pre-emptive scheduling (my recent attempt to add a very simple time-quanta based preemption failed due to deadlocks in tests. I didn't figure out why but I think this is it)

Further thoughts: actually we won't be pre-empting inside Miri shims:

 * CPU 0                               CPU 1
 * // the following two accesses are in MIR and preemptable - though if they are reordered the wait call will return immediately
 * val = *futex;
 *                                     *futex = newval;
 *                                     // cannot preempt
 *                                     sys_futex(WAKE, futex);
 *                                       futex_wake(futex);
 *                                       if (queue_empty())
 *                                         return;
 * // cannot preempt
 * sys_futex(WAIT, futex, val);
 *   futex_wait(futex, val);
 *   uval = *futex;           // guaranteed to read newval and not val if we use RMW
 * 
 *   if (uval == val)
 *      lock(hash_bucket(futex));
 *      queue();
 *     unlock(hash_bucket(futex));
 *     schedule();

So is this a bug in how stdlib uses the futex API, or in how Miri implements it?

How Miri implemented it. I think changing the atomic read to a fetch_add(0) will do the trick

[RalfJung]

Yes, Miri shims are atomic.

Thanks for the detailed analysis! I cannot say I have fully understood it yet; I'll need more time to carefully re-read everything you wrote. But if you know what the fix is, then if you could create a PR (with a testcase, of course), that would be great. :)