experiment

examples/Cargo.toml

@@ -95,3 +95,6 @@ path = "named-pipe-multi-client.rs"
 [[example]]
 name = "dump"
 path = "dump.rs"
+[[example]]
+name = "flamegraph_mpsc"
+path = "flamegraph_mpsc.rs"

examples/flamegraph_mpsc.rs

@@ -0,0 +1,43 @@
+use std::hint::black_box;
+use tokio::runtime::Runtime;
+use tokio::sync::mpsc::{reclaim_called_count, reuse_failed_count};
+use tokio::task::JoinSet;
+
+fn multi_rt() -> Runtime {
+    tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap()
+}
+
+async fn run() {
+    let (tx, mut rx) = tokio::sync::mpsc::channel::<usize>(100_000);
+
+    let mut set = JoinSet::new();
+    for _ in 0..10 {
+        let tx = tx.clone();
+        set.spawn(async move {
+            for i in 0..100_ {
+                tx.send(i).await.unwrap();
+            }
+        });
+    }
+
+    black_box(for _ in 0..100_0 {
+        rx.recv().await.unwrap();
+    });
+    while let Some(res) = set.join_next().await {
+        let _ = res;
+    }
+}
+fn main() {
+    let rt = multi_rt();
+    rt.block_on(async {
+        tokio::spawn(async {
+            run().await;
+        })
+        .await
+        .unwrap();
+    });
+    println!("{} / {}", reuse_failed_count(), reclaim_called_count())
+}

tokio/src/sync/mpsc/bounded.rs

@@ -11,6 +11,8 @@ cfg_time! {
 use std::fmt;
 use std::task::{Context, Poll};
 
+use super::list::{RECLAIM_CALLED_COUNT, REUSE_FAILED_COUNT};
+
 /// Sends values to the associated `Receiver`.
 ///
 /// Instances are created by the [`channel`] function.
@@ -743,6 +745,12 @@ impl<T> fmt::Debug for Receiver<T> {
 
 impl<T> Unpin for Receiver<T> {}
 
+pub fn reclaim_called_count() -> usize {
+    RECLAIM_CALLED_COUNT.load(std::sync::atomic::Ordering::Acquire)
+}
+pub fn reuse_failed_count() -> usize {
+    REUSE_FAILED_COUNT.load(std::sync::atomic::Ordering::Acquire)
+}
 impl<T> Sender<T> {
     pub(crate) fn new(chan: chan::Tx<T, Semaphore>) -> Sender<T> {
         Sender { chan }

tokio/src/sync/mpsc/list.rs

@@ -179,6 +179,7 @@ impl<T> Tx<T> {
     }
 
     pub(crate) unsafe fn reclaim_block(&self, mut block: NonNull<Block<T>>) {
+        RECLAIM_CALLED_COUNT.fetch_add(1, Relaxed);
         // The block has been removed from the linked list and ownership
         // is reclaimed.
         //
@@ -215,6 +216,7 @@ impl<T> Tx<T> {
         }
 
         if !reused {
+            REUSE_FAILED_COUNT.fetch_add(1, Relaxed);
             let _ = Box::from_raw(block.as_ptr());
         }
     }
@@ -238,6 +240,9 @@ impl<T> fmt::Debug for Tx<T> {
     }
 }
 
+pub(crate) static RECLAIM_CALLED_COUNT: AtomicUsize = AtomicUsize::new(0);
+pub(crate) static REUSE_FAILED_COUNT: AtomicUsize = AtomicUsize::new(0);
+
 impl<T> Rx<T> {
     pub(crate) fn is_empty(&self, tx: &Tx<T>) -> bool {
         let block = unsafe { self.head.as_ref() };

tokio/src/sync/mpsc/mod.rs

@@ -116,7 +116,8 @@ pub(super) mod block;
 
 mod bounded;
 pub use self::bounded::{
-    channel, OwnedPermit, Permit, PermitIterator, Receiver, Sender, WeakSender,
+    channel, reclaim_called_count, reuse_failed_count, OwnedPermit, Permit, PermitIterator,
+    Receiver, Sender, WeakSender,
 };
 
 mod chan;