From db2e652ddced96408ddb8c00bb08584454541006 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Hermann=20Bj=C3=B6rgvin=20Haraldsson?= Date: Sat, 14 Nov 2026 05:36:43 +0000 Subject: [PATCH] fix(corebluetooth): Stop the CoreBluetooth thread once its adapter and peripherals are dropped Each Adapter started a thread that looped forever, so every Manager::adapters() call leaked a thread and its CBCentralManager (#500). The message channel can't signal the end, because the adapter's event task and peripheral map hold senders for as long as the thread runs. The adapter, every peripheral it hands out and the event and notification streams from them now hold a shared KeepAlive (the sender of a oneshot channel). The thread stops when the last of them is dropped. The adapter's internal copies of peripherals don't hold one, so they don't keep the thread alive. --- src/corebluetooth/adapter.rs | 65 ++++++++++++++++++++++++++++++--- src/corebluetooth/internal.rs | 17 ++++++++- src/corebluetooth/peripheral.rs | 19 +++++++++- 3 files changed, 93 insertions(+), 8 deletions(-) diff --git a/src/corebluetooth/adapter.rs b/src/corebluetooth/adapter.rs index b0eb21b2..afb9ffbb 100644 --- a/src/corebluetooth/adapter.rs +++ b/src/corebluetooth/adapter.rs @@ -11,6 +11,7 @@ use crate::common::adapter_manager::AdapterManager; use crate::{Error, Result}; use async_trait::async_trait; use futures::channel::mpsc::{self, Sender}; +use futures::channel::oneshot; use futures::sink::SinkExt; use futures::stream::{Stream, StreamExt}; use log::*; @@ -18,6 +19,7 @@ use objc2_core_bluetooth::CBManagerState; use std::collections::HashMap; use std::pin::Pin; use std::sync::Arc; +use std::task::{Context, Poll}; use tokio::task; /// Implementation of [api::Central](crate::api::Central). @@ -25,6 +27,43 @@ use tokio::task; pub struct Adapter { manager: Arc>, sender: Sender, + keep_alive: KeepAlive, +} + +/// Keeps an adapter's CoreBluetooth thread running. The adapter, every +/// peripheral it hands out and the event and notification streams from them +/// hold one; the thread stops when the last is dropped. +/// The adapter's own bookkeeping (its event task and peripheral map) doesn't +/// hold one, since it lives as long as the thread does. +pub(crate) type KeepAlive = Arc>; + +/// Wraps a stream handed to the application so that it holds a `KeepAlive`, +/// and keeps delivering items after the adapter it came from is dropped. +pub(crate) fn keep_alive_stream( + stream: Pin + Send>>, + keep_alive: Option, +) -> Pin + Send>> { + Box::pin(KeepAliveStream { + stream, + _keep_alive: keep_alive, + }) +} + +struct KeepAliveStream { + stream: Pin + Send>>, + _keep_alive: Option, +} + +impl Stream for KeepAliveStream { + type Item = T; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.stream.as_mut().poll_next(cx) + } + + fn size_hint(&self) -> (usize, Option) { + self.stream.size_hint() + } } fn get_central_state(state: CBManagerState) -> CentralState { @@ -38,7 +77,8 @@ fn get_central_state(state: CBManagerState) -> CentralState { impl Adapter { pub(crate) async fn new() -> Result { let (sender, mut receiver) = mpsc::channel(256); - let adapter_sender = run_corebluetooth_thread(sender)?; + let (shutdown_sender, shutdown_receiver) = oneshot::channel(); + let adapter_sender = run_corebluetooth_thread(sender, shutdown_receiver)?; // Since init currently blocked until the state update, we know the // receiver is dropped after that. We can pick it up here and make it // part of our event loop to update our peripherals. @@ -153,6 +193,7 @@ impl Adapter { Ok(Adapter { manager, sender: adapter_sender, + keep_alive: Arc::new(shutdown_sender), }) } } @@ -162,7 +203,10 @@ impl Central for Adapter { type Peripheral = Peripheral; async fn events(&self) -> Result + Send>>> { - Ok(self.manager.event_stream()) + Ok(keep_alive_stream( + self.manager.event_stream(), + Some(self.keep_alive.clone()), + )) } async fn start_scan(&self, filter: ScanFilter) -> Result<()> { @@ -182,7 +226,12 @@ impl Central for Adapter { } async fn peripherals(&self) -> Result> { - Ok(self.manager.peripherals()) + Ok(self + .manager + .peripherals() + .into_iter() + .map(|p| p.with_keep_alive(&self.keep_alive)) + .collect()) } async fn retrieve_peripherals( @@ -206,7 +255,10 @@ impl Central for Adapter { }) .await?; match fut.await { - CoreBluetoothReply::Peripherals(peripherals) => Ok(peripherals), + CoreBluetoothReply::Peripherals(peripherals) => Ok(peripherals + .into_iter() + .map(|p| p.with_keep_alive(&self.keep_alive)) + .collect()), CoreBluetoothReply::Err(msg) => Err(Error::RuntimeError(msg)), CoreBluetoothReply::Ok => Ok(Vec::new()), _ => Err(Error::RuntimeError( @@ -216,7 +268,10 @@ impl Central for Adapter { } async fn peripheral(&self, id: &PeripheralId) -> Result { - self.manager.peripheral(id).ok_or(Error::DeviceNotFound) + self.manager + .peripheral(id) + .map(|p| p.with_keep_alive(&self.keep_alive)) + .ok_or(Error::DeviceNotFound) } async fn add_peripheral(&self, _address: &PeripheralId) -> Result { diff --git a/src/corebluetooth/internal.rs b/src/corebluetooth/internal.rs index 5c03c8c2..1db0e738 100644 --- a/src/corebluetooth/internal.rs +++ b/src/corebluetooth/internal.rs @@ -24,6 +24,8 @@ use crate::api::{ WriteType, }; use futures::channel::mpsc::{self, Receiver, Sender}; +use futures::channel::oneshot; +use futures::future::{self, Either}; use futures::select; use futures::sink::SinkExt; use futures::stream::{Fuse, StreamExt}; @@ -42,6 +44,7 @@ use std::{ fmt::{self, Debug, Formatter}, future::Future, ops::Deref, + pin::pin, thread, }; use tokio::runtime; @@ -2013,8 +2016,11 @@ async fn within_autorelease_pool(future: F) -> F::Output { future.await } +/// Starts the thread that owns the `CBCentralManager`. It runs until `shutdown` +/// completes, which happens when the adapter's last `KeepAlive` is dropped. pub fn run_corebluetooth_thread( event_sender: Sender, + mut shutdown: oneshot::Receiver<()>, ) -> Result, Error> { let authorization = unsafe { CBManager::authorization_class() }; if authorization != CBManagerAuthorization::AllowedAlways @@ -2031,9 +2037,18 @@ pub fn run_corebluetooth_thread( let runtime = runtime::Builder::new_current_thread().build().unwrap(); runtime.block_on(async move { let mut cbi = CoreBluetoothInternal::new(receiver, event_sender); + // The adapter's event task and peripheral map hold message senders + // for as long as this thread runs, so the message channel never + // closes on its own. Stop when the last `KeepAlive` is dropped. loop { - within_autorelease_pool(cbi.wait_for_message()).await; + let message = pin!(within_autorelease_pool(cbi.wait_for_message())); + // Shutdown first, so a steady stream of delegate events can't + // keep the thread running. + if let Either::Left(_) = future::select(&mut shutdown, message).await { + break; + } } + trace!("Adapter and its peripherals dropped, stopping the CoreBluetooth thread"); }) }); Ok(sender) diff --git a/src/corebluetooth/peripheral.rs b/src/corebluetooth/peripheral.rs index 6c1723c0..9e8acd34 100644 --- a/src/corebluetooth/peripheral.rs +++ b/src/corebluetooth/peripheral.rs @@ -5,6 +5,7 @@ // Licensed under the BSD 3-Clause license. See LICENSE file in the project root // for full license information. +use super::adapter::{KeepAlive, keep_alive_stream}; use super::internal::{ CoreBluetoothMessage, CoreBluetoothReply, CoreBluetoothReplyFuture, PeripheralEventInternal, }; @@ -55,6 +56,9 @@ impl Display for PeripheralId { #[derive(Clone)] pub struct Peripheral { shared: Arc, + // Set on the copies the adapter hands out, so the CoreBluetooth thread + // keeps running while the application still holds this peripheral. + keep_alive: Option, } struct Shared { @@ -194,7 +198,15 @@ impl Peripheral { } } }); - Self { shared } + Self { + shared, + keep_alive: None, + } + } + + pub(super) fn with_keep_alive(mut self, keep_alive: &KeepAlive) -> Self { + self.keep_alive = Some(keep_alive.clone()); + self } pub(super) fn update_name( @@ -599,7 +611,10 @@ impl api::Peripheral for Peripheral { async fn notifications(&self) -> Result + Send>>> { let receiver = self.shared.notifications_channel.subscribe(); - Ok(notifications_stream_from_broadcast_receiver(receiver)) + Ok(keep_alive_stream( + notifications_stream_from_broadcast_receiver(receiver), + self.keep_alive.clone(), + )) } async fn write_descriptor(&self, descriptor: &Descriptor, data: &[u8]) -> Result<()> {