Mark everything in tokio::sync::mpsc as UnwindSafe

Since these types are safe in the face of panics and cannot cause data
corruption, they're `UnwindSafe` irrespective of the channel's element
type.

CC #4657.

tokio/src/sync/mpsc/chan.rs

@@ -9,6 +9,7 @@ use crate::sync::notify::Notify;
 use crate::util::cacheline::CachePadded;
 
 use std::fmt;
+use std::panic;
 use std::process;
 use std::sync::atomic::Ordering::{AcqRel, Acquire, Relaxed, Release};
 use std::task::Poll::{Pending, Ready};
@@ -108,6 +109,8 @@ impl<T> fmt::Debug for RxFields<T> {
 
 unsafe impl<T: Send, S: Send> Send for Chan<T, S> {}
 unsafe impl<T: Send, S: Sync> Sync for Chan<T, S> {}
+impl<T, S> panic::RefUnwindSafe for Chan<T, S> {}
+impl<T, S> panic::UnwindSafe for Chan<T, S> {}
 
 pub(crate) fn channel<T, S: Semaphore>(semaphore: S) -> (Tx<T, S>, Rx<T, S>) {
     let (tx, rx) = list::channel();

tokio/tests/sync_mpsc.rs

@@ -12,6 +12,7 @@ use tokio::test as maybe_tokio_test;
 
 use std::fmt;
 use std::sync::Arc;
+use std::panic;
 use tokio::sync::mpsc;
 use tokio::sync::mpsc::error::{TryRecvError, TrySendError};
 use tokio_test::*;
@@ -23,8 +24,36 @@ mod support {
 
 #[allow(unused)]
 trait AssertSend: Send {}
-impl AssertSend for mpsc::Sender<i32> {}
 impl AssertSend for mpsc::Receiver<i32> {}
+impl AssertSend for mpsc::Sender<i32> {}
+impl AssertSend for mpsc::UnboundedReceiver<i32> {}
+impl AssertSend for mpsc::UnboundedSender<i32> {}
+impl AssertSend for mpsc::WeakSender<i32> {}
+impl AssertSend for mpsc::WeakUnboundedSender<i32> {}
+
+#[allow(unused)]
+trait AssertRefUnwindSafe: panic::RefUnwindSafe {}
+impl<T> AssertRefUnwindSafe for mpsc::OwnedPermit<T> {}
+impl<'a, T> AssertRefUnwindSafe for mpsc::Permit<'a, T> {}
+impl<'a, T> AssertRefUnwindSafe for mpsc::PermitIterator<'a, T> {}
+impl<T> AssertRefUnwindSafe for mpsc::Receiver<T> {}
+impl<T> AssertRefUnwindSafe for mpsc::Sender<T> {}
+impl<T> AssertRefUnwindSafe for mpsc::UnboundedReceiver<T> {}
+impl<T> AssertRefUnwindSafe for mpsc::UnboundedSender<T> {}
+impl<T> AssertRefUnwindSafe for mpsc::WeakSender<T> {}
+impl<T> AssertRefUnwindSafe for mpsc::WeakUnboundedSender<T> {}
+
+#[allow(unused)]
+trait AssertUnwindSafe: panic::UnwindSafe {}
+impl<T> AssertUnwindSafe for mpsc::OwnedPermit<T> {}
+impl<'a, T> AssertUnwindSafe for mpsc::Permit<'a, T> {}
+impl<'a, T> AssertUnwindSafe for mpsc::PermitIterator<'a, T> {}
+impl<T> AssertUnwindSafe for mpsc::Receiver<T> {}
+impl<T> AssertUnwindSafe for mpsc::Sender<T> {}
+impl<T> AssertUnwindSafe for mpsc::UnboundedReceiver<T> {}
+impl<T> AssertUnwindSafe for mpsc::UnboundedSender<T> {}
+impl<T> AssertUnwindSafe for mpsc::WeakSender<T> {}
+impl<T> AssertUnwindSafe for mpsc::WeakUnboundedSender<T> {}
 
 #[maybe_tokio_test]
 async fn send_recv_with_buffer() {