thread sanitizer warnings using channel operations

[tones111]

I'm integrating a rust library into a c++ binary (via c FFI) and began noticing some very strange behavior. Compiling the c++ code with thread sanitizer indicates potential data races in the rust standard library. I've created a reproducer that triggers the same tsan backtrace for the version we're using (1.39.0), but it also triggers warnings on nightly when building the rust side with thread sanitizer enabled (see Makefile). The code can be found here and run with make && ./main

I've further simplified the interaction by reproducing what I believe to be the same issue in safe Rust (below). Also below is the first tsan warning. I see 2 to 4 warnings from the pure rust version depending on the run.

use std::{
    sync::mpsc::{channel, sync_channel, Sender, SyncSender},
    thread,
};

fn main() {
    let mut m = Model::new();
    m.serve();

    let s1 = m.new_sender();

    // These also generate tsan warnings
    //let s2 = m.new_sender();
    //let t = thread::spawn(move || {
    //    send_job1(&s2);
    //    Model::delete_sender(s2);
    //});

    send_job1(&s1);
    send_job2(&s1);

    //let _ = t.join().unwrap();
    Model::delete_sender(s1);

    m.stop();
    m.delete();
}

#[derive(Debug)]
struct Model {
    tx: Option<SyncSender<Request>>,
    iot: Option<thread::JoinHandle<()>>,
}

#[derive(Debug)]
enum Request {
    JOB1(),
    JOB2(Sender<()>),
    End(Sender<()>),
    Stop,
}

impl Model {
    fn new() -> Self {
        Model {
            tx: None,
            iot: None,
        }
    }

    fn delete(self) {}

    fn serve(&mut self) {
        let (tx, rx) = sync_channel(5);
        let mut iot_sender = Some(tx.clone());
        self.tx = Some(tx);

        self.iot = Some(thread::spawn(move || {
            for job in rx.iter() {
                match job {
                    Request::JOB1() => {}
                    Request::JOB2(tx) => {
                        if let Some(iot_sender) = iot_sender.as_ref() {
                            let _ = iot_sender.send(Request::End(tx));
                        }
                    }
                    Request::End(tx) => {
                        let _ = tx.send(());
                    }
                    Request::Stop => {
                        iot_sender = None;
                    }
                }
            }
        }));
    }

    fn stop(&mut self) {
        if let Some(tx) = self.tx.as_ref() {
            let _ = tx.send(Request::Stop);
        }
        self.tx.take();

        if let Some(iot) = self.iot.take() {
            iot.join().unwrap();
        }
    }

    fn new_sender(&self) -> SyncSender<Request> {
        self.tx.as_ref().unwrap().clone()
    }

    fn delete_sender(_sender: SyncSender<Request>) {}
}

fn send_job1(sender: &SyncSender<Request>) -> bool {
    let mut status = false;
    match sender.send(Request::JOB1()) {
        Ok(_) => status = true,
        Err(e) => println!("send err: {:?}", e),
    }
    status
}

fn send_job2(sender: &SyncSender<Request>) -> bool {
    let (tx, rx) = channel();
    let mut status = false;
    match sender.send(Request::JOB2(tx)) {
        Ok(()) => match rx.recv() {
            Ok(_) => status = true,
            Err(e) => println!("error receiving response {}", e),
        },
        Err(e) => println!("send err: {:?}", e),
    }
    status
}

Meta

$ uname -a
Linux mobile 5.14.8-arch1-1 #1 SMP PREEMPT Sun, 26 Sep 2021 19:36:15 +0000 x86_64 GNU/Linux

$ cargo +nightly --version --verbose
cargo 1.57.0-nightly (d56b42c54 2021-09-27)
release: 1.57.0
commit-hash: d56b42c549dbb7e7d0f712c51b39400260d114d4
commit-date: 2021-09-27

first tsan warning...
$ RUSTFLAGS="-Z sanitizer=thread" cargo +nightly run

WARNING: ThreadSanitizer: data race (pid=9313)
  Write of size 8 at 0x7b0800000020 by main thread:
    #0 free /rustc/llvm/src/llvm-project/compiler-rt/lib/tsan/rtl/tsan_interceptors_posix.cpp:714 (threads+0x1425a)
    #1 <alloc::sync::Arc<T> as core::ops::drop::Drop>::drop /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/alloc/src/sync.rs:1626 (threads+0xd4565)
    #2 core::ptr::drop_in_place<alloc::sync::Arc<std::sync::mpsc::blocking::Inner>> /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/ptr/mod.rs:188 (threads+0xd4565)
    #3 core::ptr::drop_in_place<std::sync::mpsc::blocking::WaitToken> /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/ptr/mod.rs:188 (threads+0xd4565)
    #4 std::sync::mpsc::blocking::WaitToken::wait /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/sync/mpsc/blocking.rs:70 (threads+0xd4565)
    #5 std::sync::mpsc::Receiver<T>::recv /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/sync/mpsc/mod.rs:1158 (threads+0xa9664)
    #6 threads::send_job2 /home/paul/src/threads/src/main.rs:109 (threads+0x9ee14)
    #7 threads::main /home/paul/src/threads/src/main.rs:20 (threads+0x9e3b6)
    #8 core::ops::function::FnOnce::call_once /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/ops/function.rs:227 (threads+0xacc8e)
    #9 std::sys_common::backtrace::__rust_begin_short_backtrace /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/sys_common/backtrace.rs:125 (threads+0x9a231)
    #10 std::rt::lang_start::{{closure}} /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/rt.rs:146 (threads+0xbfcde)
    #11 core::ops::function::impls::<impl core::ops::function::FnOnce<A> for &F>::call_once /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/ops/function.rs:259 (threads+0xd5100)
    #12 std::panicking::try::do_call /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panicking.rs:403 (threads+0xd5100)
    #13 std::panicking::try /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panicking.rs:367 (threads+0xd5100)
    #14 std::panic::catch_unwind /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panic.rs:133 (threads+0xd5100)
    #15 std::rt::lang_start_internal::{{closure}} /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/rt.rs:128 (threads+0xd5100)
    #16 std::panicking::try::do_call /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panicking.rs:403 (threads+0xd5100)
    #17 std::panicking::try /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panicking.rs:367 (threads+0xd5100)
    #18 std::panic::catch_unwind /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panic.rs:133 (threads+0xd5100)
    #19 std::rt::lang_start_internal /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/rt.rs:128 (threads+0xd5100)
    #20 main ??:? (threads+0x9f0c6)

  Previous atomic write of size 8 at 0x7b0800000020 by thread T1:
    #0 __tsan_atomic64_fetch_sub /rustc/llvm/src/llvm-project/compiler-rt/lib/tsan/rtl/tsan_interface_atomic.cpp:643 (threads+0x57d85)
    #1 core::sync::atomic::atomic_sub /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/sync/atomic.rs:2406 (threads+0xbbd99)
    #2 core::sync::atomic::AtomicUsize::fetch_sub /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/sync/atomic.rs:1774 (threads+0x927c9)
    #3 <alloc::sync::Arc<T> as core::ops::drop::Drop>::drop /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/alloc/src/sync.rs:1591 (threads+0xb1433)
    #4 core::ptr::drop_in_place<alloc::sync::Arc<std::sync::mpsc::blocking::Inner>> /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/ptr/mod.rs:188 (threads+0xafef1)
    #5 core::ptr::drop_in_place<std::sync::mpsc::blocking::SignalToken> /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/ptr/mod.rs:188 (threads+0xaea11)
    #6 std::sync::mpsc::oneshot::Packet<T>::send /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/sync/mpsc/oneshot.rs:111 (threads+0xa4aea)
    #7 std::sync::mpsc::Sender<T>::send /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/sync/mpsc/mod.rs:805 (threads+0xa84e2)
    #8 threads::Model::serve::{{closure}} /home/paul/src/threads/src/main.rs:68 (threads+0xb5033)
    #9 std::sys_common::backtrace::__rust_begin_short_backtrace /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/sys_common/backtrace.rs:125 (threads+0x9a2d5)
    #10 std::thread::Builder::spawn_unchecked::{{closure}}::{{closure}} /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/thread/mod.rs:483 (threads+0xbb365)
    #11 <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/panic/unwind_safe.rs:271 (threads+0xa7c45)
    #12 std::panicking::try::do_call /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/panicking.rs:403 (threads+0xa1a47)
    #13 __rust_try 3bw9yl3jylpgppbo:? (threads+0xa1e38)
    #14 std::panicking::try /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/panicking.rs:367 (threads+0xa1939)
    #15 std::panic::catch_unwind /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/panic.rs:133 (threads+0xbb765)
    #16 std::thread::Builder::spawn_unchecked::{{closure}} /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/thread/mod.rs:482 (threads+0xbb0b3)
    #17 core::ops::function::FnOnce::call_once{{vtable.shim}} /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/ops/function.rs:227 (threads+0xacb51)
    #18 <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/alloc/src/boxed.rs:1638 (threads+0xd9da2)
    #19 <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/alloc/src/boxed.rs:1638 (threads+0xd9da2)
    #20 std::sys::unix::thread::Thread::new::thread_start /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/sys/unix/thread.rs:106 (threads+0xd9da2)

  Thread T1 (tid=9319, running) created by main thread at:
    #0 pthread_create /rustc/llvm/src/llvm-project/compiler-rt/lib/tsan/rtl/tsan_interceptors_posix.cpp:977 (threads+0x150ad)
    #1 std::sys::unix::thread::Thread::new /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/sys/unix/thread.rs:85 (threads+0xd9c24)
    #2 std::thread::Builder::spawn /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/thread/mod.rs:388 (threads+0xbb3f7)
    #3 std::thread::spawn /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/thread/mod.rs:630 (threads+0xba57a)
    #4 threads::Model::serve /home/paul/src/threads/src/main.rs:58 (threads+0x9e6fe)
    #5 threads::main /home/paul/src/threads/src/main.rs:8 (threads+0x9e343)
    #6 core::ops::function::FnOnce::call_once /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/ops/function.rs:227 (threads+0xacc8e)
    #7 std::sys_common::backtrace::__rust_begin_short_backtrace /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/sys_common/backtrace.rs:125 (threads+0x9a231)
    #8 std::rt::lang_start::{{closure}} /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/std/src/rt.rs:146 (threads+0xbfcde)
    #9 core::ops::function::impls::<impl core::ops::function::FnOnce<A> for &F>::call_once /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/core/src/ops/function.rs:259 (threads+0xd5100)
    #10 std::panicking::try::do_call /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panicking.rs:403 (threads+0xd5100)
    #11 std::panicking::try /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panicking.rs:367 (threads+0xd5100)
    #12 std::panic::catch_unwind /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panic.rs:133 (threads+0xd5100)
    #13 std::rt::lang_start_internal::{{closure}} /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/rt.rs:128 (threads+0xd5100)
    #14 std::panicking::try::do_call /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panicking.rs:403 (threads+0xd5100)
    #15 std::panicking::try /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panicking.rs:367 (threads+0xd5100)
    #16 std::panic::catch_unwind /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/panic.rs:133 (threads+0xd5100)
    #17 std::rt::lang_start_internal /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2//library/std/src/rt.rs:128 (threads+0xd5100)
    #18 main ??:? (threads+0x9f0c6)

SUMMARY: ThreadSanitizer: data race /rustc/c02371c442f811878ab3a0f5a813402b6dfd45d2/library/alloc/src/sync.rs:1626 in <alloc::sync::Arc<T> as core::ops::drop::Drop>::drop

[SkiFire13]

Did you set -Zsanitizer=thread and -Zbuild-std when compiling the rust code?

Edit: looking at the Makefile it looks like you tried -Zsanitizer=thread, however you forgot -Zbuild-std.

[tones111]

You're right. Sorry, this is my first experience running sanitizers in Rust. Setting that flag silences the warnings.

With that information I tried the same trick with a nighty closer to the 1.39.0 release and back to my FFI reproducer. Thread sanitzer seg faults, but some println debugging showed the race coming from Sender::recv() which is supposed to block until the request has been processed. Switching to a SyncSender looks like it might be a potential workaround, so I'll give that a shot.

Thanks for the tip!

[SkiFire13]

Note that #65097, which "fixed" the interaction with thread sanitizers, was merged before the release of rust 1.42