sources/channel: Send Closed for SendSender, not only Sender; fix possible race

src/sources/channel.rs

@@ -10,7 +10,8 @@
 
 use std::cmp;
 use std::fmt;
-use std::sync::mpsc;
+use std::ops;
+use std::sync::{mpsc, Arc};
 
 use crate::{EventSource, Poll, PostAction, Readiness, Token, TokenFactory};
 
@@ -30,21 +31,40 @@ pub enum Event<T> {
     Closed,
 }
 
+#[derive(Debug)]
+struct PingOnDrop(Ping);
+
+impl ops::Deref for PingOnDrop {
+    type Target = Ping;
+
+    fn deref(&self) -> &Ping {
+        &self.0
+    }
+}
+
+impl Drop for PingOnDrop {
+    fn drop(&mut self) {
+        self.0.ping();
+    }
+}
+
 /// The sender end of a channel
 ///
 /// It can be cloned and sent accross threads (if `T` is).
 #[derive(Debug)]
 pub struct Sender<T> {
     sender: mpsc::Sender<T>,
-    ping: Ping,
+    // Dropped after `sender` so receiver is guaranteed to get `Disconnected`
+    // after ping.
+    ping: PingOnDrop,
 }
 
 impl<T> Clone for Sender<T> {
     #[cfg_attr(feature = "nightly_coverage", coverage(off))]
     fn clone(&self) -> Sender<T> {
         Sender {
             sender: self.sender.clone(),
-            ping: self.ping.clone(),
+            ping: PingOnDrop(self.ping.clone()),
         }
     }
 }
@@ -59,20 +79,15 @@ impl<T> Sender<T> {
     }
 }
 
-impl<T> Drop for Sender<T> {
-    fn drop(&mut self) {
-        // ping on drop, to notify about channel closure
-        self.ping.ping();
-    }
-}
-
 /// The sender end of a synchronous channel
 ///
 /// It can be cloned and sent accross threads (if `T` is).
 #[derive(Debug)]
 pub struct SyncSender<T> {
     sender: mpsc::SyncSender<T>,
-    ping: Ping,
+    // Dropped after `sender` so receiver is guaranteed to get `Disconnected`
+    // after ping.
+    ping: Arc<PingOnDrop>,
 }
 
 impl<T> Clone for SyncSender<T> {
@@ -164,7 +179,7 @@ pub fn channel<T>() -> (Sender<T>, Channel<T>) {
     (
         Sender {
             sender,
-            ping: ping.clone(),
+            ping: PingOnDrop(ping.clone()),
         },
         Channel {
             receiver,
@@ -182,7 +197,7 @@ pub fn sync_channel<T>(bound: usize) -> (SyncSender<T>, Channel<T>) {
     (
         SyncSender {
             sender,
-            ping: ping.clone(),
+            ping: Arc::new(PingOnDrop(ping.clone())),
         },
         Channel {
             receiver,