How can I avoid any blocking when using Observable.FromEventPattern in Reactive Extensions for .NET?

Viewed 1778

I'm struggling with some concurrency issues in relation subscribing to an Observable.FromEventPattern() on the TaskPoolScheduler.

Let me illustrate with a code example:

var dataStore = new DataStore();

Observable.FromEventPattern<DataChangedEventArgs>(dataStore, nameof(dataStore.DataChanged))
    .SubscribeOn(TaskPoolScheduler.Default)
    .Select(x => x.EventArgs)
    .StartWith(new DataChangedEventArgs())
    .Throttle(TimeSpan.FromMilliseconds(25))
    .Select(x => 
    {
        Thread.Sleep(5000); // Simulate long-running calculation.
        var result = 42;
        return result;
    })
    .ObserveOn(new SynchronizationContextScheduler(SynchronizationContext.Current))
    .Subscribe(result =>
    {
        // Do some interesting work with the result.
        // ...

        // Do something that makes the DataStore raise another event.
        dataStore.RaiseDataChangedEvent(); // <- DEADLOCK!
    });

dataStore.RaiseDataChangedEvent(); // <- Returns immediately, i.e. does NOT wait for long-running calculation.
dataStore.RaiseDataChangedEvent(); // <- Blocks while waiting for the previous long-running calculation to complete, then returns eventually.

My issue is that, when any new items are emitted by the original observable Observable.FromEventPattern() (i.e. when the DataStore object raises new DataChanged events), then they appear to be blocked waiting for the previous items to finish flowing through the entire pipeline.

Since the subscribing is done on the TaskPoolScheduler I had expected every new item emitted to simply spin up a new task, but actually, the source of the event instead seems to block on the event invocation if the pipeline is busy.

How can I accomplish a subscription that executes every new emitted item (raised event) on it's own task/thread, such that the source object never blocks on its internal DataChangedEvent.Invoke() call?

(Except of course the Subscribe() lambda which should execute on the UI thread - which is already the case.)

As a side-note: @jonstodle mentioned in the #rxnet Slack channel that the TaskPoolScheduler might have different semantics than what I assumed. Specifically he said it probably creates one task and does both the subscribing and the producing of values in an event loop inside of that one task. But if that's the case, then I find it a bit strange that the first event invocation doesn't block (since the second one does). Seems to me that if the task pool task doing the subscription is asynchronous enough that the souce doesn't have to block on the first invocation, there shouldn't be a need to make it block on the second call either?

1 Answers
Related