mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-27 11:05:19 +00:00
Stop bleQueue maintenance and status paths from blocking on collectionsQueue
The traffic-burst tracker becomes a lock-backed monitor (written by the receive pipeline, read by scan-duty adaptation and announce pacing on bleQueue), the status-log peer summary and topology refresh read the already lock-backed registry directly, and the stalled-fragment reap moves to an async collections hop with the gossip resync request inside it. bleQueue no longer sync-waits on the collections queue anywhere. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -77,6 +77,26 @@ struct BLEReceivePipeline {
|
||||
}
|
||||
}
|
||||
|
||||
/// Lock-backed traffic-level signal: the receive pipeline records packets,
|
||||
/// and the radio layer (maintenance and scan-duty adaptation on bleQueue)
|
||||
/// reads the level without crossing onto a transport queue.
|
||||
final class BLERecentTrafficMonitor: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var tracker = BLERecentTrafficTracker()
|
||||
|
||||
func recordPacket(at now: Date) {
|
||||
lock.withLock { tracker.recordPacket(at: now) }
|
||||
}
|
||||
|
||||
func hasTraffic(within seconds: TimeInterval, now: Date) -> Bool {
|
||||
lock.withLock { tracker.hasTraffic(within: seconds, now: now) }
|
||||
}
|
||||
|
||||
func removeAll() {
|
||||
lock.withLock { tracker.removeAll() }
|
||||
}
|
||||
}
|
||||
|
||||
struct BLERecentTrafficTracker: Equatable {
|
||||
private var packetTimestamps: [Date] = []
|
||||
|
||||
|
||||
@@ -426,7 +426,8 @@ final class BLEService: NSObject {
|
||||
// Relay jitter scheduling to reduce redundant floods
|
||||
private var scheduledRelays = BLEScheduledRelayStore()
|
||||
// Track short-lived traffic bursts to adapt announces/scanning under load
|
||||
private var recentTrafficTracker = BLERecentTrafficTracker()
|
||||
// (lock-backed: written by the receive pipeline, read on bleQueue)
|
||||
private let recentTrafficTracker = BLERecentTrafficMonitor()
|
||||
|
||||
// Ingress link tracking for duplicate and last-hop suppression
|
||||
private var ingressLinks = BLEIngressLinkRegistry()
|
||||
@@ -4661,13 +4662,11 @@ extension BLEService {
|
||||
let peripheralState = peripheralManager?.state ?? .unknown
|
||||
let isAdvertising = peripheralManager?.isAdvertising ?? false
|
||||
|
||||
let peerSummary = collectionsQueue.sync {
|
||||
(
|
||||
let peerSummary = (
|
||||
connected: peerRegistry.connectedCount,
|
||||
known: peerRegistry.count,
|
||||
candidates: connectionScheduler.candidateCount
|
||||
)
|
||||
}
|
||||
|
||||
#if os(iOS)
|
||||
// INVARIANT: bleQueue must NEVER sync-dispatch to the main thread.
|
||||
@@ -4701,10 +4700,7 @@ extension BLEService {
|
||||
}
|
||||
|
||||
private func refreshLocalTopology() {
|
||||
let neighbors: [Data] = collectionsQueue.sync {
|
||||
peerRegistry.connectedRoutingData
|
||||
}
|
||||
meshTopology.updateNeighbors(for: myPeerIDData, neighbors: neighbors)
|
||||
meshTopology.updateNeighbors(for: myPeerIDData, neighbors: peerRegistry.connectedRoutingData)
|
||||
}
|
||||
|
||||
private func computeRoute(to peerID: PeerID) -> [Data]? {
|
||||
@@ -7932,9 +7928,7 @@ extension BLEService {
|
||||
let now = Date()
|
||||
let connectedCount = peerRegistry.connectedCount
|
||||
let elapsed = announceThrottle.elapsed(since: now)
|
||||
let recentSeen = collectionsQueue.sync { () -> Bool in
|
||||
recentTrafficTracker.hasTraffic(within: 5.0, now: now)
|
||||
}
|
||||
let recentSeen = recentTrafficTracker.hasTraffic(within: 5.0, now: now)
|
||||
let hasNoPeers = peerRegistry.isEmpty
|
||||
let plan = BLEMaintenancePolicy.plan(
|
||||
cycle: maintenanceCounter,
|
||||
@@ -8061,18 +8055,20 @@ extension BLEService {
|
||||
// Clean old fragments (> configured seconds old), then ask peers for
|
||||
// the specific fragment streams whose reassembly has stalled instead
|
||||
// of waiting for the next periodic GCS fragment round.
|
||||
let stalledFragmentIDs = collectionsQueue.sync(flags: .barrier) { () -> [Data] in
|
||||
collectionsQueue.async(flags: .barrier) { [weak self] in
|
||||
guard let self else { return }
|
||||
let cutoff = now.addingTimeInterval(-TransportConfig.bleFragmentLifetimeSeconds)
|
||||
fragmentAssemblyBuffer.removeExpired(before: cutoff)
|
||||
sourceRouteFailures.prune(now: now)
|
||||
return fragmentAssemblyBuffer.stalledBroadcastFragmentIDs(
|
||||
self.fragmentAssemblyBuffer.removeExpired(before: cutoff)
|
||||
self.sourceRouteFailures.prune(now: now)
|
||||
let stalledFragmentIDs = self.fragmentAssemblyBuffer.stalledBroadcastFragmentIDs(
|
||||
stalledAfter: TransportConfig.bleFragmentResyncStallSeconds,
|
||||
retryAfter: TransportConfig.bleFragmentResyncRetrySeconds,
|
||||
now: now
|
||||
)
|
||||
}
|
||||
if !stalledFragmentIDs.isEmpty {
|
||||
gossipSyncManager?.requestMissingFragments(fragmentIDs: stalledFragmentIDs)
|
||||
// GossipSyncManager serializes on its own internal queue.
|
||||
self.gossipSyncManager?.requestMissingFragments(fragmentIDs: stalledFragmentIDs)
|
||||
}
|
||||
}
|
||||
|
||||
// Clean old connection timeout backoff entries (> window)
|
||||
@@ -8117,12 +8113,10 @@ extension BLEService {
|
||||
let active = true
|
||||
#endif
|
||||
// Force full-time scanning if we have very few neighbors or very recent traffic
|
||||
let hasRecentTraffic: Bool = collectionsQueue.sync {
|
||||
recentTrafficTracker.hasTraffic(
|
||||
let hasRecentTraffic = recentTrafficTracker.hasTraffic(
|
||||
within: TransportConfig.bleRecentTrafficForceScanSeconds,
|
||||
now: Date()
|
||||
)
|
||||
}
|
||||
let scanPlan = BLEScanDutyPolicy.plan(
|
||||
dutyEnabled: dutyEnabled,
|
||||
appIsActive: active,
|
||||
|
||||
Reference in New Issue
Block a user