Run Android players concurrently and fall back between decoders

2026-10-20 16:58:41 +02:00
1 parent 4e386fc8e3
commit 75806249bd
@@ -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)
@@ -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<StreamPlayerEngine>() // Только 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<Frame>
        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}")
@@ -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 <jni.h>
 #include <stdint.h>
@@ -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); }