[Solved] Help: Atomic operation goes wrong in this situation

[kingwingfly]

Version

cargo tree | grep tokio
└── tokio v1.27.0

Platform
Both MacBook and Windows PC.

Darwin louismac.mshome.net 22.5.0 Darwin Kernel Version 22.5.0: Sun Apr  2 19:24:16 PDT 2023; root:xnu-8796.120.31~10/RELEASE_ARM64_T8103 arm64

Description
In this situation:
async function A is fetch_adding an AtomicUsize and may live longer than B,
function B spawns A, and loads the AtomicUsize while the async function A didn't finish.

It's hard to describe, so I write a minimal reproducible example.

I tried this code:

use std::sync::{
    atomic::{AtomicUsize, Ordering},
    Arc,
};

// A helper function to create a Runtime
fn create_rt() -> tokio::runtime::Runtime {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
}

// Add 100 to a AtomicUsize
async fn async_add_100(num: Arc<AtomicUsize>) {
    // Some jobs need await, so using async is unavoidable
    num.fetch_add(100, Ordering::SeqCst);
}

pub struct Adder {
    num: Arc<AtomicUsize>,
}

impl Adder {
    pub fn new() -> Self {
        Self {
            num: Arc::new(AtomicUsize::new(0)),
        }
    }

    // do add_100 for 10 times
    pub async fn run(&self) {
        for _ in 0..10 {
            tokio::spawn(async_add_100(self.num.clone()));
        }
        // because I have some more jobs to do after changing, so:
        loop {
            if self.num.load(Ordering::Relaxed) == 1000 {
                break;
            }
        }
        // I know there are better ways, but this way can lead to unexpected result
        // Also, just sleeping 2 seconds leads to the same consequence.
        // std::thread::sleep(std::time::Duration::from_secs(2));


        // some other jobs
    }
}

pub struct Worker {
    rt: tokio::runtime::Runtime,
}

impl Worker {
    pub fn new() -> Self {
        Self { rt: create_rt() }
    }

    pub fn exe(&self) -> usize {
        let adder = Arc::new(Adder::new());
        let adder_c = adder.clone();
        self.rt.spawn(async move {
            adder_c.run().await;
        });
        // sleep 1 second to ensure adder.num finished changing
        std::thread::sleep(std::time::Duration::from_secs(1));
        // query and return the result, expected is 1000
        adder.num.load(Ordering::Relaxed)
        // However,the result is 900 as the last adding goes wrong,
        // which is different from the similar std::thread implement's behavior
        // ps: I'm sure once there's possibility that run() still running while load() happens,  
        // the last spawned adding will never execute
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn it_works() {
        let adder = Adder::new();
        let rt = create_rt();
        rt.block_on(adder.run());
        let res = adder.num.load(Ordering::Relaxed);
        assert_eq!(res, 1000);
    }

    #[test]
    fn it_not_works() {
        let worker = Worker::new();
        let res = worker.exe();
        assert_eq!(res, 1000);
    }
}

I expected to see this happen:

The result should be 1000 in the second test.

Instead, this happened:

However, the result is 900.

I also did the same thing using std::thread:

use std::sync::{
    atomic::{AtomicUsize, Ordering},
    Arc,
};
use std::thread;

fn add_100(num: Arc<AtomicUsize>) {
    num.fetch_add(100, Ordering::SeqCst);
}

pub struct Adder {
    num: Arc<AtomicUsize>,
}

impl Adder {
    pub fn new() -> Self {
        Self {
            num: Arc::new(AtomicUsize::new(0)),
        }
    }

    pub fn run(&self) {
        for _ in 0..10 {
            let num_c = self.num.clone();
            thread::spawn(move || add_100(num_c));
        }
        loop {
            if self.num.load(Ordering::Relaxed) == 1000 {
                break;
            }
        // or just
        // std::thread::sleep(std::time::Duration::from_secs(2));
        }
    }
}

pub struct Worker {}

impl Worker {
    pub fn new() -> Self {
        Self {}
    }

    pub fn exe(&self) -> usize {
        let adder = Arc::new(Adder::new());
        let adder_c = adder.clone();
        thread::spawn(move || adder_c.run());
        // sleep 1 second to ensure adder.num finished changing
        std::thread::sleep(std::time::Duration::from_secs(1));
        adder.num.load(Ordering::Relaxed)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn it_works() {
        let adder = Adder::new();
        adder.run();
        let res = adder.num.load(Ordering::Relaxed);
        assert_eq!(res, 1000);
    }

    #[test]
    fn it_also_works() {
        let worker = Worker::new();
        let res = worker.exe();
        assert_eq!(res, 1000);
    }
}

It works as I want.

[Darksonn]

This is because you are blocking the thread. One of the tasks you spawned was scheduled to run next on the current thread (via the LIFO slot), and the other tasks were sent to other worker threads in the runtime. Since your task never yields, the task in the LIFO slot is never scheduled, meaning that it's add is not seen when the timeout triggers.

[Darksonn]

To fix this, you need to change your code to not block the thread (see my link for an explanation). The simplest would be to use yield_now.

loop {
    if self.num.load(Ordering::Relaxed) == 1000 {
        break;
    }
    tokio::task::yield_now().await;
}

However, this solution is very inefficient.

Here are some other possible solutions:

1. Replace the atomic with a watch channel. It lets you wait for the value of the counter to change in a non-blocking manner.
2. Replace the atomic with a semaphore. If you replace the loop with acquire(1000).await and have add_100 call add_permits(100), then you would get the correct behavior.
3. Use a Notify to wait for changes. Any time you change the atomic, you check if it has reached the value 1000, and if so, you emit a wakeup. The sleeping task will receive the wakeup and check the atomic.

[kingwingfly]

Do you think this is a good way?

pub async fn run(&self) {
        let num = self.num.clone();
        tokio::task::spawn_blocking(move || {
            for _ in 0..10 {
                tokio::spawn(async_add_100(num.clone()));
            }
        });
        std::thread::sleep(std::time::Duration::from_secs(2));
        dbg!(&self.num.load(Ordering::Relaxed));

        // some other jobs

[Darksonn]

I wonder how to explicitly spawn a job onto a non-current thread.

Your mistake is that run is spawned on the runtime, but is blocking the thread. You should not attempt to fix this by disabling the lifo slot or by forcing the tasks to end up on a different worker thread. That's extremely fragile and bad design.

Either, run should not be async and should be executed outside of Tokio, or you should stop it from blocking the thread.

[Darksonn]

Do you think this is a good way?

No, it's terrible.

[kingwingfly]

My CPU is broken, ahhhh ahhhh ahhh ……

All in all, I should delete loop instead of keeping it. And blocking is silly in an async block and may cause a deadlock.

I learnt a lot from you. Thanks a lot!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!