Reimplement TimeoutScheduler with timer thread and pool dispatch

[rmahfoud]

Summary

• TimeoutScheduler no longer spawns a threading.Timer (OS thread) per pending timeout; one timer thread tracks due times and dispatches fired actions onto a ThreadPoolExecutor.
• All existing TimeoutScheduler.singleton() call sites (timeout, debounce, delay, timer, …) benefit without per-operator changes.
• Addresses review feedback on Reimplement TimeoutScheduler with timer thread and pool dispatch #748: avoids EventLoopScheduler deadlock (user code never runs on the timer thread) and queue growth (cancel evicts by identity).

Test plan

• pytest tests/test_scheduler/test_timeoutscheduler.py (singleton, schedule, cancel, nested/blocking handler, queue eviction, short-vs-long, no thread-per-timer)
• Confirm heavy concurrent timeout() use no longer exhausts threads
• Spot-check nested timeout / blocking handlers on the default scheduler

[coveralls]

coverage: 93.453%. remained the same
when pulling d9e33a4 on rmahfoud:timeout-on-event-loop
into d0b472b on ReactiveX:master.

[dbrattli]

Thanks for the PR — the underlying problem is real: TimeoutScheduler spawns a threading.Timer (a full OS thread) per pending timeout, so a process with many concurrent timeout() subscriptions can exhaust threads.

That said, I think this particular fix introduces two concrete regressions, and it also only addresses one of ~20 places with the same issue.

Blocking issues

1. Deadlock: a single shared thread runs user callbacks inline

EventLoopScheduler executes actions serially on one thread. The timeout action calls obs.subscribe(observer, scheduler=scheduler), passing that same scheduler down, so the fallback observable's on_error/on_next — i.e. arbitrary user code — now runs on the one global timer thread shared by every timeout() in the process. Anything that blocks there stalls all other timeouts.

Reproduced against current master, simulating the PR by passing a shared EventLoopScheduler explicitly:

import threading
import reactivex
from reactivex import operators as ops
from reactivex.scheduler import EventLoopScheduler

sched = EventLoopScheduler()  # simulates the PR's module-level default
done = threading.Event()

def on_err(e):
    print("outer timeout fired, now running a nested timeout...", flush=True)
    try:
        reactivex.never().pipe(ops.timeout(0.5, scheduler=sched)).run()
    except Exception as ex:
        print("nested completed:", ex, flush=True)
    done.set()

reactivex.never().pipe(ops.timeout(0.5, scheduler=sched)).subscribe(on_error=on_err)
print("nested finished in time:", done.wait(5), flush=True)

outer timeout fired, now running a nested timeout...
nested finished in time: False

The same code with TimeoutScheduler.singleton() prints control nested completed: Timeout / True. A .run() (or any blocking call) inside a timeout handler deadlocks permanently under this PR, because the inner timeout is queued behind the outer handler on the very thread it is waiting for.

2. Unbounded queue growth on hot streams

timeout calls create_timer() on every on_next. SerialDisposable cancels the superseded timer, but ScheduledItem.cancel() only flags the item — cancelled items stay in the PriorityQueue until their duetime passes (eventloopscheduler.py:184). So pending entries grow as event-rate × timeout-duration, process-wide:

sched = EventLoopScheduler()
s = Subject()
sub = s.pipe(ops.timeout(60.0, scheduler=sched)).subscribe(lambda v: None)
for i in range(20000):
    s.on_next(i)
print("pending items in event loop queue:", len(sched._queue))
# pending items in event loop queue: 20001

Every on_next also takes the scheduler's global lock and issues a notify(), waking the shared loop — a process-wide serialization point across all timeout() users. threading.Timer.cancel() releases its thread promptly, so this is a regression in exactly the "heavy use" scenario the PR targets.

Secondary concerns

• Never-exiting thread. EventLoopScheduler() defaults to exit_if_empty=False, so once the thread starts it lives for the process lifetime with no way to reclaim it. exit_if_empty=True would let it stop when idle and respawn on demand — that's what NewThreadScheduler already does at newthreadscheduler.py:57.
• Eager module-level global. Constructed at import time and never disposed. It's also reachable; if anything disposes it, schedule_absolute raises DisposedException and every timeout() in the process breaks permanently. A lazily-created singleton (matching the TimeoutScheduler.singleton() / ImmediateScheduler.singleton() convention used elsewhere) would fit the existing style better than a bare module global.
• Fixes 1 of ~20 call sites. TimeoutScheduler.singleton() is the default in _debounce.py, _delay.py, _windowwithtime.py, observable/timer.py, _timeoutwithmapper.py, and others. If thread-per-timer is the problem, timeout isn't special — and the inconsistency is confusing to read.
• No tests / changelog. All 15 tests in tests/test_observable/test_timeout.py inject TestScheduler via scheduler.start(), so the default-scheduler path has no coverage and nothing here would catch either issue above.
• Docs. The scheduler: docstring isn't updated to mention the new default or its threading model.

Suggested direction

Note: ThreadPoolScheduler is not a drop-in default either

Worth flagging before anyone reaches for it. ThreadPoolScheduler extends NewThreadScheduler, whose schedule_relative creates a fresh EventLoopScheduler(exit_if_empty=True) per scheduled action (newthreadscheduler.py:57) with pool workers as the thread factory. So a pending timer parks a worker in _condition.wait(seconds) for its entire duration — the pool bounds the thread count but doesn't stop a timer from holding a thread. With max_workers=2:

pool = ThreadPoolScheduler(max_workers=2)
pool.schedule_relative(5.0, fire("long-A"))
pool.schedule_relative(5.0, fire("long-B"))
pool.schedule_relative(0.5, fire("SHORT"))

long-A (want 5.0s) fired at 5.00s
SHORT   (want 0.5s) fired at 5.00s      # 10x late
long-B (want 5.0s) fired at 5.00s

Both workers were parked in the 5s waits, so the 0.5s timeout couldn't fire until one freed up. Head-of-line blocking, and it degrades silently — timeouts just arrive late. (Easy to miss in a quick test: duetimes are absolute and computed on the calling thread, so if all timers are created at once with equal duetimes, the ones that waited fire immediately as workers free up and everything looks fine.)

What would work

The fix seems to belong at the scheduler layer rather than in one operator: one thread that does nothing but track duetimes and never runs user code, dispatching each fired action via the existing ThreadPoolScheduler for execution. That would:

• eliminate thread-per-pending-timeout for all call sites at once,
• preserve today's concurrency semantics (no deadlock, no head-of-line blocking),
• allow cancellation that actually evicts entries from the queue.

Since the dispatch half already exists, the work is the timing wheel plus proper cancellation. That could ship as a new TimerScheduler, or as a reimplementation of TimeoutScheduler itself so existing defaults benefit without churn.

If a narrower change is preferred as a stopgap, at minimum: use a lazily-created singleton with exit_if_empty=True, don't pass the timer scheduler into obs.subscribe, and add a regression test covering nested/blocking handlers on the default scheduler. Even then the queue-growth behaviour on high-frequency sources remains.

[rmahfoud]

Thanks for the detailed review — agreed on both blocking issues and on fixing this at the scheduler layer rather than swapping timeout() to a shared EventLoopScheduler.

I've reworked the approach locally along the lines you suggested:

• Reverted the operator change; timeout() again defaults to TimeoutScheduler.singleton().
• Reimplemented TimeoutScheduler: one dedicated timer thread tracks due times; fired actions are dispatched via a ThreadPoolExecutor so user/fallback callbacks never run on the timer thread.
• Cancellation evicts queue entries by identity (not due-time equality), so high-frequency sources don't accumulate cancelled items.
• Existing TimeoutScheduler.singleton() call sites benefit without per-operator churn.
• Added regression tests for nested/blocking handlers, queue eviction, short-vs-long concurrency, and no thread-per-pending-timer.

I'll push an updated commit to this branch shortly (title/description will reflect the scheduler rewrite).