Retain stream audio owners through partial startup and failed release

2026-10-29 13:34:24 +01:00
1 parent f28f95a501
commit 26348b35e8
@@ -16,7 +16,12 @@ internal object StreamAudioMixer {
        snapshot = inputs.toList()
        if (worker == null) {
            output = OutputState()
-            worker = Thread(::run, "stream-mixer").apply { start() }
+            val next = Thread(::run, "stream-mixer"); worker = next
+            try { next.start() } catch (e: Exception) {
+                worker = null; inputs.remove(input); snapshot = inputs.toList()
+                LogManager.addLog("ERROR", "stream", "mixer worker start failed input=${input.id}; registration rolled back: $e")
+                throw e
+            }
        }
        LogManager.addLog("INFO", "stream", "mixer input attached id=${input.id} count=${inputs.size}")
    }
@@ -33,28 +38,32 @@ internal object StreamAudioMixer {
            check(inputs.isEmpty()) { "Stream mixer still has ${inputs.size} inputs after viewer release" }
            worker
        }
-        closing?.join()
+        check(closing !== Thread.currentThread()) { "Mixer cannot join itself" }
+        closing?.join(250)
+        check(closing?.isAlive != true) {
+            LogManager.addLog("ERROR", "stream", "mixer output cleanup pending; worker retains device")
+            "Mixer output cleanup pending"
+        }
        LogManager.addLog("INFO", "stream", "shared audio output released for call")
    }

    private fun run() {
-        val audio = StreamAudioOutput(log = { LogManager.addLog("INFO", "stream", "mixer $it") })
-        val mix = StreamMixBuffer()
+        var audio: StreamAudioOutput? = null
        var generated = 0L
        try {
+            val device = StreamAudioOutput(log = { LogManager.addLog("INFO", "stream", "mixer $it") }); audio = device
+            val mix = StreamMixBuffer()
            try { android.os.Process.setThreadPriority(android.os.Process.THREAD_PRIORITY_AUDIO) }
            catch (e: Exception) { LogManager.addLog("WARN", "stream", "mixer audio priority unavailable: ${e.message}") }
            LogManager.addLog("INFO", "stream", "mixer started: ready PCM only, blockFrames=${mix.frames}")
            while (true) {
                val current = synchronized(lock) {
                    if (inputs.isEmpty()) {
-                        // Закрытие под lock исключает открытие второго AudioTrack новым attach.
-                        audio.close(); output = OutputState(); worker = null
                        return
                    }
                    snapshot
                }
-                audio.step(System.nanoTime(), hold = false, pull = {
+                device.step(System.nanoTime(), hold = false, pull = {
                    mix.clear()
                    var active = false
                    for (input in current) active = input.mix(mix, generated) || active
@@ -63,19 +72,39 @@ internal object StreamAudioMixer {
                        mix.finish()
                    } else null
                }, finished = { false })
-                output = OutputState(audio.played, audio.leadUs)
+                output = OutputState(device.played, device.leadUs)
                Thread.sleep(5)
            }
        } catch (e: Exception) {
-            LogManager.addLog("ERROR", "stream", "shared audio output failed: ${e.message}; ${audio.description}")
+            LogManager.addLog("ERROR", "stream", "shared audio output failed: $e")
            synchronized(lock) {
                for (input in inputs) input.outputFailure = e.message ?: "Audio output failed"
                inputs.clear()
                snapshot = emptyList()
-                try { audio.close() }
-                catch (close: Exception) { LogManager.addLog("WARN", "stream", "mixer release: ${close.message}") }
-                output = OutputState(); worker = null
            }
+        } finally {
+            var delay = 250L
+            if (Thread.interrupted()) LogManager.addLog("WARN", "stream", "mixer interrupted before cleanup; finishing device release")
+            while (true) try { audio?.close(); break } catch (e: Exception) {
+                LogManager.addLog("ERROR", "stream", "mixer output release failed; owner retained delayMs=$delay: $e")
+                try { Thread.sleep(delay) } catch (interrupted: InterruptedException) {
+                    LogManager.addLog("WARN", "stream", "mixer cleanup wait interrupted; retrying")
+                }
+                delay = (delay * 2).coerceAtMost(30_000)
+            }
+            synchronized(lock) {
+                output = OutputState(); worker = null
+                // Новые inputs могли прийти во время cleanup; до SDK ACK второго устройства не было.
+                if (inputs.isNotEmpty()) {
+                    val next = Thread(::run, "stream-mixer"); worker = next
+                    try { next.start() } catch (e: Exception) {
+                        worker = null; inputs.forEach { it.outputFailure = "Mixer worker start failed: $e" }
+                        inputs.clear(); snapshot = emptyList()
+                        LogManager.addLog("ERROR", "stream", "mixer restart after cleanup failed: $e")
+                    }
+                }
+            }
+            LogManager.addLog("INFO", "stream", "shared audio output cleanup confirmed")
        }
    }
 }
@@ -86,7 +115,8 @@ internal object StreamAudioMixer {
 internal class StreamMixedPlayback(val id: Long, pts: Long, private val audioEnabled: Boolean, private val wake: () -> Unit) {
    data class Diagnostics(val pcmFrames: Int, val underruns: Long, val silenceFrames: Long)
    private val lock = Any()
-    private var handle = NativeLib.streamPlaybackCreate(id, pts, audioEnabled).also { check(it != 0L) }
+    private val initialPts = pts
+    private var handle = 0L
    private val timeline = StreamMixTimeline()
    private var started = false
    private var ended = false
@@ -97,7 +127,13 @@ internal class StreamMixedPlayback(val id: Long, pts: Long, private val audioEna
    private var silentFrames = 0L
    private var observedUnderruns = 0L
    @Volatile var outputFailure: String? = null
-    init { if (audioEnabled) StreamAudioMixer.attach(this) }
+    /** Владелец сохраняет пассивный объект до native create и регистрации mixer input. */
+    fun start() {
+        check(handle == 0L) { "Mixed playback already started id=$id" }
+        handle = NativeLib.streamPlaybackCreate(id, initialPts, audioEnabled)
+        check(handle != 0L) { "Stream playback create failed id=$id" }
+        if (audioEnabled) StreamAudioMixer.attach(this)
+    }

    val leadUs: Long get() = if (audioEnabled) StreamAudioMixer.output.leadUs else 0
    val empty: Boolean get() = synchronized(lock) { handle == 0L || NativeLib.streamPlaybackEmpty(handle) }
@@ -6,7 +6,9 @@ import android.media.AudioTrack
 import android.os.Build

 /** Один mixer worker владеет устройством и единственным недописанным PCM.
- * Native уже учёл выданные кадры: остаток нельзя терять или переставлять до сброса всей шкалы. */
+ * Native уже учёл выданные кадры: остаток нельзя терять или переставлять до сброса всей шкалы.
+ * Sink сохраняется до initialize/play, close снимает ссылку только после успешного освобождения.
+ */
 internal class StreamAudioOutput(
    private val open: () -> Sink = { AndroidSink() },
    private val log: (String) -> Unit = {}
@@ -17,6 +19,7 @@ internal class StreamAudioOutput(
        val threshold: Int
        val head: Long
        val description: String
+        fun initialize() {}
        fun write(pcm: ShortArray, offset: Int, count: Int): Int
        fun pause()
        fun play()
@@ -33,13 +36,11 @@ internal class StreamAudioOutput(
                .setEncoding(AudioFormat.ENCODING_PCM_16BIT).build())
            .setBufferSizeInBytes(AudioTrack.getMinBufferSize(48000, AudioFormat.CHANNEL_OUT_STEREO,
                AudioFormat.ENCODING_PCM_16BIT).coerceAtLeast(3840)).setTransferMode(AudioTrack.MODE_STREAM).build()
-        init {
-            try {
-                check(track.state == AudioTrack.STATE_INITIALIZED) { "AudioTrack initialization failed" }
-                // API 26..30 не умеет уменьшать порог отдельно от аппаратного буфера.
-                if (Build.VERSION.SDK_INT >= 31) track.setStartThresholdInFrames(1)
-                route = MediaAudioController.attachOutput(track)
-            } catch (e: Exception) { track.release(); throw e }
+        override fun initialize() {
+            check(track.state == AudioTrack.STATE_INITIALIZED) { "AudioTrack initialization failed" }
+            // API 26..30 не умеет уменьшать порог отдельно от аппаратного буфера.
+            if (Build.VERSION.SDK_INT >= 31) track.setStartThresholdInFrames(1)
+            route = MediaAudioController.attachOutput(track)
        }
        override val bufferFrames: Int get() = track.bufferSizeInFrames
        override val threshold: Int get() = if (Build.VERSION.SDK_INT >= 31)
@@ -54,6 +55,7 @@ internal class StreamAudioOutput(
    }

    private var sink: Sink? = null
+    private var closing = false
    private var pending: ShortArray? = null
    private var offset = 0
    private var padding = false
@@ -96,6 +98,7 @@ internal class StreamAudioOutput(

    /** Ровно одна неблокирующая запись за итерацию; ожидание устройства не задерживает video pump. */
    fun step(now: Long, hold: Boolean, pull: () -> ShortArray?, finished: () -> Boolean) {
+        check(!closing) { "Stream audio output already closing" }
        poll(now)
        if (paused != hold) {
            if (hold) sink?.pause() else sink?.play()
@@ -109,7 +112,7 @@ internal class StreamAudioOutput(
            if (pcm != null) {
                check(pcm.isNotEmpty() && pcm.size <= 960 && pcm.size % 2 == 0) { "Invalid stream PCM size=${pcm.size}" }
                if (sink == null) {
-                    sink = open(); openedAt = now; sink!!.play()
+                    sink = open(); openedAt = now; sink!!.initialize(); sink!!.play()
                    log("audio opened ${sink!!.description}")
                }
                pending = pcm; offset = 0; padding = false
@@ -144,8 +147,8 @@ internal class StreamAudioOutput(
    }

    fun close() {
-        val s = sink
+        closing = true
+        sink?.close()
        sink = null; pending = null; offset = 0; padding = false
-        s?.close()
    }
 }
@@ -149,6 +149,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
            if (playback != null) return
            audioEnabled = meta[5] and 2 != 0L
            playback = StreamMixedPlayback(id, meta[3], audioEnabled) { NativeLib.streamSignalNotify(signal) }
+            playback!!.start()
            position = meta[3]; generation = meta[4]
            LogManager.addLog("INFO", "stream", "playback starts id=$id PTS=$position generation=$generation fixedPrefill=0 " +
                "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} readyUs=$ready videoUs=${meta[1]} audioUs=${meta[2]} " +