Poor interaction with channels

[sftse]

We've been using rayon in a configuration with two thread pools connected by a channel, the first stage relying on some dynamic conditions to send tasks to the second thread pool. This always seemed a bit of a strange setup, so we attempted to refactor this to run on the same thread pool.

This minimal example shows how we tried to do this. By tweaking the constants it should be easy to achieve a deadlock, where no progression is visible.

use crossbeam_channel::Receiver;
use crossbeam_channel::SendError;
use crossbeam_channel::Sender;
use crossbeam_channel::TryRecvError;
use crossbeam_channel::TrySendError;
use rayon::ThreadPool;
use rayon::prelude::*;

const A: usize = 3;
const B: usize = 5;
const C: usize = 2;

struct Snd<'pool>(&'pool ThreadPool, Sender<()>);

fn bounded_channel<'pool>(
    pool: &'pool ThreadPool,
    capacity: usize,
) -> (Snd<'pool>, Recv<'pool>) {
    let (snd, rcv) = crossbeam_channel::bounded(capacity);
    (Snd(pool, snd), Recv(pool, rcv))
}

impl<'pool> Snd<'pool> {
    fn send(&self) -> Result<(), SendError<()>> {
        let mut t = ();
        while let Err(e) = self.1.try_send(t) {
            match e {
                TrySendError::Full(elt) => {
                    t = elt;
                    self.0.yield_now();
                }
                TrySendError::Disconnected(_) => return Err(SendError(())),
            }
        }
        Ok(())
    }
}

struct Recv<'pool>(&'pool ThreadPool, Receiver<()>);

impl Iterator for Recv<'_> {
    type Item = ();
    fn next(&mut self) -> Option<Self::Item> {
        self.try_recv()
    }
}

impl<'pool> Recv<'pool> {
    fn try_recv(&self) -> Option<()> {
        loop {
            match self.1.try_recv() {
                Ok(t) => return Some(t),
                Err(TryRecvError::Empty) => {
                    self.0.yield_now();
                }
                Err(TryRecvError::Disconnected) => return None,
            };
        }
    }
}

fn foo(snd: Snd<'_>) {
    [(); A]
        .par_iter()
        .for_each(|_| {
            for _ in 0..B {
                snd.send().unwrap();
            }
        })
}

fn main() {
    let pool = rayon::ThreadPoolBuilder::new().build().unwrap();
    let (snd, rcv) = bounded_channel(&pool, C);
    pool.join(
        || foo(snd),
        || rcv.par_bridge().for_each(|()| {}),
    );
}

[sftse]

Further minimized

use crossbeam_channel::Receiver;
use crossbeam_channel::SendError;
use crossbeam_channel::Sender;
use crossbeam_channel::TryRecvError;
use crossbeam_channel::TrySendError;
use rayon::ThreadPool;
use rayon::prelude::*;

struct Snd<'pool>(&'pool ThreadPool, Sender<()>);

fn bounded_channel<'pool>(
    pool: &'pool ThreadPool,
    capacity: usize,
) -> (Snd<'pool>, Recv<'pool>) {
    let (snd, rcv) = crossbeam_channel::bounded(capacity);
    (Snd(pool, snd), Recv(pool, rcv))
}

impl<'pool> Snd<'pool> {
    fn send(&self) -> Result<(), SendError<()>> {
        let mut t = ();
        while let Err(e) = self.1.try_send(t) {
            match e {
                TrySendError::Full(elt) => {
                    t = elt;
                    self.0.yield_now();
                }
                TrySendError::Disconnected(_) => return Err(SendError(())),
            }
        }
        Ok(())
    }
}

struct Recv<'pool>(&'pool ThreadPool, Receiver<()>);

impl Iterator for Recv<'_> {
    type Item = ();
    fn next(&mut self) -> Option<Self::Item> {
        self.try_recv()
    }
}

impl<'pool> Recv<'pool> {
    fn try_recv(&self) -> Option<()> {
        loop {
            match self.1.try_recv() {
                Ok(t) => return Some(t),
                Err(TryRecvError::Empty) => {
                    self.0.yield_now();
                }
                Err(TryRecvError::Disconnected) => return None,
            };
        }
    }
}

fn main() {
    let pool = rayon::ThreadPoolBuilder::new().build().unwrap();
    let (snd, rcv) = bounded_channel(&pool, 1);
    pool.join(
        || {
            for _ in 0..2 {
                snd.send().unwrap();
            }
        },
        || rcv.par_bridge().for_each(|()| {}),
    );
}

[sftse]

Further minimized

use crossbeam_channel::TryRecvError;
use crossbeam_channel::TrySendError;

fn main() {
    let pool = rayon::ThreadPoolBuilder::new().build().unwrap();
    let (snd, rcv) = crossbeam_channel::bounded(1);
    pool.join(
        || {
            for _ in 0..2 {
                while let Err(e) = snd.try_send(()) {
                    match e {
                        TrySendError::Full(_) => pool.yield_now(),
                        TrySendError::Disconnected(_) => break,
                    };
                }
            }
        },
        || {
            loop {
                match rcv.try_recv() {
                    Ok(()) => continue,
                    Err(TryRecvError::Empty) => pool.yield_now(),
                    Err(TryRecvError::Disconnected) => break,
                };
            }
        },
    );
}

[cuviper]

Rayon doesn't do well with "outside" dependency edges like a channel send/recv pair. The work-stealing algorithm is greedy, and it has no idea that one of these closure depends on the completion of another.

In the case of your third minimized example, the send closure quickly blocks and calls yield_now, which enters work-stealing. If it does this before any other thread has woken up to steal that second join closure, then the "send" thread may get that job itself, becoming the "recv" thread. So then it will receive the one item already sent, and loop forever waiting for another item (when it should be sending that itself).

Note that the recv yield_now does not help matters, because these are plain closures, not like co-routines that could be swapped around on the stack. The send closure is underneath the recv closure on the real stack, and it can't continue until the recv closure returns.

We've been using rayon in a configuration with two thread pools connected by a channel, the first stage relying on some dynamic conditions to send tasks to the second thread pool. This always seemed a bit of a strange setup, so we attempted to refactor this to run on the same thread pool.

The original setup worked because they have independent threads (and call stacks) to work on send and recv. I think this could only get you in trouble if the "dynamic conditions" of sending have any dependency on the results of the receiving thread, forming a dependency loop.

[cuviper]

BTW, a simpler problem with the second and third examples is that they don't consume the sender in the join closure, only capture it by reference. So even if you don't have any work-stealing-induced nested calls, the receiver is never going to see a disconnected state. You need to move in those closures to make sure the sender is dropped when that's done.

(The first example doesn't have this problem because foo takes the sender by value.)

[sftse]

Thanks. We should probably expand the documentation to warn against this.

You need to move in those closures to make sure the sender is dropped when that's done.

Thanks, I forgot about that when brute-minimizing the example.

[sftse]

Renamed issue for now, would have helped me discover the problem.