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

Question

I am struggling with some concurrency issues related to subscribing Observable.FromEventPattern()to TaskPoolScheduler.

Let me illustrate a sample code:

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 problem is that when any new elements are emitted by the original observable Observable.FromEventPattern()(i.e. when the object DataStoretriggers new events DataChanged), then they seem to be blocked, waiting for the end of the previous elements flowing through the pipeline.

Since subscribing is done on TaskPoolScheduler, I expected every new item to be released to just deploy a new task, but in fact the source of the event instead is blocked when the event is called, if the pipeline is busy.

, ( ) /, DataChangedEvent.Invoke()?

( , , Subscribe() , - .)

: @jonstodle #rxnet Slack, TaskPoolScheduler , . , , , , , . , , ( ). , , , , , ?

Answer 1

, , - , Rx - , Rx, , . Rx-, FromEventPattern<DataChangedEventArgs> , (-) , .

, . , TaskPoolScheduler.

, , - , .

:

Observable
    .FromEventPattern<DataChangedEventArgs>(dataStore, nameof(dataStore.DataChanged))
    .SubscribeOn(TaskPoolScheduler.Default)
    .Select(x => x.EventArgs)
    .StartWith(new DataChangedEventArgs())
    .Throttle(TimeSpan.FromMilliseconds(25))
    .SelectMany(x =>
        Observable.Start(() =>
        {
            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!
    });

.SelectMany(x => Observable.Start(() => .Select(x =>, , , .

.Select(x => Observable.Start(() => ...)).Merge().

, , :

var source = new Subject<int>();

source
    .SelectMany(x =>
        Observable.Start(() =>
        {
            Thread.Sleep(1000);
            return x * 2;
        }))
    .Subscribe(result =>
    {
        Console.WriteLine(result);
        source.OnNext(result);
        source.OnNext(result + 1);
    });

source.OnNext(1);

:

2
4
6
14
12
8
10
24
28
30
26
16
20
22
18
48
50
56
52
58
60
62
54
32
34
46
44
40
42