diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioMixer.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioMixer.kt index a9f7ee0a..25737297 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioMixer.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioMixer.kt @@ -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) } diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioOutput.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioOutput.kt index f22d3b4c..8aeac612 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioOutput.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioOutput.kt @@ -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() } } diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt index edde1fb4..951beca4 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt @@ -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]} " +