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 dd689a53..13333f96 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 @@ -225,7 +225,10 @@ object NativeLib { external fun streamSignalClose(handle: Long) 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 streamPlaybackPrepare(handle: Long): Int = nativeStreamPlaybackPrepare(handle) + fun streamPlaybackBuffered(handle: Long): Int = nativeStreamPlaybackBuffered(handle) + fun streamPlaybackPull(handle: Long, pcm: ShortArray, frames: Int, limit: Long): Int = + nativeStreamPlaybackPull(handle, pcm, 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) @@ -251,7 +254,9 @@ object NativeLib { 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 nativeStreamPlaybackPrepare(handle: Long): Int + private external fun nativeStreamPlaybackBuffered(handle: Long): Int + private external fun nativeStreamPlaybackPull(handle: Long, pcm: ShortArray, frames: Int, limit: Long): Int 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? diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioMixer.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioMixer.kt index 5013fdd6..f254546a 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioMixer.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioMixer.kt @@ -1,17 +1,19 @@ package com.utun.chat.data -/** Один AudioTrack для всех трансляций. Worker не удерживает общий lock при обращении к decoder. - * Input закрывается после исключения из списка; его собственный lock защищает native DSP от release. */ +/** Один AudioTrack для всех трансляций. Mixer только копирует и смешивает готовый PCM. + * Input закрывается после исключения из списка; его короткий lock защищает consumer от release. */ internal object StreamAudioMixer { data class OutputState(val played: Long = 0, val leadUs: Long = 0) @Volatile var output = OutputState() private set private val lock = Any() private val inputs = linkedSetOf() + private var snapshot = emptyList() private var worker: Thread? = null fun attach(input: StreamMixedPlayback) = synchronized(lock) { check(inputs.add(input)) + snapshot = inputs.toList() if (worker == null) { output = OutputState() worker = Thread(::run, "stream-mixer").apply { start() } @@ -21,6 +23,7 @@ internal object StreamAudioMixer { fun detach(input: StreamMixedPlayback) = synchronized(lock) { inputs.remove(input) + snapshot = inputs.toList() LogManager.addLog("INFO", "stream", "mixer input detached id=${input.id} count=${inputs.size}") } @@ -39,6 +42,9 @@ internal object StreamAudioMixer { val mix = StreamMixBuffer() var generated = 0L try { + try { android.os.Process.setThreadPriority(android.os.Process.THREAD_PRIORITY_AUDIO) } + catch (e: Exception) { LogManager.addLog("WARN", "stream", "mixer audio priority unavailable: ${e.message}") } + LogManager.addLog("INFO", "stream", "mixer started: ready PCM only, blockFrames=${mix.frames}") while (true) { val current = synchronized(lock) { if (inputs.isEmpty()) { @@ -46,7 +52,7 @@ internal object StreamAudioMixer { audio.close(); output = OutputState(); worker = null return } - inputs.toList() + snapshot } audio.step(System.nanoTime(), hold = false, pull = { mix.clear() @@ -65,6 +71,7 @@ internal object StreamAudioMixer { synchronized(lock) { for (input in inputs) input.outputFailure = e.message ?: "Audio output failed" inputs.clear() + snapshot = emptyList() try { audio.close() } catch (close: Exception) { LogManager.addLog("WARN", "stream", "mixer release: ${close.message}") } output = OutputState(); worker = null @@ -73,19 +80,21 @@ internal object StreamAudioMixer { } } -/** Независимые Opus/SoundTouch, темп и PTS. Video worker и mixer сериализуют только свой DSP. - * Каждый 10ms блок общего выхода отражён в своей шкале, включая тишину при starvation/hold. +/** Независимые Opus/SoundTouch, темп и PTS. Audio worker готовит PCM вне lock; mixer потребляет SPSC. + * Каждый 10ms блок общего выхода отражён в своей шкале, включая тишину при starvation. * Reset создаёт новый Input с новой origin; AudioTrack и остальные источники продолжают играть. */ internal class StreamMixedPlayback(val id: Long, pts: Long, private val audioEnabled: Boolean, private val wake: () -> Unit) { private val lock = Any() private var handle = NativeLib.streamPlaybackCreate(id, pts, audioEnabled).also { check(it != 0L) } private val timeline = StreamMixTimeline() private var started = false - private var hold = false - private var ready = -1L private var ended = false private var endFrame: Long? = null private var fadeFrames = 0 + private val pcm = ShortArray(960) + private var underruns = 0L + private var silentFrames = 0L + private var observedUnderruns = 0L @Volatile var outputFailure: String? = null init { if (audioEnabled) StreamAudioMixer.attach(this) } @@ -98,31 +107,40 @@ internal class StreamMixedPlayback(val id: Long, pts: Long, private val audioEna } val description: String get() = synchronized(lock) { "mixed=$audioEnabled origin=${timeline.origin} written=${timeline.frames} " + - "played=${timeline.played(StreamAudioMixer.output.played)} hold=$hold leadUs=$leadUs" + "played=${timeline.played(StreamAudioMixer.output.played)} leadUs=$leadUs " + + "pcmFrames=${NativeLib.streamPlaybackBuffered(handle)} underruns=$underruns silentFrames=$silentFrames" } - fun feed(pts: Long, packet: ByteArray): Int = synchronized(lock) { - check(handle != 0L); NativeLib.streamPlaybackFeed(handle, pts, packet) + // Только audio worker; он же закрывает handle, поэтому DSP не нуждается в consumer lock. + fun prepare(): Int { check(handle != 0L); return NativeLib.streamPlaybackPrepare(handle) } + fun feed(pts: Long, packet: ByteArray): Int { + check(handle != 0L); return NativeLib.streamPlaybackFeed(handle, pts, packet) } - fun finish() = synchronized(lock) { - check(handle != 0L); ended = true; NativeLib.streamPlaybackFinish(handle) + fun finish() { + synchronized(lock) { check(handle != 0L); ended = true } + NativeLib.streamPlaybackFinish(handle) } - fun update(now: Long, available: Long, guard: Long, videoHeld: Boolean): LongArray? = synchronized(lock) { + fun update(now: Long, available: Long, guard: Long): LongArray? = synchronized(lock) { check(handle != 0L); check(outputFailure == null) { outputFailure!! } - ready = available; hold = videoHeld; started = true - NativeLib.streamPlaybackUpdate(handle, now, timeline.played(StreamAudioMixer.output.played), available, guard, videoHeld) + started = true + val starved = underruns != observedUnderruns + observedUnderruns = underruns + NativeLib.streamPlaybackUpdate(handle, now, timeline.played(StreamAudioMixer.output.played), available, guard, starved) } /** Mixer не блокирует чужой input ожиданием media; недостающее заменяется тишиной только этого источника. */ fun mix(buffer: StreamMixBuffer, firstOutputFrame: Long): Boolean = synchronized(lock) { if (handle == 0L || !started || !audioEnabled || outputFailure != null) return false try { - val pcm = if (hold) null else NativeLib.streamPlaybackPull(handle, buffer.frames, if (ended) Long.MAX_VALUE else ready) - val count = (pcm?.size ?: 0) / 2 - if (pcm != null) fadeFrames = buffer.add(pcm, fadeFrames) + val count = NativeLib.streamPlaybackPull(handle, pcm, buffer.frames, Long.MAX_VALUE) + check(count >= 0) { "Ready PCM pull failed" } + fadeFrames = buffer.add(pcm, fadeFrames, count) if (ended && endFrame == null && NativeLib.streamPlaybackEmpty(handle)) endFrame = timeline.frames + count - if (count < buffer.frames) NativeLib.streamPlaybackSilence(handle, buffer.frames - count) + if (count < buffer.frames) { + NativeLib.streamPlaybackSilence(handle, buffer.frames - count) + if (!ended) { underruns++; silentFrames += buffer.frames - count } + } timeline.append(firstOutputFrame, buffer.frames) wake() true 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 a4db9b1f..f5c603b6 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 @@ -277,7 +277,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv if (!running) break video.step(position, System.nanoTime() / 1000) needsAudio = !audioInputEnded && (playback?.needsInput ?: true) - val audioAhead = if (!audioInputEnded && (!audioEnabled || needsAudio) && records.audios == 0) deviceLead + 80000 else -1L + 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 val batch = if (ended) null else NativeLib.streamPull(id, position, videoBudget, audioAhead, 16) if (batch != null) { batches++; batchMax = maxOf(batchMax, records.add(batch)) } @@ -345,14 +345,13 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv if (!ended && (!opened || waitingKey) && System.nanoTime() > deadline) error("Viewer open timeout") if (!playbackStarted && video.firstOutputPts != null) playbackStarted = true if (playbackStarted && playback != null && ready >= 0) { - val lateVideo = !video.ended && (video.firstOutputPts == null - || (video.oldestSubmittedPts >= 0 && video.oldestSubmittedPts < position - 10000)) clockState = playback!!.update(System.nanoTime() / 1000, if (ended) ready.coerceAtLeast(position) else ready, - maxOf(playback!!.leadUs + 10000, video.decoderLeadUs + video.framePeriodUs), lateVideo) + maxOf(40000, playback!!.leadUs + 20000)) check(clockState != null) { "Playback timeline failed" } 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) { diff --git a/tools/chatgui-android/jni_bridge/android_stream_jni.c b/tools/chatgui-android/jni_bridge/android_stream_jni.c index 38208e75..aebbdbcd 100644 --- a/tools/chatgui-android/jni_bridge/android_stream_jni.c +++ b/tools/chatgui-android/jni_bridge/android_stream_jni.c @@ -1,9 +1,10 @@ -/* JNI live media: session IDs для ядра. Playback handles сериализуются своим StreamMixedPlayback lock - * между video worker и общим микшером; capture Opus handles принадлежат одному audio worker. */ +/* JNI live media: session IDs для ядра. Playback DSP принадлежит audio worker; mixer берёт готовый PCM. + * Только output map/clock сериализуются StreamMixedPlayback lock; release выполняет audio worker после detach. */ #ifdef __ANDROID__ #include #include #include +#include #include "../../../lib/debug_config.h" #include "../../../lib/mem.h" @@ -124,7 +125,7 @@ struct android_stream_playback { struct stream_playback_state state; struct stream_playback_audio *audio; int64_t position, time_us; - int finished; + atomic_int finished; uint64_t end_frame; }; @@ -135,7 +136,8 @@ JNIEXPORT jlong JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackCr struct android_stream_playback *p = u_calloc(1, sizeof(*p)); if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI playback allocation failed"); return 0; } stream_playback_clock_reset(&p->clock, (uint64_t)id); p->position = pts; p->time_us = -1; - if (audio && !(p->audio = stream_playback_audio_create(pts, 40))) { u_free(p); return 0; } + atomic_init(&p->finished, 0); + if (audio && !(p->audio = stream_playback_audio_create(pts, 250))) { u_free(p); return 0; } return (jlong)(intptr_t)p; } @@ -150,22 +152,34 @@ JNIEXPORT jint JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackFee return stream_playback_audio_feed(p->audio, pts, bytes, (size_t)length); } -JNIEXPORT jshortArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackPull( - JNIEnv *env, jobject obj, jlong handle, jint frames, jlong limit) { +JNIEXPORT jint JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackPrepare(JNIEnv *env, jobject obj, jlong handle) { + (void)env; (void)obj; struct android_stream_playback *p = (struct android_stream_playback *)(intptr_t)handle; + if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI playback prepare invalid handle"); return -1; } + if (!p->audio) return 0; + stream_playback_audio_set_tempo(p->audio, p->clock.tempo_milli); + return stream_playback_audio_prepare(p->audio); +} + +JNIEXPORT jint JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackBuffered(JNIEnv *env, jobject obj, jlong handle) { + (void)env; (void)obj; struct android_stream_playback *p = (struct android_stream_playback *)(intptr_t)handle; + return p && p->audio ? (jint)stream_playback_audio_buffered(p->audio) : 0; +} + +JNIEXPORT jint JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackPull( + JNIEnv *env, jobject obj, jlong handle, jshortArray output, jint frames, jlong limit) { (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 PCM invalid frames=%d", frames); return NULL; + if (!p || !p->audio || !output || frames <= 0 || frames > 8192 || (*env)->GetArrayLength(env, output) < frames * 2) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI playback PCM invalid frames=%d", frames); return -1; } - if (limit < 0) return NULL; + if (limit < 0) return 0; int16_t pcm[8192 * 2]; int count = stream_playback_audio_pull(p->audio, pcm, (unsigned)frames, limit); if (p->finished && !p->end_frame && stream_playback_audio_empty(p->audio)) p->end_frame = stream_playback_audio_written(p->audio) + 1; /* 0 — конец ещё не поставлен устройству. */ - if (count <= 0) return NULL; - jshortArray output = (*env)->NewShortArray(env, count * 2); - if (output) (*env)->SetShortArrayRegion(env, output, 0, count * 2, pcm); - else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI playback PCM allocation failed frames=%d", count); - return output; + if (count <= 0) return count; + (*env)->SetShortArrayRegion(env, output, 0, count * 2, pcm); + if ((*env)->ExceptionCheck(env)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI playback PCM copy failed"); return -1; } + return count; } JNIEXPORT jlongArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackUpdate( @@ -178,7 +192,6 @@ JNIEXPORT jlongArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlayb int64_t position = stream_playback_audio_position(p->audio, (uint64_t)played); if (position > p->position) p->position = position; pending = stream_playback_audio_pending_gain(p->audio, (uint64_t)played); - if (!p->finished && stream_playback_audio_starved(p->audio, (uint64_t)played)) starved = JNI_TRUE; } if ((!p->audio || audio_done) && p->time_us >= 0 && !starved) { int64_t elapsed = now - p->time_us; @@ -189,7 +202,6 @@ JNIEXPORT jlongArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlayb } p->time_us = now; if (stream_playback_clock_update(&p->clock, now, p->position, ready, guard, pending, starved, &p->state) < 0) return NULL; - if (p->audio) stream_playback_audio_set_tempo(p->audio, p->state.tempo_milli); jlong values[] = {p->position, p->state.reserve_us, p->state.minimum_us, p->state.tempo_milli, p->state.waiting, p->audio ? stream_playback_audio_needs_input(p->audio) : 0, p->audio ? (jlong)stream_playback_audio_written(p->audio) : 0}; @@ -210,7 +222,7 @@ JNIEXPORT void JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackSil 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); } + if (p) { stream_playback_audio_finish(p->audio); p->finished = 1; } } JNIEXPORT jboolean JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlaybackNeedsInput(JNIEnv *env, jobject obj, jlong handle) {