Receivers don't always get the broadcasted item

[lespea]

Took me all day but I finally got a simple test case. If you cargo run this a few times it will eventually hang (on my machine anyway).

extern crate bus;

use std::thread;

use bus::Bus;

fn main() {
    println!("Starting");

    let strings = vec!["hi", "bye", "one", "two"];
    let mut bus = Bus::new(2);

    {
        let recv = bus.add_rx();

        thread::spawn(move || {
            println!("Looping 1");
            for rec in recv {
                println!("Got 1 :: {}", rec);
            }
        });
    };

    {
        let recv = bus.add_rx();

        thread::spawn(move || {
            println!("Looping 2");
            for rec in recv {
                println!("Got a 2 :: {}", rec);
            }
        });
    };

    for f in strings {
        println!("bcast :: {}", f);
        bus.broadcast(0);
    }
    println!("Done");
}

Example output

Finishes:

    Finished dev [unoptimized + debuginfo] target(s) in 0.08s
     Running `target/debug/fse_dump`
Starting
Looping 1
bcast :: hi
bcast :: bye
Looping 2
Got 1 :: 0
bcast :: one
Got 1 :: 0
Got 1 :: 0
Got a 2 :: 0
Got a 2 :: 0
Got a 2 :: 0
bcast :: two
Got 1 :: 0
Got a 2 :: 0
Done

Stuck:

    Finished dev [unoptimized + debuginfo] target(s) in 0.06s
     Running `target/debug/fse_dump`
Starting
Looping 1
bcast :: hi
Looping 2
Got a 2 :: 0
bcast :: bye
bcast :: one
Got 1 :: 0
Got 1 :: 0
Got 1 :: 0
bcast :: two

Machine info:

$ uname -a
Darwin 34363bc7dc9c 18.2.0 Darwin Kernel Version 18.2.0: Thu Dec 20 20:46:53 PST 2018; root:xnu-4903.241.1~1/RELEASE_X86_64 x86_64 i386 MacBookPro11,3 Darwin

$ rustc --version
rustc 1.33.0 (2aa4c46cf 2019-02-28)

[jonhoo]

Hmm, that's weird. I cannot for the life of me get the code you posted to get stuck. I've run it literally a thousand times, and it always completes. You may have to do a little debugging for me here.. The log you gave is a little disconcerting, because it looks like what gets stuck is the sender side. Specifically, here's how the tail, the heads, and the seats should look based on that log:

Starting
> head[0] = head[1] = tail = 0
> tail = 0, seat[x].read = 0

bcast :: hi
> seat[tail + 1].read == 0?
> yes! allow write. and then:
> seat[0].read = 0
> tail = 1

> head[1] != tail, so receive in head[1] (= 0)
> head[1] = 1
> seat[0].read = 1
Got a 2 :: 0

bcast :: bye
> seat[tail + 1].read == 0?
> yes! allow write. and then:
> tail = 2
> seat[0].read = 1, seat[1].read = 0

bcast :: one
> seat[tail + 1].read = seat[0].read == 2?
> NO! block writer.

> head[0] != tail, so receive in head[0] (= 0)
> head[0] = 1
> seat[0].read = 2
> seat.read == nreaders (2), so wake up writer
Got 1 :: 0

> head[0] != tail, so receive in head[0] (= 1)
> head[0] = 2
> seat[1].read = 1
Got 1 :: 0

> head[0] != tail, so receive in head[0] (= 2)
> head[0] = 0
> seat[2].read = 1
Got 1 :: 0

> [in writer] seat[tail + 1].read = seat[0].read == self.expected(tail + 1), so allow "one" broadcast to go through
> tail = 0
> seat[0].read = 0
bcast :: two
> seat[tail + 1].read = seat[1].read == 2?
> NO! block writer.

What seems wrong here is that the second reader is never unblocked so that it realizes it should read. In theory that shouldn't be possible, since every park in BusReader::recv_inner has a timeout associated with it, so it should always eventually wake up. This suggests to me that it's somehow blocked somewhere else, or has entered an infinite loop. The trick is going to be to figure out where that thread is stuck, and why.

Here's a diff I'd try that may make debugging easier:

diff --git a/src/lib.rs b/src/lib.rs
index 730a85a..793df60 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -216,6 +216,7 @@ impl<T: Clone + Sync> Seat<T> {
 
             // since we're the last reader, no-one else will be cloning this value, so we can
             // safely take a mutable reference, and just take the val instead of cloning it.
+            eprintln!("?: EMPTY SEAT");
             unsafe { &mut *self.state.get() }.val.take().unwrap()
         } else {
             state
@@ -354,6 +355,7 @@ impl<T> Bus<T> {
     /// Note that broadcasts will succeed even if there are no consumers!
     fn broadcast_inner(&mut self, val: T, block: bool) -> Result<(), T> {
         let tail = self.state.tail.load(atomic::Ordering::Relaxed);
+        eprintln!(">: TRY SEND @ {}", tail);
 
         // we want to check if the next element over is free to ensure that we always leave one
         // empty space between the head and the tail. This is necessary so that readers can
@@ -529,6 +531,7 @@ impl<T> Bus<T> {
         self.readers += 1;
 
         BusReader {
+            i: self.readers,
             bus: Arc::clone(&self.state),
             head: self.state.tail.load(atomic::Ordering::Relaxed),
             leaving: self.leaving.0.clone(),
@@ -580,6 +583,7 @@ enum RecvCondition {
 /// assert_eq!(tx.try_broadcast(true), Ok(()));
 /// ```
 pub struct BusReader<T> {
+    i: usize,
     bus: Arc<BusInner<T>>,
     head: usize,
     leaving: mpsc::Sender<usize>,
@@ -595,6 +599,8 @@ impl<T: Clone + Sync> BusReader<T> {
     /// parked until there is another broadcast on the bus, at which point the receive will be
     /// performed.
     fn recv_inner(&mut self, block: RecvCondition) -> Result<T, std_mpsc::RecvTimeoutError> {
+        eprintln!("{}: TRY RECV @ {}", self.i, self.head);
+
         if self.closed {
             return Err(std_mpsc::RecvTimeoutError::Disconnected);
         }
@@ -638,6 +644,7 @@ impl<T: Clone + Sync> BusReader<T> {
 
             // park and tell writer to notify on write
             if first {
+                eprintln!("{}: send to waiting", self.i);
                 if let Err(..) = self.waiting.send((thread::current(), self.head)) {
                     // writer has gone away, but somehow we _just_ missed the close signal (in
                     // self.bus.closed). iterate again to ensure the channel is _actually_ empty.
@@ -652,6 +659,7 @@ impl<T: Clone + Sync> BusReader<T> {
                     RecvCondition::Timeout(t) => {
                         match t.checked_sub(start.as_ref().unwrap().elapsed()) {
                             Some(left) => {
+                                eprintln!("{}: park", self.i);
                                 if left < spintime {
                                     thread::park_timeout(left);
                                 } else {
@@ -666,6 +674,7 @@ impl<T: Clone + Sync> BusReader<T> {
                         }
                     }
                     RecvCondition::Block => {
+                        eprintln!("{}: blocking", self.i);
                         thread::park_timeout(spintime);
                     }
                     RecvCondition::Try => unreachable!(),
@@ -673,6 +682,8 @@ impl<T: Clone + Sync> BusReader<T> {
             }
         }
 
+        eprintln!("{}: GOT", self.i);
+
         let head = self.head;
         let ret = self.bus.ring[head].take();
 

[lespea]

Okay I'll try to see where it's getting stuck tomorrow when I get to work. Thanks again for the quick response.

[lespea]

Sorry today's been very busy. I brought in the log crate so we get timestamps and all that. The blocking log repeats forever.

cargo run                                                                                                                                                                                    [21:38:24]
    Finished dev [optimized + debuginfo] target(s) in 0.08s
     Running `target/debug/bad`
[2019-03-08T03:38:25.609746000Z INFO  bad] Starting
[2019-03-08T03:38:25.610060000Z INFO  bad] Looping 1
[2019-03-08T03:38:25.610099000Z WARN  bus] 1: Try Recv @ 0
[2019-03-08T03:38:25.610101000Z WARN  bad] bcast :: hi
[2019-03-08T03:38:25.610130000Z WARN  bus] 1: send to waiting
[2019-03-08T03:38:25.610148000Z INFO  bad] Looping 2
[2019-03-08T03:38:25.610172000Z WARN  bus] 2: Try Recv @ 0
[2019-03-08T03:38:25.610183000Z WARN  bus] 2: send to waiting
[2019-03-08T03:38:25.610149000Z WARN  bus] >: Try Send @ 0
[2019-03-08T03:38:25.610207000Z ERROR bus] 2: blocking
[2019-03-08T03:38:25.610194000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.610252000Z WARN  bad] bcast :: bye
[2019-03-08T03:38:25.610266000Z WARN  bus] >: Try Send @ 1
[2019-03-08T03:38:25.610275000Z WARN  bad] bcast :: one
[2019-03-08T03:38:25.610297000Z INFO  bus] 1: GOT
[2019-03-08T03:38:25.610331000Z WARN  bus] >: Try Send @ 2
[2019-03-08T03:38:25.610354000Z WARN  bad] Got 1 :: 0
[2019-03-08T03:38:25.610378000Z WARN  bus] 1: Try Recv @ 1
[2019-03-08T03:38:25.610387000Z INFO  bus] 1: GOT
[2019-03-08T03:38:25.610396000Z WARN  bad] Got 1 :: 0
[2019-03-08T03:38:25.610403000Z WARN  bus] 1: Try Recv @ 2
[2019-03-08T03:38:25.610411000Z WARN  bus] 1: send to waiting
[2019-03-08T03:38:25.610426000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.610588000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.610773000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.610948000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.611118000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.611282000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.611420000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.611586000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.611755000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.611921000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.612089000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.612250000Z ERROR bus] 1: blocking
[2019-03-08T03:38:25.612418000Z ERROR bus] 1: blocking

[lespea]

https://git.target.com/z001j31/busBlockingTest -> test repo I'm using.

I was messing around with trying 2018 edition and updated crates to see if it would matter (it doesn't). Also tried compiling with -C target-cpu=native and nightly.

[jonhoo]

Wow, that's really weird. Specifically, it looks like 2 never wakes up again after this call:

bus/src/lib.rs

Line 669 in 10719f1

thread::park_timeout(spintime);

Can you try to add a print statement after that park too just to verify? That's definitely not supposed to happen. Even if it's not woken up directly for some reason, the timeout should ensure that eventually checks again!

[lespea]

[2019-03-08T15:59:19.318217000Z INFO  bad] Looping 1
[2019-03-08T15:59:19.318246000Z WARN  bus] 1: Try Recv @ 0
[2019-03-08T15:59:19.318267000Z WARN  bus] 1: send to waiting
[2019-03-08T15:59:19.318281000Z INFO  bad] Looping 2
[2019-03-08T15:59:19.318293000Z ERROR bus] 1: blocking
[2019-03-08T15:59:19.318302000Z WARN  bus] 2: Try Recv @ 0
[2019-03-08T15:59:19.318312000Z WARN  bus] 2: send to waiting
[2019-03-08T15:59:19.318274000Z WARN  bad] bcast :: hi
[2019-03-08T15:59:19.318335000Z WARN  bus] >: Try Send @ 0
[2019-03-08T15:59:19.318337000Z ERROR bus] 2: blocking
[2019-03-08T15:59:19.318368000Z WARN  bad] bcast :: bye
[2019-03-08T15:59:19.318379000Z WARN  bus] >: Try Send @ 1
[2019-03-08T15:59:19.318393000Z WARN  bad] bcast :: one
[2019-03-08T15:59:19.318402000Z WARN  bus] >: Try Send @ 2
[2019-03-08T15:59:19.318407000Z ERROR bus] 1: after park
[2019-03-08T15:59:19.318418000Z INFO  bus] 1: GOT
[2019-03-08T15:59:19.318426000Z WARN  bad] Got 1 :: 0
[2019-03-08T15:59:19.318436000Z WARN  bus] 1: Try Recv @ 1
[2019-03-08T15:59:19.318443000Z INFO  bus] 1: GOT
[2019-03-08T15:59:19.318451000Z WARN  bad] Got 1 :: 0
[2019-03-08T15:59:19.318459000Z WARN  bus] 1: Try Recv @ 2
[2019-03-08T15:59:19.318466000Z WARN  bus] 1: send to waiting
[2019-03-08T15:59:19.318479000Z ERROR bus] 1: blocking
[2019-03-08T15:59:19.318660000Z ERROR bus] 1: after park
[2019-03-08T15:59:19.318676000Z ERROR bus] 1: blocking
[2019-03-08T15:59:19.318875000Z ERROR bus] 1: after park
[2019-03-08T15:59:19.318891000Z ERROR bus] 1: blocking
[2019-03-08T15:59:19.319085000Z ERROR bus] 1: after park
[2019-03-08T15:59:19.319101000Z ERROR bus] 1: blocking
[2019-03-08T15:59:19.319271000Z ERROR bus] 1: after park
[2019-03-08T15:59:19.319287000Z ERROR bus] 1: blocking
[2019-03-08T15:59:19.319448000Z ERROR bus] 1: after park
[2019-03-08T15:59:19.319464000Z ERROR bus] 1: blocking
[2019-03-08T15:59:19.319634000Z ERROR bus] 1: after park
[2019-03-08T15:59:19.319649000Z ERROR bus] 1: blocking
[2019-03-08T15:59:19.319806000Z ERROR bus] 1: after park
[2019-03-08T15:59:19.319822000Z ERROR bus] 1: blocking
[2019-03-08T15:59:19.319963000Z ERROR bus] 1: after park```

[lespea]

So I changed the park_timeout to this: thread::sleep(Duration::from_millis(100)); and can no longer get it to hang!

[lespea]

Just double checked the spin duration didn't somehow get crazy and it looks okay to me:

[2019-03-08T16:02:48.642960000Z INFO  bad] Looping 1
[2019-03-08T16:02:48.642983000Z WARN  bus] 1: Try Recv @ 0
[2019-03-08T16:02:48.642999000Z WARN  bus] 1: send to waiting
[2019-03-08T16:02:48.643002000Z WARN  bad] bcast :: hi
[2019-03-08T16:02:48.643013000Z INFO  bad] Looping 2
[2019-03-08T16:02:48.643044000Z WARN  bus] 2: Try Recv @ 0
[2019-03-08T16:02:48.643057000Z WARN  bus] 2: send to waiting
[2019-03-08T16:02:48.643049000Z ERROR bus] 1: blocking :: timeout=100µs
[2019-03-08T16:02:48.643023000Z WARN  bus] >: Try Send @ 0
[2019-03-08T16:02:48.643086000Z ERROR bus] 2: blocking :: timeout=100µs
[2019-03-08T16:02:48.643121000Z WARN  bad] bcast :: bye
[2019-03-08T16:02:48.643140000Z WARN  bus] >: Try Send @ 1
[2019-03-08T16:02:48.643153000Z WARN  bad] bcast :: one
[2019-03-08T16:02:48.643155000Z ERROR bus] 1: after park
[2019-03-08T16:02:48.643167000Z WARN  bus] >: Try Send @ 2
[2019-03-08T16:02:48.643172000Z INFO  bus] 1: GOT
[2019-03-08T16:02:48.643187000Z WARN  bad] Got 1 :: 0
[2019-03-08T16:02:48.643201000Z WARN  bus] 1: Try Recv @ 1
[2019-03-08T16:02:48.643212000Z INFO  bus] 1: GOT
[2019-03-08T16:02:48.643220000Z WARN  bad] Got 1 :: 0
[2019-03-08T16:02:48.643227000Z WARN  bus] 1: Try Recv @ 2
[2019-03-08T16:02:48.643235000Z WARN  bus] 1: send to waiting
[2019-03-08T16:02:48.643249000Z ERROR bus] 1: blocking :: timeout=100µs
[2019-03-08T16:02:48.643397000Z ERROR bus] 1: after park
[2019-03-08T16:02:48.643412000Z ERROR bus] 1: blocking :: timeout=100µs
[2019-03-08T16:02:48.643554000Z ERROR bus] 1: after park
[2019-03-08T16:02:48.643564000Z ERROR bus] 1: blocking :: timeout=100µs```

[lespea]

This didn't help: thread::park_timeout(spintime * 1000);

[lespea]

Okay so most of the time this will only print Parking 1\nParking 2 and then sometimes it works fine. This should always work right?

use std::thread;
use std::time::Duration;

fn main() {
    let t1 = thread::spawn(|| {
        let sleep = Duration::from_micros(50);
        for _ in 0..100 {
            println!("Parking1");
            thread::park_timeout(sleep);
        }
    });

    let t2 = thread::spawn(|| {
        let sleep = Duration::from_micros(50);
        for _ in 0..100 {
            println!("Parking2");
            thread::park_timeout(sleep);
        }
    });

    t1.join().expect("Couldn't join thread 1");
    t2.join().expect("Couldn't join thread 2");
}

Edit: I should note that if I only spawn a single thread I can never get it to block.

[jonhoo]

That definitely seems wrong. I would file an issue on https://github.com/rust-lang/rust/ !

[lespea]

Okay will do. Thanks for all the help!

[jonhoo]

Thanks for doing all that debugging! :D Would you mind linking the reported issue back here so we can keep an eye on it?

[lespea]

Yup. This has been a pretty frustrating bug haha. (I linked this issue in the one I filed)