Support for Condition Variables

[kaimast]

Is your feature request related to a problem? Please describe.
Imho, it is fairly unintuitive that there is no equivalence to std::sync::CondVar in tokio.

Describe the solution you'd like
Provide a tokio::sync::CondVar that works with tokio::sync::Mutex.

Describe alternatives you've considered
I tried using the other primitives in tokio::sync and ended up using an AtomicUsize (to track the number of waiting tasks) and a tokio::sync::Semaphore to wake them up. It works but it is kind of ugly.

Additional context
Is this something on the todo list for tokio? If not, would patches be accepted? I could probably work on this.

[carllerche]

Condvars don’t really fit here. Notify would be the strategy to use. Does that fit your requirement?

[carllerche]

This is now tracked by #3066

[kaimast]

Correct me if I am wrong, but I do not think this is implemented. Condition variables work in conjunction with a mutex. There is no straightforward way to do this with Notifier.

There needs to be a way to pass a mutex guard to Notifier::notified. Releasing the lock before calling notified or notify_all is not an option as it leads to various race conditions...

[Darksonn]

I think something like this makes sense to have. Not sure how it fits in with our async mutex though.

[kaimast]

I wrote something like this for my particular use case. Ideally, a similar implementation would be part of tokio or tokio-util.

use std::sync::atomic;
use tokio::sync::{Mutex, MutexGuard, Notify};

pub struct Condvar {
    notifier: Notify,
    /// Atomic here is only used to make Convdar Send and Sync
    /// Callers still need to hold the associated mutex to avoid data races
    wait_count: atomic::AtomicU32,
}

impl Condvar {
    pub fn new() -> Self {
        Self{ wait_count: atomic::AtomicU32::new(0), notifier: Notify::new() }
    }

    pub async fn wait<'a, 'b, T>(&self, lock: MutexGuard<'a, T>, mutex: &'b Mutex<T>) -> MutexGuard<'b, T> {
        self.wait_count.fetch_add(1, atomic::Ordering::SeqCst);
        drop(lock);

        self.notifier.notified().await;
        mutex.lock().await
    }

    pub fn notify_one<'a, T>(&self, lock: MutexGuard<'a, T>) {
        let count = self.wait_count.load(atomic::Ordering::SeqCst);

        if count > 0 {
            self.wait_count.store(count-1, atomic::Ordering::SeqCst);
            drop(lock);

            self.notifier.notify_one();
        }
    }

    pub fn notify_all<'a, T>(&self, lock: MutexGuard<'a, T>) {
        let count = self.wait_count.swap(0, atomic::Ordering::SeqCst);
        drop(lock);

        for _ in 0..count {
            self.notifier.notify_one();
        }
    }
}

[Darksonn]

Typically you would not pass the lock to the notify methods. By using some different atomic instructions, you can make remove the need for locking around them.

Feel free to open a new issue or PR if you have a proposal for adding them.

[kaimast]

I created pull requests for my implementation: #3741 and #3742.