Fix wake_by_ref() weak memory issue

[frostyplanet]

Version
tokio 1.47.1

Platform
github runner macos-15

rustc 1.89.0 (29483883e 2025-08-04)

Default host: aarch64-apple-darwin
rustup home:  /Users/runner/.rustup

installed toolchains
--------------------
stable-aarch64-apple-darwin (active, default)

active toolchain
----------------
name: stable-aarch64-apple-darwin
active because: it's the default toolchain
installed targets:
  aarch64-apple-darwin

Description

There is still dead lock after issue #7589, calling waker_by_ref from another thread, does not wake up the task. So I added more debug log and came up with a patch. (tracked by frostyplanet/crossfire-rs#39)

The test code
https://github.com/frostyplanet/crossfire-rs/blob/ae7a430416ea6488f698454c5ebdda050ba126c1/src/tests/test_async.rs#L857
case(mpmc::bounded_async::(1)

In this scenario, there are 2 senders (one async task and one thread), 2 receivers (one async task and one thread) .
hang at lease three times while running under a current-thread tokio runtime.

The debug log was added inside tokio:
https://github.com/frostyplanet/tokio/compare/5e509ca7dc75112f2a6e2840c364f07b2e7f691d..6a96ec96cecbef5ba37a82a196501ffa14f25021

Original log can be downloaded from https://github.com/frostyplanet/crossfire-rs/actions/runs/17850277588/job/50757354826
The last part of the log before hang.

[2025-09-19 06:45:00.469416][DEBUG][ThreadId(2)][async_rx.rs:230] rxSome(Id(2)): recv waker(8227)
[2025-09-19 06:45:00.469418][DEBUG][ThreadId(2)][waker_registry.rs:209] rxSome(Id(2)): reg waker(8228)
[2025-09-19 06:45:00.469419][DEBUG][ThreadId(2)][async_rx.rs:265] rxSome(Id(2)): commit_waiting waker(8228) 1
[2025-09-19 06:45:00.469419][DEBUG][ThreadId(2)][harness.rs:219] task 2 transition_to_idle Ok
[2025-09-19 06:45:00.469448][DEBUG][ThreadId(6)][waker_registry.rs:209] rxNone: reg waker(8229)
[2025-09-19 06:45:00.469450][DEBUG][ThreadId(6)][blocking_rx.rs:133] rx: waker(8229) commit_waiting state=1
[2025-09-19 06:45:00.469454][DEBUG][ThreadId(5)][state.rs:271] waker.wake_by_ref (idle)
[2025-09-19 06:45:00.469456][DEBUG][ThreadId(5)][mod.rs:661] task Id(2) outside push
[2025-09-19 06:45:00.469457][DEBUG][ThreadId(5)][waker_registry.rs:392] wake rx waker(8228) Waked
[2025-09-19 06:45:00.469460][DEBUG][ThreadId(5)][waker_registry.rs:52] txNone: reg waker(9253)
[2025-09-19 06:45:00.469460][DEBUG][ThreadId(5)][blocking_tx.rs:149] tx: sender_reg_and_try Some(waker(9253)) state=1

Thread 5 is a sender in blocking context, it tried to wake task Id(2) by waker(8228). We can see task Id(2) transit from idle to notified.

[2025-09-19 06:45:00.469454][DEBUG][ThreadId(5)][state.rs:271] waker.wake_by_ref (idle)
[2025-09-19 06:45:00.469456][DEBUG][ThreadId(5)][mod.rs:661] task Id(2) outside push
[2025-09-19 06:45:00.469457][DEBUG][ThreadId(5)][waker_registry.rs:392] wake rx waker(8228) Waked

And then pushed to schedule. My guess is that the worker thread will unpark, but saw nothing to run.

src/runtime/scheduler/current_thread/mod.rs

    fn schedule(&self, task: task::Notified<Self>) {                                                                        
        use scheduler::Context::CurrentThread;                                                                              
    
        context::with_scheduler(|maybe_cx| match maybe_cx {                                                                 
            Some(CurrentThread(cx)) if Arc::ptr_eq(self, &cx.handle) => {                                                   
                let mut core = cx.core.borrow_mut();
    
                // If `None`, the runtime is shutting down, so there is no need                                             
                // to schedule the task.                                                                                    
                if let Some(core) = core.as_mut() {
                    tracing::debug!("task {:?} push cx", task.task_id());
                    core.push_task(self, task);
                } else {
                    tracing::debug!("task {:?} push cx None", task.task_id());                                              
                }
            }   
            _ => {
                // Track that a task was scheduled from **outside** of the runtime.                                         
                self.shared.scheduler_metrics.inc_remote_schedule_count();                                                  
                tracing::debug!("task {:?} outside push", task.task_id());                                                  
                // Schedule the task                                                                                        
                self.shared.inject.push(task);
                self.driver.unpark();                                                                                       
            }
        }); 
    }

In src/runtime/scheduler/inject.rs, push() is protected with lock, but pop() uses is_empty() before lock.

    /// Pushes a value into the queue.
    ///
    /// This does nothing if the queue is closed.
    pub(crate) fn push(&self, task: task::Notified<T>) {
        let mut synced = self.synced.lock();
        // safety: passing correct `Synced`
        unsafe { self.shared.push(&mut synced, task) }
    }

    pub(crate) fn pop(&self) -> Option<task::Notified<T>> {
        if self.shared.is_empty() {
            return None;
        }

        let mut synced = self.synced.lock();
        // safety: passing correct `Synced`
        unsafe { self.shared.pop(&mut synced) }
    }

In src/runtime/scheduler/inject/shared.rs, len() is using Acquire, and push is using Release.
So I think the problem is here, because I had a similar piece of code which reported a deadlock by Miri.
frostyplanet/crossfire-rs@0e790a6

Although Miri emulates weak memory by delaying the effect on load() and store(),
and Arm is other-multi-copy arch, from which load and store effects are immediate.
My theory is that it's possible self.driver.unpark() to be re-order inside the mutex scope, wake up the worker before setting len, therefore pop() might not see the change in len.

So I have changed load() and store() to SeqCst, the tests passed.
The patch:
frostyplanet@2fb5aaf
Test results:
https://github.com/frostyplanet/crossfire-rs/actions/runs/17851772944
https://github.com/frostyplanet/crossfire-rs/actions/runs/17852889652
(NOTE: #7622 was opt-out from the dep in the workflow, in order to reproduce hang more frequently)

[Darksonn]

The task is pushed like this:

self.shared.inject.push(task);
self.driver.unpark();

So when the runtime wakes up from the unpark() call, the following call to pop is guaranteed to see everything that happened before unpark(). This means that pop is guaranteed to see a value.

So I don't see any bug. You will have to provide more information about the issue.

[frostyplanet]

It's possible instruction get re-order by cpu or compiler, if there's not a SeqCst fence for the len value. For example, execute in the order

self.driver.unpark();
self.shared.inject.push(task);

so there would be a deadlock.

[Darksonn]

No, that's not how it works. When calling unpark() it will use a release operation even if the thread is already unparked, so when the thread receives the notification it will synchronize-with the notification. I looked at the unpark() logic, and it does not have the same issue as #7622.

[frostyplanet]

The problem is not in unpark(), but in self.shared.inject.push() and self.shared.inject.pop(), here's my patch frostyplanet@2fb5aaf

[frostyplanet]

in push(), the order is as follow:

self.synced.lock();  // acquire
read len without atomic
len.store(Release) 
guard drop // release
unpark()

both guard dropping and len.store is Release, it does not stop the code below (like unpark) from being re-order to previous position.
So replacing len.store(Release) with a SeqCst fence should guarantee the expected behavior.

[frostyplanet]

and in pop()

if is_empty() -> len() == 0  -> len.load(Acquire) 
return None

if len reads old value on platform weaker than Arm (for example Power), and skip the lock , like #7622, there's also problem.

[Darksonn]

unpark() itself performs a release operation, so things above unpark() must not be moved below unpark(). It's not possible for pop() to read an old value in this scenario.

[frostyplanet]

Because in this scenario, push() and pop() execute on two different cpu cores, there's no sync-with relationship between is_empty() reads true and push(), even though push did use store(Release), but is_empty() is not in a loop, so no guarantee of that.
According to the author of Miri. I asked the question with my code (which has similar logic), put in one thread but the other thread think it's empty.

    #[inline(always)]
    pub fn put(&self, item: Weak<T>) {
        let old_ptr = self.ptr.swap(item.into_raw() as *mut T, Ordering::SeqCst);
        if old_ptr != ptr::null_mut() {
            let _ = unsafe { Weak::from_raw(old_ptr) };
        }
    }

    pub fn pop(&self) -> Option<Arc<T>> {
        if self.ptr.load(Ordering::Acquire) == ptr::null_mut() {
            return None;
        }
        let ptr = self.ptr.swap(ptr::null_mut(), Ordering::SeqCst);
        if ptr != ptr::null_mut() {
            return unsafe { Weak::from_raw(ptr) }.upgrade();
        } else {
            None
        }
    }

He replied that rust-lang/miri#4521 (comment)

For your last example ("example 2" in your second post), note that the A.load in thread b is allowed to return an outdated value -- even if the A.store "already happened" in another thread, there's no guarantee that that store has propagated to thread b. That's how both load can return false. This is a classic weak memory example.

Does weak memory emulation re-order operations of atomic load/store, or just delay the result to be seen by RNG?

Note that nothing gets reordered anywhere in Miri. Thinking of the memory model in terms of reorderings is sometimes a useful approximation, but it's not how the memory model actually works so you will likely be confused if you stick to that way of thinking.

See https://marabos.nl/atomics/ for a book with a lot more information on weak memory in Rust.

Using Release ordering to store a value to atomic in one thread, and then using Acquire to load the value from another thread, will it read the new value?

Acquire loads can return outdated values. Even SeqCst loads can sometimes do that but the SC total order heavily restricts this. Acquire loads have no such restriction (beyond the usual "will return a value that's at least as up-to-date as as the most recent write of the current thread and everything acquired from other threads").

A fetch_add(0, AcqRel) can be used to force a load of the latest value in some sense, but this is highly unusual so you should make sure you actually understand the underlying model before resorting to such tricks. Miri can find some weak memory issues, but it can't (yet) find all of them, so Miri being happy cannot replace your own analysis of your code.

[frostyplanet]

The patch did resolve my problem, you can see that test workflow on macos-15 had not been hang after applying the fix https://github.com/frostyplanet/crossfire-rs/actions/workflows/cron_dev_arm_trace.yml

[frostyplanet]

Additional explaination: waky_by_ref() called outside tokio context (it's executed in a pure thread), try to wake a task. When worker get's notified to unpark(), it will use pop() to check if there's task in the queue. I can unstandard that the purpose of is_empty() checkout without lock is to optimise for an empty condition, but len.load(Acquire) can not establish sync-with relationship with store(Release) from other thread, because it will return immediately (and go to park again) when reading 0, there might not be another one to wake it after it missed the task, (not like the logic in #7622 there's a loop with load + CAS to retry). So the only solution is use SeqCst to get a strong enough isolation

[Darksonn]

The difference between this thread and the example you shared with @RalfJung is that there is a happens-before relationship from push to pop due to the use of unpark.

[frostyplanet]

I am unfamiliar with the driver code, in order to move on with this issue, so I'd just add more log to find out what happens