Fix double drop in StreamExt::cycle

[taiki-e]

StreamExt::cycle drop source field without updating field if source field's destructor panicked.

async-std/src/stream/stream/cycle.rs

Lines 47 to 48 in 11196c8

	ManuallyDrop::drop(&mut this.source);
	this.source = ManuallyDrop::new(this.orig.clone());

This is unsound because it causes double drop if the user catches panic in the scope where the Cycle is alive.

Repro:

#[test]
#[should_panic(expected = "second")]
fn test() {
    use async_std::stream::*;
    use futures::task::{noop_waker, Context, Poll};
    use std::{panic, pin::Pin, thread};

    #[derive(Clone, Default)]
    struct S {
        dropped: bool,
    }

    impl Stream for S {
        type Item = ();

        fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            Poll::Ready(None)
        }
    }

    impl Drop for S {
        fn drop(&mut self) {
            if !thread::panicking() {
                if std::mem::replace(&mut self.dropped, true) {
                    panic!("second")
                } else {
                    panic!("first")
                }
            }
        }
    }

    let mut s = Box::pin(S::default().cycle());
    let w = noop_waker();
    let cx = &mut Context::from_waker(&w);
    let _ = panic::catch_unwind(panic::AssertUnwindSafe(|| {
        let _ = s.as_mut().poll_next(cx); // <-- dropped the first clone of `s`
    }));
    // <-- dropped the first clone of`s` again since `source` field wasn't updated
}

[yoshuawuyts]

Oh, thanks for catching this!