futures-channel fails under Miri

[GoldsteinE]

Discovered while discussing rust-lang/unsafe-code-guidelines#380, probably related to rust-lang/unsafe-code-guidelines#148.

This code (playground):

use std::thread;
use futures::{channel::mpsc, executor::block_on, sink::SinkExt, stream::StreamExt};

async fn send_sequence(n: u32, mut sender: mpsc::Sender<u32>) {
    for x in 0..n {
        sender.send(n - x).await.unwrap();
    }
}

fn main() {
    let (tx, rx) = mpsc::channel(1);
    let t = thread::spawn(move || block_on(send_sequence(20, tx)));
    let _list: Vec<_> = block_on(rx.collect());
    t.join().unwrap();
}

fails under Miri.

[RalfJung]

Preliminary debugging results:

• the problematic tag gets created here

  --> /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.25/src/sink/feed.rs:20:16
   |
20 |         Feed { sink, item: Some(item) }
   |                ^^^^ created tag 12798 for unique reference (pointee type futures::futures_channel::mpsc::Sender<u32>) at alloc4773[0x10..0x28] derived from <12797>
   |
   = note: inside `futures::sink::Feed::<'_, futures::futures_channel::mpsc::Sender<u32>, u32>::new` at /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.25/src/sink/feed.rs:20:16
   = note: inside `futures::sink::Send::<'_, futures::futures_channel::mpsc::Sender<u32>, u32>::new` at /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.25/src/sink/send.rs:20:22
   = note: inside `<futures::futures_channel::mpsc::Sender<u32> as futures::SinkExt<u32>>::send` at /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.25/src/sink/mod.rs:224:53
note: inside closure at src/main.rs:6:9
  --> src/main.rs:6:9
   |
6  |         sender.send(n - x).await.unwrap();
   |         ^^^^^^^^^^^^^^^^^^

• it then gets invalidated here

note: tracking was triggered
   --> /home/r/.rustup/toolchains/miri/lib/rustlib/src/rust/library/core/src/ops/deref.rs:183:9
    |
183 |         *self
    |         ^^^^^ popped tracked tag for item [Unique for <12798>] due to SharedReadWrite retag from <10022>
    |
    = note: inside `<&mut impl Future<Output = ()> as std::ops::DerefMut>::deref_mut` at /home/r/.rustup/toolchains/miri/lib/rustlib/src/rust/library/core/src/ops/deref.rs:183:9
    = note: inside `std::pin::Pin::<&mut impl Future<Output = ()>>::as_mut` at /home/r/.rustup/toolchains/miri/lib/rustlib/src/rust/library/core/src/pin.rs:703:42
    = note: inside closure at /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-executor-0.3.25/src/local_pool.rs:317:23
    = note: inside closure at /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-executor-0.3.25/src/local_pool.rs:90:37
    = note: inside `std::thread::LocalKey::<std::sync::Arc<futures_executor::local_pool::ThreadNotify>>::try_with::<[closure@futures_executor::local_pool::run_executor<(), [closure@futures::executor::block_on<impl Future<Output = ()>>::{closure#0}]>::{closure#0}], ()>` at /home/r/.rustup/toolchains/miri/lib/rustlib/src/rust/library/std/src/thread/local.rs:446:16
    = note: inside `std::thread::LocalKey::<std::sync::Arc<futures_executor::local_pool::ThreadNotify>>::with::<[closure@futures_executor::local_pool::run_executor<(), [closure@futures::executor::block_on<impl Future<Output = ()>>::{closure#0}]>::{closure#0}], ()>` at /home/r/.rustup/toolchains/miri/lib/rustlib/src/rust/library/std/src/thread/local.rs:422:9
    = note: inside `futures_executor::local_pool::run_executor::<(), [closure@futures::executor::block_on<impl Future<Output = ()>>::{closure#0}]>` at /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-executor-0.3.25/src/local_pool.rs:86:5
    = note: inside `futures::executor::block_on::<impl Future<Output = ()>>` at /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-executor-0.3.25/src/local_pool.rs:317:5
note: inside closure at src/main.rs:12:35
   --> src/main.rs:12:35
    |
12  |     let t = thread::spawn(move || block_on(send_sequence(20, tx)));
    |                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

• and then we get UB here

error: Undefined Behavior: trying to retag from <12798> for SharedReadWrite permission at alloc4773[0x10], but that tag does not exist in the borrow stack for this location
  --> /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.25/src/sink/feed.rs:24:18
   |
24 |         Pin::new(self.sink)
   |                  ^^^^^^^^^
   |                  |
   |                  trying to retag from <12798> for SharedReadWrite permission at alloc4773[0x10], but that tag does not exist in the borrow stack for this location
   |                  this error occurs as part of two-phase retag at alloc4773[0x10..0x28]
   |
   = help: this indicates a potential bug in the program: it performed an invalid operation, but the Stacked Borrows rules it violated are still experimental
   = help: see https://github.com/rust-lang/unsafe-code-guidelines/blob/master/wip/stacked-borrows.md for further information
help: <12798> was created by a Unique retag at offsets [0x10..0x28]
  --> src/main.rs:6:9
   |
6  |         sender.send(n - x).await.unwrap();
   |         ^^^^^^^^^^^^^^^^^^
help: <12798> was later invalidated at offsets [0x0..0x58] by a SharedReadWrite retag
  --> src/main.rs:12:35
   |
12 |     let t = thread::spawn(move || block_on(send_sequence(20, tx)));
   |                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   = note: BACKTRACE:
   = note: inside `futures::sink::Feed::<'_, futures::futures_channel::mpsc::Sender<u32>, u32>::sink_pin_mut` at /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.25/src/sink/feed.rs:24:18
   = note: inside `<futures::sink::Send<'_, futures::futures_channel::mpsc::Sender<u32>, u32> as futures::Future>::poll` at /home/r/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-util-0.3.25/src/sink/send.rs:37:16
note: inside closure at src/main.rs:6:27
  --> src/main.rs:6:27
   |
6  |         sender.send(n - x).await.unwrap();
   |                           ^^^^^^

This sink reference has type &mut futures::futures_channel::mpsc::Sender<u32> which is Unpin and hence requires full alias control. And yet it seems like another pointer to that sender exists in the executor?

[RalfJung]

The error occurs even without block_on and in only one thread:

#![feature(pin_macro)]

use futures::{channel::mpsc, sink::SinkExt, stream::StreamExt};
use std::future::Future;
use std::pin::pin;
use std::sync::Arc;
use std::task::*;

async fn send_sequence(n: u32, mut sender: mpsc::Sender<u32>) {
    for x in 0..n {
        sender.send(n - x).await.unwrap();
    }
}

fn main() {
    let (tx, rx) = mpsc::channel(1);

    let mut sender = pin!(send_sequence(20, tx));
    let mut receiver = pin!(rx.collect::<Vec<_>>());

    struct MyWaker;
    impl Wake for MyWaker {
        fn wake(self: Arc<Self>) {}
    }
    let waker = Waker::from(Arc::new(MyWaker));
    let mut context = Context::from_waker(&waker);

    let _ = sender.as_mut().poll(&mut context);
    let _ = receiver.as_mut().poll(&mut context);
    let _ = sender.as_mut().poll(&mut context);
}

[RalfJung]

So this is what happens:

• sender.send(n - x) creates a unique mutable reference to sender (sender has an Unpin type so there is no exception here)
• but sender is stored directly in-line in the future, so when we reborrow the future the next time, since fix handling of spurious accesses during retag miri#2694 we consider this a 'write' of that entire memory range
• that in turn invalidates the previously created unique reference (if the reborrow was considered a 'read', that would not help either)

So... well we could go back to Miri not doing 'fake' accesses on !Unpin reborrows, which would be consistent with !Freeze reborrows on shared references. The difference is that !Freeze reborrows do not get dereferenceable. Here we would be marking a pointer as dereferenceable where doing a read could actually introduce UB due to the aliasing model.

Either way, this is not a bug in futures-rs.

[RalfJung]

The one thing I am still struggling to do is reproduce this without futures-rs. For some reason this one does not fail.

Cc @saethlin -- an interesting Stacked Borrows issue

[RalfJung]

All right, I think I dug (mostly) to the bottom of this. See rust-lang/unsafe-code-guidelines#381 for details.

[taiki-e]

Addressed by rust-lang/miri#2713.