diff --git a/tokio/src/sync/mpsc/bounded.rs b/tokio/src/sync/mpsc/bounded.rs
index 2d2006d5883..542eae2782b 100644
--- a/tokio/src/sync/mpsc/bounded.rs
+++ b/tokio/src/sync/mpsc/bounded.rs
@@ -320,6 +320,41 @@ impl<T> Sender<T> {
         }
     }
 
+    /// Completes when the receiver has dropped.
+    ///
+    /// This allows the producers to get notified when interest in the produced
+    /// values is canceled and immediately stop doing work.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::sync::mpsc;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let (mut tx1, rx) = mpsc::channel::<()>(1);
+    ///     let mut tx2 = tx1.clone();
+    ///     let mut tx3 = tx1.clone();
+    ///     let mut tx4 = tx1.clone();
+    ///     let mut tx5 = tx1.clone();
+    ///     tokio::spawn(async move {
+    ///         drop(rx);
+    ///     });
+    ///
+    ///     futures::join!(
+    ///         tx1.closed(),
+    ///         tx2.closed(),
+    ///         tx3.closed(),
+    ///         tx4.closed(),
+    ///         tx5.closed()
+    ///     );
+    ////     println!("Receiver dropped");
+    /// }
+    /// ```
+    pub async fn closed(&mut self) {
+        self.chan.closed().await
+    }
+
     /// Attempts to immediately send a message on this `Sender`
     ///
     /// This method differs from [`send`] by returning immediately if the channel's
diff --git a/tokio/src/sync/mpsc/chan.rs b/tokio/src/sync/mpsc/chan.rs
index 2d3f014996a..ebe0f54058b 100644
--- a/tokio/src/sync/mpsc/chan.rs
+++ b/tokio/src/sync/mpsc/chan.rs
@@ -1,13 +1,15 @@
 use crate::loom::cell::UnsafeCell;
 use crate::loom::future::AtomicWaker;
+use crate::loom::sync::atomic::AtomicBool;
 use crate::loom::sync::atomic::AtomicUsize;
 use crate::loom::sync::Arc;
 use crate::sync::mpsc::error::TryRecvError;
 use crate::sync::mpsc::list;
+use crate::sync::Notify;
 
 use std::fmt;
 use std::process;
-use std::sync::atomic::Ordering::{AcqRel, Relaxed};
+use std::sync::atomic::Ordering::{AcqRel, Relaxed, SeqCst};
 use std::task::Poll::{Pending, Ready};
 use std::task::{Context, Poll};
 
@@ -44,6 +46,12 @@ pub(crate) trait Semaphore {
 }
 
 struct Chan<T, S> {
+    /// Notifies all tasks listening for the receiver being dropped
+    notify_rx_closed: Notify,
+
+    /// Indicates whether the receiver has been dropped
+    rx_closed: AtomicBool,
+
     /// Handle to the push half of the lock-free list.
     tx: list::Tx<T>,
 
@@ -102,6 +110,8 @@ pub(crate) fn channel<T, S: Semaphore>(semaphore: S) -> (Tx<T, S>, Rx<T, S>) {
     let (tx, rx) = list::channel();
 
     let chan = Arc::new(Chan {
+        rx_closed: AtomicBool::new(false),
+        notify_rx_closed: Notify::new(),
         tx,
         semaphore,
         rx_waker: AtomicWaker::new(),
@@ -131,6 +141,38 @@ impl<T, S> Tx<T, S> {
         self.inner.send(value);
     }
 
+    pub(crate) async fn closed(&mut self) {
+        use std::future::Future;
+        use std::pin::Pin;
+        use std::task::Poll;
+
+        // In order to avoid a race condition, we first request a notification,
+        // **then** check the current value's version. If a new version exists,
+        // the notification request is dropped. Requesting the notification
+        // requires polling the future once.
+        let notified = self.inner.notify_rx_closed.notified();
+        pin!(notified);
+
+        // Polling this for first time will register the waiter and
+        // return `Pending` or return `Ready` right away. If `Ready`
+        // is returned we are done
+        let aquired_lost_notification =
+            crate::future::poll_fn(|cx| match Pin::new(&mut notified).poll(cx) {
+                Poll::Ready(()) => Poll::Ready(true),
+                Poll::Pending => Poll::Ready(false),
+            })
+            .await;
+
+        if aquired_lost_notification {
+            return;
+        }
+
+        if self.inner.rx_closed.load(SeqCst) {
+            return;
+        }
+        notified.await;
+    }
+
     /// Wake the receive half
     pub(crate) fn wake_rx(&self) {
         self.inner.rx_waker.wake();
@@ -182,6 +224,8 @@ impl<T, S: Semaphore> Rx<T, S> {
         });
 
         self.inner.semaphore.close();
+        self.inner.rx_closed.store(true, SeqCst);
+        self.inner.notify_rx_closed.notify_waiters();
     }
 
     /// Receive the next value
diff --git a/tokio/src/sync/mpsc/unbounded.rs b/tokio/src/sync/mpsc/unbounded.rs
index 59456375297..09f71f218b4 100644
--- a/tokio/src/sync/mpsc/unbounded.rs
+++ b/tokio/src/sync/mpsc/unbounded.rs
@@ -210,4 +210,39 @@ impl<T> UnboundedSender<T> {
             }
         }
     }
+
+    /// Completes when the receiver has dropped.
+    ///
+    /// This allows the producers to get notified when interest in the produced
+    /// values is canceled and immediately stop doing work.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::sync::mpsc;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///     let (mut tx1, rx) = mpsc::unbounded_channel::<()>();
+    ///     let mut tx2 = tx1.clone();
+    ///     let mut tx3 = tx1.clone();
+    ///     let mut tx4 = tx1.clone();
+    ///     let mut tx5 = tx1.clone();
+    ///     tokio::spawn(async move {
+    ///         drop(rx);
+    ///     });
+    ///
+    ///     futures::join!(
+    ///         tx1.closed(),
+    ///         tx2.closed(),
+    ///         tx3.closed(),
+    ///         tx4.closed(),
+    ///         tx5.closed()
+    ///     );
+    ////     println!("Receiver dropped");
+    /// }
+    /// ```
+    pub async fn closed(&mut self) {
+        self.chan.closed().await
+    }
 }
diff --git a/tokio/src/sync/tests/loom_mpsc.rs b/tokio/src/sync/tests/loom_mpsc.rs
index e8db2dea4ca..330e798bcdf 100644
--- a/tokio/src/sync/tests/loom_mpsc.rs
+++ b/tokio/src/sync/tests/loom_mpsc.rs
@@ -40,6 +40,34 @@ fn closing_unbounded_tx() {
     });
 }
 
+#[test]
+fn closing_bounded_rx() {
+    loom::model(|| {
+        let (mut tx1, rx) = mpsc::channel::<()>(16);
+        let mut tx2 = tx1.clone();
+        thread::spawn(move || {
+            drop(rx);
+        });
+
+        block_on(tx1.closed());
+        block_on(tx2.closed());
+    });
+}
+
+#[test]
+fn closing_unbounded_rx() {
+    loom::model(|| {
+        let (mut tx1, rx) = mpsc::unbounded_channel::<()>();
+        let mut tx2 = tx1.clone();
+        thread::spawn(move || {
+            drop(rx);
+        });
+
+        block_on(tx1.closed());
+        block_on(tx2.closed());
+    });
+}
+
 #[test]
 fn dropping_tx() {
     loom::model(|| {