Don't instantiate new CurrentThreadScheduler._Local with each scheduler instance

[erikkemperman]

This fixes the problem pointed out by @MichaelSchneeberger in #358.

[coveralls]

Coverage remained the same at 90.57% when pulling 10e0996 on erikkemperman:currentthreadscheduler-local into b0d75e4 on ReactiveX:master.

[MichaelSchneeberger]

Do you actually still need the _global field and the __new__ method?

[erikkemperman]

@MichaelSchneeberger Well, only if it is decided that we want to enforce, not just internally but for end-users as well, that there is at most a single instance of CurrentThreadScheduler per thread. Unless, do you know a better way to enforce this?

[erikkemperman]

I've added a note which I hope clears up some of the not-so-intuitive behaviour of the CurrentThreadScheduler, as it currently stands.

@MichaelSchneeberger Do you think this is sufficiently clear?
3175465

[MichaelSchneeberger]

the best in my opinion would be to have backwards compatibility with the CurrentThreadScheduler v1.6, which allows to have nested CurrentThreadSchedulers as the factorial example in #358 illustrates.

A singleton behavior could then be added by introducing a classmethod singleton for instance.

scheduler = CurrentThreadScheduler.singleton()

Another point I mentioned before is that the current implementation of CurrentThreadScheduler also works without the __new__ method and _global field.

class CurrentThreadScheduler(SchedulerBase):
    """Represents an object that schedules units of work on the current thread.
    You never want to schedule timeouts using the CurrentThreadScheduler since
    that will block the current thread while waiting.
    """
    class Local(threading.local):
        __slots__ = 'idle', 'queue'

        def __init__(self):
            self.idle: bool = True
            self.queue: PriorityQueue[ScheduledItem[typing.TState]] = PriorityQueue()

    _local = Local()

    def __init__(self, local: 'CurrentThreadScheduler.Local' = None):
        """Creates a scheduler that schedules work as soon as possible
        on the current thread."""
        super().__init__()

To test the CurrentThreadScheduler, I used the following code ...

scheduler1 = EventLoopScheduler()
scheduler2 = EventLoopScheduler()

scheduler = CurrentThreadScheduler()
print(scheduler._local.queue)
scheduler = CurrentThreadScheduler()
print(scheduler._local.queue)

def show_current_thread_scheduler(_):
    def gen1():
        for _ in range(4):
            scheduler = CurrentThreadScheduler()
            yield scheduler._local.queue
    print(list(gen1()))

rx.just(None, scheduler=scheduler1).pipe(
    rx.operators.map(show_current_thread_scheduler),
    rx.operators.zip(rx.just(None, scheduler=scheduler2).pipe(rx.operators.map(show_current_thread_scheduler))),
).run()

... which prints ...

<rx.internal.priorityqueue.PriorityQueue object at 0x7f84d8a7e390>
<rx.internal.priorityqueue.PriorityQueue object at 0x7f84d8a7e390>
[<rx.internal.priorityqueue.PriorityQueue object at 0x7f84d89e0e10>, <rx.internal.priorityqueue.PriorityQueue object at 0x7f84d89e0e10>, <rx.internal.priorityqueue.PriorityQueue object at 0x7f84d89e0e10>, <rx.internal.priorityqueue.PriorityQueue object at 0x7f84d89e0e10>]
[<rx.internal.priorityqueue.PriorityQueue object at 0x7f84d89e0898>, <rx.internal.priorityqueue.PriorityQueue object at 0x7f84d89e0898>, <rx.internal.priorityqueue.PriorityQueue object at 0x7f84d89e0898>, <rx.internal.priorityqueue.PriorityQueue object at 0x7f84d89e0898>]

[erikkemperman]

Well, speaking just for myself, I have a feeling that using "nested CurrentThreadSchedulers" in this fashion, to achieve a kind of synchronization, is perhaps not the prettiest pattern...

I am not sure if scenarios like these ought to be facilitated, and if they do if there isn't a better, cleaner way to accomplish that. My feeling is that it was never actually explicitly intended to work this way -- i.e. that it works like it does in 1.6 more or less by accident. But, since I wasn't around at the time, I might very well be wrong about that!

However, having said all that, you are absolutely correct that the current behaviour of CurrentThreadScheduler is, in this respect, different from 1.x.

Since I am the author of this change, and also new around here, I think I should not weigh in (more than I already have) on what the resolution should be here.

So I'll wait what others might propose? If it is decided to keep this change, or something like it, I hope the doc I added here might be sufficient in the way of announcing this slight departure from 1.x semantics.

[jcafhe]

That looks good to me, at least, it should consolidate the expected behavior of 1 queue per thread.

[erikkemperman]

I think this should be merged now — the question of @MichaelSchneeberger is not yet resolved, but this PR fixes a legitimate bug (local constructor) and documents the current state of the master branch.

I propose to continue discussion of how to proceed from there in the issue #358.

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.