From ff1f7a4bb815ecedc7d99f8d3ca243ca6e2a2716 Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 29 Oct 2026 09:22:02 +0100 Subject: [PATCH] Keep call audio main thread responsive while device workers are blocked --- .../com/utun/chat/data/CallAudioEngine.kt | 108 +++++++++++------- .../main/kotlin/android/media/HostTracks.kt | 7 +- .../utun/chat/data/CallAudioLifecycleTest.kt | 44 ++++++- 3 files changed, 116 insertions(+), 43 deletions(-) diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioEngine.kt index 0c95c5aa..57364e83 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioEngine.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallAudioEngine.kt @@ -14,6 +14,8 @@ import android.os.Handler import android.os.Looper import android.os.SystemClock import java.util.concurrent.atomic.AtomicBoolean +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock import com.utun.chat.ChatApplication /** @@ -50,9 +52,9 @@ class CallAudioEngine { const val AEC_ACOUSTIC_SAMPLES = 150L } - private val recordLock = Any() - private val trackLock = Any() - private var audioRecord: AudioRecord? = null + private val recordLock = ReentrantLock() + private val trackLock = ReentrantLock() + @Volatile private var audioRecord: AudioRecord? = null private var captureThread: Thread? = null @Volatile private var audioTrack: CallAudioOutput? = null private var playThread: Thread? = null @@ -92,7 +94,7 @@ class CallAudioEngine { @Volatile private var captureElapsedMs = 0L private val mainHandler = Handler(Looper.getMainLooper()) private val trackRouteListener = AudioRouting.OnRoutingChangedListener { output -> - synchronized(trackLock) { + withDeviceLock(trackLock) { val track = audioTrack if (active && track != null && output === track.track) { val device = track.routedDevice @@ -103,7 +105,7 @@ class CallAudioEngine { } } private val recordRouteListener = AudioRouting.OnRoutingChangedListener { input -> - synchronized(recordLock) { + withDeviceLock(recordLock) { val record = audioRecord if (active && record != null && input === record) { val device = record.routedDevice @@ -142,10 +144,23 @@ class CallAudioEngine { } } - val diagnostics: String get() = synchronized(recordLock) { + val diagnostics: String get() = "active=$active ready=$captureReady/$playReady capture=$capFramesRead play=$playFramesWritten " + "workers=${captureThread?.isAlive}/${playThread?.isAlive} " + - "record=${audioRecord?.recordingState} track=${audioTrack?.playState} route=${router.route}" + "recordOwned=${audioRecord != null} trackOwned=${audioTrack != null} route=${router.route}" + + /** SDK-вызовы workers не задерживают main: маршрут/замер повторяется только для того же активного звонка. */ + private fun withDeviceLock(lock: ReentrantLock, work: () -> Unit) { + if (Looper.myLooper() != Looper.getMainLooper()) { + lock.withLock(work) + } else if (lock.tryLock()) { + try { work() } finally { lock.unlock() } + } else { + val id = callId + if (active && !mainHandler.postDelayed({ if (active && callId == id) withDeviceLock(lock, work) }, 50)) { + LogManager.addLog("ERROR", "CallAudio", "device action retry rejected id=$id") + } + } } /** Mute применяется в native после AEC/AGC; настоящий микрофон продолжает обучать обработку. */ @@ -221,14 +236,14 @@ class CallAudioEngine { } router.onOutputDevice = { dev -> applyTrackOutputDevice(dev) } router.onRouteRequested = { - synchronized(trackLock) { playbackBuffer?.settle(SystemClock.elapsedRealtime()) } + withDeviceLock(trackLock) { playbackBuffer?.settle(SystemClock.elapsedRealtime()) } } /* Замер AEC — через 300мс после готовности обоих направлений; RX этого не ждёт. */ val ready = Runnable { if (callId == id && active && !failed.get() && captureReady && playReady && !duplexReady) { duplexReady = true - mainHandler.postDelayed(aecDelayRunnable, 300) + scheduleAecDelay(300) LogManager.addLog("INFO", "CallAudio", "duplex ready id=$id totalMs=${SystemClock.elapsedRealtime() - startElapsedMs}") } } @@ -250,7 +265,7 @@ class CallAudioEngine { // Каждый поток готовит только своё устройство. Отмена ждёт оба до release Android/native. playThread = Thread({ runWorker(id, "play") { - synchronized(trackLock) { + trackLock.withLock { if (!active) return@runWorker outputPosition.reset() outputRecovery.route(confirmedDeviceId) @@ -260,12 +275,13 @@ class CallAudioEngine { "playback started id=$id totalMs=${SystemClock.elapsedRealtime() - startElapsedMs} " + "backend=${audioTrack?.name} captureReady=$captureReady") } - mainHandler.post(ready) + if (!mainHandler.post(ready)) LogManager.addLog("ERROR", "CallAudio", "playback ready callback rejected id=$id") playLoop() - } }, "utun-call-play").apply { start() } + } }, "utun-call-play") + playThread!!.start() captureThread = Thread({ runWorker(id, "capture") { - synchronized(recordLock) { + recordLock.withLock { if (!active) return@runWorker val createdAt = SystemClock.elapsedRealtime() wiredActive = wantedWired @@ -289,9 +305,10 @@ class CallAudioEngine { "recordCreateMs=$recordCreateMs recordStartMs=${SystemClock.elapsedRealtime() - captureElapsedMs} " + "playReady=$playReady") } - mainHandler.post(ready) + if (!mainHandler.post(ready)) LogManager.addLog("ERROR", "CallAudio", "capture ready callback rejected id=$id") captureLoop() - } }, "utun-call-capture").apply { start() } + } }, "utun-call-capture") + captureThread!!.start() LogManager.addLog("INFO", "CallAudio", "preparing RX/TX independently id=$id totalMs=${SystemClock.elapsedRealtime() - startElapsedMs} " + "nativeMs=$nativeMs modeMs=$modeMs routeMs=$routeMs") @@ -319,7 +336,9 @@ class CallAudioEngine { private fun reportFailure(id: Long, detail: String) { if (!active || !failed.compareAndSet(false, true)) return LogManager.addLog("ERROR", "CallAudio", "$detail id=$id route=${router.route} capture=$capFramesRead play=$playFramesWritten") - mainHandler.post { if (callId == id && active) onFailure?.invoke(id, detail) } + if (!mainHandler.post { if (callId == id && active) onFailure?.invoke(id, detail) }) { + LogManager.addLog("ERROR", "CallAudio", "failure callback rejected id=$id detail=$detail") + } } /** Main: отменить работу, разблокировать read и подтвердить join/release каждой независимой части. @@ -348,20 +367,22 @@ class CallAudioEngine { } } } - synchronized(recordLock) { + if (recordLock.tryLock()) try { audioRecord?.takeIf { captureThread != null }?.let { rec -> try { if (rec.recordingState == AudioRecord.RECORDSTATE_RECORDING) rec.stop() } catch (e: Exception) { LogManager.addLog("ERROR", "CallAudio", "record stop failed id=$id: $e") } } - } - synchronized(trackLock) { + } finally { recordLock.unlock() } + else LogManager.addLog("WARN", "CallAudio", "record stop deferred id=$id; worker holds device lock") + if (trackLock.tryLock()) try { audioTrack?.takeIf { playThread != null }?.let { track -> try { track.removeOnRoutingChangedListener(trackRouteListener) } catch (e: Exception) { LogManager.addLog("WARN", "CallAudio", "track listener removal failed id=$id: $e") } try { if (track.playState == AudioTrack.PLAYSTATE_PLAYING) track.pause() } catch (e: Exception) { LogManager.addLog("WARN", "CallAudio", "track pause failed id=$id: $e") } } - } + } finally { trackLock.unlock() } + else LogManager.addLog("WARN", "CallAudio", "track pause deferred id=$id; worker holds device lock") playThread?.interrupt() var interrupted = false for (worker in listOfNotNull(captureThread, playThread)) { @@ -375,17 +396,17 @@ class CallAudioEngine { if (captureThread?.isAlive != true) captureThread = null if (playThread?.isAlive != true) playThread = null if (captureThread == null && playThread == null) { - synchronized(recordLock) { + recordLock.withLock { try { audioRecord?.release(); audioRecord = null } catch (e: Exception) { LogManager.addLog("ERROR", "CallAudio", "record release failed id=$id: $e") } } - synchronized(trackLock) { + trackLock.withLock { try { audioTrack?.release(); audioTrack = null } catch (e: Exception) { LogManager.addLog("ERROR", "CallAudio", "track release failed id=$id: $e") } + playbackBuffer = null } } captureReady = false; playReady = false; duplexReady = false - playbackBuffer = null try { router.stop() } catch (e: Exception) { LogManager.addLog("ERROR", "CallAudio", "router stop failed id=$id; retained for retry: $e") } @@ -432,7 +453,7 @@ class CallAudioEngine { while (active) { /* проводная гарнитура появилась/исчезла — меняем источник захвата */ if (wantedWired != wiredActive) { - synchronized(recordLock) { + recordLock.withLock { if (!active) return rec?.apply { stop() @@ -552,10 +573,10 @@ class CallAudioEngine { LogManager.addLog("ERROR", "CallAudio", "record setPreferredDevice(wired) threw: ${e.message}") false } - LogManager.addLog("WARN", "CallAudio", + LogManager.addLog(if (ok) "INFO" else "WARN", "CallAudio", "record input -> VOICE_COMMUNICATION dev=${dev?.type ?: "-"} setPreferredDevice=$ok") } else { - LogManager.addLog("WARN", "CallAudio", "record input -> VOICE_COMMUNICATION (default)") + LogManager.addLog("INFO", "CallAudio", "record input -> VOICE_COMMUNICATION (default)") } LogManager.addLog("INFO", "CallAudio", "capture device ready id=$callId minQueryMs=$queryMs constructorMs=$constructorMs " + "routingMs=${SystemClock.elapsedRealtime() - routeAt} minBytes=$recMin") @@ -581,7 +602,7 @@ class CallAudioEngine { } private fun replacePlaybackOutput(epoch: Long, reason: String) { - synchronized(trackLock) { + trackLock.withLock { if (!active) return CallOutputNative.suspendClock(callId, epoch) routeGeneration++ @@ -606,7 +627,7 @@ class CallAudioEngine { try { val stamp = AudioTimestamp() while (!Thread.currentThread().isInterrupted) { - synchronized(trackLock) { + trackLock.withLock { audioTrack?.let { track -> if (active) { trace.device = samplePlaybackDevice(track, stamp) @@ -803,7 +824,7 @@ class CallAudioEngine { /** Reporter управляет устройством независимо от уровня debug; playback не берёт trackLock. */ private fun adjustPlaybackBuffer(track: CallAudioOutput) { - synchronized(trackLock) { + trackLock.withLock { if (!active || audioTrack !== track) return val policy = playbackBuffer ?: return try { @@ -846,12 +867,12 @@ class CallAudioEngine { "performanceRequested=${AudioTrack.PERFORMANCE_MODE_LOW_LATENCY} performanceActual=$performance pacing=device") if (playReady && (previous != actual || change == null)) { // Очередь сначала принимает новый размер; устаревшая задача не затронет следующий звонок. - mainHandler.post { + if (!mainHandler.post { if (active && audioTrack === track) { mainHandler.removeCallbacks(aecDelayRunnable) - mainHandler.postDelayed(aecDelayRunnable, 1_000) + scheduleAecDelay(1_000) } - } + }) LogManager.addLog("ERROR", "CallAudio", "buffer AEC callback rejected id=$callId") } } catch (e: Exception) { playbackBuffer = null @@ -867,11 +888,17 @@ class CallAudioEngine { NativeLib.callAudioSetAec(on) } + private fun scheduleAecDelay(delayMs: Long) { + if (!mainHandler.postDelayed(aecDelayRunnable, delayMs)) { + LogManager.addLog("ERROR", "CallAudio", "AEC timer rejected id=$callId delayMs=$delayMs") + } + } + /** Замер задержки рендер→захват после старта, смены выхода или размера его очереди. */ private fun measureAecDelay() { - synchronized(recordLock) { - if (!active || !captureReady || !playReady) return - val rec = audioRecord ?: return + withDeviceLock(recordLock) { + if (!active || !captureReady || !playReady) return@withDeviceLock + val rec = audioRecord ?: return@withDeviceLock try { val output = outputLatency(callId) val rTs = AudioTimestamp() @@ -880,7 +907,7 @@ class CallAudioEngine { if (output.outputMs < 0 || inputFrames == null) { LogManager.addLog("WARN", "CallAudio", "getTimestamp unavailable; AEC delay = $AEC_DEFAULT_DELAY_FRAMES frames") NativeLib.callAudioSetAecDelay(AEC_DEFAULT_DELAY_FRAMES) - return + return@withDeviceLock } // output уже включает HAL и локальную очередь; локальный frame counter нового выхода здесь не используется. val trackLatency = output.outputMs * 48L @@ -892,15 +919,16 @@ class CallAudioEngine { NativeLib.callAudioSetAecDelay(delayFrames) } catch (e: Exception) { LogManager.addLog("WARN", "CallAudio", "AEC delay measure failed: ${e.message}") - NativeLib.callAudioSetAecDelay(AEC_DEFAULT_DELAY_FRAMES) + try { NativeLib.callAudioSetAecDelay(AEC_DEFAULT_DELAY_FRAMES) } + catch (fallback: Exception) { LogManager.addLog("ERROR", "CallAudio", "AEC fallback failed id=$callId: $fallback") } } } } private fun applyTrackOutputDevice(dev: AudioDeviceInfo?) { - synchronized(trackLock) { - val track = audioTrack ?: return - if (!active) return + withDeviceLock(trackLock) { + val track = audioTrack ?: return@withDeviceLock + if (!active || router.currentOutputDevice != dev) return@withDeviceLock val startedAt = SystemClock.elapsedRealtime() try { val ok = track.setPreferredDevice(dev) diff --git a/tools/chatgui-android/router-tests/src/main/kotlin/android/media/HostTracks.kt b/tools/chatgui-android/router-tests/src/main/kotlin/android/media/HostTracks.kt index e1b935b0..cab9054d 100644 --- a/tools/chatgui-android/router-tests/src/main/kotlin/android/media/HostTracks.kt +++ b/tools/chatgui-android/router-tests/src/main/kotlin/android/media/HostTracks.kt @@ -41,7 +41,7 @@ class AudioTrack(val id: Int) : AudioRouting { fun pause() { access("pause"); playState = PLAYSTATE_STOPPED } fun setBufferSizeInFrames(value: Int): Int { access("buffer"); bufferSizeInFrames = value; return value } fun setStartThresholdInFrames(value: Int): Int { access("threshold"); startThresholdInFrames = value; return value } - fun getTimestamp(stamp: AudioTimestamp): Boolean { access("timestamp"); return false } + fun getTimestamp(stamp: AudioTimestamp): Boolean { onTimestamp(); access("timestamp"); return false } var interruptOnStop = false fun stop() { access("stop"); if (interruptOnStop) Thread.currentThread().interrupt() } var releaseCalls = 0 @@ -70,6 +70,7 @@ class AudioTrack(val id: Int) : AudioRouting { data class Write(val offset: Int, val count: Int, val result: Int, val pcm: ShortArray) val writes = LinkedBlockingDeque(64) @Volatile var onWrite: () -> Unit = {} + @Volatile var onTimestamp: () -> Unit = {} @Volatile var maxWriteSamples = Int.MAX_VALUE @Volatile var forcedWriteResult: Int? = null @Volatile var failures = emptySet() @@ -148,6 +149,7 @@ class AudioRecord(source: Int, rate: Int, channels: Int, encoding: Int, size: In const val STATE_INITIALIZED = 1; const val RECORDSTATE_RECORDING = 3; const val READ_BLOCKING = 0; const val SUCCESS = 0 val created = mutableListOf() var nextFailures = emptySet() + var nextOnStart: () -> Unit = {} fun getMinBufferSize(rate: Int, channels: Int, encoding: Int) = 1920 } @Volatile var failures = nextFailures @@ -164,7 +166,8 @@ class AudioRecord(source: Int, rate: Int, channels: Int, encoding: Int, size: In val listeners = mutableSetOf() init { check("create" !in failures); created.add(this) } private fun access(operation: String) { check(!released && operation !in failures) { "record failed: $operation" } } - fun startRecording() { access("start"); recordingState = RECORDSTATE_RECORDING } + private val onStart = nextOnStart + fun startRecording() { onStart(); access("start"); recordingState = RECORDSTATE_RECORDING } fun stop() { access("stop"); synchronized(monitor) { recordingState = 1; monitor.notifyAll() } } fun release() { releaseCalls++ diff --git a/tools/chatgui-android/router-tests/src/test/kotlin/com/utun/chat/data/CallAudioLifecycleTest.kt b/tools/chatgui-android/router-tests/src/test/kotlin/com/utun/chat/data/CallAudioLifecycleTest.kt index 936edd11..256423c3 100644 --- a/tools/chatgui-android/router-tests/src/test/kotlin/com/utun/chat/data/CallAudioLifecycleTest.kt +++ b/tools/chatgui-android/router-tests/src/test/kotlin/com/utun/chat/data/CallAudioLifecycleTest.kt @@ -33,6 +33,7 @@ class CallAudioLifecycleTest { BluetoothAdapter.current = null AudioTrack.created.clear(); AudioTrack.nextFailures = emptySet(); AudioTrack.minimumBuffer = 1920 AudioRecord.created.clear(); AudioRecord.nextFailures = emptySet() + AudioRecord.nextOnStart = {} NativeLib.callAudioCommands.clear(); NativeLib.callAudioFailures = emptySet(); NativeLib.onCallFeed = {} TelecomCallManager.connection = null Handler.rejectImmediate = false; Handler.rejectDelayed = false @@ -47,7 +48,8 @@ class CallAudioLifecycleTest { NativeLib.callAudioFailures = emptySet(); NativeLib.onCallFeed = {} manager.failures = emptySet() AudioRecord.created.forEach { it.failures = emptySet() } - AudioTrack.created.forEach { it.failures = emptySet(); it.onWrite = {} } + AudioTrack.created.forEach { it.failures = emptySet(); it.onWrite = {}; it.onTimestamp = {} } + AudioRecord.nextOnStart = {} engine.stop(); main.advanceBy(10_000) assertTrue(AudioRecord.created.all { it.released }) assertTrue(AudioTrack.created.all { it.released }) @@ -174,6 +176,46 @@ class CallAudioLifecycleTest { engine.stop(); assertEquals(1, stops()) } + @Test(timeout = 5_000) fun blockedRecordStartupCannotHoldMainBeforeTheBoundedJoin() { + val entered = CountDownLatch(1) + val unblock = CountDownLatch(1).also { gates.add(it) } + AudioRecord.nextOnStart = { entered.countDown(); unblock.await() } + assertTrue(engine.start(7)); assertTrue(entered.await(3, TimeUnit.SECONDS)) + assertTrue(engine.diagnostics.contains("recordOwned=true")) + val began = System.nanoTime() + rejected { engine.stop() } + assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - began) < 1_500) + assertFalse(AudioRecord.created.single().released); assertEquals(0, stops()) + assertTrue(diagnostic("record stop deferred")) + unblock.countDown() + await("record startup exits") { (field("captureThread") as Thread).isAlive.not() } + engine.stop(); assertEquals(1, stops()) + } + + @Test(timeout = 5_000) fun blockedTimestampReporterCannotHoldMainOrReleaseItsOutputEarly() { + start() + val entered = CountDownLatch(1) + val unblock = CountDownLatch(1).also { gates.add(it) } + val track = AudioTrack.created.single() + track.onTimestamp = { + entered.countDown() + var interrupted = false + while (unblock.count != 0L) { + try { unblock.await() } catch (_: InterruptedException) { interrupted = true } + } + if (interrupted) Thread.currentThread().interrupt() + } + assertTrue(entered.await(3, TimeUnit.SECONDS)) + val began = System.nanoTime() + rejected { engine.stop() } + assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - began) < 1_500) + assertFalse(track.released); assertEquals(0, stops()); assertTrue(diagnostic("track pause deferred")) + unblock.countDown() + await("reporter exits with playback") { (field("playThread") as Thread).isAlive.not() } + engine.stop(); assertTrue(track.released); assertEquals(0, track.accessAfterRelease) + assertEquals(1, stops()) + } + @Test fun recordSetupFailureRetainsEvenAnUninitializedDeviceUntilReleaseSucceeds() { for (failure in listOf("uninitialized", "state", "add listener", "add listener after", "start")) { AudioRecord.nextFailures = setOf(failure, "release")