Keep call audio main thread responsive while device workers are blocked

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)

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<Write>(64)
     @Volatile var onWrite: () -> Unit = {}
+    @Volatile var onTimestamp: () -> Unit = {}
     @Volatile var maxWriteSamples = Int.MAX_VALUE
     @Volatile var forcedWriteResult: Int? = null
     @Volatile var failures = emptySet<String>()
@@ -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<AudioRecord>()
         var nextFailures = emptySet<String>()
+        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<AudioRouting.OnRoutingChangedListener>()
     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++

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")