How is thread safety achieved for observable::subscribe? Who should care about lifetime?

[daixtrose]

I am curious how rpp solves the lifetime issues during subscription. I am referring to this code:

    class observable
    {
        // ...

        template<constraint::observer_strategy<Type> ObserverStrategy>
        void subscribe(observer<Type, ObserverStrategy>&& observer) const
        {
            if (!observer.is_disposed())
                // ==> I see a sequence point here <==
                m_strategy.subscribe(std::move(observer));
        }

Do we really need the is_disposed() check? Does it save us from lifetime issues? Is it thread-safe?

I may be wrong with what I say here, so please forgive the noise in that case.

AFAICS at the marked sequence point anything could happen to observer, especially a dispose call coming through anyway. Or a destructor kicking in, making the move not very successful.

Generally speaking, IMHO any Reactive Extension C++ library should make it clear that lifetime issues are not its business, but the user's responsibility. So for me, it would be fine if observable::subscribe returned a (movable) subscription object which I need to keep alive myself. The destructor of subscription should unsubscribe its observers from the observable. I believe that emulating the lifetime management of Reactive Extension implementations in managed programming languages is a huge burden for rx C++ library authors.

Saving CPU cycles like this:

        template<constraint::observer_strategy<Type> ObserverStrategy>
        auto subscribe(observer<Type, ObserverStrategy>&& observer) const
        {
                return m_strategy.subscribe(std::move(observer));
        }

I am pretty sure there are good arguments for your current design and I would like to hear your opinion on this.

[victimsnino]

Hmmmm, really good catch and really good question. Good mark for me that is should be covered in docs =)

Lets move one by one:

1. Do we really need the is_disposed() check? Does it save us from lifetime issues? Is it thread-safe?
   Yes, we need it, but ONLY to be sure that observer is (still) interested in this emissions. Why do we need it? Let's imagine we have something like this:

disposable.dispose();
observable
| rpp::ops::subscribe_on(rpp::schedulers::new_thread{})
| rpp::ops::subscribe_on(rpp::schedulers::new_thread{})
| rpp::ops::subscribe_on(rpp::schedulers::new_thread{})
| rpp::ops::subscribe(disposable, [](int){});

^^^^^ really synthetic example, but: there obsefvable is pretty "heavy to subscribe" - it starts 3 threads for each subscriptions. In this case observer is JUST disposed for some reason before subscribe. Do we really need to move through all subscriptions chain to understand later that observer is not interested it that fact? I think no. It is pretty arguable decision, but main idea was: when observer reaches observable's subscribe - observer was interested in this action (at least, at the moment of begining of this call)

One more thing about

Is it thread-safe?

Technically observer is SINGLE-THREADED object. All calls to observer HAVE to be done serially (it is why all observers are not copyable, but just movable by default). All rpp's operators handling it properly (not doing copies of observer and etc. If we need it, then access observer under lock).
Only exception are subjects, but they are thread-safe.

2. AFAICS at the marked sequence point anything could happen to observer, especially a dispose call coming through anyway. Or a destructor kicking in, making the move not very successful.

Yeah, it is true, but we can't handle it somehow properly

3. About subscription and subscribe: there such an overloading calling subscribe_with_disposable (technically, subscription == disposable only another internal meaning). Why it is not default behavior?? creation of disposable is forcing to use stack due to disposable have to be created via std::make_shared. But in practice user don't need disposable/subscription in most cases, so, we can omit it and use local disposable instead (it is 1 of the reasons WHY rpp is so much faster). for example,

rpp::source::just(rpp::schedulers::immediate{}, 1) | rpp::ops::subscribe([](int){});

could happens without ANY heap allocations (in contrast to rxcpp)

So, in summary, it means that in most cases you behave like "hey, observable, take this observer and just call its methods, i don't care" (== default method without return value), while if you want to control observable-observer relationships, you need to call subscribe_with_disposable method or just use overloading accepting disposable + observer.

[daixtrose]

creation of disposable is forcing to use stack due to disposable have to be created via std::make_shared

Did you mean heap?

[victimsnino]

Yeah, my bad, of course i mean heap =)

[victimsnino]

So, shorter summary:

• observer should be single-threaded (at least RPP expecting and using them as single-threaded NOT guarded/safe objects)
• if you are initiate such an situation where observer used as multi-threaded object (for example, pass same observer to multiple observables) -> you are violating this rule and YOU have to solve this somehow

[daixtrose]

I am super unhappy that dispose() and the destructor are two different things. I am yet to be convinced this is necessary.

[victimsnino]

It is how rxcpp also works. and how other RX implementations also works. Kirk described it there: ReactiveX/RxCpp#215

In short: would it be expected behavior?

auto observer = ....;
{
    auto subscription = observable.subscribe(observer);
} // subscription calls unsubscribe there

For me it is not expected. In most cases i'm event don't use subscription object at all, so, in my case it would be like:

auto observer = ....;
observable.subscribe(observer);
// there instant unsubscribe due to temporary object

we can add scoped_guard or something like this, but i'm not sure if this should be default behavior

[victimsnino]

Minor update:
Actually dispose happens automatically inside destructor, but inside destructor of underlying shared_ptr state, not wrapper. https://github.com/victimsnino/ReactivePlusPlus/blob/v2/src/tests/rpp/test_disposables.cpp#L139

It means, it would auto-dispose when last instance of disposable is dying. In most cases it is when observer is dying (in my cases above)

In my understanding it is enough in terms of life-time management and etc

It was recent discussion in our discord

So, such an solution has advantages of both:

• no auto dispose when "just some instance is destruction"
• 100% guarantees to call dispose when disposable (internal state) is dying

Is it what are you talking about? Or are you talking about auto-dispose in case of dying first ever intance?