Thread model questions

[Robert-M-Muench]

Am I right that rx uses threads?

Is every observer running in its own thread? What other threads do exists?

[lempiji]

By default it will run in own thread.

The thread model is abstracted as Scheduler. If you want to switch, you can switch with TaskPoolScheduler, ThreadScheduler, observeOn or subscribeOn.

However, if you always want to run on UI threads, write the scheduler yourself.
An example of this is the DlangUIScheduler.

https://github.com/lempiji/rx/blob/dev/examples/scheduler-dlangui/source/app.d

[Robert-M-Muench]

So, it's pretty simple to start some long-running tasks on a specific event and utilize multi-core machines? That's pretty cool...

What's the best way to exchange data between threads in RX manner? Or just use plain D infrastructure for this?

[Robert-M-Muench]

From #33 "But keeping everything lock-free is really hard and tedious..."

Can you explain the thread concept a bit more? If I just use RX as is it's single-threaded? So, all the lock-free handling is there for cases, where I use a scheduler?

[Robert-M-Muench]

Need to come back to this:

• I just use RX as is. Nothing special, no scheduler, etc. Just out-of-the-box.

• I have a single-threaded D app, which a couple of streams and many observers per stream.

• The app crashes when there is a heavy load on the streams. I have the suspicion that this has something to do with multiple threads and some race-condition, that might origin from RX.

Some questions:

1. When I put a value into a stream, will this start a new thread?

2. When observers are notified, is this done sequentially or via threads?

[lempiji]

I'm back now.

A1. With simple usage, no threads are created. Always works on a single thread.

A2. Observers will be notified in the order they were subscribed. Internally, the Subject is just an array of Observers.

At the moment, threads are only created when using "debounce", "subscribeOn" or "observeOn".

If you suspect a crash due to data races between threads, look at "Thread.id" for something.

[Robert-M-Muench]

Ok, thanks.

Any reason why you don't use std.signals for implementation? IMO that would make RX a bit more standard compatible.

[lempiji]

The reason is "lack of multi-thread support".

I think the method of unsubscribing in ReactiveX is superior to Signal's disconnect in terms of encapsulation and ownership.