Improve singleton scheduler pattern, add TrampolineScheduler

[erikkemperman]

I hope this might resolve issue #383.

This improves the singleton pattern, so it uses .instance() instead of a separately imported variable. I've kept the trick with __new__, to make sure that the singleton is enforced even if people call the constructor (I think this is the closest we can get to achieving what a "private" constructor might otherwise). This now also accounts for inheritance, if you extend one of these schedulers you'll still get a singleton, but of the right type.

Tests are added for all of this as well, including for how I interpret the difference in execution order of actions scheduled on "nested" schedulers.

Then, as suggested, I've added a new TrampolineScheduler, which replicates the old behaviour of the CurrentThreadScheduler (or that was the intention, anyway). I'm not too excited about that name, but don't have any better suggestions either.

Would it make sense to switch this around, so the CurrentThreadScheduler essentially stays the same as in 1.x, and we'd introduce a new scheduler (which would be used internally)?

@MichaelSchneeberger Could you see if something like this would satisfy your requirements?

[coveralls]

Coverage increased (+0.09%) to 92.902% when pulling 8b27dc5 on erikkemperman:currentthread-trampoline-scheduler into 43f54f4 on ReactiveX:master.

[dbrattli]

I think we should keep CurrentThreadScheduler a singleton as in this PR since it now matches dotnet/reactive. My initial thought was to remove TrampolineScheduler for now (to get v3 out), but why not show the relation/similarity between them by letting CurrentThreadScheduler inherit TrampolineScheduler? Then we could remove schedule, schedule_relative, ensure_trampoline from CurrentThreadScheduler. With some work we could perhaps also remove schedule_absolute, just leaving the difference in _get_queue and _set_queue? Right now there is a difference in schedule_relative i.e max(DELTA_ZERO is missing from TrampolineScheduler, that we would then avoid.

[erikkemperman]

@dbrattli Good suggestion, thanks! I've managed to get things working with TrampolineScheduler as base class of CurrentThreadScheduler -- the only slightly awkward point is that we have to explicitly avoid calling super().__init__(). Otherwise this is pretty clean, imho, no need to override any of the schedule methods.

[dbrattli]

I think this is much better than having a single scheduler trying to be 2 different things which can be confusing for users. Some expect a singleton, other don't. Great work to resolve this issue!

[erikkemperman]

Thanks for review! The distinction between these schedulers is fairly subtle, I hope that won’t be too confusing...

[MichaelSchneeberger]

@erikkemperman thanks for taking the time to solve the issue.

I think making two schedulers (CurrentThreadScheduler and TrampolineScheduler) out of one makes sense.

Maybe one question is why enforcing the "current thread" behavior for the TrampolineScheduler? Another approach could be to run actions on an object's local queue (see my implementation trampolinescheduler.py as an example). That way it is not defined on which thread the scheduler is currently running on. On the contrary, depending on the threads that call the schedule method, it might wander from thread to thread. That way, however, the order of execution (of actions) is guaranteed.

You get the following classes:

CurrentThreadScheduler.singleton     # a classmethod
CurrentThreadScheduler               # the current TrampolineScheduler inplementation
TrampolineScheduler                  # trampoline behavior with a single queue

[jcafhe]

It's not mandatory, but could be simplified by using set_default method from dict, i.e.:

local = self._local.setdefault(thread, _Local())

[jcafhe]

Also should writing to self._local weak dict be protected by a lock?

[erikkemperman]

@MichaelSchneeberger I think your suggestions makes a lot of sense, but I'll have to see if it's easy to tack onto this PR. I suspect the sub-class structure proposed by @dbrattli might no longer be a natural one.

@jcafhe Thanks for review, I'll take your remarks into account!

[erikkemperman]

OK, I think I more or less have it... I've moved the logic of checking idle state and managing the queue into the Trampoline class, that seems to make things a bit cleaner.

The CurrentThreadScheduler is still a subclass of TrampolineScheduler, but the latter is no longer tied to a particular thread. In other words, it hijacks the first thread that calls a schedule method to run the trampoline until the queue is empty, and in the mean time other threads could schedule items on the same queue.

Note that this requires a slightly more complicated Trampoline, with locks and a means to interrupt waits. The downside is that for the singleton CurrentThreadScheduler instance, which is heavily used internally, this added complexity is not strictly necessary. I am wondering if that's sufficient reason to have two versions of the Trampoline class, one being threadsafe and the other being faster for single thread use?

I've made a "private" subclass of CurrentThreadScheduler which is returned by the singleton method -- this has the semantics we've discussed before, meaning that each thread will have an exclusive queue due to the threading.local trick.

In the middle, as it were, are regular CurrentThreadScheduler instances which have their own queue, and you could have multiple instances of the scheduler for any given thread.

I believe this covers pretty much all bases, although I am slightly worried that users might not properly understand the subtle distinctions here...

@jcafhe
I've decided I don't really like the setdefault method here, it would create lots of instances that are never used. A WeakKeyDefaultDictionary is what we would want, but that doesn't exist as far as I know. And I think that locking the dict is not necessary in the case you mentioned, we don't have to worry about something happening in between the check for a key and then inserting it if absent, because the key in this particular case is the current thread itself.

[jcafhe]

@erikkemperman Ok, thanks for the insight 👍 .

[MichaelSchneeberger]

I am wondering if that's sufficient reason to have two versions of the Trampoline class, one being threadsafe and the other being faster for single thread use?

That's a good question. Maybe one that can be dealt with later?

I added a new issue #445 that reveals a problem with threading.local. Maybe it is safer to just not use it. Implementing the behavior by ourselves means that we have more control of what actually happens.

[erikkemperman]

@dbrattli You approved this PR earlier, so I just wanted to check if the later changes are all right with you?

[dbrattli]

Yes, please merge

[erikkemperman]

Done. Thanks!