Proposed Scheduler Interface Change for 0.18 (yes, again)

[benjchristensen]

Reviewing the Scheduler interface changes of 0.17 with @headinthebox revealed that we're not 100% happy with the outcome, particularly after learning that Java 8 does not allow referencing this from within a lambda.

The Scheduler interface as of 0.17 is:

class Scheduler {
    public abstract Subscription schedule(Action1<Scheduler.Inner> action);
    public abstract Subscription schedule(Action1<Scheduler.Inner> action, final long delayTime, final TimeUnit unit);
    public Subscription scheduleRecursive(Action1<Recurse> action);
    public Subscription schedulePeriodically(Action1<Scheduler.Inner> action, long initialDelay, long period, TimeUnit unit);
    public int degreeOfParallelism();
    public long now();

    public static final class Recurse {
        public void schedule() {
        public void schedule(long delay, TimeUnit unit) {
    }

    public abstract static class Inner implements Subscription {
        public abstract void schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Scheduler.Inner> action);
        public long now();
    }
}

We have determined two problems with this:

1. Inner/Outer Dance

In practice we have found that usage is always one of two things, either you just interact with the outer and don't care about the Inner, or you immediately need the Inner and have to do an awkward first scheduling just to get access to the Inner. (See here and weep.)

2. Recursion

The Action1<Scheduler.Inner> signature was chosen and put on both outer and inner so that an inner class could refer to itself using this to simply reschedule itself from the outer onto the inner.

It was assumed this would work in Java 8 lambdas but unfortunately we did not prove it.

This works with anonymous classes:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        System.out.println("do stuff");
        // recurse
        inner.schedule(this);
    }

});

but this does not with lambdas:

Schedulers.newThread().schedule((inner) -> {
    System.out.println("do stuff");
    inner.schedule(this); // doesn't compile
});

So we end up with this:

Schedulers.newThread().scheduleRecursive((recurse) -> {
    System.out.println("do stuff");
    recurse.schedule();
});

At that point it's clear that Inner is not working well and we have Recurse to fix the problem.

Thus, the proposed changes (breaking again) are:

class Scheduler {
    public final Subscription schedule(Action1<Recurse> action);
    public final Subscription schedule(Action1<Recurse> action, final long delayTime, final TimeUnit unit);
    public final Subscription schedulePeriodically(Action1<Recurse> action, long initialDelay, long period, TimeUnit unit);
    public abstract Inner createInner(); // for advanced use cases like `observeOn`
    public int degreeOfParallelism();
    public long now();

    // now the primary interface
    public static final class Recurse {
        public final void schedule();
        public final void schedule(long delay, TimeUnit unit);
        public final void schedule(Action1<Recurse> action);
        public final void schedule(Action1<Recurse> action, final long delayTime, final TimeUnit unit);
    }

    // now mostly an implementation detail except for advanced use cases
    public abstract static class Inner implements Subscription {
        public abstract void schedule(Action1<Recurse> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Recurse> action);
        public long now();
    }
}

The name of Recurse is up for debate. It may be possible to merge Recurse and Inner but I haven't figured it out yet. The reason is that Inner is a single instance representing a thread or event-loop whereas Recurse represents an Action or work. Thus a given Inner could have multiple Recurse actions scheduled on to it. It is being an Action that allows it to recurse by invoking schedule() that just reschedules itself.

This would make it better support Java 8 lambdas and simply recursion, while also better supporting (via the createInner() method) the more complicated use cases like observeOn where current code is very awkward.

This needs to be the last refactor of this so we nail it down and stop breaking things and can get to 1.0.

Let the discussion begin ...

[kirkshoop]

Here is the state of scheduler in rxcpp v2.

In order to allow tight tail-recursion, the rxcpp (v2) scheduler has a type currently called schedulable

class schedulable : public schedulable_base
{
    composite_subscription lifetime;
    scheduler controller;
    action activity;

    recursed_scope_type recursed_scope;

public:
    schedulable()
    {
    }
    schedulable(composite_subscription cs, scheduler q, action a)
        : lifetime(std::move(cs))
        , controller(std::move(q))
        , activity(std::move(a))
    {
    }

    inline composite_subscription& get_subscription() ;
    inline scheduler& get_scheduler() ;
    inline action& get_action() ;

    inline auto set_recursed(const recurse& r) const;

    // recursed - request tail-recursion
    //
    inline void operator()() const ;

    // composite_subscription
    //
    inline bool is_subscribed() const ;
    inline weak_subscription add(shared_subscription s) const ;
    inline weak_subscription add(dynamic_subscription s) const ;
    inline void remove(weak_subscription w) const ;
    inline void clear() const ;
    inline void unsubscribe() const ;

    // scheduler
    //
    inline clock_type::time_point now() const ;
    inline void schedule() const ;
    inline void schedule(clock_type::duration when) const ;
    inline void schedule(clock_type::time_point when) const ;

    // action
    //
    inline action_duration::type get_duration() const;
    inline void operator()(const schedulable& scbl, const recurse& r) const;
};

The tail-recursion can only be requested in the callback from the scheduler. So for this implementation, if you transport the schedulable out and then request tail-recursion, the process will exit.

If I run into the Inner issue in observe_on, I will just make schedulable extract it from the scheduler in the constructor through a private or detail:: accessor.

A scheduler takes a schedulable and ignores the scheduler it contains.

class scheduler : public scheduler_base
{
public:
    typedef scheduler_base::clock_type clock_type;

    inline clock_type::time_point now() const ;

    inline void schedule(const schedulable& scbl) const ;
    inline void schedule(clock_type::duration when, const schedulable& scbl) const ;
    inline void schedule(clock_type::time_point when, const schedulable& scbl) const ;
};

The range operator requests tail-recursion

template<class T>
struct range : public source_base<T>
{
//...
    template<class Subscriber>
    void on_subscribe(Subscriber o) {
        auto state = std::make_shared<state_type>(init);
        state->sc.schedule(make_schedulable(
            o, // share the same subscription with the scheduler
            [=](const rxsc::schedulable& self){
                if (state->remaining == 0) {
                    o.on_completed();
                    // o is unsubscribed
                }
                if (!o.is_subscribed()) {
                    // terminate loop
                    return;
                }

                // send next value
                --state->remaining;
                o.on_next(state->next);
                state->next = static_cast<T>(state->step + state->next);

                // tail recurse this same action to continue loop
                self();
            }));
    }
};

In C++ it is essential to keep virtual function calls out of an inner loop. To make tail-recursion work efficiently the recursion objects create a space on the stack inside the virtual function call in the actor that allows the callback and the scheduler to share stack space that records the request and the allowance without any virtual calls in the loop.

template<class F>
inline action make_action(F&& f, action_duration::type d = action_duration::runs_short) {
    auto fn = std::forward<F>(f);
    return action(std::make_shared<detail::action_type>(
        d,
        // tail-recurse inside of the virtual function call
        // until a new action, lifetime or scheduler is returned
        [fn](const schedulable& s, const recurse& r) {
            auto scope = s.set_recursed(r);
            while (s.is_subscribed()) {
                r.reset();
                fn(s);
                if (!r.is_allowed() || !r.is_requested()) {
                    if (r.is_requested()) {
                        s.schedule();
                    }
                    break;
                }
            }
        }));
}

The current_thread uses a thread-local recursion to indicate that tail-recursion is only allowed when the queue is empty

//...
        const auto& recursor = queue::get_recursion().get_recurse();

        // loop until queue is empty
        for (
             auto when = queue::top().when;
             std::this_thread::sleep_until(when), true;
             when = queue::top().when
             )
        {
            auto what = queue::top().what;

            queue::pop();

            what(recursor);

            if (queue::empty()) {
                break;
            }
        }
//...

[benjchristensen]

Based on these implementation requirements for C++ would you recommend a different signature from this?

class Scheduler {
    public final Subscription schedule(Action1<Recurse> action);
    public final Subscription schedule(Action1<Recurse> action, final long delayTime, final TimeUnit unit);
    public final Subscription schedulePeriodically(Action1<Recurse> action, long initialDelay, long period, TimeUnit unit);
    public abstract Inner createInner(); // for advanced use cases like `observeOn`
    public int degreeOfParallelism();
    public long now();

    // now the primary interface
    public static final class Recurse {
        public final void schedule();
        public final void schedule(long delay, TimeUnit unit);
        public final void schedule(Action1<Recurse> action);
        public final void schedule(Action1<Recurse> action, final long delayTime, final TimeUnit unit);
    }

    // now mostly an implementation detail except for advanced use cases
    public abstract static class Inner implements Subscription {
        public abstract void schedule(Action1<Recurse> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Recurse> action);
        public long now();
    }
}

@headinthebox Does this change your perspective on the Java design at all?

[benjchristensen]

I have submitted a pull request with these changes.

Usage looks like this:

import java.util.concurrent.TimeUnit;

import rx.Scheduler.Inner;
import rx.Scheduler.Recurse;
import rx.functions.Action1;
import rx.schedulers.Schedulers;

public class Test {

    public static void main(String args[]) {

        Schedulers.newThread().schedule(new Action1<Recurse>() {

            @Override
            public void call(Recurse r) {
                System.out.println("do stuff");
                // recurse
                r.schedule(this);
            }

        });

        Schedulers.newThread().schedule(recurse -> {
            System.out.println("do stuff");
            recurse.schedule();
        });

        Schedulers.newThread().schedule(recurse -> {
            System.out.println("do stuff");
            recurse.schedule(1000, TimeUnit.MILLISECONDS);
        });

        Schedulers.newThread().schedule(recurse -> {
            recurse.schedule(re -> {
                System.out.println("do more stuff");
            });
        });

        Inner inner = Schedulers.newThread().createInner();
        inner.schedule(re -> {
            System.out.println("do stuff");
            re.schedule(r -> {
                System.out.println("do more stuff");
            });
        });

    }
}

Code outline:

[benjchristensen]

This is the time to voice opinions and affect change. Once we make these changes we're headed towards 1.0 and really do not want to change these signatures again.

I welcome bike shedding and arguing over names, signatures, etc for the next couple days but would like to wrap it up and move forward by the weekend unless a glaring issue is found.

I would appreciate suggestions for better names than Recurse and Inner, neither of which I particularly like.

[benjchristensen]

@kirkshoop If you have a chance to answer my previous question, I'd appreciate it so we make sure these changes address broad use cases and are not bound to a specific language (since even though we're on the JVM, there are several different languages we support, and it would be preferable for Rx across platforms to be similar).

Based on these implementation requirements for C++ would you recommend a different signature from this?

[kirkshoop]

Oh. My.

Well I had a long reply but it just got lost. I know better than to use the web interface. :(

I can't write it again so here is the summary.

RxCpp took the insight from a comment by @akarnokd, I think, to another RxJava issue that the parameter to subscribe is like subscriber which is also like subject in that they are all objects whose single-concern is to bind other single-concern objects together.

In this case the parameter would be a type that binds the scheduler (with private access to its inner) the action and the subscription together and then expose the functionality of Recurse, Inner, Subscription and Action1. The subscription would be removed from the Action1 in this case. A name that comes to mind is ScheduleAction.

I can still count the lines of Java that I have written so be gentle :)

class ScheduleAction {
    public final Action1<ScheduleAction> get_action();
    public final Subscription get_subscription();
    public final Scheduler get_scheduler();

    public final bool isUnsubscribed();
    public final void unsubscribe();

    // action methods - Sorry, I didn't look them up
    :::

    public long now();

    // uses scheduler
    public final void schedule();
    public final void schedule(long delay, TimeUnit unit);

    // uses inner
    public final void schedule(ScheduleAction action);
    public final void schedule(ScheduleAction action, final long delayTime, final TimeUnit unit);
}

class Scheduler {
    public final Subscription schedule(ScheduleAction action);
    public final Subscription schedule(ScheduleAction action, final long delayTime, final TimeUnit unit);
    public final Subscription schedulePeriodically(ScheduleAction action, long initialDelay, long period, TimeUnit unit);
    <private to ScheduleAction> abstract Inner createInner(); // I would use a friend decl in C++
    public int degreeOfParallelism();
    public long now();
}

[benjchristensen]

Thank you @kirkshoop for the feedback. If I understand correctly, the primary change would be hiding the Inner from public use. I'm going to repost a comment from #1014 (comment) about a use case that requires Inner being accessible publicly:

---

The problem I've had with combining them (Inner and Recurse) is that when first creating an Inner there is no Action to invoke. Using the ThreadLocal to store the current action doesn't solve this, it would just make it non-obvious why inner.schedule() doesn't work the first time, and the API would be odd that I can get an Inner with a schedule() method even though nothing has been invoked yet.

The reason for this scenario is that retrieving an Inner via createInner() is needed to make use cases like observeOn less awkward. Here is the example where Inner is retrieved before an Action is executed in observeOn:

        protected void schedule() {
            if (counter.getAndIncrement() == 0) {
                if (recursiveScheduler == null) {
                    recursiveScheduler = scheduler.createInner();
                    add(recursiveScheduler);
                }
                recursiveScheduler.schedule(new Action1<Recurse>() {

                    @Override
                    public void call(Recurse inner) {
                        pollQueue();
                    }

                });
            }
        }

The reason is that the recursion happens externally (the operator is doing it) rather than internally (inside the Action<Inner>/Action<Recurse>.

Therefore, we have use cases where Inner is used before Recurse makes sense, so the API is not appropriate when they are combined.

---

Thus, if we have ScheduledAction/Recurse as the only public type and hide Inner as an implementation detail, we can not gracefully solve the observeOn use case without doing what it does now: https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/operators/OperatorObserveOn.java#L94 =>

        protected void schedule() {
            if (counter.getAndIncrement() == 0) {
                if (recursiveScheduler == null) {
                    add(scheduler.schedule(new Action1<Inner>() {

                        @Override
                        public void call(Inner inner) {
                            recursiveScheduler = inner;
                            pollQueue();
                        }

                    }));
                } else {
                    recursiveScheduler.schedule(new Action1<Inner>() {

                        @Override
                        public void call(Inner inner) {
                            pollQueue();
                        }

                    });
                }
            }
        }

Is solving this use case gracefully a good enough reason for exposing Inner via the createInner() method? Most people will never need to use it and only interact with Recurse (or ScheduledAction or whatever we call it) and thus only see the Inner type on the createInner() method that they ignore.

The Scheduler.createInner() method feels similar to the Observable.unsafeSubscribe() method.

[kirkshoop]

Thanks @benjchristensen!

What prevents this?

        protected void schedule() {
            if (counter.getAndIncrement() == 0) {
                ScheduleAction action(scheduler, new Action1<ScheduleAction>() {

                    @Override
                    public void call(ScheduleAction inner) {
                        pollQueue();
                    }

                });
                action.schedule();
            }
        }

[benjchristensen]

The scheduler being passed in there is the "outer" one which means it would create a new "inner" each time and could (depending on which Scheduler it is) spawn a new thread for every ScheduledAction.

[benjchristensen]

Also, by doing it that way it makes the Subscription logic unclear. A ScheduledAction.schedule() should not return a Subscription each time, as unsubscribing from it would shut down the entire "inner" scheduler, so unsubscribing on one should not shut down all actions.

This is why the Scheduler.schedule methods return Subscription as it represents the Inner thread or event-loop, but the Recurse.schedule and Inner.schedule methods return void.

[kirkshoop]

I would like to pick up these particular points, in-person at React, if possible :)

EDIT: I am 'kirk dot shoop at microsoft com'

[benjchristensen]

I had the chance to meet @kirkshoop in person while in London and we reviewed the needs of CPP and Java and came to agreement on the proposed model. We are considering slightly changed names though to improve the semantic meaning:

class Scheduler {
    public final Subscription schedule(Action1<Schedulable> action);
    public final Subscription schedule(Action1<Schedulable> action, final long delayTime, final TimeUnit unit);
    public final Subscription schedulePeriodically(Action1<Schedulable> action, long initialDelay, long period, TimeUnit unit);
    public abstract EventLoop createEventLoop(); // for advanced use cases like `observeOn`
    public int degreeOfParallelism();
    public long now();

    // now the primary interface
    public static final class Schedulable {
        public final void schedule();
        public final void schedule(long delay, TimeUnit unit);
        public final void schedule(Action1<Schedulable> action);
        public final void schedule(Action1<Schedulable> action, final long delayTime, final TimeUnit unit);
    }

    // now mostly an implementation detail except for advanced use cases
    public abstract static class EventLoop implements Subscription {
        public abstract void schedule(Action1<Schedulable> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Schedulable> action);
        public long now();
    }
}

Neither of us are sold on Schedulable or EventLoop (nor Recurse or `Inner) so please let us know if you have better ideas.