lifetime issue about rxcpp

[kirkshoop]

I looked at this and found the cause.

The range source schedules each number. The subscribe_on operator schedules the subscribe and the unsubscribe. This follows the .Net pattern and explains why the the code in #127 works after the fix in #128.

The create in this issue does not schedule at all, it uses a loop. Thus it blocks the subscribe_on worker until the whole loop is finished. The unsubscribe is scheduled in the same worker, but does not run until the loop exits.

RxJava has taken a different path. range is a loop that does not schedule and subscribeOn only schedules the subscribe, not the unsubscribe. Thus the unsubscribe would not be scheduled and would execute immediately. This has different consequences on usage.

I have favored the highly granular scheduling up to now, but would be open to discussion and pull-requests to shift to the chunky schedule model.

[kirkshoop]

unrelated to the bug, the above can be written using existing operators:

        auto published_observable =
            rx::sources::interval(std::chrono::milliseconds(100), rx::observe_on_new_thread())
            .publish();

        auto lifetime = rx::composite_subscription();
        lifetime.add([](){
            std::cout << "unsubscribed" << std::endl;
        });

        published_observable.
            ref_count().
            take_until(rx::sources::timer(std::chrono::seconds(1))).
            as_blocking().
            subscribe(lifetime, [](int i){
                std::cout << i << std::endl;
            });

as_blocking().subscribe ensures that the console app does not exit until the sequence is finished.

[daixtrose]

Just dropping in from the side. I heavily used C# Rx for some years now and I found that the scheduler part in the last version of Rx somehow feels right. From the source code and the changes one can see that there were many discussions about the schedulers. If there is some decision to take in which direction RxCpp should go I think that following the scheduler design for the C# version whereever possible might be a GoodIdea (TM) - also for those living in both worlds.

[jpattersonz]

I'd love to dig into the scheduling, but unfortunately won't be able to in the near future. I've reworked my av_source to use the existing scheduling functionality for now. Thanks.

[daixtrose]

Side question @kirkshoop (compiler currently not available):

Can I use auto?

auto published_observable =
    rxcpp::sources::create<int>([](auto & s) // use auto here?

[jpattersonz]

Auto declarations for lambda parameters are part of C++14.

http://en.m.wikipedia.org/wiki/C%2B%2B14

Look for generic lambdas.

Everything in rxcpp thus far is targeting C++11.
On May 20, 2015 3:54 AM, "daixtrose" [email protected] wrote:

Side question @kirkshoop https://github.com/kirkshoop (compiler
currently not available):

Can I use auto?

auto published_observable =
rxcpp::sources::create([](auto & s) // use auto here?

—
Reply to this email directly or view it on GitHub
#129 (comment)
.

[daixtrose]

Oops! I falsely assumed we are using C++14 here.
Is this C++11 decision cast in stone or is there a roadmap for the future when to switch?

[daixtrose]

@kirkshoop wrt your question whether to go the JAVA or C# path here I would like to refer to one of the 101 Rx C# examples:

IObservable<int> ob =
    Observable.Create<int>(o =>
        {
            var cancel = new CancellationDisposable(); // internally creates a new CancellationTokenSource
            NewThreadScheduler.Default.Schedule(() =>
                {
                    int i = 0;
                    for (; ; )
                    {
                        Thread.Sleep(200);  // here we do the long lasting background operation
                        if (!cancel.Token.IsCancellationRequested)    // check cancel token periodically
                            o.OnNext(i++);
                        else
                        {
                            Console.WriteLine("Aborting because cancel event was signaled!");
                            o.OnCompleted();
                            return;
                        }
                    }
                }
            );

            return cancel;
        }
    );

IMHO it is acceptable for a library user to get used to the fact that rxcpp::sources::create should be passed a non-blocking lambda. Admittedly it first was kind of suprising to me (because the tutorials are misleading!), but perfectly logical once I had worked with Rx for some time - and read introtorx: The non-blocking vs. blocking issue is clearly addressd in the C# docs (http://www.introtorx.com/Content/v1.0.10621.0/04_CreatingObservableSequences.html). I think one can live with it for now.

OTOH there might be overloads to create that allow .subscribe_on(/* */) to be more intrusive. Maybe those methods could return something that can be transformed by subscribe_on. Since I am just starting to learn the C++ variant of Rx I currently have no proposal at hand, but I am thinking about something like

rxcpp::sources::create<int>([](const rxcpp::subscriber<int>& s, const scheduler & s) { /* ... */ } 

[kirkshoop]

rxcpp is written to support C++11, but will compile in C++14 - so code using rxcpp can use C++14, just set the right compiler switch on clang/gcc.

[kirkshoop]

for the best current docs for rxcpp scheduling check out this http://stackoverflow.com/a/30294037/599231

[kirkshoop]

@daixtrose That C# sample will ignore the scheduler supplied to subscribe_on and always leapfrog from it to the NewThreadScheduler. the new thread it uses will be blocked until the loop finishes since it does not recursively schedule the next value.

The scheduler passed to subscribe_on will not be blocked and cancellation will work, but it also will not be used to run the loop which might surprise users as well.

[kirkshoop]

@daixtrose Piping the scheduler used in subscribe_on to its source is essentially done implicitly in rxcpp today. since the source subscribe is run in the supplied scheduler and the scheduler always registers its queue with the thread local variable that the current_thread scheduler uses to queue work, any use of the current_thread scheduler reuses the scheduler passed to subscribe_on.