1use super::discovery::{
2 BusyOperationGuard, DiscoveryRole, DiscoveryState, DiscoveryWindow, LiveLinkGuard,
3};
4use super::*;
5
6#[derive(Debug)]
7pub struct Closed;
8
9#[derive(Clone, Copy)]
10pub(super) struct SeenPeer {
11 pub(super) kind: AddrKind,
12 pub(super) addr: BdAddr,
13 pub(super) rssi: i8,
14}
15
16#[derive(Clone, Copy)]
17pub(super) struct DialTarget {
18 pub(super) kind: AddrKind,
19 pub(super) addr: BdAddr,
20 transient_client_retries: TransientClientRetries,
21}
22
23#[derive(Clone, Copy)]
24enum TransientClientRetries {
25 EightRemaining,
26 SevenRemaining,
27 SixRemaining,
28 FiveRemaining,
29 FourRemaining,
30 ThreeRemaining,
31 TwoRemaining,
32 OneRemaining,
33 Exhausted,
34}
35
36enum TransientClientRetry {
37 Retry(DialTarget),
38 Backoff,
39}
40
41pub(super) enum TransientClientRetryOutcome {
42 Queued,
43 Exhausted,
44 QueueBusy,
45}
46
47#[derive(Clone, Copy)]
48struct RecentSighting {
49 address: [u8; 6],
50 emitted_at_ms: u64,
51}
52
53#[derive(Debug, Eq, PartialEq)]
54enum SightingAdmissionOutcome {
55 Admit,
56 Coalesce,
57}
58
59struct SightingAdmission {
60 recent: [Option<RecentSighting>; SIGHTING_DEPTH],
61}
62
63impl SightingAdmission {
64 const fn new() -> Self {
65 Self {
66 recent: [None; SIGHTING_DEPTH],
67 }
68 }
69
70 fn classify(&mut self, address: [u8; 6], now_ms: u64) -> SightingAdmissionOutcome {
71 if let Some(recent) = self
72 .recent
73 .iter_mut()
74 .flatten()
75 .find(|recent| recent.address == address)
76 {
77 if now_ms.saturating_sub(recent.emitted_at_ms) < SIGHTING_COALESCE_MS {
78 return SightingAdmissionOutcome::Coalesce;
79 }
80 recent.emitted_at_ms = now_ms;
81 return SightingAdmissionOutcome::Admit;
82 }
83 let sighting = RecentSighting {
84 address,
85 emitted_at_ms: now_ms,
86 };
87 if let Some(slot) = self.recent.iter_mut().find(|entry| entry.is_none()) {
88 *slot = Some(sighting);
89 return SightingAdmissionOutcome::Admit;
90 }
91 if let Some(slot) = self.recent.iter_mut().min_by_key(|entry| {
92 entry
93 .as_ref()
94 .map_or(u64::MAX, |recent| recent.emitted_at_ms)
95 }) {
96 *slot = Some(sighting);
97 }
98 SightingAdmissionOutcome::Admit
99 }
100}
101
102impl DialTarget {
103 fn new(kind: AddrKind, addr: BdAddr) -> Self {
104 Self {
105 kind,
106 addr,
107 transient_client_retries: TransientClientRetries::EightRemaining,
108 }
109 }
110
111 fn after_transient_client_disconnect(self) -> TransientClientRetry {
112 let transient_client_retries = match self.transient_client_retries {
113 TransientClientRetries::EightRemaining => TransientClientRetries::SevenRemaining,
114 TransientClientRetries::SevenRemaining => TransientClientRetries::SixRemaining,
115 TransientClientRetries::SixRemaining => TransientClientRetries::FiveRemaining,
116 TransientClientRetries::FiveRemaining => TransientClientRetries::FourRemaining,
117 TransientClientRetries::FourRemaining => TransientClientRetries::ThreeRemaining,
118 TransientClientRetries::ThreeRemaining => TransientClientRetries::TwoRemaining,
119 TransientClientRetries::TwoRemaining => TransientClientRetries::OneRemaining,
120 TransientClientRetries::OneRemaining => TransientClientRetries::Exhausted,
121 TransientClientRetries::Exhausted => return TransientClientRetry::Backoff,
122 };
123 TransientClientRetry::Retry(Self {
124 kind: self.kind,
125 addr: self.addr,
126 transient_client_retries,
127 })
128 }
129}
130
131pub(super) enum SlotJob {
132 Accept {
133 connection: Connection<'static, DefaultPacketPool>,
134 slot: BleSlotLease,
135 },
136 Dial {
137 connection: Connection<'static, DefaultPacketPool>,
138 slot: BleSlotLease,
139 target: DialTarget,
140 },
141}
142
143pub(super) struct SlotChannels {
144 pub(super) control_in: Channel<BridgeMutex, Control, CONTROL_QUEUE_DEPTH>,
145 pub(super) control_out: Channel<BridgeMutex, Control, CONTROL_QUEUE_DEPTH>,
146 pub(super) data_in: Channel<BridgeMutex, BleFrameLease, FRAME_QUEUE_DEPTH>,
147 pub(super) data_out: Channel<BridgeMutex, BleFrameLease, FRAME_QUEUE_DEPTH>,
148 pub(super) identity_in: Signal<BridgeMutex, BleIdentity>,
149 pub(super) identity_out: Channel<BridgeMutex, BleIdentity, 1>,
150 pub(super) data_plane: Signal<BridgeMutex, L2capPlan>,
151 pub(super) shutdown: Signal<BridgeMutex, ()>,
152 peer_addr: BlockingMutex<BridgeMutex, Cell<[u8; 6]>>,
153 peer_protocol: BlockingMutex<BridgeMutex, Cell<PeerProtocol>>,
154}
155
156impl SlotChannels {
157 const fn new() -> Self {
158 Self {
159 control_in: Channel::new(),
160 control_out: Channel::new(),
161 data_in: Channel::new(),
162 data_out: Channel::new(),
163 identity_in: Signal::new(),
164 identity_out: Channel::new(),
165 data_plane: Signal::new(),
166 shutdown: Signal::new(),
167 peer_addr: BlockingMutex::new(Cell::new([0u8; 6])),
168 peer_protocol: BlockingMutex::new(Cell::new(PeerProtocol::Native)),
169 }
170 }
171
172 pub(super) fn set_peer_addr(&self, bytes: [u8; 6]) {
173 self.peer_addr.lock(|cell| cell.set(bytes));
174 }
175
176 fn addr(&self) -> [u8; 6] {
177 self.peer_addr.lock(|cell| cell.get())
178 }
179
180 pub(super) fn set_peer_protocol(&self, peer_protocol: PeerProtocol) {
181 self.peer_protocol.lock(|cell| cell.set(peer_protocol));
182 }
183
184 fn peer_protocol(&self) -> PeerProtocol {
185 self.peer_protocol.lock(|cell| cell.get())
186 }
187
188 pub(super) fn clear_lanes(&self) {
189 self.data_plane.reset();
190 self.shutdown.reset();
191 self.control_in.clear();
192 self.control_out.clear();
193 self.data_in.clear();
194 self.data_out.clear();
195 self.identity_in.reset();
196 self.identity_out.clear();
197 }
198
199 fn link(
200 &'static self,
201 slot: BleSlotLink,
202 outbound_frames: &'static BleFramePool,
203 ) -> EmbeddedBleLink {
204 EmbeddedBleLink {
205 peer_protocol: self.peer_protocol(),
206 control_in: self.control_in.receiver(),
207 control_out: self.control_out.sender(),
208 data_in: self.data_in.receiver(),
209 data_out: self.data_out.sender(),
210 identity_in: &self.identity_in,
211 identity_out: self.identity_out.sender(),
212 data_plane: &self.data_plane,
213 plan: L2capPlan::None,
214 address: self.addr(),
215 outbound_frames,
216 slot,
217 }
218 }
219}
220
221pub struct BleHub {
222 pub(super) slots: [SlotChannels; PEER_CAPACITY],
223 pub(super) connection_slots: BleSlotPool,
224 pub(super) assign: [Channel<BridgeMutex, SlotJob, 1>; PEER_CAPACITY],
225 pub(super) ready: Channel<BridgeMutex, BleReadySlot, PEER_CAPACITY>,
226 pub(super) dial_failed: Channel<BridgeMutex, [u8; 6], PEER_CAPACITY>,
227 pub(super) sightings: Channel<BridgeMutex, SeenPeer, SIGHTING_DEPTH>,
228 pub(super) dial_request: Channel<BridgeMutex, DialTarget, PEER_CAPACITY>,
229 pub(super) inbound_frames: BleFramePool,
230 pub(super) outbound_frames: BleFramePool,
231 sighting_admission: BlockingMutex<BridgeMutex, RefCell<SightingAdmission>>,
232 radio: RadioArbiter,
233 discovery_turn: DiscoveryTurnArbiter,
234 pub(super) advertise: Signal<BridgeMutex, bool>,
235 pub(super) scan_enabled: Signal<BridgeMutex, bool>,
236 pub(super) radio_enabled: AtomicBool,
237 discovery: DiscoveryState,
238 pub(super) local_address: BlockingMutex<BridgeMutex, Cell<[u8; 6]>>,
239 status: BluetoothAutoStatus<PEER_CAPACITY>,
240}
241
242impl BleHub {
243 pub const fn new(status: BluetoothAutoStatus<PEER_CAPACITY>) -> Self {
244 Self {
245 slots: [const { SlotChannels::new() }; PEER_CAPACITY],
246 connection_slots: ConnectionSlotPool::new(),
247 assign: [const { Channel::new() }; PEER_CAPACITY],
248 ready: Channel::new(),
249 dial_failed: Channel::new(),
250 sightings: Channel::new(),
251 dial_request: Channel::new(),
252 inbound_frames: SharedFramePool::new(),
253 outbound_frames: SharedFramePool::new(),
254 sighting_admission: BlockingMutex::new(RefCell::new(SightingAdmission::new())),
255 radio: FairSemaphore::new(1),
256 discovery_turn: FairSemaphore::new(1),
257 advertise: Signal::new(),
258 scan_enabled: Signal::new(),
259 radio_enabled: AtomicBool::new(false),
260 discovery: DiscoveryState::new(),
261 local_address: BlockingMutex::new(Cell::new([0; 6])),
262 status,
263 }
264 }
265
266 pub fn set_local_address(&self, local_address: [u8; 6]) {
267 self.local_address.lock(|cell| cell.set(local_address));
268 }
269
270 pub(super) async fn acquire_radio(&self) -> RadioPermit<'_> {
271 loop {
272 match self.radio.acquire(1).await {
273 Ok(permit) => return permit,
274 Err(_) => yield_now().await,
275 }
276 }
277 }
278
279 fn admit_sighting(&self, address: [u8; 6], now_ms: u64) -> SightingAdmissionOutcome {
280 self.sighting_admission
281 .lock(|admission| admission.borrow_mut().classify(address, now_ms))
282 }
283
284 pub(super) async fn acquire_discovery_turn(&self) -> DiscoveryTurnPermit<'_> {
285 loop {
286 match self.discovery_turn.acquire(1).await {
287 Ok(permit) => return permit,
288 Err(_) => yield_now().await,
289 }
290 }
291 }
292
293 pub(super) fn note_ingress_pressure(&self) {
294 self.status.note_ingress_pressure();
295 }
296
297 pub(super) fn track_live_link(&self) -> LiveLinkGuard<'_> {
298 self.discovery.track_live_link()
299 }
300
301 pub(super) fn begin_busy_operation(&self) -> BusyOperationGuard<'_> {
302 self.discovery.begin_busy_operation()
303 }
304
305 pub(super) fn note_link_activity(&self) {
306 self.discovery.note_link_activity();
307 }
308
309 pub(super) fn retry_transient_client_disconnect(
310 &self,
311 target: DialTarget,
312 ) -> TransientClientRetryOutcome {
313 let target = match target.after_transient_client_disconnect() {
314 TransientClientRetry::Retry(target) => target,
315 TransientClientRetry::Backoff => return TransientClientRetryOutcome::Exhausted,
316 };
317 match self.dial_request.try_send(target) {
318 Ok(()) => TransientClientRetryOutcome::Queued,
319 Err(_) => TransientClientRetryOutcome::QueueBusy,
320 }
321 }
322
323 pub(super) async fn await_discovery_turn(
324 &self,
325 enabled: &Signal<BridgeMutex, bool>,
326 role: DiscoveryRole,
327 ) -> Result<DiscoveryWindow, bool> {
328 self.discovery.await_turn(enabled, role).await
329 }
330
331 pub(super) fn finish_discovery_turn(&self, window: DiscoveryWindow) {
332 self.discovery.finish_turn(window);
333 }
334
335 pub fn backend(&'static self) -> EmbeddedBleBackend {
336 EmbeddedBleBackend {
337 hub: self,
338 ready: self.ready.receiver(),
339 dial_failed: self.dial_failed.receiver(),
340 sightings: self.sightings.receiver(),
341 dial_request: self.dial_request.sender(),
342 seen: heapless::Vec::new(),
343 }
344 }
345}
346
347pub struct EmbeddedBleBackend {
348 hub: &'static BleHub,
349 ready: Receiver<'static, BridgeMutex, BleReadySlot, PEER_CAPACITY>,
350 dial_failed: Receiver<'static, BridgeMutex, [u8; 6], PEER_CAPACITY>,
351 sightings: Receiver<'static, BridgeMutex, SeenPeer, SIGHTING_DEPTH>,
352 dial_request: Sender<'static, BridgeMutex, DialTarget, PEER_CAPACITY>,
353 seen: heapless::Vec<DialTarget, SEEN_CAP>,
354}
355
356impl EmbeddedBleBackend {
357 fn remember(&mut self, peer: SeenPeer) {
358 let target = DialTarget::new(peer.kind, peer.addr);
359 if self
360 .seen
361 .iter()
362 .any(|seen| seen.addr.into_inner() == peer.addr.into_inner())
363 {
364 return;
365 }
366 if self.seen.push(target).is_err() {
367 self.seen.remove(0);
368 let _ = self.seen.push(target);
369 }
370 }
371
372 fn resolve(&self, address: BleAddress) -> Option<DialTarget> {
373 self.seen
374 .iter()
375 .find(|seen| seen.addr.into_inner() == *address.octets())
376 .copied()
377 }
378}
379
380impl BleBackend<PEER_CAPACITY> for EmbeddedBleBackend {
381 type Error = Closed;
382 type Link = EmbeddedBleLink;
383
384 async fn set_advertising(&mut self, mode: AdvertisingMode) -> Result<(), Closed> {
385 self.hub.advertise.signal(mode.is_on());
386 Ok(())
387 }
388
389 async fn set_scanning(&mut self, mode: ScanningMode) -> Result<(), Closed> {
390 self.hub.scan_enabled.signal(mode.is_on());
391 Ok(())
392 }
393
394 async fn set_radio_mode(&mut self, mode: RadioMode) -> Result<(), Closed> {
395 let enabled = mode.is_on();
396 self.hub.radio_enabled.store(enabled, Ordering::Relaxed);
397 if !enabled {
398 self.hub.advertise.signal(false);
399 self.hub.scan_enabled.signal(false);
400 self.hub.dial_request.clear();
401 self.hub.dial_failed.clear();
402 self.hub.ready.clear();
403 for (assign, slot) in self.hub.assign.iter().zip(self.hub.slots.iter()) {
404 assign.clear();
405 slot.shutdown.signal(());
406 }
407 }
408 Ok(())
409 }
410
411 async fn next_event(&mut self) -> BleEvent<EmbeddedBleLink> {
412 match select3(
413 self.ready.receive(),
414 self.sightings.receive(),
415 self.dial_failed.receive(),
416 )
417 .await
418 {
419 Either3::First(ready) => {
420 let ReadyConnectionSlotParts { origin, link } = ready.into_parts();
421 let index = link.index();
422 match origin {
423 Origin::Accepted => BleEvent::Inbound(
424 self.hub.slots[index].link(link, &self.hub.outbound_frames),
425 ),
426 Origin::Dialed => BleEvent::LinkReady {
427 link: self.hub.slots[index].link(link, &self.hub.outbound_frames),
428 origin: Origin::Dialed,
429 peer_rssi: None,
430 },
431 }
432 }
433 Either3::Second(peer) => {
434 self.remember(peer);
435 BleEvent::Sighting {
436 address: BleAddress::new(peer.addr.into_inner()),
437 rssi: Some(peer.rssi),
438 }
439 }
440 Either3::Third(bytes) => BleEvent::DialFailed {
441 address: BleAddress::new(bytes),
442 },
443 }
444 }
445
446 async fn dial(&mut self, address: BleAddress) -> DialOutcome {
447 if !self.hub.radio_enabled.load(Ordering::Relaxed) {
448 return DialOutcome::RadioOff;
449 }
450 let Some(target) = self.resolve(address) else {
451 return DialOutcome::UnknownPeer;
452 };
453 if self.dial_request.try_send(target).is_ok() {
454 DialOutcome::Started
455 } else {
456 DialOutcome::Busy
457 }
458 }
459}
460
461pub struct EmbeddedBleLink {
462 peer_protocol: PeerProtocol,
463 control_in: Receiver<'static, BridgeMutex, Control, CONTROL_QUEUE_DEPTH>,
464 control_out: Sender<'static, BridgeMutex, Control, CONTROL_QUEUE_DEPTH>,
465 data_in: Receiver<'static, BridgeMutex, BleFrameLease, FRAME_QUEUE_DEPTH>,
466 data_out: Sender<'static, BridgeMutex, BleFrameLease, FRAME_QUEUE_DEPTH>,
467 identity_in: &'static Signal<BridgeMutex, BleIdentity>,
468 identity_out: Sender<'static, BridgeMutex, BleIdentity, 1>,
469 data_plane: &'static Signal<BridgeMutex, L2capPlan>,
470 plan: L2capPlan,
471 address: [u8; 6],
472 outbound_frames: &'static BleFramePool,
473 slot: BleSlotLink,
474}
475
476impl BleLink for EmbeddedBleLink {
477 type Error = Closed;
478 type Source = EmbeddedBleSource;
479 type Sink = EmbeddedBleSink;
480
481 fn peer_protocol(&self) -> PeerProtocol {
482 self.peer_protocol
483 }
484
485 fn address(&self) -> BleAddress {
486 BleAddress::new(self.address)
487 }
488
489 async fn control_send(&mut self, msg: &Control) -> Result<(), Closed> {
490 match select(self.control_out.send(*msg), self.slot.wait_for_close()).await {
491 Either::First(()) => Ok(()),
492 Either::Second(()) => Err(Closed),
493 }
494 }
495
496 async fn control_recv(&mut self) -> Result<Control, Closed> {
497 match select(self.control_in.receive(), self.slot.wait_for_close()).await {
498 Either::First(msg) => Ok(msg),
499 Either::Second(()) => Err(Closed),
500 }
501 }
502
503 async fn receive_columba_peer_identity(&mut self) -> Result<BleIdentity, Closed> {
504 match select(self.identity_in.wait(), self.slot.wait_for_close()).await {
505 Either::First(identity) => Ok(identity),
506 Either::Second(()) => Err(Closed),
507 }
508 }
509
510 async fn send_columba_identity(&mut self, identity: BleIdentity) -> Result<(), Closed> {
511 match select(self.identity_out.send(identity), self.slot.wait_for_close()).await {
512 Either::First(()) => Ok(()),
513 Either::Second(()) => Err(Closed),
514 }
515 }
516
517 async fn upgrade(&mut self, plan: &L2capPlan) -> Result<(), Closed> {
518 self.plan = *plan;
519 Ok(())
520 }
521
522 fn into_data(self) -> (EmbeddedBleSource, EmbeddedBleSink) {
523 self.data_plane.signal(self.plan);
524 let ConnectionSlotDataOwners {
525 source: source_slot,
526 sink: sink_slot,
527 } = self.slot.into_data();
528 (
529 EmbeddedBleSource {
530 data_in: self.data_in,
531 slot: source_slot,
532 },
533 EmbeddedBleSink {
534 data_out: self.data_out,
535 frames: self.outbound_frames,
536 slot: sink_slot,
537 },
538 )
539 }
540}
541
542pub struct EmbeddedBleSource {
543 data_in: Receiver<'static, BridgeMutex, BleFrameLease, FRAME_QUEUE_DEPTH>,
544 slot: BleSlotSource,
545}
546
547impl BleSource for EmbeddedBleSource {
548 type Error = Closed;
549
550 async fn recv_frame(&mut self, out: &mut [u8]) -> Result<usize, Closed> {
551 match select(self.data_in.receive(), self.slot.wait_for_close()).await {
552 Either::First(frame) => {
553 let frame = frame.lock().await;
554 let len = frame.len().min(out.len());
555 out[..len].copy_from_slice(&frame[..len]);
556 Ok(len)
557 }
558 Either::Second(()) => Err(Closed),
559 }
560 }
561}
562
563pub struct EmbeddedBleSink {
564 data_out: Sender<'static, BridgeMutex, BleFrameLease, FRAME_QUEUE_DEPTH>,
565 frames: &'static BleFramePool,
566 slot: BleSlotSink,
567}
568
569impl BleSink for EmbeddedBleSink {
570 type Error = Closed;
571
572 async fn send_frame(&mut self, frame: &[u8]) -> Result<(), Closed> {
573 let lease = match select(self.frames.lease(), self.slot.wait_for_close()).await {
574 Either::First(Ok(lease)) => lease,
575 Either::First(Err(error)) => {
576 crate::diagnostic_log::warn!("ble frame lease failed: {error:?}");
577 return Err(Closed);
578 }
579 Either::Second(()) => return Err(Closed),
580 };
581 lease.fill(frame).await.map_err(|_| Closed)?;
582 match select(self.data_out.send(lease), self.slot.wait_for_close()).await {
583 Either::First(()) => Ok(()),
584 Either::Second(()) => Err(Closed),
585 }
586 }
587}
588
589pub(super) struct ScanFunnel {
590 pub(super) hub: &'static BleHub,
591 pub(super) local_address: BleAddress,
592}
593
594impl EventHandler for ScanFunnel {
595 fn on_adv_reports(&self, reports: LeAdvReportsIter) {
596 for report in reports {
597 let Ok(report) = report else { continue };
598 let peer_address = BleAddress::from_hci_bytes(report.addr.into_inner());
599 let capabilities =
600 columba_role_capabilities(report.data).unwrap_or(BleRoleCapabilities::DualRole);
601 let should_dial = columba_connection_role(
602 self.local_address,
603 BleRoleCapabilities::DualRole,
604 peer_address,
605 capabilities,
606 ) == ColumbaConnectionRole::Dial;
607 if contains_service(report.data) && should_dial {
608 let address = report.addr.into_inner();
609 let outcome = self.hub.admit_sighting(address, Instant::now().as_millis());
610 if outcome == SightingAdmissionOutcome::Admit {
611 let _ = self.hub.sightings.try_send(SeenPeer {
612 kind: report.addr_kind,
613 addr: report.addr,
614 rssi: report.rssi,
615 });
616 }
617 }
618 }
619 }
620}
621
622#[cfg(test)]
623mod tests {
624 use super::*;
625
626 #[test]
627 fn transient_client_disconnect_retry_is_bounded() {
628 let mut target = DialTarget::new(AddrKind::PUBLIC, BdAddr::new([1, 2, 3, 4, 5, 6]));
629
630 for _ in 0..8 {
631 target = match target.after_transient_client_disconnect() {
632 TransientClientRetry::Retry(target) => target,
633 TransientClientRetry::Backoff => panic!("transient retry exhausted early"),
634 };
635 }
636
637 assert!(matches!(
638 target.after_transient_client_disconnect(),
639 TransientClientRetry::Backoff
640 ));
641 }
642
643 #[test]
644 fn sightings_coalesce_per_address_until_the_retry_window() {
645 let mut admission = SightingAdmission::new();
646 let first = [1, 2, 3, 4, 5, 6];
647 let second = [6, 5, 4, 3, 2, 1];
648
649 assert_eq!(
650 admission.classify(first, 1_000),
651 SightingAdmissionOutcome::Admit
652 );
653 assert_eq!(
654 admission.classify(first, 2_999),
655 SightingAdmissionOutcome::Coalesce
656 );
657 assert_eq!(
658 admission.classify(second, 2_999),
659 SightingAdmissionOutcome::Admit
660 );
661 assert_eq!(
662 admission.classify(first, 3_000),
663 SightingAdmissionOutcome::Admit
664 );
665 }
666}