Thread model questions - Githubissues

lempiji / rx

Reactive Extensions for D Programming Language

MIT License

53 stars 8 forks source link

Thread model questions #20

Open Robert-M-Muench opened 6 years ago

Robert-M-Muench commented 6 years ago

Am I right that rx uses threads?

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

lempiji commented 6 years ago

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 commented 6 years ago

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 commented 5 years ago

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 commented 4 years ago

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:

When I put a value into a stream, will this start a new thread?
When observers are notified, is this done sequentially or via threads?

lempiji commented 4 years ago

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 commented 4 years ago

Ok, thanks.

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

lempiji commented 4 years ago

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.