Quadratic slowdown in latency of tokio::sync::broadcast::Sender::send as the number of Receivers increases, when each receiver is running in its own single-threaded runtime

[glittershark]

Version

❯ cargo tree | grep tokio
└── tokio v1.30.0
    └── tokio-macros v2.1.0 (proc-macro)

Platform

❯ uname -a
Linux ogopogo 6.1.41 #1-NixOS SMP PREEMPT_DYNAMIC Mon Jul 24 16:55:35 UTC 2023 x86_64 GNU/Linux

Description

In an application that launches many threads (up to hundreds) each of which runs a select loop which polls on tokio::sync::broadcast::Receiver::recv(), calls to the send half of that channel sporadically get quadratically slower, up to minutes, as the number of receivers starts to increase past around 40. This behavior is worse (by about half an order of magnitude) if running in release mode. To demonstrate, I've written the following repro, which outputs a CSV :

use std::thread;
use std::time::Instant;
use std::{sync::Arc, time::Duration};

use tokio::select;
use tokio::sync::broadcast;
use tokio::time::sleep;

const BUFFER_SIZE: usize = 64;

async fn recv_loop(tx: Arc<broadcast::Sender<usize>>) {
    let mut rx = tx.subscribe();
    loop {
        select! {
            _ = sleep(Duration::from_millis(100)) => {}
            res = rx.recv() => {
                if matches!(res, Err(broadcast::error::RecvError::Closed)) {
                    break;
                }
            }
        }
    }
}

fn run_test(num_subscribers: usize) -> Duration {
    let (tx, _) = broadcast::channel(BUFFER_SIZE);
    let tx = Arc::new(tx);

    for _ in 0..num_subscribers {
        let tx = tx.clone();

        thread::spawn(move || {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .max_blocking_threads(1)
                .build()
                .unwrap();

            runtime.block_on(recv_loop(tx));
        });
    }

    let start = Instant::now();
    let _ = tx.send(0);
    start.elapsed()
}

fn main() {
    for num_subscribers in 30..120 {
        let dur = run_test(num_subscribers);
        println!("{num_subscribers},{}", dur.as_nanos());
    }
}

And plotted the results in both debug and release mode when run on my machine:

[glittershark]

some extra info: it looks like whatever send is doing that's taking so long, it's doing it after the receivers all receive the message - somewhere in this loop

[Darksonn]

It's probably that receivers keep being re-added during that loop. We can most likely fix it by limiting the number of wakers woken to whatever the total number was before it started iterating.

Thanks for reporting this.

[glittershark]

@Darksonn something like this? This does fix the perf issue, which is great!

diff --git a/tokio/src/sync/broadcast.rs b/tokio/src/sync/broadcast.rs
index 246ec212..ce503996 100644
--- a/tokio/src/sync/broadcast.rs
+++ b/tokio/src/sync/broadcast.rs
@@ -820,6 +820,8 @@ fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
 impl<T> Shared<T> {
     fn notify_rx<'a, 'b: 'a>(&'b self, mut tail: MutexGuard<'a, Tail>) {
         let mut wakers = WakeList::new();
+        let active_receivers = tail.rx_cnt;
+        let mut woken = 0usize;
         'outer: loop {
             while wakers.can_push() {
                 match tail.waiters.pop_back() {
@@ -832,6 +834,7 @@ impl<T> Shared<T> {
 
                         if let Some(waker) = waiter.waker.take() {
                             wakers.push(waker);
+                            woken += 1;
                         }
                     }
                     None => {
@@ -852,6 +855,10 @@ impl<T> Shared<T> {
 
             // Acquire the lock again.
             tail = self.tail.lock();
+
+            if woken >= active_receivers {
+                break;
+            }
         }
 
         // Release the lock before waking.

[glittershark]

I was worried that wouldn't fix the issue if there was a receiver that wasn't trying to recv(), but it looks like the perf issue is fixed even with this patch to my repro above:

diff --git a/src/main.rs b/src/main.rs
index d35832b..fbe9e4b 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -23,7 +23,7 @@ async fn recv_loop(tx: Arc<broadcast::Sender<usize>>) {
 }
 
 fn run_test(num_subscribers: usize) -> Duration {
-    let (tx, _) = broadcast::channel(BUFFER_SIZE);
+    let (tx, rx_that_never_recvs) = broadcast::channel(BUFFER_SIZE);
     let tx = Arc::new(tx);
 
     for _ in 0..num_subscribers {
@@ -40,6 +40,8 @@ fn run_test(num_subscribers: usize) -> Duration {
         });
     }
 
+    drop(rx_that_never_recvs);
+
     let start = Instant::now();
     let _ = tx.send(0);
     start.elapsed()