Observer pattern in Rust

ivstas · November 21, 2017, 12:27pm

This problem was already mentioned (How can I correctly implement observer pattern in Rust?), however, I didn’t find any appropriate solution.

Here is the basic code structure I wrote to implement observable pattern. I omitted rest of the methods for simplicity and compactness.

pub struct Observable<'l, T> {
    value: T,
    observers: Vec<Box<FnMut(&T) + 'l>>,
}

impl <'l, T> Observable<'l, T> {
    pub fn subscribe<Callback>(&mut self, observer: Callback) where Callback: FnMut(&T) + 'l {
        self.observers.push(Box::new(observer));
    }

    fn notify_observers(&mut self) {
        for observer in &mut self.observers {
            observer(&self.value);
        }
    }
}

However, implementation above is broken by design. And here is why:
This is what I’m trying to achieve in terms of lifetime:

            lifetime of observable
|--------------------------------------------|

          |-----------------------|
            lifetime of observer

Since I store reference to observers in my observable implementation, these references must outlive observable itself (so Rust can guarantee refs would be valid).
So here is what I got:

            lifetime of observable
          |-----------------------|

|--------------------------------------------|
            lifetime of observer

This is definitely the opposite of what I wanted to achieve.

It is because observable pattern implies having a circular reference: you should have a reference to observable to subscribe to it’s changes and, vice versa, after you subscribed you should have a reference to observer so you can notify it about the updates.

The most close solution that solves a problem mentioned: https://stackoverflow.com/a/39691337/1730696

It uses Rc and Weak. However, the code looks weird and it is definitely not an elegant solution we usually have in Rust.

One of solutions I though about is to have some kind of Subscription struct that automatically unsubscribes when being droped. However, I don’t see any possibility to implement this without using unsafe code.

Are there any opportunities I missed? Or any other approach that makes possibile to achieve what I want w/o using unsafe?

vitalyd · November 21, 2017, 12:46pm

You’re owning the observers, not storing references to them. The 'l lifetime parameter is allowing your observers to capture a reference to their environment, which means the environment they capture must outlive the Observable itself. Is that what you don’t like?

Maybe you provide a sample usecase that you’d like to achieve.

ivstas · November 21, 2017, 1:12pm

something like this:

let mut observable = Observable::new(0);

let mut counter = 0;
{
    let mut increment = |_| {
        counter += 1;
    };

    let subscription = observable.subscribe(increment);

    observable.set(5);
    assert_eq!(1, counter);
} // subscription is dropped

observable.set(10);
assert_eq!(1, counter);

frehberg · November 21, 2017, 1:42pm

I googled for “boost::signal for Rust” and found references to frappe - FRP library for Rust. Looks up2date and well maintained

vitalyd · November 21, 2017, 2:21pm

Yeah ok, you want the observer to borrow an environment that lives for less than the observable. I’m not sure how you’d implement that without unsafety in play because I don’t think this lifetime/scope restriction/constraint is expressible.

Maybe you can also decouple the observer and observable by putting a channel in the middle and communicating notifications over it rather than direct method calls.

ivstas · November 21, 2017, 2:41pm

You’re right, the issue is to have an environment that lives less than the observable.
I hope that I’m just not experienced enough in Rust and it is possible to implement this avoiding unsafe code.

Did you mean std::sync::mpsc::channel?
Unfortunatelly, it requires a call from reciever to obtain changes rather than notifying reciever from producer.

kornel · November 21, 2017, 2:50pm

If this isn’t absolutely performance-critical to the point of needing to use only memory on stack, consider using Rc/Arc instead. This will make all references easier to work with.

vitalyd · November 21, 2017, 4:31pm

Is it a requirement to use (ie modify) stack variables from the observer? This is basically what @kornel is asking. If not, you can put the shared state into an Rc<RefCell<...>>> (or Rc<Cell<...>>) and keep it alive that way.

I meant channel in the abstract sense - haven’t thought about the particular implementation.

ivstas · November 21, 2017, 5:20pm

No, definitely not.

I'll try this way. Thank you!

ivstas · December 9, 2017, 11:41am

I eventually tried to implement Observer using raw pointer to store callbacks.

pub struct Observable<T> {
    value: T,
    callbacks: Vec<*const FnMut(&T)>,
}

However, when I try to add callback:

impl <T> Observable<T> {
    // 's (callback) lifetime is shorter than 'l (self) lifetime
    pub fn subscribe<'s, 'l: 's, C>(&'l self, callback: C) where C: FnMut(&T) + 's {
        let ptr = &callback as *const C;
        self.callbacks.push(ptr);
    }
}

compilation fails: error[E0310]: the parameter type C may not live long enough.

It seems that Vec<*const FnMut(&T)> in structure definition has some implicit 'static lifetime on FnMut.

Is there a way to store raw pointer to FnMut w/o lifetime?

vitalyd · December 9, 2017, 3:41pm

Did you try the Rc<RefCell<...>> approach?

What you have above with the raw ptr is incorrect - the callback is moved into the subscribe method and will be dropped at the end of the method if nothing retakes ownership of it. If you just add a raw ptr of it to the Vec you’ll be holding a dangling pointer.

ivstas · December 9, 2017, 4:05pm

No, since I’m planning to return Subscription from this method (where I will own passed callback).
In Rc<RefCell<..>> approach i would return Rc<RefCell<Subscription>>, that seemed to me as weird API, so I decided to try raw pointers first.

The intention was to store raw pointer as they should have no lifetime guarantess and don’t place any restruction on callbacks lifetime. However, I can’t find any way to store Vec of callbacks without mentioning any lifetime.

vitalyd · December 9, 2017, 4:12pm

You can try hiding that inside some type that internally has the Rc and RefCell and return that type. This is a common approach in Rust where you have a Foo type that your API exposes and then Foo has some Inner type that contains the guts.

But who’s going to own the closure if the Vec only has a raw ptr to it? You can allocate the closure on the heap and then leak it and store the ptr in the Vec. You’d need to find a place in code to drop that closure manually so you don’t leak forever.

If you allocate a closure on the heap then you just pass a Box to subscribe, leak the box there and store the ptr. This closure would have no lifetime parameters - Box<FnMut...> is essentially Box<FnMut... + 'static>.

ivstas · December 9, 2017, 4:16pm

Subscription, that will implement Drop and will remove raw pointer to itself from observable when being dropped.

vitalyd · December 9, 2017, 4:46pm

Ah ok, I was going by your sample code a bit too much but ok - sounds doable.

Edit: re-reading this thread from the top, I see you want to have a closure that captures a reference to a stack variable. You’ll need to transmute your closure to have a 'static lifetime before adding to the Vec. That’s obviously a lie we’re telling the compiler (and hence unsafe is required). So just be careful in how you use this code to make sure you don’t hit UB due to dangling references. I’d still explore a safe approach before resorting to this but you’ll know better.

lacourc · December 9, 2017, 10:49pm

Could you please explain what exactly you are trying to achieve with your Observer? I ask because I want to think about if there is an overall better solution in Rust.

ivstas · December 10, 2017, 10:05am

I’m trying to create some primitives for UI state.
Imagine a UI form with some inputs and View-Model for It.
Lets assume we have to block from submitting if user entered invalid data.

I want to have observable model with auto-derivable parts (e.g. you change input and validation is automatically recomputed).

fasihrana · September 4, 2018, 10:10am

Did you ever get to an acceptable answer for this question?

ivstas · September 4, 2018, 7:45pm

I have tried to use several ideas suggested here, but unfortunately I didn’t manage to get a compiling code.
Every time I try a new approach, I get some new issues with compiling it.

farnbams · September 10, 2018, 6:49am

To give a summarize to someone who looks at this topic and to mix in my experience. Here are two working implementations of patterns working like the observer pattern:

Observer has its own thread

If the observing objects have their own worker threads, I prefer to use the sync::mspc channels. The Observabel holds a list of mspc::Senders. If an Observer goes out of scope Sender::send() will fail and the Sender will be removed from the Observable's list.

Sharing `Weak` references

If the Observable's thread should call the update function I prefer giving the Observabel a list of Weak<Mutex<Observer>> (or RefCell, RwLock, depending on the use case). The Weak-reference decouples Observable and Observer. Like in the channel solution a reference is removed if upgrading the reference fails. In this case Observer is a trait with some function like update() that is called by the Observable for each element of its list of listeners.

Using Closures

Like the author of this topic I haven’t encountered a way to store closures at the Observable. Maybe their would be a way in moving Weak references inside the closures. If someone has a solution for this I would be very interested in it.

Observer pattern in Rust

Observer has its own thread

Sharing Weak references

Using Closures

Sharing `Weak` references