From 2887932340756716d0440cb79498941a2d1d8b41 Mon Sep 17 00:00:00 2001
From: Updawg <10038272+updawg@users.noreply.github.com>
Date: Mon, 1 Jun 2026 19:32:44 -0400
Subject: [PATCH] Remove the per-Bus unparking thread

Bus::new spawns a thread whose only job is to call unpark() so that
broadcast_inner doesn't have to. Its JoinHandle is dropped, so the thread
lives for the whole life of the Bus and can never be joined.

Unpark inline in broadcast_inner instead and drop the thread. That puts the
unpark back on the broadcast path (only when there's a blocked receiver to
wake), but a Bus no longer holds a background thread it can't clean up.
---
 src/lib.rs | 20 +-------------------
 1 file changed, 1 insertion(+), 19 deletions(-)

diff --git a/src/lib.rs b/src/lib.rs
index 53a0f47..fa3b10f 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -294,9 +294,6 @@ pub struct Bus<T> {
         mpsc::Receiver<(thread::Thread, usize)>,
     ),
 
-    // channel used to communicate to unparker that a given thread should be woken up
-    unpark: mpsc::Sender<thread::Thread>,
-
     // cache used to keep track of threads waiting for next write.
     // this is only here to avoid allocating one on every broadcast()
     cache: Vec<(thread::Thread, usize)>,
@@ -310,7 +307,6 @@ impl<T> fmt::Debug for Bus<T> {
             .field("rleft", &self.rleft)
             .field("leaving", &self.leaving)
             .field("waiting", &self.waiting)
-            .field("unpark", &self.unpark)
             .field("cache", &self.cache)
             .finish()
     }
@@ -340,26 +336,12 @@ impl<T> Bus<T> {
             let _ = time::Instant::now().elapsed();
         }
 
-        // we run a separate thread responsible for unparking
-        // so we don't have to wait for unpark() to return in broadcast_inner
-        // sending on a channel without contention is cheap, unparking is not
-        let (unpark_tx, unpark_rx) = mpsc::unbounded::<thread::Thread>();
-        let _ = thread::Builder::new()
-            .name("bus_unparking".to_owned())
-            .spawn(move || {
-                for t in unpark_rx.iter() {
-                    t.unpark();
-                }
-            });
-
         Bus {
             state: inner,
             readers: 0,
             rleft: iter::repeat(0).take(len).collect(),
             leaving: mpsc::unbounded(),
             waiting: mpsc::unbounded(),
-            unpark: unpark_tx,
-
             cache: Vec::new(),
         }
     }
@@ -479,7 +461,7 @@ impl<T> Bus<T> {
             if at == tail {
                 self.cache.push((t, at))
             } else {
-                self.unpark.send(t).unwrap();
+                t.unpark();
             }
         }
         for w in self.cache.drain(..) {