Undefined Behaviour in tests/stream_chain.rs

[tiif]

Version
Tokio master branch commit 27539ae3

Platform

Linux thinkpad 6.8.0-40-generic #40~22.04.3-Ubuntu SMP PREEMPT_DYNAMIC Tue Jul 30 17:30:19 UTC 2 x86_64 x86_64 x86_64 GNU/Linux

rustc version:

rustc 1.83.0-nightly (12b26c13f 2024-09-07)
binary: rustc
commit-hash: 12b26c13fba25c9e1bc2fdf05f3c2dbb851c83de
commit-date: 2024-09-07
host: x86_64-unknown-linux-gnu
release: 1.83.0-nightly
LLVM version: 19.1.0

Description
Miri reported UB with MIRIFLAGS="-Zmiri-disable-isolation -Zmiri-strict-provenance -Zmiri-retag-fields" cargo +nightly miri test --features full --test stream_chain on tokio master branch 27539ae3:

Full trace:

    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.05s
     Running tests/stream_chain.rs (target/miri/x86_64-unknown-linux-gnu/debug/deps/stream_chain-fe570dc7ff6e1d67)

running 3 tests
test basic_usage ... ok
test pending_first ... error: Undefined Behavior: trying to retag from <464381> for SharedReadWrite permission at alloc170919[0x0], but that tag does not exist in the borrow stack for this location
   --> /home/byt/Documents/tokio/tokio/src/sync/mpsc/unbounded.rs:170:22
    |
170 |         poll_fn(|cx| self.poll_recv(cx)).await
    |                      ^^^^
    |                      |
    |                      trying to retag from <464381> for SharedReadWrite permission at alloc170919[0x0], but that tag does not exist in the borrow stack for this location
    |                      this error occurs as part of two-phase retag at alloc170919[0x0..0x8]
    |
    = 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: <464381> was created by a Unique retag at offsets [0x0..0x8]
   --> /home/byt/Documents/tokio/tokio/src/sync/mpsc/unbounded.rs:170:9
    |
170 |         poll_fn(|cx| self.poll_recv(cx)).await
    |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
help: <464381> was later invalidated at offsets [0x0..0x70] by a SharedReadOnly retag
   --> /home/byt/Documents/tokio/tokio-test/src/task.rs:153:9
    |
153 |         self.future.size_hint()
    |         ^^^^^^^^^^^^^^^^^^^^^^^
    = note: BACKTRACE (of the first span) on thread `pending_first`:
    = note: inside closure at /home/byt/Documents/tokio/tokio/src/sync/mpsc/unbounded.rs:170:22: 170:26
note: inside `<tokio::future::poll_fn::PollFn<{closure@tokio::sync::mpsc::UnboundedReceiver<i32>::recv::{closure#0}::{closure#0}}> as std::future::Future>::poll`
   --> /home/byt/Documents/tokio/tokio/src/future/poll_fn.rs:58:9
    |
58  |         (me.f)(cx)
    |         ^^^^^^^^^^
note: inside closure
   --> /home/byt/Documents/tokio/tokio/src/sync/mpsc/unbounded.rs:170:42
    |
170 |         poll_fn(|cx| self.poll_recv(cx)).await
    |                                          ^^^^^
note: inside closure
   --> tokio-stream/tests/support/mpsc.rs:9:42
    |
9   |         while let Some(item) = rx.recv().await {
    |                                          ^^^^^
    = note: inside `<async_stream::__private::AsyncStream<i32, {async block@/home/byt/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59}> as tokio_stream::Stream>::poll_next` at /home/byt/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/async_stream.rs:56:13: 56:34
note: inside `<tokio_stream::adapters::Fuse<async_stream::__private::AsyncStream<i32, {async block@/home/byt/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59}>> as tokio_stream::Stream>::poll_next`
   --> /home/byt/Documents/tokio/tokio-stream/src/stream_ext/fuse.rs:35:36
    |
35  |             Some(stream) => ready!(stream.poll_next(cx)),
    |                                    ^^^^^^^^^^^^^^^^^^^^
note: inside `<tokio_stream::adapters::Chain<async_stream::__private::AsyncStream<i32, {async block@/home/byt/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59}>, async_stream::__private::AsyncStream<i32, {async block@/home/byt/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59}>> as tokio_stream::Stream>::poll_next`
   --> /home/byt/Documents/tokio/tokio-stream/src/stream_ext/chain.rs:40:33
    |
40  |         if let Some(v) = ready!(me.a.poll_next(cx)) {
    |                                 ^^^^^^^^^^^^^^^^^^
note: inside closure
   --> /home/byt/Documents/tokio/tokio-test/src/task.rs:133:30
    |
133 |         self.task.enter(|cx| stream.poll_next(cx))
    |                              ^^^^^^^^^^^^^^^^^^^^
note: inside `tokio_test::task::MockTask::enter::<{closure@tokio_test::task::Spawn<tokio_stream::adapters::Chain<async_stream::__private::AsyncStream<i32, {async block@/home/byt/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59}>, async_stream::__private::AsyncStream<i32, {async block@/home/byt/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59}>>>::poll_next::{closure#0}}, std::task::Poll<std::option::Option<i32>>>`
   --> /home/byt/Documents/tokio/tokio-test/src/task.rs:177:9
    |
177 |         f(&mut cx)
    |         ^^^^^^^^^^
note: inside `tokio_test::task::Spawn::<tokio_stream::adapters::Chain<async_stream::__private::AsyncStream<i32, {async block@/home/byt/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59}>, async_stream::__private::AsyncStream<i32, {async block@/home/byt/.cargo/registry/src/index.crates.io-6f17d22bba15001f/async-stream-0.3.5/src/lib.rs:193:9: 193:59}>>>::poll_next`
   --> /home/byt/Documents/tokio/tokio-test/src/task.rs:133:9
    |
133 |         self.task.enter(|cx| stream.poll_next(cx))
    |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside closure
   --> tokio-stream/tests/stream_chain.rs:77:39
    |
77  |     assert_eq!(Some(2), assert_ready!(stream.poll_next()));
    |                                       ^^^^^^^^^^^^^^^^^^
    = note: inside `<std::pin::Pin<&mut dyn std::future::Future<Output = ()>> as std::future::Future>::poll` at /home/byt/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9: 123:61
    = note: inside `<std::pin::Pin<&mut std::pin::Pin<&mut dyn std::future::Future<Output = ()>>> as std::future::Future>::poll` at /home/byt/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/future.rs:123:9: 123:61
note: inside closure
   --> /home/byt/Documents/tokio/tokio/src/runtime/scheduler/current_thread/mod.rs:696:57
    |
696 |                         crate::runtime::coop::budget(|| future.as_mut().poll(&mut cx))
    |                                                         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside `tokio::runtime::coop::with_budget::<std::task::Poll<()>, {closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::block_on<std::pin::Pin<&mut std::pin::Pin<&mut dyn std::future::Future<Output = ()>>>>::{closure#0}::{closure#0}::{closure#0}}>`
   --> /home/byt/Documents/tokio/tokio/src/runtime/coop.rs:107:5
    |
107 |     f()
    |     ^^^
note: inside `tokio::runtime::coop::budget::<std::task::Poll<()>, {closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::block_on<std::pin::Pin<&mut std::pin::Pin<&mut dyn std::future::Future<Output = ()>>>>::{closure#0}::{closure#0}::{closure#0}}>`
   --> /home/byt/Documents/tokio/tokio/src/runtime/coop.rs:73:5
    |
73  |     with_budget(Budget::initial(), f)
    |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside closure
   --> /home/byt/Documents/tokio/tokio/src/runtime/scheduler/current_thread/mod.rs:696:25
    |
696 |                         crate::runtime::coop::budget(|| future.as_mut().poll(&mut cx))
    |                         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside `tokio::runtime::scheduler::current_thread::Context::enter::<std::task::Poll<()>, {closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::block_on<std::pin::Pin<&mut std::pin::Pin<&mut dyn std::future::Future<Output = ()>>>>::{closure#0}::{closure#0}}>`
   --> /home/byt/Documents/tokio/tokio/src/runtime/scheduler/current_thread/mod.rs:423:19
    |
423 |         let ret = f();
    |                   ^^^
note: inside closure
   --> /home/byt/Documents/tokio/tokio/src/runtime/scheduler/current_thread/mod.rs:695:36
    |
695 |                       let (c, res) = context.enter(core, || {
    |  ____________________________________^
696 | |                         crate::runtime::coop::budget(|| future.as_mut().poll(&mut cx))
697 | |                     });
    | |______________________^
note: inside closure
   --> /home/byt/Documents/tokio/tokio/src/runtime/scheduler/current_thread/mod.rs:774:68
    |
774 |         let (core, ret) = context::set_scheduler(&self.context, || f(core, context));
    |                                                                    ^^^^^^^^^^^^^^^^
note: inside `tokio::runtime::context::scoped::Scoped::<tokio::runtime::scheduler::Context>::set::<{closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::enter<{closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::block_on<std::pin::Pin<&mut std::pin::Pin<&mut dyn std::future::Future<Output = ()>>>>::{closure#0}}, std::option::Option<()>>::{closure#0}}, (std::boxed::Box<tokio::runtime::scheduler::current_thread::Core>, std::option::Option<()>)>`
   --> /home/byt/Documents/tokio/tokio/src/runtime/context/scoped.rs:40:9
    |
40  |         f()
    |         ^^^
note: inside closure
   --> /home/byt/Documents/tokio/tokio/src/runtime/context.rs:180:26
    |
180 |         CONTEXT.with(|c| c.scheduler.set(v, f))
    |                          ^^^^^^^^^^^^^^^^^^^^^
    = note: inside `std::thread::LocalKey::<tokio::runtime::context::Context>::try_with::<{closure@tokio::runtime::context::set_scheduler<(std::boxed::Box<tokio::runtime::scheduler::current_thread::Core>, std::option::Option<()>), {closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::enter<{closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::block_on<std::pin::Pin<&mut std::pin::Pin<&mut dyn std::future::Future<Output = ()>>>>::{closure#0}}, std::option::Option<()>>::{closure#0}}>::{closure#0}}, (std::boxed::Box<tokio::runtime::scheduler::current_thread::Core>, std::option::Option<()>)>` at /home/byt/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:283:12: 283:27
    = note: inside `std::thread::LocalKey::<tokio::runtime::context::Context>::with::<{closure@tokio::runtime::context::set_scheduler<(std::boxed::Box<tokio::runtime::scheduler::current_thread::Core>, std::option::Option<()>), {closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::enter<{closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::block_on<std::pin::Pin<&mut std::pin::Pin<&mut dyn std::future::Future<Output = ()>>>>::{closure#0}}, std::option::Option<()>>::{closure#0}}>::{closure#0}}, (std::boxed::Box<tokio::runtime::scheduler::current_thread::Core>, std::option::Option<()>)>` at /home/byt/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:260:9: 260:25
note: inside `tokio::runtime::context::set_scheduler::<(std::boxed::Box<tokio::runtime::scheduler::current_thread::Core>, std::option::Option<()>), {closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::enter<{closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::block_on<std::pin::Pin<&mut std::pin::Pin<&mut dyn std::future::Future<Output = ()>>>>::{closure#0}}, std::option::Option<()>>::{closure#0}}>`
   --> /home/byt/Documents/tokio/tokio/src/runtime/context.rs:180:9
    |
180 |         CONTEXT.with(|c| c.scheduler.set(v, f))
    |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside `tokio::runtime::scheduler::current_thread::CoreGuard::<'_>::enter::<{closure@tokio::runtime::scheduler::current_thread::CoreGuard<'_>::block_on<std::pin::Pin<&mut std::pin::Pin<&mut dyn std::future::Future<Output = ()>>>>::{closure#0}}, std::option::Option<()>>`
   --> /home/byt/Documents/tokio/tokio/src/runtime/scheduler/current_thread/mod.rs:774:27
    |
774 |         let (core, ret) = context::set_scheduler(&self.context, || f(core, context));
    |                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside closure
   --> /home/byt/Documents/tokio/tokio/src/runtime/scheduler/current_thread/mod.rs:191:28
    |
191 |                     return core.block_on(future);
    |                            ^^^^^^^^^^^^^^^^^^^^^
note: inside `pending_first`
   --> tokio-stream/tests/stream_chain.rs:84:5
    |
84  |     assert_eq!(None, assert_ready!(stream.poll_next()));
    |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside closure
   --> tokio-stream/tests/stream_chain.rs:52:25
    |
51  | #[tokio::test]
    | -------------- in this procedural macro expansion
52  | async fn pending_first() {
    |                         ^
    = note: this error originates in the attribute macro `::core::prelude::v1::test` which comes from the expansion of the attribute macro `tokio::test` (in Nightly builds, run with -Z macro-backtrace for more info)

note: some details are omitted, run with `MIRIFLAGS=-Zmiri-backtrace=full` for a verbose backtrace

error: aborting due to 1 previous error

error: test failed, to rerun pass `-p tokio-stream --test stream_chain`

Caused by:
  process didn't exit successfully: `/home/byt/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/bin/cargo-miri runner /home/byt/Documents/tokio/target/miri/x86_64-unknown-linux-gnu/debug/deps/stream_chain-fe570dc7ff6e1d67` (exit status: 1)
note: test exited abnormally; to see the full output pass --nocapture to the harness.

cc @Darksonn

[Darksonn]

This seems like another case of #6744.

tokio/tokio-stream/src/stream_ext/fuse.rs

Lines 8 to 13 in 542197c

	/// Stream returned by [`fuse()`][super::StreamExt::fuse].
	#[derive(Debug)]
	pub struct Fuse<T> {
	#[pin]
	stream: Option<T>,
	}

with the extra complication that an immutable reference is used to read the Option discriminant via size_hint.

[taiki-e]

It should be fixed in the latest nightly (rust-lang/rust#129313).

[tiif]

I tried again with the latest nightly

rustc 1.83.0-nightly (7042c269c 2024-09-23)
binary: rustc
commit-hash: 7042c269c166191cd5d8daf0409890903df7af57
commit-date: 2024-09-23
host: x86_64-unknown-linux-gnu
release: 1.83.0-nightly
LLVM version: 19.1.0

but miri still produced the same error.