unsoundness if cancelled future was in readiness queue

[StoicDeveloper]

This crate is unsound. If a future is cancelled while its task is in the readiness queue, then the task will be dropped, and then can be dereferenced when the MappedFutures is next polled.

Solutions may include:

• add to Task Option and Option fields that can be used to set remove an item from the queue if it gets cancelled
• drop the future on cancellation, but don't drop the task. Instead check during ReadyToRunQueue dequeue() whether the future was dropped. If so, dequeue again and drop the current task.

[StoicDeveloper]

Here is the miri output:

running 3 tests
test stream::mapped_futures::tests::map_futures ... error: Undefined Behavior: out-of-bounds pointer arithmetic: alloc76643 has been freed, so this pointer is dangling
    --> futures-util/src/stream/futures_unordered_internal/ready_to_run_queue.rs:52:28
     |
52   |             let mut next = (*tail).next_ready_to_run.load(Acquire);
     |                            ^^^^^^^^^^^^^^^^^^^^^^^^^ out-of-bounds pointer arithmetic: alloc76643 has been freed, so this pointer is dangling
     |
     = help: this indicates a bug in the program: it performed an invalid operation, and caused Undefined Behavior
     = help: see https://doc.rust-lang.org/nightly/reference/behavior-considered-undefined.html for further information
help: alloc76643 was allocated here:
    --> futures-util/src/stream/futures_unordered_internal/mod.rs:173:20
     |
173  |           let task = Arc::new(Task {
     |  ____________________^
174  | |             future: UnsafeCell::new(Some(future)),
175  | |             key: UnsafeCell::new(Some(key)),
176  | |             next_all: AtomicPtr::new(self.pending_next_all()),
...    |
182  | |             woken: AtomicBool::new(false),
183  | |         });
     | |__________^
help: alloc76643 was deallocated here:
    --> futures-util/src/stream/mapped_futures/mod.rs:175:9
     |
175  |         }
     |         ^
     = note: BACKTRACE (of the first span) on thread `stream::mapped_`:
     = note: inside `stream::futures_unordered_internal::ready_to_run_queue::ReadyToRunQueue::<u32, futures_timer::Delay>::dequeue` at futures-util/src/stream/futures_unordered_internal/ready_to_run_queue.rs:52:28: 52:53
note: inside `<stream::futures_unordered_internal::FuturesUnorderedInternal<u32, futures_timer::Delay, stream::mapped_futures::TaskSet<u32, futures_timer::Delay>> as futures_core::Stream>::poll_next`
    --> futures-util/src/stream/futures_unordered_internal/mod.rs:416:39
     |
416  |             let task = match unsafe { self.ready_to_run_queue.dequeue() } {
     |                                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside `<stream::mapped_futures::MappedFutures<u32, futures_timer::Delay> as futures_core::Stream>::poll_next`
    --> futures-util/src/stream/mapped_futures/mod.rs:308:33
     |
308  |         match std::task::ready!(Pin::new(&mut self.inner).poll_next(cx)) {
     |                                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside `<stream::mapped_futures::MappedFutures<u32, futures_timer::Delay> as stream::stream::StreamExt>::poll_next_unpin`
    --> futures-util/src/stream/stream/mod.rs:1758:9
     |
1758 |         Pin::new(self).poll_next(cx)
     |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
note: inside `<stream::stream::next::Next<'_, stream::mapped_futures::MappedFutures<u32, futures_timer::Delay>> as futures_core::Future>::poll`
    --> futures-util/src/stream/stream/next.rs:32:9
     |
32   |         self.stream.poll_next_unpin(cx)
     |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
     = note: inside closure at /home/user/Projects/futures-rs/futures-executor/src/local_pool.rs:316:23: 316:42
     = note: inside closure at /home/user/Projects/futures-rs/futures-executor/src/local_pool.rs:90:37: 90:47
     = note: inside `std::thread::LocalKey::<alloc::sync::Arc<futures_executor::local_pool::ThreadNotify>>::try_with::<{closure@futures_executor::local_pool::run_executor<core::option::Option<(u32, ())>, {closure@futures::executor::block_on<stream::stream::next::Next<'_, stream::mapped_futures::MappedFutures<u32, futures_timer::Delay>>>::{closure#0}}>::{closure#0}}, core::option::Option<(u32, ())>>` at /home/user/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:284:16: 284:31
     = note: inside `std::thread::LocalKey::<alloc::sync::Arc<futures_executor::local_pool::ThreadNotify>>::with::<{closure@futures_executor::local_pool::run_executor<core::option::Option<(u32, ())>, {closure@futures::executor::block_on<stream::stream::next::Next<'_, stream::mapped_futures::MappedFutures<u32, futures_timer::Delay>>>::{closure#0}}>::{closure#0}}, core::option::Option<(u32, ())>>` at /home/user/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:260:9: 260:25
     = note: inside `futures_executor::local_pool::run_executor::<core::option::Option<(u32, ())>, {closure@futures::executor::block_on<stream::stream::next::Next<'_, stream::mapped_futures::MappedFutures<u32, futures_timer::Delay>>>::{closure#0}}>` at /home/user/Projects/futures-rs/futures-executor/src/local_pool.rs:86:5: 102:7
     = note: inside `futures::executor::block_on::<stream::stream::next::Next<'_, stream::mapped_futures::MappedFutures<u32, futures_timer::Delay>>>` at /home/user/Projects/futures-rs/futures-executor/src/local_pool.rs:316:5: 316:43
note: inside `stream::mapped_futures::tests::map_futures`
    --> futures-util/src/stream/mapped_futures/mod.rs:431:20
     |
431  |         assert_eq!(block_on(futures.next()).unwrap().0, 4);
     |                    ^^^^^^^^^^^^^^^^^^^^^^^^
note: inside closure
    --> futures-util/src/stream/mapped_futures/mod.rs:421:21
     |
420  |     #[test]
     |     ------- in this procedural macro expansion
421  |     fn map_futures() {
     |                     ^
     = note: this error originates in the attribute macro `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; 10 warnings emitted

error: test failed, to rerun pass `--lib`

Caused by:
  process didn't exit successfully: `/home/user/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/bin/cargo-miri runner /home/user/Projects/futures-rs/target/miri/x86_64-unknown-linux-gnu/debug/deps/futures_util-ef6660b6f3390346 mapped_futures` (exit status: 1)
note: test exited abnormally; to see the full output pass --nocapture to the harness.

[StoicDeveloper]

I've determined that the unsoundness is due to diverergence between the PR branch in futures-rs, and does not exist in this crate. It was cause by failing to call realease_task() when removing a future. This crate is therefore sound, as far as I (or Miri) can tell.