High handles count growth when using ObserveOn(TaskPoolScheduler.Default)

[petedavis]

Version: 4.1.0

Given this basic subscription

provider.Updates
              .ObserveOn(TaskPoolScheduler.Default)
              .Subscribe(PublishUpdate)

Having an issue where each time an item is pushed through the observable subject, we get new threads created for each subscriber, rather than being invoked on a thread pool thread.

As an example, if we have 10 clients connecting asking for updates which come from an IObservable, we end up with 10 threads created to service each update that comes in. Given in our case we just want to make 10 WCF calls back over a client connection, this is a little heavy. Also with server mode GC the handle count is going to about 20,000+ in a handful of seconds before the GC finally kicks in.

This is due to the fact that the ScheduledObserver will always using the TaskPoolScheduler as ISchedulerLongRunning, and then calling ISchedulerLongRunning.ScheduleLongRunning, which results in a task with TaskCreationOptions.LongRunning being set.

I cannot find a way to indicate to RX that the subscriber is not long running, and even though the scheduler supports long running, I really dont want it to happen.

To fix our issue I have made a copy of the TaskPoolScheduler and removed the ISchedulerLongRunning interface. Using this scheduler has flat-lined the handle count of the service.

[bartdesmet]

IScheduler has a DisableOptimizations extension method that can be used to suppress optimization interfaces from being used. See

reactive/Rx.NET/Source/src/System.Reactive/Concurrency/Scheduler.Wrappers.cs

Line 32 in 525b719

public static IScheduler DisableOptimizations(this IScheduler scheduler, params Type[] optimizationInterfaces)

Does this help?

[tfabian]

Hi,
our Visual Studio output is full of messages:
The thread 0x13d8 has exited with code 0 (0x0).
The thread 0x190c has exited with code 0 (0x0).
...

It is caused by .ObserveOn(TaskPoolScheduler.Default) in version 4.1.3

It doesn't happen in version 4.0.0.
DisableOptimizations didn't help.

This pull request seems suspicious #508
Isn't there a breaking change?

Thx Thomas

[bartdesmet]

Yeah, that seems like a breaking change.

I'm not too familiar with all the changes to Rx post the v2.1 release, so I just had a look at that PR. It diverges from the initial design point where ISchedulerLongRunning should be used in an O(1) fashion with respect to the operator's lifecycle. That is, it offers up a single long-running execution unit (e.g. a thread, akin to TaskCreationOptions.LongRunning) for the lifetime of the operator (thus not O(n) in event volume) and prefers to block on it rather than spawn/exit new such long-running execution units, under the assumption that spawn/exit is more expensive than blocking. As such, it's the extreme complete opposite of fine-grained scheduling. The ICancelable parameter is used to wire cancellation down to blocking operations, to enable shutdown.

Examples include Delay and the original ObserveOn.

The irony of this change is that ISchedulerLongRunning was introduced exactly to get people out of this type of issue with Scheduler.NewThread in the v1.x days, where recursive scheduling resulted in O(n) thread creation with the same outcome as what you're observing (and terrible performance). In addition, there were use cases where one wants to rely on thread-local state on an ObserveOn-ed thread, per operator instance. Seeing that guarantee removed here is a breaking change as well, where an observer relying on TLS will see random behavior as the thread gets destroyed and recreated.

I'll defer to @akarnokd and @onovotny to look into this and restore the original behavior for ISchedulerLongRunning (and add much needed regression tests).

In your case, you should (have) be(en) fine with either original long-running behavior (no churn in thread creation, given you have a single subscription using ObserveOn) or short-time task pool work item creation (enqueue/dequeue in task pool work queues). For the second case, we also need to review why DisableOptimizations didn't provide you with a workaround (i.e. xs.ObserveOn(TaskPoolScheduler.Default.DisableOptimizations(typeof(ISchedulerLongRunning)))). Either way, there are two issues here:

• ISchedulerLongRunning behavior should be as described above.
• DisableOptimizations should be granted.

[akarnokd]

Yes, it was due to my changes to ObserveOn. I did not understand the purpose of the semaphore logic nor the implications of ISchedulerLongRunning and it did not trip any unit tests as far as I remember. I'll post a fix and regression test tomorrow.

[tfabian]

Thank you for the answers bartdesmet and akarnokd. We have recently updated our solutions from Rx-2.2.5 to System.Reactive 4.1.3. As a solution we will rollback to version 4.0.0, due to performance implications. New threads are created, not preserved as before, as you have mentioned Bart.

[clairernovotny]

v4.1.5 is out on NuGet now with the fix.

[tfabian]

Does it really behave in v 4.1.5 the same as in version 4.0.0?

Here is an example for demonstration purposes:

  var d = Observable.Timer(TimeSpan.Zero, TimeSpan.FromMilliseconds(25))
    .ObserveOn(TaskPoolScheduler.Default)
    .Subscribe(_ =>
    {
      Thread.Sleep(25);
      Console.WriteLine(_);
    });

In debug output window new messages are written during the app lifecycle:
.....
The thread 0x30b0 has exited with code 0 (0x0).
The thread 0x5708 has exited with code 0 (0x0).
....

Visual Studio output with System.Reactive 4.0.0 does not contain these messages.
Thx

[akarnokd]

Maybe this affects your debug output: https://stackoverflow.com/a/18713944/61158

[tfabian]

Szia Dávid. It seems that the older version (4.1.3) was loaded from GAC. I removed it from there with GACBrowser.
Now it works perfect. I'm confirming that the fix is correct.
Thank you all. You are great.