Retain stream decoder ownership until worker and SDK cleanup finish

This commit is contained in:
evgeny committed 2026-10-29 13:34:24 +01:00
1 parent aa27892c12
commit 2f353bf35e
2 files changed
+37 -17

No files matched your search

@@ -151,6 +151,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
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 })
video.start()
LogManager.addLog("INFO", "stream", "viewer audio worker started id=$id preparedPcmMs=250 audioAheadUs=500000")
while (running && surface.isValid) {
@@ -11,7 +11,8 @@ import android.view.Surface
import java.util.ArrayDeque
/** MediaCodec и pump принадлежат только video worker. Audio worker передаёт кадры через ограниченный
* mailbox; reset меняет поколение, close дожидается release до освобождения Surface/notification.
* mailbox; владелец сохраняет пассивный объект до start. reset меняет поколение.
* close ждёт worker до 250 мс и подтверждает release; ошибка сохраняет decoder для повторного close.
* Ни один вызов MediaCodec не выполняется под mailbox lock или на потоке подготовки звука. */
internal class StreamVideoPlayback(
private val id: Long, private val surface: Surface, private val requestedAtUs: Long,
@@ -38,7 +39,8 @@ internal class StreamVideoPlayback(
private set
private var inputEnded = false // mailbox lock
private var activeGeneration = -1L // Далее поля только video worker.
private var decoder: MediaCodec? = null
private var decoderResources: VideoCodecResources? = null
private val decoder: MediaCodec? get() = decoderResources?.codec
private var video: StreamDecodePump<Frame> = createVideoPump()
private val info = MediaCodec.BufferInfo()
private var videoRevision = 0L
@@ -49,7 +51,9 @@ internal class StreamVideoPlayback(
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() }
private val worker = Thread(::run, "stream-video-$id")
fun start() = worker.start()
val wantsInput: Boolean get() = synchronized(lock) { running && !inputEnded && input.size < 2 }
@@ -73,15 +77,23 @@ internal class StreamVideoPlayback(
fun close() {
synchronized(lock) { running = false; lock.notifyAll() }
worker.join()
check(Thread.currentThread() !== worker) {
LogManager.addLog("ERROR", "video", "video close attempted self join id=$id"); "Video close cannot join its worker"
}
try { worker.join(250) } catch (e: InterruptedException) {
Thread.currentThread().interrupt()
LogManager.addLog("ERROR", "video", "video close join interrupted id=$id; owner retained: $e")
throw e
}
check(!worker.isAlive) {
LogManager.addLog("ERROR", "video", "video close pending id=$id worker=${worker.state}; owner retained")
"Video worker cleanup pending id=$id"
}
releaseDecoder()
}
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
decoderResources?.close(); decoderResources = null
}
private fun run() {
@@ -152,8 +164,11 @@ internal class StreamVideoPlayback(
wake()
}
} finally {
releaseDecoder()
LogManager.addLog("INFO", "video", "viewer video worker released id=$id")
running = false
try { releaseDecoder() } catch (e: Exception) {
LogManager.addLog("ERROR", "video", "viewer decoder cleanup pending id=$id; retained until close: $e")
}
LogManager.addLog("INFO", "video", "viewer video worker exited id=$id decoderPending=${decoderResources != null}")
}
}
@@ -170,19 +185,20 @@ internal class StreamVideoPlayback(
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
val attempt = VideoCodecResources(MediaCodec.createByCodecName(candidate.name), candidate.name)
decoderResources = attempt
val codec = attempt.codec
codec.configure(format, surface, null, 0); codec.start()
var firstRendered = true
val codecRevision = frame.revision
try {
attempt.setOnFrameRenderedListener({ codec, pts, renderedNs ->
codec.setOnFrameRenderedListener({ codec, pts, renderedNs ->
if (firstRendered) {
firstRendered = false
val callbackUs = System.nanoTime() / 1000
@@ -211,8 +227,11 @@ internal class StreamVideoPlayback(
} 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}") }
try { releaseDecoder() } catch (release: Exception) {
e.addSuppressed(release)
LogManager.addLog("ERROR", "video", "decoder fallback cleanup pending id=$id; aborting replacement: $release")
throw e
}
if (!surface.isValid) throw e
}
}