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 cb18de52..bf208bf6 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 @@ -207,7 +207,18 @@ object NativeLib { width: Int = 0, height: Int = 0, rotation: Int = 0, audio: Boolean = true): Int = nativeStreamFeed(id, video, pts, key, data, revision, width, height, rotation, audio) fun streamStop(id: Long) { nativeStreamStop(id) } - fun streamPull(id: Long): ByteArray? = nativeStreamPull(id) + fun streamPull(id: Long, positionUs: Long = -1, videoAheadUs: Long = 80000, audioAheadUs: Long = 80000): ByteArray? = + nativeStreamPull(id, positionUs, videoAheadUs, audioAheadUs) + fun streamStatus(id: Long): LongArray? = nativeStreamStatus(id) + fun streamResync(id: Long): Boolean = nativeStreamResync(id) + 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 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) } + fun streamPlaybackEmpty(handle: Long): Boolean = nativeStreamPlaybackEmpty(handle) + fun streamPlaybackClose(handle: Long) { nativeStreamPlaybackClose(handle) } fun streamOpusCreate(encode: Boolean): Long = nativeStreamOpusCreate(encode) fun streamOpusEncode(handle: Long, pcm: ShortArray, compressor: RadioCompressorSettings): ByteArray? = nativeStreamOpusEncode(handle, pcm, compressor.enabled, compressor.maxGainDb, compressor.riseRate) @@ -219,7 +230,17 @@ object NativeLib { private external fun nativeStreamFeed(id: Long, video: Boolean, pts: Long, key: Boolean, data: ByteArray, revision: Long, width: Int, height: Int, rotation: Int, audio: Boolean): Int private external fun nativeStreamStop(id: Long) - private external fun nativeStreamPull(id: Long): ByteArray? + private external fun nativeStreamPull(id: Long, position: Long, videoAhead: Long, audioAhead: Long): ByteArray? + private external fun nativeStreamStatus(id: Long): LongArray? + private external fun nativeStreamResync(id: Long): Boolean + 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 nativeStreamPlaybackUpdate(handle: Long, now: Long, played: Long, ready: Long, + guard: Long, starved: Boolean): LongArray? + private external fun nativeStreamPlaybackFinish(handle: Long) + private external fun nativeStreamPlaybackEmpty(handle: Long): Boolean + private external fun nativeStreamPlaybackClose(handle: Long) private external fun nativeStreamOpusCreate(encode: Boolean): Long private external fun nativeStreamOpusEncode(handle: Long, pcm: ShortArray, compressor: Boolean, maxGainDb: Int, riseRate: Int): ByteArray? private external fun nativeStreamOpusDecode(handle: Long, packet: ByteArray): ShortArray? 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 4e6bc19b..c8b7f429 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,15 @@ import java.nio.ByteOrder import java.util.ArrayDeque import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock -import kotlin.math.roundToInt -/** Один worker владеет декодерами и AudioTrack. Hardware audio clock управляет временем показа видео. - * Просроченные P-кадры декодируются без показа, чтобы сохранить references; при переполнении ждём KEY. - * viewerLock сериализует stop/open разных экранов до подтверждения освобождения native owner. */ +/** Запас кодированного потока остаётся в общем RS-кольце. Worker запрашивает ближайшие кадры, + * общий native controller ведёт минутную историю и темп, AudioTrack задаёт фактические часы. + * Временная нехватка сохраняет decoder references; GAP/вытеснение начинают новую точку KEY. */ class StreamPlayerEngine(private val group: Long, private val source: Long, private val epoch: Long, private val localPreview: StreamPreview? = null, private val displayDegrees: Int = 0) { companion object { private val viewerLock = ReentrantLock() } - private data class Frame(val pts: Long, val key: Boolean, val width: Int, val height: Int, val rotation: Int, val top: Int?, val revision: Long, val data: ByteArray) + private data class Frame(val pts: Long, val key: Boolean, val width: Int, val height: Int, + val rotation: Int, val top: Int?, val revision: Long, val data: ByteArray) private val mutableAspect = MutableStateFlow(16f / 9f) val aspect: StateFlow = mutableAspect private val mutableTop = MutableStateFlow(null) @@ -39,294 +39,238 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv running = true worker = Thread({ viewerLock.withLock { if (running) play(surface) } }, "stream-player").apply { start() } } - fun stop() { running = false; NativeLib.streamStop(id) } private fun play(surface: Surface) { var decoder: MediaCodec? = null var track: AudioTrack? = null - var opus = 0L - val videos = ArrayDeque(); val audios = ArrayDeque() - var videoBytes = 0 + var playback = 0L + val videos = ArrayDeque() + val submitted = ArrayDeque() var waitingKey = true - var audioBase = -1L - var written = 0L - var videoBase = -1L - var videoStartNs = 0L + var audioEnabled = false + var opened = false var ended = false - val audioTimestamp = AudioTimestamp() - val bufferInfo = MediaCodec.BufferInfo() - var report = 0L + var pausedAudio = false var videoRevision = 0L - var reportedDecoded = 0L + var position = -1L + var ready = -1L + var needsAudio = true + var written = 0L + var previousHead = 0L + var headEpoch = 0L + var generation = 0L var decoded = 0L var discarded = 0L - var lastHead = -1L - var headChangedNs = 0L - var endDrainUntil = 0L - var headEpoch = 0L - var previousRawHead = 0L - var prefillUntil = 0L - var firstAudioSkipFrames = 0 - var prerollFrames = 0L - var warmAudioPts = -1L - var connectDeadline = System.nanoTime() + 10_000_000_000 + var decodedThrough = -1L + var pendingOutput = -1 + var pendingPts = 0L + var eosQueued = false + var decoderEnded = false + var endUntil = 0L + var reported = 0L + var deadline = System.nanoTime() + 10_000_000_000 + val timestamp = AudioTimestamp() + val info = MediaCodec.BufferInfo() + var clockState: LongArray? = null fun resetVideo() { - try { decoder?.stop() } + try { decoder?.stop(); decoder?.release() } catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder reset id=$id: ${e.message}") } - try { decoder?.release() } - catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") } - decoder = null; videos.clear(); videoBytes = 0; waitingKey = true + decoder = null; videos.clear(); submitted.clear(); pendingOutput = -1; decodedThrough = -1 + waitingKey = true; eosQueued = false; decoderEnded = false mutableStatus.value = R.string.stream_waiting_key } - fun resetAudio() { - try { track?.pause(); track?.flush() } + fun resetTimeline(request: Boolean) { + try { track?.pause(); track?.flush(); track?.release() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") } - try { track?.release() } - catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") } - track = null - if (opus != 0L) NativeLib.streamOpusClose(opus) - opus = 0; audioBase = -1; written = 0; lastHead = -1; headEpoch = 0; previousRawHead = 0; audios.clear() - firstAudioSkipFrames = 0; prerollFrames = 0; warmAudioPts = -1 - } - fun resetTimeline() { - resetAudio(); resetVideo(); videoBase = -1; prefillUntil = 0 - connectDeadline = System.nanoTime() + 10_000_000_000 - } - // Прогрев допускает джиттер PTS и короткий PLC; большой разрыв начинает новое состояние Opus. - fun prepareAudio(pts: Long) { - if (warmAudioPts >= 0 && (pts < warmAudioPts - 10000 || pts > warmAudioPts + 100000)) { - LogManager.addLog("WARN", "stream", "preroll audio discontinuity id=$id expected=$warmAudioPts pts=$pts") - if (opus != 0L) NativeLib.streamOpusClose(opus) - opus = 0; warmAudioPts = -1 - } - if (opus == 0L) { opus = NativeLib.streamOpusCreate(false); check(opus != 0L) } - val missing = if (warmAudioPts < 0) 0 else ((pts - warmAudioPts + 10000) / 20000).toInt() - repeat(missing) { - check(NativeLib.streamOpusDecode(opus, byteArrayOf()) != null) { "Preroll Opus PLC failed" } - prerollFrames += 960 - } + track = null; pausedAudio = false; written = 0; previousHead = 0; headEpoch = 0 + if (playback != 0L) NativeLib.streamPlaybackClose(playback) + playback = 0; position = -1; generation = 0; needsAudio = true; clockState = null + resetVideo(); deadline = System.nanoTime() + 10_000_000_000 + if (request) check(NativeLib.streamResync(id)) { "Media resync refused" } } - // Отброшенный звук всё равно прогревает Opus; остающиеся пакеты декодируются один раз при воспроизведении. - fun warmAudio(frame: Frame) { - prepareAudio(frame.pts) - check(NativeLib.streamOpusDecode(opus, frame.data) != null) { "Preroll Opus decode failed" } - warmAudioPts = frame.pts + 20000; prerollFrames += 960 - } - // PlaybackHeadPosition — unsigned 32 bit: при 48kHz оборачивается примерно через 25 часов. - fun playedFrames(t: AudioTrack): Long { + fun playedFrames(now: Long): Long { + val t = track ?: return 0 val raw = t.playbackHeadPosition.toLong() and 0xffffffffL - if (raw < previousRawHead) headEpoch += 0x100000000L - previousRawHead = raw - return headEpoch + raw - } - fun currentPts(now: Long): Long { - val t = track - if (t != null && audioBase >= 0) { - val head = playedFrames(t) - val played = if (t.getTimestamp(audioTimestamp)) { - var position = audioTimestamp.framePosition - if (position <= 0xffffffffL) { - position += headEpoch - if (position - head > 0x80000000L) position -= 0x100000000L - if (head - position > 0x80000000L) position += 0x100000000L - } - position + (now - audioTimestamp.nanoTime).coerceAtLeast(0) * 48000 / 1_000_000_000 - } else { - if (head != lastHead) { lastHead = head; headChangedNs = now } - head + if (head >= written) (now - headChangedNs) * 48000 / 1_000_000_000 else 0 - } - return audioBase + played * 1_000_000 / 48000 + if (raw < previousHead) headEpoch += 0x100000000L + previousHead = raw + val head = headEpoch + raw + if (pausedAudio || !t.getTimestamp(timestamp)) return head.coerceAtMost(written) + var frames = timestamp.framePosition + if (frames <= 0xffffffffL) { + frames += headEpoch + if (frames - head > 0x80000000L) frames -= 0x100000000L + if (head - frames > 0x80000000L) frames += 0x100000000L } - return if (videoBase >= 0) videoBase + (now - videoStartNs) / 1000 else 0 + return (frames + (now - timestamp.nanoTime).coerceAtLeast(0) * 48000 / 1_000_000_000).coerceIn(0, written) } - fun writePcm(t: AudioTrack, pcm: ShortArray, initialOffset: Int = 0) { - val deadline = System.nanoTime() + 100_000_000 - var offset = initialOffset + fun writePcm(pcm: ShortArray) { + if (track == null) { + track = AudioTrack.Builder().setAudioAttributes(AudioAttributes.Builder().setUsage(AudioAttributes.USAGE_MEDIA) + .setContentType(AudioAttributes.CONTENT_TYPE_MOVIE).build()).setAudioFormat(AudioFormat.Builder() + .setSampleRate(48000).setChannelMask(AudioFormat.CHANNEL_OUT_STEREO).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() + check(track!!.state == AudioTrack.STATE_INITIALIZED); track!!.play() + LogManager.addLog("INFO", "stream", "audio output id=$id bufferFrames=${track!!.bufferSizeInFrames} stereo=1") + } + val until = System.nanoTime() + 100_000_000 + var offset = 0 while (running && offset < pcm.size) { - val count = t.write(pcm, offset, pcm.size - offset, AudioTrack.WRITE_NON_BLOCKING) - check(count >= 0) { "AudioTrack write=$count" } - if (count > 0) { - check(count % 2 == 0) { "AudioTrack returned an incomplete stereo frame: samples=$count" } - offset += count; written += count / 2 - } - else { - check(System.nanoTime() < deadline) { "AudioTrack stalled for 100ms" } - Thread.sleep(2) - } + val count = track!!.write(pcm, offset, pcm.size - offset, AudioTrack.WRITE_NON_BLOCKING) + check(count >= 0 && count % 2 == 0) { "AudioTrack write=$count" } + if (count > 0) { offset += count; written += count / 2 } + else { check(System.nanoTime() < until) { "AudioTrack stalled" }; Thread.sleep(2) } } } try { - check(NativeLib.streamOpen(id, group, source, epoch = epoch)) { "Viewer open refused" } - var opened = false + check(NativeLib.streamOpen(id, group, source, audio = localPreview == null, epoch = epoch)) { "Viewer open refused" } while (running && surface.isValid) { - if (!NativeLib.isReady()) error("Core stopped") - repeat(64) { - val raw = if (!ended) NativeLib.streamPull(id) else null - if (raw != null) { - check(raw.size >= 24) - val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN) - val kind = b.get().toInt(); val flags = b.get().toInt() and 255; val key = flags and 1 != 0 - val width = b.short.toInt() and 65535; val height = b.short.toInt() and 65535 - val broadcastRotation = b.short.toInt() and 65535; val pts = b.long; val size = b.int - val revision = b.int.toLong() and 0xffffffffL - check(size == b.remaining()) - when (kind) { - 1, 2 -> { - val camera = if (kind == 1) localPreview?.camera(revision) else null - if (kind == 1 && localPreview != null && camera == null) { - LogManager.addLog("WARN", "video", "preview camera unavailable id=$id revision=$revision; wait KEY") - resetVideo(); return@repeat - } - val display = StreamVideoDisplay.forFrame(broadcastRotation, camera, displayDegrees) - val rotation = display.rotation - val frame = Frame(pts, key, width, height, rotation, display.top, revision, ByteArray(size).also { b.get(it) }) - if (kind == 1) { - if (revision != videoRevision) { - check(revision > videoRevision) { "Video revision regressed" } - LogManager.addLog("INFO", "video", "viewer configuration id=$id revision=$videoRevision->$revision ${width}x$height " + - "rotation=$rotation broadcastRotation=$broadcastRotation preview=${localPreview != null} top=${display.top}") - resetVideo(); videoRevision = revision - } - if (videos.size >= 128 || videoBytes + size > H264AnnexB.MAX_FRAME || - (videos.isNotEmpty() && pts - videos.first.pts > 250000)) { - LogManager.addLog("WARN", "video", "viewer video overflow id=$id frames=${videos.size} bytes=$videoBytes revision=$revision") - resetVideo() - } - if (!waitingKey || key) { - if (waitingKey) { - check(H264AnnexB.nals(frame.data).map { it.first }.toSet().containsAll(setOf(5, 7, 8))) - waitingKey = false - mutableAspect.value = if (rotation % 180 == 0) width.toFloat() / height else height.toFloat() / width - mutableTop.value = frame.top - } - if (videoBase < 0) { - while (audios.isNotEmpty() && audios.first.pts + 20000 <= pts) warmAudio(audios.removeFirst()) - if (audios.isNotEmpty() && audios.first.pts < pts) - firstAudioSkipFrames = ((pts - audios.first.pts) * 48000 + 999999).div(1000000).toInt() - prefillUntil = System.nanoTime() + 150_000_000 - videoBase = pts; videoStartNs = prefillUntil - LogManager.addLog("INFO", "stream", "viewer playback start id=$id pts=$pts prefillMs=150 " + - "prerollFrames=$prerollFrames skipFrames=$firstAudioSkipFrames audioQ=${audios.size}") - } - videos.addLast(frame); videoBytes += size - } - } else { - if (localPreview != null) return@repeat - if (audios.size >= 16) { - if (videoBase < 0) warmAudio(audios.removeFirst()) - else { - LogManager.addLog("WARN", "stream", "audio queue overflow id=$id; reset timeline") - resetTimeline() - } - } - audios.addLast(frame) - } + check(NativeLib.isReady()) { "Core stopped" } + val meta = NativeLib.streamStatus(id) + if (meta != null && meta[5] and 1 != 0L) { + ready = if (localPreview != null) meta[1] else meta[0] + if (playback == 0L) { + audioEnabled = localPreview == null && meta[5] and 2 != 0L + playback = NativeLib.streamPlaybackCreate(id, meta[3], audioEnabled) + check(playback != 0L); position = meta[3]; generation = meta[4] + LogManager.addLog("INFO", "stream", "playback starts id=$id PTS=$position generation=$generation fixedPrefill=0") + } + } + val now = System.nanoTime() + val played = playedFrames(now) + val deviceLead = (written - played).coerceAtLeast(0) * 1_000_000 / 48000 + val videoAhead = if (videos.size < 2 && submitted.size < 3) 80000L else -1L + val audioAhead = if (audioEnabled && needsAudio) deviceLead + 80000 else -1L + repeat(4) { + val raw = if (ended) null else NativeLib.streamPull(id, position, videoAhead, audioAhead) + if (raw == null) return@repeat + check(raw.size >= 24) + val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN) + val kind = b.get().toInt(); val flags = b.get().toInt() and 255 + val width = b.short.toInt() and 65535; val height = b.short.toInt() and 65535 + val broadcastRotation = b.short.toInt() and 65535; val pts = b.long; val size = b.int + val revision = b.int.toLong() and 0xffffffffL + check(size == b.remaining()) + when (kind) { + 1 -> { + val camera = localPreview?.camera(revision) + if (localPreview != null && camera == null) { + LogManager.addLog("WARN", "video", "preview camera unavailable id=$id revision=$revision") + resetTimeline(true); return@repeat + } + val display = StreamVideoDisplay.forFrame(broadcastRotation, camera, displayDegrees) + if (revision != videoRevision) { + check(revision > videoRevision); resetVideo(); videoRevision = revision + deadline = System.nanoTime() + 10_000_000_000 } - 3 -> { LogManager.addLog("INFO", "stream", "viewer GAP id=$id"); resetTimeline() } - 4 -> opened = true - 5 -> { ended = true; if (videoBase < 0) audios.clear() } - 6 -> { - mutableStatus.value = when (flags) { - 1 -> R.string.stream_group_closed - 2 -> R.string.chat_service_stopped - 3 -> R.string.stream_source_changed - 4 -> R.string.stream_member_removed - else -> R.string.stream_failed + val key = flags and 1 != 0 + if (!waitingKey || key) { + val bytes = ByteArray(size).also { b.get(it) } + if (waitingKey) { + check(H264AnnexB.nals(bytes).map { it.first }.toSet().containsAll(setOf(5, 7, 8))) + waitingKey = false; mutableTop.value = display.top + mutableAspect.value = if (display.rotation % 180 == 0) width.toFloat() / height else height.toFloat() / width } - LogManager.addLog("WARN", "stream", "native viewer closed id=$id group=$group source=$source epoch=$epoch reason=$flags") - ended = true; running = false + check(videos.size < 4) { "Video preparation queue exceeded" } + videos.addLast(Frame(pts, key, width, height, display.rotation, display.top, revision, bytes)) + } + } + 2 -> if (audioEnabled && playback != 0L) { + val result = NativeLib.streamPlaybackFeed(playback, 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) } + 6 -> { + mutableStatus.value = when (flags) { + 1 -> R.string.stream_group_closed; 2 -> R.string.chat_service_stopped + 3 -> R.string.stream_source_changed; 4 -> R.string.stream_member_removed + else -> R.string.stream_failed } - else -> error("Unknown media event=$kind") + LogManager.addLog("WARN", "stream", "native viewer closed id=$id reason=$flags"); running = false } + else -> error("Unknown media event=$kind") } } - if (!ended && (!opened || videoBase < 0) && System.nanoTime() > connectDeadline) error("Viewer open timeout") - val now = System.nanoTime() - if (prefillUntil != 0L && now >= prefillUntil) { - if (audios.isNotEmpty() && track == null) { - prepareAudio(audios.first.pts) - track = AudioTrack.Builder().setAudioAttributes(AudioAttributes.Builder().setUsage(AudioAttributes.USAGE_MEDIA) - .setContentType(AudioAttributes.CONTENT_TYPE_MOVIE).build()).setAudioFormat(AudioFormat.Builder() - .setSampleRate(48000).setChannelMask(AudioFormat.CHANNEL_OUT_STEREO).setEncoding(AudioFormat.ENCODING_PCM_16BIT).build()) - .setBufferSizeInBytes(AudioTrack.getMinBufferSize(48000, AudioFormat.CHANNEL_OUT_STEREO, AudioFormat.ENCODING_PCM_16BIT) - .coerceAtLeast(19200)).setTransferMode(AudioTrack.MODE_STREAM).build() - check(track!!.state == AudioTrack.STATE_INITIALIZED) - audioBase = audios.first.pts + firstAudioSkipFrames * 1_000_000L / 48000 - track!!.play() - LogManager.addLog("INFO", "stream", "audio playback starts id=$id PTS=$audioBase rate=48000 channels=${track!!.channelCount}") - } + if (!ended && (!opened || waitingKey) && System.nanoTime() > deadline) error("Viewer open timeout") + if (playback != 0L && audioEnabled && !pausedAudio) { + val head = playedFrames(System.nanoTime()) val t = track - if (t != null && audios.isNotEmpty()) { - val head = playedFrames(t) - if (written - head < 3840) { - val f = audios.removeFirst() - val pts = f.pts + firstAudioSkipFrames * 1_000_000L / 48000 - val missing = ((pts - audioBase - written * 1_000_000 / 48000).toDouble() / 20000).roundToInt() - if (missing > 5 || missing < -5) { - LogManager.addLog("WARN", "stream", "audio discontinuity id=$id missing=$missing packets") - resetTimeline(); audios.addFirst(f) - } else if (missing >= 0) { - repeat(missing) { - val conceal = NativeLib.streamOpusDecode(opus, byteArrayOf()) ?: error("Opus PLC failed") - writePcm(t, conceal) - } - val pcm = NativeLib.streamOpusDecode(opus, f.data) ?: error("Opus decode failed") - writePcm(t, pcm, firstAudioSkipFrames * 2); firstAudioSkipFrames = 0 - } else LogManager.addLog("DEBUG", "stream", "late audio discarded id=$id pts=${f.pts}") + val prime = t == null || (head == 0L && written < t.bufferSizeInFrames) + if (prime || written - head < 960) { + NativeLib.streamPlaybackPull(playback, 480, if (ended) Long.MAX_VALUE else ready)?.let { writePcm(it) } + } + } + if (playback != 0L && ready >= 0) { + val lateVideo = submitted.isNotEmpty() && submitted.first < position - 10000 + if (lateVideo && track != null && !pausedAudio) { track!!.pause(); pausedAudio = true } + if (pausedAudio && !lateVideo) { track!!.play(); pausedAudio = false } + val head = playedFrames(System.nanoTime()) + val audioEmpty = audioEnabled && written > 0 && head >= written && !ended + clockState = NativeLib.streamPlaybackUpdate(playback, System.nanoTime() / 1000, head, + if (ended) ready.coerceAtLeast(position) else ready, + (deviceLead + 10000).coerceAtLeast(33333), lateVideo || audioEmpty) + check(clockState != null) { "Playback timeline failed" } + position = clockState!![0]; needsAudio = clockState!![5] != 0L + } + repeat(4) { + if (videos.isEmpty()) return@repeat + val frame = videos.first + if (decoder == null) { + check(frame.key) + val nals = H264AnnexB.nals(frame.data) + val format = MediaFormat.createVideoFormat(MediaFormat.MIMETYPE_VIDEO_AVC, frame.width, frame.height).apply { + setByteBuffer("csd-0", ByteBuffer.wrap(nals.first { it.first == 7 }.second)) + setByteBuffer("csd-1", ByteBuffer.wrap(nals.first { it.first == 8 }.second)) + setInteger(MediaFormat.KEY_ROTATION, frame.rotation); setInteger(MediaFormat.KEY_MAX_INPUT_SIZE, H264AnnexB.MAX_FRAME) } + decoder = MediaCodec.createDecoderByType(MediaFormat.MIMETYPE_VIDEO_AVC) + decoder!!.configure(format, surface, null, 0); decoder!!.start() + } + val index = decoder!!.dequeueInputBuffer(0) + if (index >= 0) { + val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input") + check(frame.data.size <= input.capacity()); input.clear(); input.put(frame.data) + decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0) + submitted.addLast(frame.pts); videos.removeFirst(); decoded++ + } + } + val codec = decoder + if (codec != null) { + if (ended && videos.isEmpty() && !eosQueued) { + val index = codec.dequeueInputBuffer(0) + if (index >= 0) { codec.queueInputBuffer(index, 0, 0, decodedThrough.coerceAtLeast(0) + 1, + MediaCodec.BUFFER_FLAG_END_OF_STREAM); eosQueued = true } } - val clock = currentPts(now) repeat(8) { - if (videos.isEmpty() || videos.first.pts > clock + 80000) return@repeat - val frame = videos.first - if (decoder == null) { - check(frame.key) - val nals = H264AnnexB.nals(frame.data) - val format = MediaFormat.createVideoFormat(MediaFormat.MIMETYPE_VIDEO_AVC, frame.width, frame.height).apply { - setByteBuffer("csd-0", ByteBuffer.wrap(nals.first { it.first == 7 }.second)) - setByteBuffer("csd-1", ByteBuffer.wrap(nals.first { it.first == 8 }.second)) - setInteger(MediaFormat.KEY_ROTATION, frame.rotation); setInteger(MediaFormat.KEY_MAX_INPUT_SIZE, H264AnnexB.MAX_FRAME) + if (pendingOutput < 0) { + val index = codec.dequeueOutputBuffer(info, 0) + if (index < 0) return@repeat + if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0 && info.size == 0) { + codec.releaseOutputBuffer(index, false); decoderEnded = true; return@repeat } - decoder = MediaCodec.createDecoderByType(MediaFormat.MIMETYPE_VIDEO_AVC) - decoder!!.configure(format, surface, null, 0); decoder!!.start() - LogManager.addLog("INFO", "video", "decoder started id=$id ${frame.width}x${frame.height} rotation=${frame.rotation}") - } - val index = decoder!!.dequeueInputBuffer(0) - if (index >= 0) { - val input = decoder!!.getInputBuffer(index) ?: error("Missing decoder input") - check(frame.data.size <= input.capacity()) { "Decoder input too small" } - input.clear(); input.put(frame.data) - decoder!!.queueInputBuffer(index, 0, frame.data.size, frame.pts, 0) - videos.removeFirst(); videoBytes -= frame.data.size; decoded++ - } else if (clock - frame.pts > 250000) { - LogManager.addLog("WARN", "video", "decoder backpressure id=$id ageUs=${clock - frame.pts}; wait KEY") - resetVideo() - } - } - val codec = decoder - if (codec != null) repeat(16) { - val index = codec.dequeueOutputBuffer(bufferInfo, 0) - if (index >= 0) { - val clockPts = currentPts(System.nanoTime()) - if (bufferInfo.presentationTimeUs < clockPts - 100000) { - codec.releaseOutputBuffer(index, false); discarded++ - } else codec.releaseOutputBuffer(index, System.nanoTime() + (bufferInfo.presentationTimeUs - clockPts).coerceAtLeast(0) * 1000) - mutableStatus.value = R.string.stream_live + pendingOutput = index; pendingPts = info.presentationTimeUs; decodedThrough = pendingPts + while (submitted.isNotEmpty() && submitted.first <= pendingPts) submitted.removeFirst() } + if (pendingPts > position) return@repeat + val render = pendingPts >= position - 100000 + codec.releaseOutputBuffer(pendingOutput, render); pendingOutput = -1 + if (!render) discarded++ else mutableStatus.value = R.string.stream_live } } - if (ended && videos.isEmpty() && audios.isEmpty()) { - if (endDrainUntil == 0L) endDrainUntil = now + 250_000_000 - if (now >= endDrainUntil) { mutableStatus.value = R.string.stream_ended; break } + if (ended && videos.isEmpty() && (decoder == null || decoderEnded) && NativeLib.streamPlaybackEmpty(playback) + && playedFrames(System.nanoTime()) >= written) { + if (endUntil == 0L) endUntil = System.nanoTime() + 100_000_000 + if (System.nanoTime() >= endUntil) { mutableStatus.value = R.string.stream_ended; break } } - if (now - report >= 1_000_000_000) { - val seconds = if (report == 0L) 1.0 else (now - report) / 1_000_000_000.0 - report = now - LogManager.addLog("DEBUG", "stream", "viewer id=$id revision=$videoRevision decodeFps=${"%.1f".format((decoded - reportedDecoded) / seconds)} decoded=$decoded discarded=$discarded videoQ=${videos.size}/$videoBytes " + - "audioQ=${audios.size} written=$written clock=${currentPts(now)} waitingKey=$waitingKey") - reportedDecoded = decoded + if (System.nanoTime() - reported >= 1_000_000_000) { + reported = System.nanoTime() + LogManager.addLog("DEBUG", "stream", "viewer id=$id revision=$videoRevision decoded=$decoded discarded=$discarded " + + "videoQ=${videos.size} submitted=${submitted.size} clock=$position reserve=${clockState?.get(1)} " + + "minimum=${clockState?.get(2)} tempo=${clockState?.get(3)} written=$written waitingKey=$waitingKey") } Thread.sleep(5) } @@ -334,13 +278,13 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv if (running) { mutableStatus.value = R.string.stream_failed; LogManager.addLog("ERROR", "stream", "viewer failed id=$id: ${e.message}") } } finally { NativeLib.streamStop(id) - 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}") } - try { track?.stop() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio stop id=$id: ${e.message}") } - try { track?.release() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") } - if (opus != 0L) NativeLib.streamOpusClose(opus) - val deadline = System.nanoTime() + 500_000_000 - while (!ended && NativeLib.isReady() && System.nanoTime() < deadline) { + try { decoder?.stop(); decoder?.release() } + catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") } + try { track?.stop(); track?.release() } + 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) if (event != null && event[0].toInt() in 5..6) break Thread.sleep(5)