Reactive Extensions and parallel processing

Question

I am planning on using Rx within a project of mine at some point, and I have been researching into what I can do with Rx.

My project uses TPL for processing state machine transitions in parallel (utilising all available processor cores). However, to improve performance, I want to replace the current IList<> pull mechanism with an Rx push mechanism.

As I do not know very much about this technology I want to ascertain whether Rx will be compatible with pushing tokens to the parallel state transitions. I will require that all parallel state transitions subscribe to the Rx subject and retrieve the next token. Each state transition will require the same token from the subject. From what I understand from my own research is that once a token gets pushed out to an observer (a state transition in this case) that token is gone for good. If that is the case the other transitions (for the same state) will never receive this token and will end up in an error state.

Could somebody enlighten me as to whether my concerns are correct? I have no code to show as I am merely exploring my options.

Thank you.

Answer 1

You are probably looking for

IConnectableObservable<T> IObservable<T>.Publish()

A connectable observable is multicast variant of the original observable and can be subscribed to multiple times. To start you subscribe all your observers to the connectableobservable and then call connect on the connectableObservable.

Answer 2

@Tony makes a good point that an Observable Sequence (IObservable) is like an event, in that it can have a thread-safe list of registered subscribers/event handlers. So like events, many registered subscribers can receive the event payload (in a serial manner, where first subscriber is called first).

However, what is not addressed is a potential concern about time that Rx may help you with that events dont lend themselves too. For example you may have published/pushed a token before all parallel state transition handlers are set up. If you were using events or a standard obserable sequence, the payload would be lost. You could remedy this by using the caching features in Rx such as Replay. This could allow you to Replay the last value to late subscribers.

If this is of interest there is loads of information out there. You can read more about it in the section of my book: http://introtorx.com/Content/v1.0.10621.0/02_KeyTypes.html

Answer 3

It may help to think about the Rx Subject as similar to .NET event, that is, a thread-safe list of registered event handlers. In the case of Subject these may be considered equivalent to an individual Observer OnNext delegates. Once the event fires, each handler is invoked with the same event argument. Similarly, each Observer subscribed to a Subject gets OnNext invoked with the same argument object. This argument may contain a value type or a reference type of the token needed for state transition. Presumably your Observer's OnNext handlers would store this token (say in concurrent queue) for the corresponding parallel task consumption.

So nothing is gone for good unless you simply ignore OnNext calls or discard the object that they pass to you.

You may have shared state corruption issues, if one of your tasks changes the token. Or you may have synchronization issues, if one task processed several state transition tokens while another task had not had the time to dequeue even the first one. But this would not be specific to using Rx Subject, instead of some other technology to push the tokens.