From 75806249bd629b16466e66c3cc2f56874f4f15ea Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 20 Oct 2026 16:58:41 +0200 Subject: [PATCH] Run Android players concurrently and fall back between decoders --- .../main/java/com/utun/chat/data/NativeLib.kt | 2 + .../com/utun/chat/data/StreamPlayerEngine.kt | 97 ++++++++++--------- .../jni_bridge/android_stream_jni.c | 12 ++- 3 files changed, 63 insertions(+), 48 deletions(-) diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt index 941ecc14..e6e6dcb4 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt @@ -222,6 +222,7 @@ object NativeLib { fun streamPlaybackCreate(id: Long, pts: Long, audio: Boolean): Long = nativeStreamPlaybackCreate(id, pts, audio) fun streamPlaybackFeed(handle: Long, pts: Long, packet: ByteArray): Int = nativeStreamPlaybackFeed(handle, pts, packet) fun streamPlaybackPull(handle: Long, frames: Int, limit: Long): ShortArray? = nativeStreamPlaybackPull(handle, frames, limit) + fun streamPlaybackSilence(handle: Long, frames: Int) { nativeStreamPlaybackSilence(handle, frames) } fun streamPlaybackUpdate(handle: Long, now: Long, played: Long, ready: Long, guard: Long, starved: Boolean): LongArray? = nativeStreamPlaybackUpdate(handle, now, played, ready, guard, starved) fun streamPlaybackFinish(handle: Long) { nativeStreamPlaybackFinish(handle) } @@ -246,6 +247,7 @@ object NativeLib { private external fun nativeStreamPlaybackCreate(id: Long, pts: Long, audio: Boolean): Long private external fun nativeStreamPlaybackFeed(handle: Long, pts: Long, packet: ByteArray): Int private external fun nativeStreamPlaybackPull(handle: Long, frames: Int, limit: Long): ShortArray? + private external fun nativeStreamPlaybackSilence(handle: Long, frames: Int) private external fun nativeStreamPlaybackUpdate(handle: Long, now: Long, played: Long, ready: Long, guard: Long, starved: Boolean): LongArray? private external fun nativeStreamPlaybackFinish(handle: Long) 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 9a9b739e..01f6ba85 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 @@ -15,15 +15,12 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext import java.nio.ByteBuffer import java.nio.ByteOrder -import java.util.concurrent.locks.ReentrantLock -import kotlin.concurrent.withLock /** Запас кодированного потока остаётся в общем RS-кольце. Worker запрашивает ближайшие кадры, * общий native controller ведёт минутную историю и темп, AudioTrack задаёт фактические часы. * Временная нехватка сохраняет decoder references; GAP/вытеснение начинают новую точку KEY. */ class StreamPlayerEngine(private val group: Long, private val source: Long, private val epoch: Long) { companion object { - private val viewerLock = ReentrantLock() private val main = Handler(Looper.getMainLooper()) private val players = mutableSetOf() // Только main thread. private val mutableViewing = MutableStateFlow(false) @@ -64,7 +61,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv players.add(this); mutableViewing.value = true running = true worker = Thread({ - try { viewerLock.withLock { if (running) play(surface) } } + try { if (running) play(surface) } finally { main.post { players.remove(this); mutableViewing.value = players.isNotEmpty() } } }, "stream-player").apply { start() } } @@ -72,9 +69,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv private fun play(surface: Surface) { var decoder: MediaCodec? = null - fun audioOutput() = StreamAudioOutput(log = { LogManager.addLog("INFO", "stream", "viewer id=$id $it") }) - var audio = audioOutput() - var playback = 0L + var playback: StreamMixedPlayback? = null lateinit var video: StreamDecodePump var waitingKey = true var audioEnabled = false @@ -100,15 +95,37 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv if (decoder == null) { check(frame.key) val format = frame.codec.decoderFormat(frame.width, frame.height, frame.rotation, frame.config) - val name = MediaCodecList(MediaCodecList.REGULAR_CODECS).findDecoderForFormat(format) - ?: throw UnsupportedCodec("Unsupported decoder ${frame.codec.label} ${frame.width}x${frame.height}") - decoder = MediaCodec.createByCodecName(name) - val lowLatency = Build.VERSION.SDK_INT >= 30 && decoder!!.codecInfo.getCapabilitiesForType(frame.codec.mime) - .isFeatureSupported(MediaCodecInfo.CodecCapabilities.FEATURE_LowLatency) - if (lowLatency) format.setInteger(MediaFormat.KEY_LOW_LATENCY, 1) - LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=$name " + - "config=${frame.config.size} lowLatency=$lowLatency revision=$videoRevision") - decoder!!.configure(format, surface, null, 0); decoder!!.start() + val codecs = MediaCodecList(MediaCodecList.REGULAR_CODECS) + val preferred = codecs.findDecoderForFormat(format) + val candidates = codecs.codecInfos.filter { !it.isEncoder && frame.codec.mime in it.supportedTypes } + .sortedBy { if (it.name == preferred) 0 else 1 } + if (candidates.isEmpty()) throw UnsupportedCodec("Unsupported decoder ${frame.codec.label}") + var failure: Exception? = null + for (candidate in candidates) { + var attempt: MediaCodec? = null + try { + val capabilities = candidate.getCapabilitiesForType(frame.codec.mime) + if (!capabilities.isFormatSupported(format)) continue + val lowLatency = Build.VERSION.SDK_INT >= 30 && + capabilities.isFeatureSupported(MediaCodecInfo.CodecCapabilities.FEATURE_LowLatency) + if (Build.VERSION.SDK_INT >= 30) format.setInteger(MediaFormat.KEY_LOW_LATENCY, if (lowLatency) 1 else 0) + attempt = MediaCodec.createByCodecName(candidate.name) + attempt.configure(format, surface, null, 0); attempt.start(); decoder = attempt + LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=${candidate.name} " + + "config=${frame.config.size} lowLatency=$lowLatency revision=$videoRevision") + break + } catch (e: Exception) { + failure = e + LogManager.addLog("WARN", "video", "viewer decoder unavailable id=$id codec=${candidate.name}: ${e.message}") + try { attempt?.release() } + catch (release: Exception) { LogManager.addLog("WARN", "video", "decoder fallback release: ${release.message}") } + if (!surface.isValid) throw e + } + } + if (decoder == null) { + if (failure == null) throw UnsupportedCodec("Unsupported decoder format ${frame.codec.label} ${frame.width}x${frame.height}") + throw IllegalStateException("No available ${frame.codec.label} decoder", failure) + } } val index = decoder!!.dequeueInputBuffer(0) if (index < 0) false else { @@ -145,10 +162,10 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv fun observeStatus(meta: LongArray?) { if (meta == null || meta[5] and 1 == 0L) return ready = meta[0] - if (playback != 0L) return + if (playback != null) return audioEnabled = meta[5] and 2 != 0L - playback = NativeLib.streamPlaybackCreate(id, meta[3], audioEnabled) - check(playback != 0L); position = meta[3]; generation = meta[4] + playback = StreamMixedPlayback(id, meta[3], audioEnabled) + position = meta[3]; generation = meta[4] LogManager.addLog("INFO", "stream", "playback starts id=$id PTS=$position generation=$generation fixedPrefill=0") } @@ -160,11 +177,9 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv mutableStatus.value = R.string.stream_waiting_key } fun resetTimeline(request: Boolean) { - try { audio.close() } + try { playback?.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") } - audio = audioOutput() - if (playback != 0L) NativeLib.streamPlaybackClose(playback) - playback = 0; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false + playback = null; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false resetVideo(); deadline = System.nanoTime() + 10_000_000_000 if (request) check(NativeLib.streamResync(id)) { "Media resync refused" } } @@ -174,7 +189,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv while (running && surface.isValid) { check(NativeLib.isReady()) { "Core stopped" } observeStatus(NativeLib.streamStatus(id)) - val deviceLead = audio.leadUs + val deviceLead = playback?.leadUs ?: 0 for (work in 0 until 4) { video.step(position, System.nanoTime() / 1000) val audioAhead = if (audioEnabled && needsAudio) deviceLead + 80000 else -1L @@ -189,7 +204,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv val codecId = b.get().toInt() and 255 check(b.get() == 0.toByte() && b.get() == 0.toByte() && b.get() == 0.toByte()) check(size == b.remaining()) - if (kind in 1..2 && playback == 0L) observeStatus(NativeLib.streamStatus(id)) + if (kind in 1..2 && playback == null) observeStatus(NativeLib.streamStatus(id)) when (kind) { 1 -> { if (revision != videoRevision) { @@ -211,14 +226,14 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv System.nanoTime() / 1000) } } - 2 -> if (audioEnabled && playback != 0L) { + 2 -> if (audioEnabled && playback != null) { audioInputs++; lastAudioPts = pts - val result = NativeLib.streamPlaybackFeed(playback, pts, ByteArray(size).also { b.get(it) }) + val result = playback!!.feed(pts, ByteArray(size).also { b.get(it) }) if (result == -2) resetTimeline(true) else check(result >= 0) { "Opus feed failed=$result" } } 3 -> { LogManager.addLog("INFO", "stream", "playback GAP id=$id"); resetTimeline(false) } 4 -> opened = true - 5 -> { ended = true; if (playback != 0L) NativeLib.streamPlaybackFinish(playback) } + 5 -> { ended = true; playback?.finish() } 6 -> { mutableStatus.value = when (flags) { 1 -> R.string.stream_group_closed; 2 -> R.string.chat_service_stopped @@ -236,17 +251,12 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv LogManager.addLog("INFO", "video", "viewer first decoded frame id=$id revision=$videoRevision " + "PTS=${video.firstOutputPts} inputs=${video.inputCount} decoderLeadUs=${video.decoderLeadUs}") } - if (playbackStarted && playback != 0L && ready >= 0) { + if (playbackStarted && playback != null && ready >= 0) { val lateVideo = !video.ended && (video.firstOutputPts == null || (video.oldestSubmittedPts >= 0 && video.oldestSubmittedPts < position - 10000)) - if (audioEnabled) audio.step(System.nanoTime(), lateVideo, - pull = { NativeLib.streamPlaybackPull(playback, 480, if (ended) Long.MAX_VALUE else ready) }, - finished = { ended && NativeLib.streamPlaybackEmpty(playback) }) - val head = audio.played - val audioEmpty = audioEnabled && head >= audio.written && !ended - clockState = NativeLib.streamPlaybackUpdate(playback, System.nanoTime() / 1000, head, + clockState = playback!!.update(System.nanoTime() / 1000, if (ended) ready.coerceAtLeast(position) else ready, - maxOf(audio.leadUs + 10000, video.decoderLeadUs + video.framePeriodUs), lateVideo || audioEmpty) + maxOf(playback!!.leadUs + 10000, video.decoderLeadUs + video.framePeriodUs), lateVideo) check(clockState != null) { "Playback timeline failed" } position = clockState!![0]; needsAudio = clockState!![5] != 0L } @@ -260,8 +270,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv video.step(position, System.nanoTime() / 1000); video.checkProgress(System.nanoTime() / 1000, ended) check(!video.ended || video.firstOutputPts != null || video.inputCount == 0L) { "Decoder ended without producing an image" } } - if (ended && video.input == null && (decoder == null || video.ended) && NativeLib.streamPlaybackEmpty(playback) - && audio.drained) { + if (ended && video.input == null && (decoder == null || video.ended) && playback?.empty != false + && playback?.drained != false) { if (endUntil == 0L) endUntil = System.nanoTime() + 100_000_000 if (System.nanoTime() >= endUntil) { mutableStatus.value = R.string.stream_ended; break } } @@ -272,7 +282,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv "output=${video.output?.pts} inflight=${video.inflight} decoderLeadUs=${video.decoderLeadUs} " + "clock=$position ready=$ready audio=$audioEnabled audioInputs=$audioInputs audioPTS=$lastAudioPts needsAudio=$needsAudio " + "reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} " + - "tempo=${clockState?.get(3)} ${audio.description} waitingKey=$waitingKey") + "tempo=${clockState?.get(3)} ${playback?.description} waitingKey=$waitingKey") } Thread.sleep(5) } @@ -285,15 +295,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv NativeLib.streamStop(id) try { decoder?.stop(); decoder?.release() } catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") } - try { audio.close() } + try { playback?.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") } - if (playback != 0L) NativeLib.streamPlaybackClose(playback) - val until = System.nanoTime() + 500_000_000 - while (NativeLib.isReady() && System.nanoTime() < until) { - val event = NativeLib.streamPull(id, wantVideo = false, audioAheadUs = -1) - if (event != null && event[0].toInt() in 5..6) break - Thread.sleep(5) - } running = false LogManager.addLog("INFO", "stream", "viewer released id=$id inputs=${video.inputCount} " + "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded}") diff --git a/tools/chatgui-android/jni_bridge/android_stream_jni.c b/tools/chatgui-android/jni_bridge/android_stream_jni.c index 173e1a18..43de0118 100644 --- a/tools/chatgui-android/jni_bridge/android_stream_jni.c +++ b/tools/chatgui-android/jni_bridge/android_stream_jni.c @@ -1,4 +1,5 @@ -/* JNI live media: session IDs для ядра, codec handles принадлежат ровно одному audio worker. */ +/* JNI live media: session IDs для ядра. Playback handles сериализуются своим StreamMixedPlayback lock + * между video worker и общим микшером; capture Opus handles принадлежат одному audio worker. */ #ifdef __ANDROID__ #include #include @@ -179,6 +180,15 @@ JNIEXPORT jlongArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlayb return output; } +JNIEXPORT void JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackSilence( + JNIEnv *env, jobject obj, jlong handle, jint frames) { + (void)env; (void)obj; struct android_stream_playback *p = (struct android_stream_playback *)(intptr_t)handle; + if (!p || !p->audio || frames < 0 || frames > 8192) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI playback silence invalid frames=%d", frames); return; + } + stream_playback_audio_silence(p->audio, (unsigned)frames); +} + JNIEXPORT void JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackFinish(JNIEnv *env, jobject obj, jlong handle) { (void)env; (void)obj; struct android_stream_playback *p = (struct android_stream_playback *)(intptr_t)handle; if (p) { p->finished = 1; stream_playback_audio_finish(p->audio); }