Trampolining (or CurrentThread) Scheduler

[trxcllnt]

Need an implementation of a trampolining scheduler (Rx2's currentThreadScheduler). Don't need an implementation of the recursive scheduler (Rx2's immediateScheduler) if all the Observable creation functions (of, fromArray, returnValue, etc.) fallback to imperative loops in the absence of an external scheduler. This is possible since observers now return iteration results from onNext, onError, and onCompleted.

Example:

// src/observable/FromArrayObservable.js

var Observable = require("rx/Observable");

function FromArrayObservable(array, scheduler) {
    this.array = array;
    this.scheduler = this.scheduler;
}

FromArrayObservable.prototype = Object.create(Observable.prototype);
FromArrayObservable.prototype.subscribe = function subscribe(subscriber) {
    var array = this.array;
    var scheduler = this.scheduler;
    if (scheduler) {
        return subscriber.add(scheduler.schedule([{done: false }, subscriber, array, -1], dispatch));
    } else {
        var index = -1;
        var count = array.length;
        while (++index < count) {
            var result = subscriber.next(array[index]);
            if (result.done) {
                return subscriber;
            }
        }
        subscriber.return();
        return subscriber;
    }
};

function dispatch(scheduler, state) {
    var result = state[0];
    var subscriber = state[1];
    if (subscriber.disposed) {
        result.done = true;
        return subscriber;
    }
    var array = state[2];
    var index = state[3];
    if ((state[3] = ++index) < array.length) {
        var result = state[0] = subscriber.next(array[index]);
        if (result.done) {
            return subscriber;
        }
        return subscriber.add(scheduler.schedule(state, dispatch));
    } else if (!result.done) {
        subscriber.return();
    }
    return subscriber;
};

module.exports = FromArrayObservable;

@Blesh @jeffbcross

[benlesh]

I think I follow, but how does this differ from the current scheduler implementations in this library? If no scheduler is specified, I think we're currently defaulting to the currentThreadScheduler, and you're looping over an array in a fromArray observable, it should execute that loop immediately.

@trxcllnt, can you review the schedulers for me and ensure that's the case? Perhaps I've stared at them too long.

[trxcllnt]

@Blesh the current-frame-scheduler as it exists today is analogous to RxJS's ImmediateScheduler. I've opened this issue to track creation of a scheduler that's analogous to RxJS's CurrentFrameScheduler. I have a prototype of this scheduler in this gist.

The difference between the two is how they handle re-entrancy:

• The ImmediateScheduler immediately invokes each scheduled action. If a scheduled action schedules another action, that action is invoked within the scope of the original scheduled action invocation. If you think of actions that schedule more actions as a tree of scheduled actions, the ImmediateScheduler is a depth-first walk through the actions tree. This is also called the recursive scheduling strategy.
• The CurrentThreadScheduler maintains a FiFo queue of scheduled actions, which enforces a breadth-first walk through the actions tree. This is also called the trampolining scheduling strategy.

By Observable creation operators defaulting to synchronous iterative loops in the absence of a scheduler is an implicit implementation of the recursive scheduling strategy, which obviates the need for a recursive scheduler at all (i.e. the ImmediateScheduler). But we still need an implementation of the trampolining strategy (i.e. the CurrentThreadScheduler).

My prototype uses a linked-list instead of an Array stack, to avoid the array.push() GC thrashing and array.shift() reordering cost on large lists.

[benlesh]

I see what you're saying. I think I whiffed the semantics with naming current-frame-scheduler as I did, it should be immediate-scheduler. :\

[trxcllnt]

@Blesh tl;dr: If all the Observable creation operations default to iterative loops, we should only need the trampolining scheduler, which we still need to write.

[benlesh]

That's fair... However, I wonder what the performance implications of trampolining will be on the library. Is it worth the perf decrease to protect re-entrant behavior? Or is that more of an edge case? I had defaulted to immediate scheduling because I felt that trampolining had limited benefits the majority of the time.

[trxcllnt]

Trampolining is currently the default behavior in RxJS, and breadth-first vs. depth-first definitely have implications on functionality (esp. with nested flatMapLatest, expand, etc.). I don't mind switching to the recursive scheduling strategy by default, but it's important to include the trampolining scheduler.

Since creation operators will default to iterative looping (recursive strategy), implementing a trampolining scheduler won't have any performance implications except when you buy into that scheduling strategy.

[benlesh]

@trxcllnt just had a conversation with @jhusain about this. I think that we're on the right course defaulting to the immediate scheduler, but you're right, we need to add the trampolining scheduler in as well.

Jafar had a good suggestion, which is, that although I've already done some of the work of putting the schedulers together, we might just want to port what RxJS 2 already has to TypeScript and use those. They're battle-tested, after all. I'm okay with that, as long as we can look at the code and feel it's performance is up to what we want out of it.

[benlesh]

TODO:

• rename current-thread-scheduler to immediate-scheduler
• create trampoline-scheduler
• look at simply porting over schedulers from RxJS 2.

[trxcllnt]

@Blesh I talked with @mattpodwysocki about this last week. The existing CurrentThreadScheduler uses a priority queue, which isn't necessary in JS's single-threaded environment.

An alternative is to use an Array stack, but that could non-deterministically thrash the GC if it grows large enough, and there's a cost to shifting actions off the front. I used a linked-list, which is three hash-lookups on insertion, and max two on execution or disposal. I can write a more formal LinkedList class if desired.

We'll also need a scheduler that guarantees scheduled actions are executed on the next tick, and not in the current execution context. Maybe we should call it the ZalgoAgnosticScheduler? ;)

I'll file a ticket to track what our story is on the return value from Observable#subscribe, as that'll have an impact on what we return from Scheduler#schedule. Also if you could clarify, why do you think we still need the ImmediateScheduler?

[benlesh]

@trxcllnt The ZalgoAgnosticScheduler should be the next-frame-scheduler... There I'm putting it in a MicroTaskQueue, which is a custom impl that requires review.

As for the return value from Observable#subscribe ... I want to keep it to the current ES7 spec, which is now returning a Subscription object on it, with nothing more than an unsubscribe() function. As long as it has at least that contract, I'm open to other things it might support (generator functions for example?) as long as we can figure out the user stories that support the features.

[trxcllnt]

Awesome. Looks like the next-frame-scheduler is doing what I described. If perf is an issue, we could benchmark the micro task queue's Array against an LRU. We'll also need to ensure the schedulers are returning Subscription instances, so scheduled actions can be canceled.

@jhusain Any reason Disposable got renamed to Subscription? Originally I remember the Subscription was supposed to be some sort of Generator-like value, but it seems it's been relegated back strictly to disposal. Since disposal is a more universal concept than subscription, it seems we should generalize Subscription back to Disposable (or Cancelable, as raix called it).

[jhusain]

Yes. It was renamed Subscription to correspond with the subscribe method we expect the majority of devs to use. Disposable is a .NET concept that is sufficiently general as to be non-ergonomic.

We moved away from retuning a generator because we couldn't come up with any use cases for notify and error talkback. Use cases would be appreciated if you have any ideas.

JH

On Jun 1, 2015, at 5:24 PM, Paul Taylor notifications@github.com wrote:

Awesome. Looks like the next-frame-scheduler is doing what I described. If perf is an issue, we could benchmark the micro task queue's Array against an LRU. We'll also need to ensure the schedulers are returning Subscription instances, so scheduled actions can be canceled.

@jhusain Any reason Disposable got renamed to Subscription? Originally I remember the Subscription was supposed to be some sort of Generator-like value, but it seems it's been relegated to disposal only. Since disposal is a more universal concept than subscription, it seems we should generalize Subscription back to Disposable (or Cancelable, as raix called it).

—
Reply to this email directly or view it on GitHub.

[trxcllnt]

@jhusain Are you implying schedulers should return Subscriptions, or that they return an inconsistent type? Narrowing the concept of disposability to the domain of Observable subscriptions seems non-ergonomic when the full spectrum of disposable behaviors is considered.

[jhusain]

I'm not trying to imply the scheduler should return subscriptions. The only change I was talking about was the fact that we have named disposables subscriptions, and we return a subscription not a generator from subscribe.

Sent from my iPad

On Jun 2, 2015, at 1:27 PM, Paul Taylor notifications@github.com wrote:

@jhusain Are you implying schedulers should return Subscriptions, or that they return an inconsistent type? Narrowing the concept of disposability to the domain of Observable subscriptions seems non-ergonomic when the full spectrum of disposable behaviors is considered.

—
Reply to this email directly or view it on GitHub.

[trxcllnt]

@jhusain so what should schedulers return?

1. A Disposable type that's incompatible with Subscription.
2. A Disposable that's compatible with Subscription, either by
   • renaming Subscription to Disposable, or
   • define Subscription as a subtype of Disposable, with dispose aliased to unsubscribe.
3. Subscription, even though it'll be confusing to anybody outside the context of Observables.

We obviously need a type to encapsulate resource management. Subscriptions are resources, scheduled actions are resources. Shouldn't they be congruent?

If people get hung up on the name "Disposable," let's call it something else. I don't want to bikeshed on the name, but I'm curious why the spec differs from the principles established and proven by Rx.NET, RxJS, RxJava, and ReactiveCocoa.

[jhusain]

Schedulers are not a concept being introduced for standardization. Implementations are free to do what they like. My preference would be to return a subscription. It's worth noting that we are in line with RxJava here:

worker = Schedulers.newThread().createWorker(); worker.schedule(new Action0() { @OverRide public void call() { yourWork(); } }); // some time later... worker.unsubscribe();

It's my guess that returning a subscription from a scheduler will not blow any minds. However we can ask Ben Christensen if this introduced any confusion.

JH

On Jun 2, 2015, at 5:21 PM, Paul Taylor notifications@github.com wrote:

@jhusain so what should schedulers return?

A Disposable type that's incompatible with Subscription.
A Disposable that's compatible with Subscription, either by
renaming Subscription to Disposable, or
define Subscription as a subtype of Disposable, with dispose aliased to unsubscribe.
Subscription, even though it'll be confusing to anybody outside the context of Observables.
We obviously need a type to encapsulate resource management. Subscriptions are resources, scheduled actions are resources. Shouldn't they be congruent?

If people get hung up on the name "Disposable," let's call it something else. I don't want to bikeshed on the name, but I'm curious why the spec differs from the principles established and proven by Rx.NET, RxJS, RxJava, and ReactiveCocoa.

—
Reply to this email directly or view it on GitHub.

[headinthebox]

createWorker returns a subscription to (attempt to) cancel all outstanding scheduled work and clean up any resources the worker has allocated. schedule returns a subscription to (attempt to) cancel the action that is scheduled.

Disposable and Subscription are exactly the same thing, in RxJava we just choose a different name for whatever reason.

[trxcllnt]

@jhusain @headinthebox I'm 100% on the Disposable -> Subscription rename. I was concerned the spec is defining a type that isn't compatible with more use-cases than just subscription.

[benlesh]

It's my guess that returning a subscription from a scheduler will not blow any minds. However we can ask Ben Christensen if this introduced any confusion.

cc/ @benjchristensen

[benlesh]

Schedulers are not a concept being introduced for standardization.

As an aside:

@jhusain: @trxcllnt and I had a conversation just yesterday in which I postulated that there should be native schedulers built into JavaScript, given that people keep reinventing the same solutions to the problem. They seem like a solid add to any event driven system or event loop architecture.

[benlesh]

@trxcllnt can this issue be closed? Are we all satisfied with the direction we've chosen with Scheduling?