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 f5c603b6..c0f5c2f1 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 @@ -1,10 +1,5 @@ package com.utun.chat.data -import android.media.MediaCodec -import android.media.MediaCodecInfo -import android.media.MediaCodecList -import android.media.MediaFormat -import android.os.Build import android.os.Handler import android.os.Looper import android.view.Surface @@ -16,8 +11,8 @@ import kotlinx.coroutines.withContext import java.nio.ByteBuffer import java.nio.ByteOrder -/** Запас кодированного потока остаётся в общем RS-кольце. Worker запрашивает ближайшие кадры, - * общий native controller ведёт минутную историю и темп, AudioTrack задаёт фактические часы. +/** Audio worker читает независимые треки из RS-кольца и готовит PCM; MediaCodec работает отдельно. + * Общий native controller убирает избыточный запас ускорением, AudioTrack задаёт часы каждого источника. * Временная нехватка сохраняет decoder references; GAP/вытеснение начинают новую точку KEY. */ class StreamPlayerEngine(private val group: Long, private val source: Long, private val epoch: Long) { companion object { @@ -42,10 +37,6 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv LogManager.addLog("INFO", "stream", "viewers released for call count=${stopped.size}") } } - private class UnsupportedCodec(message: String) : IllegalStateException(message) - private data class Frame(override val pts: Long, val key: Boolean, val width: Int, val height: Int, - val rotation: Int, val revision: Long, val codec: StreamVideoCodec, - val config: ByteArray, val data: ByteArray) : StreamDecodePump.Frame private val mutableAspect = MutableStateFlow(0f) // Ориентация неизвестна до первого KEY с форматом. val aspect: StateFlow = mutableAspect private val mutableStatus = MutableStateFlow(R.string.stream_connecting) @@ -80,7 +71,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv catch (e: Exception) { LogManager.addLog("WARN", "stream", "viewer surface release id=$id: ${e.message}") } main.post { players.remove(this); mutableViewing.value = players.isNotEmpty() } } - }, "stream-player").apply { start() } + }, "stream-audio-$id").apply { start() } } /** Surface/SurfaceTexture освобождаются callback только после остановки decoder и native DSP. */ fun stop(onReleased: (() -> Unit)? = null) { @@ -94,9 +85,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv } private fun play(surface: Surface) { - var decoder: MediaCodec? = null var playback: StreamMixedPlayback? = null - lateinit var video: StreamDecodePump + var video: StreamVideoPlayback? = null var waitingKey = true var audioEnabled = false var opened = false @@ -115,130 +105,15 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv var lastAudioPts = -1L var generation = 0L var playbackStarted = false - var eosQueued = false var endUntil = 0L var reported = 0L + var workMaxUs = 0L var deadline = System.nanoTime() + 10_000_000_000 - val info = MediaCodec.BufferInfo() var clockState: LongArray? = null - var firstInputAtUs = 0L - var firstShown = false - var lastShownPts = -1L - - fun createVideoPump() = StreamDecodePump( - submit = { frame -> - if (decoder == null) { - val openingAtUs = System.nanoTime() / 1000 - check(frame.key) - val format = frame.codec.decoderFormat(frame.width, frame.height, frame.rotation, frame.config) - 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 - var firstRendered = true - val codecRevision = frame.revision - try { - attempt.setOnFrameRenderedListener({ codec, pts, renderedNs -> - if (firstRendered) { - firstRendered = false - val callbackUs = System.nanoTime() / 1000 - val renderedUs = renderedNs / 1000 - val renderClockValid = renderedUs in requestedAtUs..callbackUs - if (!renderClockValid) LogManager.addLog("WARN", "stream", - "viewer surface timestamp outside request interval id=$id PTS=$pts renderedNs=$renderedNs " + - "requestedUs=$requestedAtUs callbackUs=$callbackUs; using callback time") - LogManager.addLog("INFO", "stream", "android first surface frame id=$id revision=$codecRevision " + - "stage=surface_callback PTS=$pts elapsedUs=${callbackUs - requestedAtUs} " + - "renderElapsedUs=${if (renderClockValid) renderedUs - requestedAtUs else -1} " + - "callbackDelayUs=${if (renderClockValid) callbackUs - renderedUs else -1}") - try { codec.setOnFrameRenderedListener(null, null) } - catch (e: Exception) { - LogManager.addLog("WARN", "stream", "viewer surface diagnostics detach id=$id: ${e.message}") - } - } - }, main) - } catch (e: Exception) { - LogManager.addLog("WARN", "stream", "viewer presentation diagnostics unavailable id=$id: ${e.message}") - } - LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=${candidate.name} " + - "config=${frame.config.size} lowLatency=$lowLatency revision=$videoRevision " + - "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} openUs=${System.nanoTime() / 1000 - openingAtUs}") - 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 { - val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input") - check(frame.data.size <= input.capacity()) { "Decoder input is too small: ${input.capacity()}/${frame.data.size}" } - input.clear(); input.put(frame.data) - if (firstInputAtUs == 0L) firstInputAtUs = System.nanoTime() / 1000 - decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0); true - } - }, - receive = { - var output: StreamDecodePump.Output? = null - val codec = decoder - if (codec != null) { - for (attempt in 0 until 3) { - val index = codec.dequeueOutputBuffer(info, 0) - if (index == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED) { - LogManager.addLog("INFO", "video", "viewer output format id=$id revision=$videoRevision ${codec.outputFormat}") - continue - } - if (index >= 0) { - val eos = info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0 - output = StreamDecodePump.Output(index, info.presentationTimeUs, eos, eos && info.size == 0) - if (!output.empty && video.firstOutputPts == null) { - val decodedAtUs = System.nanoTime() / 1000 - LogManager.addLog("INFO", "video", "viewer first decoded frame id=$id revision=$videoRevision " + - "PTS=${output.pts} inputs=${video.inputCount} decoderLeadUs=${maxOf(0, video.lastInputPts - output.pts)} " + - "elapsedUs=${decodedAtUs - requestedAtUs} decodeUs=${decodedAtUs - firstInputAtUs}") - } - } - break - } - } - output - }, - release = { output, render -> - if (render && !firstShown) { - firstShown = true - LogManager.addLog("INFO", "stream", "android first frame id=$id stage=display_submit " + - "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} PTS=${output.pts} clockUs=$position " + - "readyUs=$ready readyAheadUs=${ready - output.pts} avUs=${output.pts - position} " + - "deviceLeadUs=${playback?.leadUs}") - } - decoder!!.releaseOutputBuffer(output.index, render) - if (render) { lastShownPts = output.pts; mutableStatus.value = R.string.stream_live } - } - ) fun observeStatus(meta: LongArray?) { if (meta == null || meta[5] and 1 == 0L) return - ready = meta[0] + ready = if (meta[5] and 2 != 0L && meta[5] and 16 == 0L) meta[2] else maxOf(meta[1], meta[2]) if (playback != null) return audioEnabled = meta[5] and 2 != 0L playback = StreamMixedPlayback(id, meta[3], audioEnabled) { NativeLib.streamSignalNotify(signal) } @@ -249,33 +124,34 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv } fun resetVideo() { - try { decoder?.stop(); decoder?.release() } - catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder reset id=$id: ${e.message}") } - decoder = null; video = createVideoPump(); firstInputAtUs = 0 - waitingKey = true; eosQueued = false + video?.reset(); waitingKey = true mutableStatus.value = R.string.stream_waiting_key } fun resetTimeline(request: Boolean) { try { playback?.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") } playback = null; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false - records.clear(); ended = false; videoInputEnded = false; audioInputEnded = false + records.clear(); ended = false; videoInputEnded = false; audioInputEnded = false; endUntil = 0 resetVideo(); deadline = System.nanoTime() + 10_000_000_000 if (request) check(NativeLib.streamResync(id)) { "Media resync refused" } } - video = createVideoPump() try { check(NativeLib.streamOpen(id, group, source, audio = true, epoch = epoch)) { "Viewer open refused" } signal = NativeLib.streamSignalCreate(id).also { check(it != 0L) { "Viewer notification initialization failed" } } + video = StreamVideoPlayback(id, surface, requestedAtUs, + wake = { NativeLib.streamSignalNotify(signal) }, onRendered = { if (running) mutableStatus.value = R.string.stream_live }) + LogManager.addLog("INFO", "stream", "viewer audio worker started id=$id preparedPcmMs=250 audioAheadUs=500000") + while (running && surface.isValid) { check(NativeLib.isReady()) { "Core stopped" } + video.failure?.let { throw it } + val workStarted = System.nanoTime() / 1000 observeStatus(NativeLib.streamStatus(id)) drainTimeoutMs = ((ready - position) / 1000 + 2000).coerceIn(2000, 30000) val deviceLead = playback?.leadUs ?: 0 var consumed = 0 for (work in 0 until 32) { if (!running) break - video.step(position, System.nanoTime() / 1000) needsAudio = !audioInputEnded && (playback?.needsInput ?: true) val audioAhead = if (!audioInputEnded && (!audioEnabled || needsAudio) && records.audios == 0) deviceLead + 500000 else -1L val videoBudget = if (!videoInputEnded && video.wantsInput && records.videos == 0) 1 else 0 @@ -311,8 +187,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv LogManager.addLog("INFO", "video", "viewer KEY accepted id=$id revision=$revision PTS=$pts bytes=$size " + "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} readyAheadUs=${ready - pts}") } - video.offer(Frame(pts, key, width, height, broadcastRotation, revision, codec, access.first, access.second), - System.nanoTime() / 1000) + video.offer(StreamVideoPlayback.Frame(pts, key, width, height, broadcastRotation, revision, codec, access.first, access.second)) } } 2 -> if (audioEnabled && playback != null) { @@ -323,7 +198,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv 3 -> { LogManager.addLog("INFO", "stream", "playback GAP id=$id"); resetTimeline(false) } 4 -> opened = true 5, 7, 8 -> { - if (kind != 8) videoInputEnded = true + if (kind != 8 && !videoInputEnded) { videoInputEnded = true; video.finish() } if (kind != 7 && !audioInputEnded) { audioInputEnded = true; playback?.finish() } if (kind == 5) ended = true LogManager.addLog("INFO", "stream", "viewer track end id=$id kind=$kind videoEnd=$videoInputEnded " + @@ -343,7 +218,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv } if (!running) break if (!ended && (!opened || waitingKey) && System.nanoTime() > deadline) error("Viewer open timeout") - if (!playbackStarted && video.firstOutputPts != null) playbackStarted = true + val decoded = video.state + if (!playbackStarted && decoded.firstOutputPts != null) playbackStarted = true if (playbackStarted && playback != null && ready >= 0) { clockState = playback!!.update(System.nanoTime() / 1000, if (ended) ready.coerceAtLeast(position) else ready, @@ -352,54 +228,43 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv position = clockState!![0]; needsAudio = clockState!![5] != 0L } check((playback?.prepare() ?: 0) >= 0) { "PCM preparation failed" } - val codec = decoder - if (codec != null) { - if (videoInputEnded && video.input == null && video.output == null && !eosQueued) { - val index = codec.dequeueInputBuffer(0) - if (index >= 0) { codec.queueInputBuffer(index, 0, 0, video.lastInputPts.coerceAtLeast(0), - MediaCodec.BUFFER_FLAG_END_OF_STREAM); eosQueued = true } - } - video.step(position, System.nanoTime() / 1000); video.checkProgress(System.nanoTime() / 1000, videoInputEnded) - check(!video.ended || video.firstOutputPts != null || video.inputCount == 0L) { "Decoder ended without producing an image" } - } - if (ended && video.input == null && (decoder == null || video.ended) && playback?.empty != false + video.update(position, ready, playback?.leadUs ?: 0) + workMaxUs = maxOf(workMaxUs, System.nanoTime() / 1000 - workStarted) + if (ended && decoded.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 } } if (System.nanoTime() - reported >= 1_000_000_000) { reported = System.nanoTime() - LogManager.addLog("DEBUG", "stream", "viewer id=$id revision=$videoRevision inputs=${video.inputCount} " + - "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded} input=${video.input != null} " + - "output=${video.output?.pts} inflight=${video.inflight} decoderLeadUs=${video.decoderLeadUs} " + + LogManager.addLog("DEBUG", "stream", "android audio worker id=$id generation=$generation revision=$videoRevision " + "clock=$position ready=$ready audio=$audioEnabled audioInputs=$audioInputs audioPTS=$lastAudioPts needsAudio=$needsAudio " + - "reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} " + - "tempo=${clockState?.get(3)} ${playback?.description} waitingKey=$waitingKey " + - "shownPTS=$lastShownPts displayLagUs=${if (lastShownPts >= 0) position - lastShownPts else -1} " + + "reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} tempo=${clockState?.get(3)} " + + "${playback?.description} workMaxUs=$workMaxUs waitingKey=$waitingKey " + "batches=$batches batchMax=$batchMax pending=${records.size} videoEnd=$videoInputEnded audioEnd=$audioInputEnded") - batches = 0; batchMax = 0 + batches = 0; batchMax = 0; workMaxUs = 0 } - // Синхронный MediaCodec требует проверки готовности; media/PCM/stop будят без ожидания deadline. + // PCM consumption, media и video state будят worker; без аудио часы идут по wall clock. if (running && consumed < 32) { - val codecPending = video.input != null || video.output != null || video.inflight > 0 || (videoInputEnded && !video.ended) - check(NativeLib.streamSignalWait(signal, if (codecPending) 5 else 50) >= 0) { "Viewer notification wait failed" } + check(NativeLib.streamSignalWait(signal, if (playbackStarted && !audioEnabled) 5 else 50) >= 0) { + "Viewer notification wait failed" + } } } } catch (e: Exception) { if (running) { - mutableStatus.value = if (e is UnsupportedCodec) R.string.stream_decoder_unsupported else R.string.stream_failed + mutableStatus.value = if (e is StreamVideoPlayback.UnsupportedCodec) R.string.stream_decoder_unsupported else R.string.stream_failed LogManager.addLog("ERROR", "stream", "viewer failed id=$id: ${e.message}") } } finally { NativeLib.streamStop(id) - try { decoder?.stop(); decoder?.release() } + try { video?.close() } catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") } try { playback?.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") } NativeLib.streamSignalClose(signal) running = false - LogManager.addLog("INFO", "stream", "viewer released id=$id inputs=${video.inputCount} " + - "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded}") + LogManager.addLog("INFO", "stream", "viewer released id=$id video=${video?.state}") } } } diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamVideoPlayback.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamVideoPlayback.kt new file mode 100644 index 00000000..35c8c49d --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamVideoPlayback.kt @@ -0,0 +1,273 @@ +package com.utun.chat.data + +import android.media.MediaCodec +import android.media.MediaCodecInfo +import android.media.MediaCodecList +import android.media.MediaFormat +import android.os.Build +import android.os.Handler +import android.os.Looper +import android.view.Surface +import java.util.ArrayDeque + +/** MediaCodec и pump принадлежат только video worker. Audio worker передаёт кадры через ограниченный + * mailbox; reset меняет поколение, close дожидается release до освобождения Surface/notification. + * Ни один вызов MediaCodec не выполняется под mailbox lock или на потоке подготовки звука. */ +internal class StreamVideoPlayback( + private val id: Long, private val surface: Surface, private val requestedAtUs: Long, + private val wake: () -> Unit, private val onRendered: () -> Unit +) { + class UnsupportedCodec(message: String) : IllegalStateException(message) + data class Frame(override val pts: Long, val key: Boolean, val width: Int, val height: Int, + val rotation: Int, val revision: Long, val codec: StreamVideoCodec, + val config: ByteArray, val data: ByteArray) : StreamDecodePump.Frame + data class State( + val firstOutputPts: Long? = null, val ended: Boolean = false, val inputs: Long = 0, + val decoded: Long = 0, val shown: Long = 0, val discarded: Long = 0, val inflight: Int = 0, + val decoderLeadUs: Long = 0, val shownPts: Long = -1, val outputPts: Long? = null + ) + private data class Clock(val position: Long = -1, val ready: Long = -1, val leadUs: Long = 0) + private val lock = Object() + private val input = ArrayDeque() + @Volatile private var running = true + @Volatile private var generation = 0L + @Volatile private var clock = Clock() + @Volatile var state = State() + private set + @Volatile var failure: Exception? = null + private set + private var inputEnded = false // mailbox lock + private var activeGeneration = -1L // Далее поля только video worker. + private var decoder: MediaCodec? = null + private var video: StreamDecodePump = createVideoPump() + private val info = MediaCodec.BufferInfo() + private var videoRevision = 0L + private var firstInputAtUs = 0L + private var firstShown = false + private var lastShownPts = -1L + private var eosQueued = false + private val main = Handler(Looper.getMainLooper()) + private val position: Long get() = clock.position + private val ready: Long get() = clock.ready + private val worker = Thread(::run, "stream-video-$id").apply { start() } + + val wantsInput: Boolean get() = synchronized(lock) { running && !inputEnded && input.size < 2 } + + fun offer(frame: Frame) = synchronized(lock) { + check(running && !inputEnded && input.size < 2) { "Video mailbox full or closed id=$id size=${input.size}" } + input.addLast(frame); lock.notifyAll() + } + + fun update(position: Long, ready: Long, leadUs: Long) { + clock = Clock(position, ready, leadUs) + synchronized(lock) { lock.notifyAll() } + } + + fun finish() = synchronized(lock) { inputEnded = true; lock.notifyAll() } + + fun reset() = synchronized(lock) { + generation++; input.clear(); inputEnded = false; state = State(); clock = Clock() + lock.notifyAll() + LogManager.addLog("INFO", "video", "viewer video reset id=$id generation=$generation") + } + + fun close() { + synchronized(lock) { running = false; lock.notifyAll() } + worker.join() + } + + private fun releaseDecoder() { + try { decoder?.stop() } + catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder stop id=$id: ${e.message}") } + try { decoder?.release() } + catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") } + decoder = null + } + + private fun run() { + var reported = 0L + var workMaxUs = 0L + try { + LogManager.addLog("INFO", "video", "viewer video worker started id=$id mailbox=2") + while (running) { + val requestedGeneration = generation + if (activeGeneration != requestedGeneration) { + releaseDecoder(); video = createVideoPump(); firstInputAtUs = 0 + eosQueued = false; videoRevision = 0; lastShownPts = -1 + activeGeneration = requestedGeneration + } + val started = System.nanoTime() / 1000 + video.step(position, started) + val frame = synchronized(lock) { + if (activeGeneration == generation && video.wantsInput) input.pollFirst() else null + } + if (frame != null) { + if (videoRevision != frame.revision) { + check(frame.key) { "Video revision does not start at KEY" } + releaseDecoder(); video = createVideoPump(); firstInputAtUs = 0 + eosQueued = false; videoRevision = frame.revision + } + video.offer(frame, started); video.step(position, System.nanoTime() / 1000) + } + val finishing = synchronized(lock) { activeGeneration == generation && inputEnded && input.isEmpty() } + val codec = decoder + if (finishing && codec != null && video.input == null && video.output == null && !eosQueued) { + val index = codec.dequeueInputBuffer(0) + if (index >= 0) { + codec.queueInputBuffer(index, 0, 0, video.lastInputPts.coerceAtLeast(0), MediaCodec.BUFFER_FLAG_END_OF_STREAM) + eosQueued = true + } + } + video.step(position, System.nanoTime() / 1000) + video.checkProgress(System.nanoTime() / 1000, finishing) + check(!video.ended || video.firstOutputPts != null || video.inputCount == 0L) { "Decoder ended without producing an image" } + val next = State(video.firstOutputPts, video.ended || (finishing && codec == null), + video.inputCount, video.outputCount, video.shown, video.discarded, video.inflight, + video.decoderLeadUs, lastShownPts, video.output?.pts) + val changed = synchronized(lock) { + if (activeGeneration == generation && state != next) { state = next; true } else false + } + if (changed || frame != null) wake() + val now = System.nanoTime() / 1000 + workMaxUs = maxOf(workMaxUs, now - started) + if (now - reported >= 1_000_000) { + reported = now + LogManager.addLog("DEBUG", "stream", "android video worker id=$id generation=$activeGeneration revision=$videoRevision " + + "inputs=${video.inputCount} decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded} " + + "inflight=${video.inflight} decoderLeadUs=${video.decoderLeadUs} workMaxUs=$workMaxUs " + + "clock=$position shownPTS=$lastShownPts displayLagUs=${if (lastShownPts >= 0) position - lastShownPts else -1}") + workMaxUs = 0 + } + synchronized(lock) { + if (running && activeGeneration == generation && (input.isEmpty() || !video.wantsInput)) { + val pending = video.input != null || video.output != null || video.inflight > 0 || (finishing && !next.ended) + lock.wait(if (pending) 5 else 50) + } + } + } + } catch (e: Exception) { + if (running) { + failure = e + LogManager.addLog("ERROR", "video", "viewer video worker failed id=$id generation=$activeGeneration: ${e.message}") + wake() + } + } finally { + releaseDecoder() + LogManager.addLog("INFO", "video", "viewer video worker released id=$id") + } + } + + private fun createVideoPump() = StreamDecodePump( + submit = { frame -> + if (decoder == null) { + val openingAtUs = System.nanoTime() / 1000 + check(frame.key) + val format = frame.codec.decoderFormat(frame.width, frame.height, frame.rotation, frame.config) + 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 + var firstRendered = true + val codecRevision = frame.revision + try { + attempt.setOnFrameRenderedListener({ codec, pts, renderedNs -> + if (firstRendered) { + firstRendered = false + val callbackUs = System.nanoTime() / 1000 + val renderedUs = renderedNs / 1000 + val renderClockValid = renderedUs in requestedAtUs..callbackUs + if (!renderClockValid) LogManager.addLog("WARN", "stream", + "viewer surface timestamp outside request interval id=$id PTS=$pts renderedNs=$renderedNs " + + "requestedUs=$requestedAtUs callbackUs=$callbackUs; using callback time") + LogManager.addLog("INFO", "stream", "android first surface frame id=$id revision=$codecRevision " + + "stage=surface_callback PTS=$pts elapsedUs=${callbackUs - requestedAtUs} " + + "renderElapsedUs=${if (renderClockValid) renderedUs - requestedAtUs else -1} " + + "callbackDelayUs=${if (renderClockValid) callbackUs - renderedUs else -1}") + try { codec.setOnFrameRenderedListener(null, null) } + catch (e: Exception) { + LogManager.addLog("WARN", "stream", "viewer surface diagnostics detach id=$id: ${e.message}") + } + } + }, main) + } catch (e: Exception) { + LogManager.addLog("WARN", "stream", "viewer presentation diagnostics unavailable id=$id: ${e.message}") + } + LogManager.addLog("INFO", "video", "viewer codec id=$id type=${frame.codec.label} decoder=${candidate.name} " + + "config=${frame.config.size} lowLatency=$lowLatency revision=$videoRevision " + + "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} openUs=${System.nanoTime() / 1000 - openingAtUs}") + 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 { + val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input") + check(frame.data.size <= input.capacity()) { "Decoder input is too small: ${input.capacity()}/${frame.data.size}" } + input.clear(); input.put(frame.data) + if (firstInputAtUs == 0L) firstInputAtUs = System.nanoTime() / 1000 + decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0); true + } + }, + receive = { + var output: StreamDecodePump.Output? = null + val codec = decoder + if (codec != null) { + for (attempt in 0 until 3) { + val index = codec.dequeueOutputBuffer(info, 0) + if (index == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED) { + LogManager.addLog("INFO", "video", "viewer output format id=$id revision=$videoRevision ${codec.outputFormat}") + continue + } + if (index >= 0) { + val eos = info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0 + output = StreamDecodePump.Output(index, info.presentationTimeUs, eos, eos && info.size == 0) + if (!output.empty && video.firstOutputPts == null) { + val decodedAtUs = System.nanoTime() / 1000 + LogManager.addLog("INFO", "video", "viewer first decoded frame id=$id revision=$videoRevision " + + "PTS=${output.pts} inputs=${video.inputCount} decoderLeadUs=${maxOf(0, video.lastInputPts - output.pts)} " + + "elapsedUs=${decodedAtUs - requestedAtUs} decodeUs=${decodedAtUs - firstInputAtUs}") + } + } + break + } + } + output + }, + release = { output, render -> + val current = running && activeGeneration == generation + if (render && current && !firstShown) { + firstShown = true + LogManager.addLog("INFO", "stream", "android first frame id=$id stage=display_submit " + + "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} PTS=${output.pts} clockUs=$position " + + "readyUs=$ready readyAheadUs=${ready - output.pts} avUs=${output.pts - position} " + + "deviceLeadUs=${clock.leadUs}") + } + decoder!!.releaseOutputBuffer(output.index, render && current) + if (render && current) { lastShownPts = output.pts; onRendered() } + } + ) + +} +