| 1 | import Foundation |
| 2 | import LiveKitWebRTC |
| 3 | import os |
| 4 | |
| 5 | // Where the audio stack's slow work runs: never on the main thread. |
| 6 | // |
| 7 | // Paul, 0.3.8, MacBook Air, AirPods Pro: "after I open the Test window, |
| 8 | // there's a pause, then a beachball, then it works." Measured on an M1 Max |
| 9 | // with a main-thread watchdog (now `CallsInterop --audio-start-stall`), |
| 10 | // 0.3.8's test held the main thread three times on its way to a running |
| 11 | // microphone: |
| 12 | // |
| 13 | // 1. Building the audio (CallAudio): 166–200 ms. WebRTC's peer connection |
| 14 | // factory with LiveKit's engine module, 100–135 ms; the module's first |
| 15 | // device list, read only to be logged, 61–68 ms. |
| 16 | // 2. The in-memory loopback's answer, applied: 188–229 ms. Applying it |
| 17 | // starts the engine's playout on WebRTC's worker thread, and |
| 18 | // `setLocalDescription` waits for that before it returns, completion |
| 19 | // handler or not — every RTCPeerConnection call is a blocking hop to |
| 20 | // WebRTC's own threads. |
| 21 | // 3. `initRecording`, which waits on the worker thread while the engine |
| 22 | // starts with the microphone. Here, with no microphone permission, 25 |
| 23 | // ms. With AirPods it is as long as the headset takes to move from A2DP |
| 24 | // to its hands-free profile, often one to two seconds: with that start |
| 25 | // simulated at 1.5 s the main thread stalled for 1.5–1.7 s. That is the |
| 26 | // beachball; 1 and 2 are the pause before it. |
| 27 | // |
| 28 | // A call's join had the same shape (1.3–1.5 s with the switch simulated), |
| 29 | // and more: while the worker thread is starting the engine, *every* call |
| 30 | // into the engine module waits for it — `isRecording`, `isPlaying`, the |
| 31 | // device lists, the factory's processing state, a track's |
| 32 | // `isEnabled = false` (mute), its processing options — 0.9 to 1.2 s each, |
| 33 | // and a call's meters read some of them ten times a second. |
| 34 | // |
| 35 | // None of it is the main thread's business. It all runs here, on one |
| 36 | // serial queue, in the order it was asked for; the main actor awaits the |
| 37 | // result and only ever sets what the window shows. What the main thread |
| 38 | // needs to know about the module it reads from ModuleState. |
| 39 | |
| 40 | /// The audio stack's serial queue: the factory and its engine module, the |
| 41 | /// engine's start and stop, and the peer connections' calls. |
| 42 | /// |
| 43 | /// Serial on purpose. A test closed while its microphone is still |
| 44 | /// starting queues its stop behind the start, so the stop always sees the |
| 45 | /// microphone it has to close; nothing is left recording. |
| 46 | enum AudioWork { |
| 47 | static let queue = DispatchQueue(label: "net.amadan.oficina.calls.audio", qos: .userInitiated) |
| 48 | |
| 49 | /// Runs `body` on the audio queue and hands its result back. |
| 50 | static func run<T>(_ body: @escaping () -> T) async -> T { |
| 51 | await withCheckedContinuation { (c: CheckedContinuation<T, Never>) in |
| 52 | queue.async { c.resume(returning: body()) } |
| 53 | } |
| 54 | } |
| 55 | |
| 56 | static func run<T>(_ body: @escaping () throws -> T) async throws -> T { |
| 57 | try await withCheckedThrowingContinuation { (c: CheckedContinuation<T, Error>) in |
| 58 | queue.async { |
| 59 | do { c.resume(returning: try body()) } catch { c.resume(throwing: error) } |
| 60 | } |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | /// Queues `body` and does not wait for it: a stop, a close, a log line |
| 65 | /// that has to read the module. |
| 66 | static func post(_ body: @escaping () -> Void) { |
| 67 | queue.async(execute: body) |
| 68 | } |
| 69 | |
| 70 | /// Waits for everything queued so far. For tests and the harness. |
| 71 | static func drain() async { |
| 72 | await run {} |
| 73 | } |
| 74 | } |
| 75 | |
| 76 | /// A peer connection's calls, made from the audio queue. |
| 77 | /// |
| 78 | /// Every RTCPeerConnection method is a blocking hop to one of WebRTC's |
| 79 | /// own threads, completion handler or not, and applying a description |
| 80 | /// waits for the worker thread — which may be busy starting the audio |
| 81 | /// engine for seconds. So a call makes them from here and awaits the |
| 82 | /// answer; the main thread never waits on WebRTC. |
| 83 | enum PeerCalls { |
| 84 | static func describe( |
| 85 | _ pc: RTCPeerConnection, answer: Bool, _ constraints: RTCMediaConstraints |
| 86 | ) async throws -> RTCSessionDescription { |
| 87 | try await withCheckedThrowingContinuation { c in |
| 88 | AudioWork.queue.async { |
| 89 | let completion: @Sendable (RTCSessionDescription?, Error?) -> Void = { desc, error in |
| 90 | if let desc { c.resume(returning: desc) } else { c.resume(throwing: error ?? CancellationError()) } |
| 91 | } |
| 92 | if answer { |
| 93 | pc.answer(for: constraints, completionHandler: completion) |
| 94 | } else { |
| 95 | pc.offer(for: constraints, completionHandler: completion) |
| 96 | } |
| 97 | } |
| 98 | } |
| 99 | } |
| 100 | |
| 101 | static func set(_ pc: RTCPeerConnection, local: Bool, _ desc: RTCSessionDescription) async throws { |
| 102 | try await withCheckedThrowingContinuation { (c: CheckedContinuation<Void, Error>) in |
| 103 | AudioWork.queue.async { |
| 104 | let done: @Sendable (Error?) -> Void = { error in |
| 105 | if let error { c.resume(throwing: error) } else { c.resume() } |
| 106 | } |
| 107 | if local { pc.setLocalDescription(desc, completionHandler: done) } else { |
| 108 | pc.setRemoteDescription(desc, completionHandler: done) |
| 109 | } |
| 110 | } |
| 111 | } |
| 112 | } |
| 113 | |
| 114 | static func add(_ pc: RTCPeerConnection, _ candidate: RTCIceCandidate) async throws { |
| 115 | try await withCheckedThrowingContinuation { (c: CheckedContinuation<Void, Error>) in |
| 116 | AudioWork.queue.async { |
| 117 | pc.add(candidate) { error in if let error { c.resume(throwing: error) } else { c.resume() } } |
| 118 | } |
| 119 | } |
| 120 | } |
| 121 | |
| 122 | static func statistics(_ pc: RTCPeerConnection) async -> RTCStatisticsReport { |
| 123 | await withCheckedContinuation { c in |
| 124 | AudioWork.queue.async { pc.statistics { c.resume(returning: $0) } } |
| 125 | } |
| 126 | } |
| 127 | |
| 128 | static func statistics(_ pc: RTCPeerConnection, for receiver: RTCRtpReceiver) async -> RTCStatisticsReport { |
| 129 | await withCheckedContinuation { c in |
| 130 | AudioWork.queue.async { pc.statistics(for: receiver) { c.resume(returning: $0) } } |
| 131 | } |
| 132 | } |
| 133 | |
| 134 | static func statistics(_ pc: RTCPeerConnection, for sender: RTCRtpSender) async -> RTCStatisticsReport { |
| 135 | await withCheckedContinuation { c in |
| 136 | AudioWork.queue.async { pc.statistics(for: sender) { c.resume(returning: $0) } } |
| 137 | } |
| 138 | } |
| 139 | |
| 140 | /// Anything else, read or done on the audio queue. |
| 141 | static func read<T>(_ pc: RTCPeerConnection, _ body: @escaping (RTCPeerConnection) -> T) async -> T { |
| 142 | await AudioWork.run { body(pc) } |
| 143 | } |
| 144 | |
| 145 | /// Closes it after whatever is queued for it; not waited for. Closing |
| 146 | /// stops the audio engine when it is the last connection, and that can |
| 147 | /// take as long as a start. |
| 148 | static func close(_ pc: RTCPeerConnection) { |
| 149 | AudioWork.post { pc.close() } |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | #if canImport(DeepFilter) |
| 154 | /// DeepFilterNet's model, loaded before anyone needs it. |
| 155 | /// |
| 156 | /// Loading takes about a third of a second on an M1 Max (and longer on |
| 157 | /// an Air) of one core, and a call or test that has to load it starts on |
| 158 | /// RNNoise and moves to DeepFilterNet when it is ready. So one is loaded |
| 159 | /// in the background a few seconds after launch, at utility priority, |
| 160 | /// and the first test or call takes it ready-made. |
| 161 | /// |
| 162 | /// Memory is the cost: a loaded model is about 88 MB resident. So only |
| 163 | /// one is kept spare; a spare nobody has taken for fifteen minutes is |
| 164 | /// let go, and the next user loads its own as before (off the main |
| 165 | /// thread, starting on RNNoise). Taking the spare does not replace it at |
| 166 | /// once — a call would then hold two — but when its user is done, a new |
| 167 | /// spare is loaded for the next one. A model is never handed from one |
| 168 | /// user to the next: its state holds the last audio it heard, and there |
| 169 | /// is no way to clear it. |
| 170 | public enum DeepFilterPreload { |
| 171 | private struct State { |
| 172 | var spare: DeepFilterSuppressor? |
| 173 | var loading = false |
| 174 | var generation = 0 |
| 175 | } |
| 176 | // The suppressor is only ever touched by whoever holds it; the lock |
| 177 | // guards the handing over. |
| 178 | nonisolated(unsafe) private static var state = State() |
| 179 | private static let lock = NSLock() |
| 180 | private static let background = DispatchQueue(label: "net.amadan.oficina.calls.deepfilter", qos: .utility) |
| 181 | static let idleRelease: TimeInterval = 15 * 60 |
| 182 | |
| 183 | /// After launch: loads a spare in `delay` seconds, if there is a model |
| 184 | /// and no spare yet. |
| 185 | public static func preload(after delay: TimeInterval = 3) { |
| 186 | background.asyncAfter(deadline: .now() + delay) { load() } |
| 187 | } |
| 188 | |
| 189 | /// The spare, if one is ready; nil otherwise, and the caller loads. |
| 190 | static func take() -> DeepFilterSuppressor? { |
| 191 | lock.lock() |
| 192 | defer { lock.unlock() } |
| 193 | let spare = state.spare |
| 194 | state.spare = nil |
| 195 | state.generation += 1 |
| 196 | if spare != nil { AudioLog.info("DeepFilterNet: using the preloaded model") } |
| 197 | return spare |
| 198 | } |
| 199 | |
| 200 | /// A user of a model is done with it: a fresh spare, a little later. |
| 201 | static func finished() { |
| 202 | preload(after: 5) |
| 203 | } |
| 204 | |
| 205 | /// Whether a spare is loaded (for the tests). |
| 206 | static var ready: Bool { |
| 207 | lock.lock() |
| 208 | defer { lock.unlock() } |
| 209 | return state.spare != nil |
| 210 | } |
| 211 | |
| 212 | private static func load() { |
| 213 | lock.lock() |
| 214 | guard state.spare == nil, !state.loading, let model = DeepFilterSuppressor.model else { |
| 215 | lock.unlock() |
| 216 | return |
| 217 | } |
| 218 | state.loading = true |
| 219 | lock.unlock() |
| 220 | let started = CaptureDenoiser.now() |
| 221 | let loaded = DeepFilterSuppressor(model: model) |
| 222 | lock.lock() |
| 223 | state.loading = false |
| 224 | state.spare = loaded |
| 225 | state.generation += 1 |
| 226 | let generation = state.generation |
| 227 | lock.unlock() |
| 228 | guard loaded != nil else { return } |
| 229 | AudioLog.info( |
| 230 | "DeepFilterNet: preloaded in \(Int((CaptureDenoiser.now() - started) * 1000)) ms, kept for the next test or call") |
| 231 | background.asyncAfter(deadline: .now() + idleRelease) { release(ifStill: generation) } |
| 232 | } |
| 233 | |
| 234 | private static func release(ifStill generation: Int) { |
| 235 | lock.lock() |
| 236 | guard state.generation == generation, state.spare != nil else { |
| 237 | lock.unlock() |
| 238 | return |
| 239 | } |
| 240 | let spare = state.spare |
| 241 | state.spare = nil |
| 242 | lock.unlock() |
| 243 | _ = spare |
| 244 | AudioLog.info("DeepFilterNet: preloaded model unused for \(Int(idleRelease / 60)) minutes; released") |
| 245 | } |
| 246 | } |
| 247 | #endif |
| 248 | |
| 249 | /// What the app does for its audio once it has launched: loads the |
| 250 | /// noise suppressor's model in the background (DeepFilterPreload), so the |
| 251 | /// first test or call does not wait for it. Nothing on iPhone, which runs |
| 252 | /// RNNoise (made in well under a millisecond). |
| 253 | @MainActor public enum AudioWarmUp { |
| 254 | private static var done = false |
| 255 | |
| 256 | /// Once per launch; later calls (a window shown again) do nothing. |
| 257 | public static func afterLaunch() { |
| 258 | guard !done else { return } |
| 259 | done = true |
| 260 | #if canImport(DeepFilter) |
| 261 | DeepFilterPreload.preload(after: 3) |
| 262 | #endif |
| 263 | } |
| 264 | } |
| 265 | |
amadan