ParBridge using too few threads concurrently

[ambiso]

I have a long running single-threaded computation, of which I'd like to run many concurrent instances:

    let n = 320;
    let iter = repeat(()).take(n).par_bridge();

    let r: u32 = iter.map(|_| {
        // Some long running computation
        sleep(Duration::from_secs(10));
        1
    }).sum();

When I run this on many cores (96) often only a random number of cores are used to perform the computation, and the rest is idle.
The number of active cores does not change, over time.
However, it doesn't appear to be a deadlock: rayon works through the iterator in chunks of some random size,
and it terminates correctly.

I have written an MWE here.

The MWE is best run with RAYON_NUM_THREADS=96 cargo run --release, which reproduces about 75% of the time on my 8 logical core system.

When I increase the item count n in the iterator to threads * threads * 2 the bug does not occur anymore.
However it does still occur with n = threads * threads * 2 - 1.

Therefore, I think some logic related to this line may be responsible.

Is this expected behavior? Is it a bug?

I will look into it further, I just wanted to get a report out before I forget about it.

Kind regards,
ambiso

[ambiso]

As a workaround - for my particular use-case - I can change from a ParBridge to this:

use rayon::iter::repeat;
let iter = repeat(()).take(n);

alternatively, if one needs different items one can use the iterator on Vecs:

let v = vec![(); n];
v.par_iter()

[ambiso]

Could this FIXME be related to this issue?

I don't really understand what is meant by "but that doesn't consider active stealers elsewhere.".

[wagnerf42]

hi, you should take a look at https://docs.rs/rayon/1.5.0/rayon/iter/fn.repeatn.html

or simply a parallel range.

par_bridge is meant to be used when no parallel iterators are available.

[ambiso]

Hi @wagnerf42 - I just found this too - Thanks!

I'll add it to the workarounds list.

However as I see it this doesn't change anything about the bug for ParBridge

[cuviper]

The input Iterator is inherently sequential because Iterator::next(&mut self) mandates exclusive mutable access. We cannot parallelize that, but par_bridge() does let the resulting items be consumed by multiple threads for the rest of the ParallelIIterator chain. So the only part that's actually parallel in your MWE is the simple summation, which wouldn't be worthwhile.

Furthermore, ParBridge simply spins while waiting for work (#795), which will be painful if your iterator can't produce items quickly enough to keep the thread pool busy.

[ambiso]

So the only part that's actually parallel in your MWE is the simple summation, which wouldn't be worthwhile.

I realize that the iterator is a bottleneck when it can't produce items quickly enough.
However, in my case the iterator shouldn't be the bottleneck since every item takes 10 seconds to process.

In the MWE multiple threads are sleeping concurrently - i.e. the .map(Fn) is also executed concurrently.
I this not intended behavior?

If you execute the MWE you will see that multiple threads are outputting the message that they are running the Map function.
However, the number of threads executing it concurrently is often lower than RAYON_NUM_THREADS when RAYON_NUM_THREADS=96 (for example).
I would expect it to execute the map-function concurrently with every available thread.

Furthermore, ParBridge simply spins while waiting for work (#795), which will be painful if your iterator can't produce items quickly enough to keep the thread pool busy.

There is enough work for every thread to be busy - however for some reason not every thread is doing work (i.e. sleeping).
Replace sleep with some meaningful CPU intensive and long-running computation, and suddenly you're only using (for example) 50 out of 96 cores.

[cuviper]

Oh wow, sorry, I completely misread your MWE, thinking the sleep was on the Iterator side.

It's not generally productive to oversubscribe threads -- 96 threads on your 8-core system is pretty extreme. But when they're actually blocking in some way (like sleep), I would hope that all threads do get a chance to participate. It's possible there's a bug in our scheduler and the wake-up handling.

Does the problem go away without par_bridge? (using repeatn or a parallel range)

[ambiso]

Oh wow, sorry, I completely misread your MWE, thinking the sleep was on the Iterator side.

No problem, good that we cleared that up :-)

It's not generally productive to oversubscribe threads

Definitely, I fully agree.
I'm only doing it here because it seems to reproduce much more easily when run on many threads.
With only 8 threads the issue only triggers very rarely.
However, I've tested it on a real 96 core system and get the essentially same behavior as when I set RAYON_NUM_THREADS=96 on my local 8 core machine.

Does the problem go away without par_bridge?

Yes, I've tested rayon::iter::repeat(()).take(n) instead of std::iter::repeat(()).take(n).par_bridge() and it works flawlessly.
I would think that it's not a scheduler bug if it's only specific to par_iter? (EDIT: I meant par_bridge)

As said before, the issue also doesn't trigger when n >= threads * threads * 2 (which likely is connected to this line).

There's a reproduce.sh script in the MWE repo which will run the MWE and count how many times it could reproduce the issue (by checking whether all threads output a message within the first second).

[cuviper]

Yes, I've tested rayon::iter::repeat(()).take(n) instead of std::iter::repeat(()).take(n).par_bridge() and it works flawlessly.
I would think that it's not a scheduler bug if it's only specific to par_iter?

I think you meant par_bridge at the end, but yes, I agree that narrows the problem.

As said before, the issue also doesn't trigger when n >= threads * threads * 2 (which likely is connected to this line).

That is curious. I think we do have an issue with iterator fusing there, something like:

• Thread A sees self.done.load(_) as false
• Thread B with the lock sees the iterator return None, sets self.done, and unlocks
• Thread A gets the lock and tries the iterator again

However, your case with Take<Repeat<()>> should be a FusedIterator anyway.

My other suspicion is that the split_count code might be misbehaving. I did change that atomic swap recently in #828, which hasn't been published yet, but I don't think it was wrong before. Another way that splitting might fail is if one of the active bridge threads blocks on a rayon call and goes work-stealing, getting another of the splits, so they won't actually be distributed fully. I don't see that in your MWE though.

[cuviper]

I think I see another race with self.done:

• Thread A tries to steal() and gets Steal::Empty
• Thread B loads the last of the items and sets self.done
• Thread A, on its way to try the iterator, sees self.done and quits.
• Thus the extra items in the deque are only consumed by the remaining threads.