Multithreaded observer pattern in Rust

Question

I often want to program in an event driven way, but classic implementations of the observer pattern often pose an ownership challenge, and they only get more difficult once multiple threads are involved. This approach uses a polling loop in a thread (but not busy waiting), and "notifies the observer" (calls a call-back function), whenever the data contained in the Observable (my ArcReactor) changes.

I'm pretty happy with this, but would like opinions from others before I start using it in projects.

The benefits of this over single-threaded implementations of the observer pattern are that it works out of the box when the observable is changed in a different thread from the observers.

use std::sync::{Arc, Mutex, Condvar};

struct ArcReactor<T>(Arc<(Mutex<T>, Condvar)>);
impl<T: 'static + Clone + Send> ArcReactor<T> {

    /// Creates a new `Arc<(Mutex<T>, Condvar)>` with data: T stored in the mutex.
    fn new(data: T) -> ArcReactor<T> {
        ArcReactor(Arc::new((Mutex::new(data), Condvar::new())))
    }
    /// Write to the value contained in the mutex, and notify the condvar
    fn write(&mut self, val: &T) {
        let &(ref lock, ref cvar) = &*self.0;
        let mut data = lock.lock().unwrap();
        *data = (*val).clone();
        cvar.notify_all();
    }
    /// Read from the value contained in the mutex
    fn read(&self) -> T {
        let &(ref lock, _) = &*self.0;
        let data = lock.lock().unwrap();
        data.clone()
    }
    /// Get a clone of this object. Because this object is a refference type (Arc), any
    /// modifications to the clone will be reflected in the original object.
    fn clone(&self) -> ArcReactor<T> {
        ArcReactor(self.0.clone())
    }

    /// Blocks the current thread until the condvar receives a notification
    /// Consumes no CPU time while waiting for an event to occur.
    fn wait_for_change(&self) {
        let &(ref lock, ref cvar) = &*self.0;
        let _ = cvar.wait(lock.lock().unwrap()).unwrap();
    }

    /// Starts a new thread that waits for the condvar to receive a notification,
    /// and calls the callback when it does.
    /// Consumes no CPU time while waiting for an event to occur
    /// This may miss a change if the value is changing faster than this loop can process
    fn on_changed(&self, callback: fn(T)) -> std::thread::JoinHandle<()> {
        let mirror = self.clone();
        let reader = std::thread::spawn( move || {
            loop {
                mirror.wait_for_change();
                callback(mirror.read());
            }
        });
        reader
    }
}


fn main() {
    
    let mut data = ArcReactor::new(0);
    let data_ref = data.clone();

    let observer_a = data_ref.on_changed(|val| println!("A: {}", val));
    let observer_b = data_ref.on_changed(|val| println!("B: {}", val));

    let writer = std::thread::spawn( move || {
        loop {
            let new = data.read() + 1;
            data.write(&new);
            std::thread::sleep(std::time::Duration::from_millis(100));
        }
    });

    writer.join().unwrap();
    observer_a.join().unwrap();
    observer_b.join().unwrap();
}

Here is a version that waits for all observers to call their call-back before allowing another write. It also has the ability to stop the reader threads:

use std::sync::{Arc, Mutex, Condvar};

struct ArcReactor<T>(Arc<(Mutex<Option<T>>, Condvar, Condvar)>);
impl<T: 'static + Clone + Send> ArcReactor<T> {

    /// Creates a new `Arc<(Mutex<Option<T>>, Condvar)>` with data: T stored in the mutex.
    fn new(data: T) -> ArcReactor<T> {
        ArcReactor(Arc::new((Mutex::new(Option::Some(data)), Condvar::new(), Condvar::new())))
    }
    /// Write to the value contained in the mutex, and notify the condvar
    /// Waits for the observables to be notified before allowing another write.
    fn write(&mut self, val: &T) {
        let &(ref lock, ref write_cvar, _) = &*self.0;
        let mut data = lock.lock().unwrap();
        *data = Option::Some((*val).clone());
        write_cvar.notify_all();
        drop(data);
        self.wait_for_callback();
    }
    /// Read from the value contained in the mutex
    fn read(&self) -> Option<T> {
        let &(ref lock, _, _) = &*self.0;
        let data = lock.lock().unwrap();
        data.clone()
    }
    /// Get a clone of this object. Because this object is a refference type (Arc), any
    /// modifications to the clone will be reflected in the original object.
    fn clone(&self) -> ArcReactor<T> {
        ArcReactor(self.0.clone())
    }

    /// Blocks the current thread until the condvar receives a notification
    /// Consumes no CPU time while waiting for an event to occur.
    fn wait_for_change(&self) {
        let &(ref lock, ref write_cvar, _) = &*self.0;
        let _ = write_cvar.wait(lock.lock().unwrap()).unwrap();
    }

    /// Notify that the callbacks have been called, a new write is now allowed.
    fn notify_callback(&self) {
        let &( _, _, ref callback_cvar) = &*self.0;
        callback_cvar.notify_all();
    }

    fn wait_for_callback(&self) {
        let &(ref lock, _, ref callback_cvar) = &*self.0;
        let _ = callback_cvar.wait(lock.lock().unwrap()).unwrap();
    }

    /// Starts a new thread that waits for the condvar to receive a notification,
    /// and calls the callback when it does.
    /// Consumes no CPU time while waiting for an event to occur
    fn on_changed(&self, callback: fn(T)) -> std::thread::JoinHandle<()> {
        let mirror = self.clone();
        let reader = std::thread::spawn( move || {
            loop {
                mirror.wait_for_change();
                match mirror.read() {
                    Some(data) => {
                        callback(data);
                        mirror.notify_callback();
                    }
                    None => break,
                }
            }
        });
        reader
    }

    ///Writes a None value to the mutex, and notifies the condvar
    fn stop_all_observers(&mut self) {
        let &(ref lock, ref write_cvar, _) = &*self.0;
        let mut data = lock.lock().unwrap();
        *data = Option::None;
        write_cvar.notify_all();
    }

}


fn main() {
    
    let mut data = ArcReactor::new(0);
    let data_ref = data.clone();

    let observer_a = data_ref.on_changed(|val| println!("A got: {}", val));
    let observer_b = data_ref.on_changed(|val| println!("B got: {}", val));

    let writer = std::thread::spawn( move || {
        loop {
            let new = match data.read() {
                Some(val) => val + 1,
                None => break,
            };
            if new < 1000 {
                println!("writing {}", new);
                data.write(&new);
            } else {
                println!("stopping");
                data.stop_all_observers();
                break;
            }
        }
    });

    writer.join().unwrap();
    observer_a.join().unwrap();
    observer_b.join().unwrap();
    
}

Answer 1

One issue is the type of your callback function:

fn on_changed(&self, callback: fn(T)) -> std::thread::JoinHandle<()> {

It is a fn(T) which is a function pointer. This is very limiting, because you can't pass a closure which captures any variables. It'll only be able to access the parameter and global variables. You probably want to take impl Fn(T) + Send + 'static which will allow the callback access to any variables it closes over.

Your code invokes clone in both the read and write functions. This could be problematic if the type in question is quite large. For reading, you should consider a function like this:

fn read(&self) -> std::sync::MutexGuard<T> {
    let &(ref lock, _) = &*self.0;
    lock.lock().unwrap()
}

This won't clone the T but will ensure that doesn't get messed with while it is being used. For write, take the value by ownership instead of borrowing it.

fn write(&mut self, val: T) {
    let &(ref lock, ref cvar) = &*self.0;
    let mut data = lock.lock().unwrap();
    *data = val;
    cvar.notify_all();
}

The caller may clone the value if it needs to. But there's a good chance the caller isn't interested in the value after calling write and can transfer ownership.

fn clone(&self) -> ArcReactor<T> {
    ArcReactor(self.0.clone())
}

This is just a default clone implementation, you use #[derive(Clone)] instead.

struct ArcReactor<T>(Arc<(Mutex<Option<T>>, Condvar, Condvar)>);

The inner tuple with three items is complex enough that I'd put it in a struct with named fields to better be able to keep track of what is what.

observer_a.join().unwrap();

Since your observers don't terminate (baring panics), it is a little strange to join on them.

    let reader = std::thread::spawn( move || {
        loop {
            mirror.wait_for_change();
            callback(mirror.read());
        }
    });

I have to question the use of threads here. Threads are relatively heavyweight. Does it really make sense to launch a thread for each individual observer for every observable? Furthermore, I suspect that with threads you'll end up getting a lot of contention over the Arc/Mutex/any other parts of the system the callbacks touch. This may be a recipe for really bad performance.

But, let's take a moment to think about your callbacks. Right now you just print stuff out. That works, but as soon as you want to do anything more interesting you'll run into difficulties. How are the callbacks going to interact with the rest of the system? Probably, what you'll need to do is wrap anything the callbacks need to touch in Arc<Mutex<?>>.

But this raises its own set of problems. How do you make sure you don't have a cycle between the Arc in the callbacks and the observer? How can you make sure all of your different threads going around locking mutexes don't create a deadlock situation?

The fundamental issue is that any use of a pattern involving callbacks fits poorly into Rust's ownership semantics. You can make it work using enough Arc and Mutex, but you'll feel like you are fighting Rust. Alternatively, don't try to use those patterns in Rust.

What to do instead? In most cases, just write straightforward code. Instead of creating an observable system and then registering callbacks with it, just write code that directly calls what those callbacks would have been.