Browse Source

Preserve stream capture settings and refine audio playback

master
evgeny 2 days ago
parent
commit
d60f8e4b1b
  1. 13
      src/stream/stream_playback_audio.cpp
  2. 4
      src/stream/stream_playback_audio.h
  3. 28
      tools/chatgui-android/app/src/main/java/com/utun/chat/data/ConfigProvider.kt
  4. 2
      tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureConfig.kt
  5. 29
      tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureSettings.kt
  6. 17
      tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamController.kt
  7. 32
      tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamScreen.kt
  8. 2
      tools/chatgui-android/jni_bridge/android_stream_jni.c
  9. 13
      tools/chatgui/src/mainwindow.cpp
  10. 43
      tools/chatgui/src/streamplayer_window.cpp
  11. 16
      tools/chatgui/src/voice_audio_io.cpp
  12. 5
      tools/chatgui/src/voice_audio_io.h
  13. 11
      tools/chatgui/src/voice_playback_buffer.h
  14. 33
      tools/chatgui/tests/test_stream_audio.cpp
  15. 15
      tools/chatgui/tests/test_stream_view.cpp
  16. 24
      tools/chatgui/tests/test_voice_audio_io.cpp

13
src/stream/stream_playback_audio.cpp

@ -30,6 +30,7 @@ struct stream_playback_audio {
soundtouch::SoundTouch stretch;
Span input[mapCapacity], output[mapCapacity];
unsigned inputHead = 0, inputCount = 0, outputHead = 0, outputCount = 0;
unsigned bufferFrames = 0;
int tempo = 1000;
int64_t startPts = 0, expectedPts = -1;
double expectedOutput = 0, producedOutput = 0, lastPts = 0;
@ -85,12 +86,15 @@ static int feedPcm(stream_playback_audio* a, int64_t pts, const uint8_t* packet,
extern "C" {
struct stream_playback_audio* stream_playback_audio_create(int64_t start) {
struct stream_playback_audio* stream_playback_audio_create(int64_t start, unsigned bufferMs) {
if (start < 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback audio invalid start=%lld", (long long)start); return nullptr; }
if (!bufferMs || bufferMs > 1000) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback audio invalid PCM buffer=%ums range=1..1000ms", bufferMs); return nullptr;
}
void* memory = u_malloc(sizeof(stream_playback_audio));
if (!memory) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback audio allocation failed"); return nullptr; }
auto* a = new (memory) stream_playback_audio;
a->startPts = start; a->lastPts = (double)start;
a->startPts = start; a->lastPts = (double)start; a->bufferFrames = rate * bufferMs / 1000;
a->opus = opus_codec_decoder_create(rate, 2);
if (!a->opus) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback audio Opus initialization failed"); stream_playback_audio_destroy(a); return nullptr;
@ -99,7 +103,8 @@ struct stream_playback_audio* stream_playback_audio_create(int64_t start) {
a->stretch.setSetting(SETTING_USE_QUICKSEEK, 1);
a->stretch.setSetting(SETTING_SEQUENCE_MS, 20); a->stretch.setSetting(SETTING_SEEKWINDOW_MS, 10);
a->stretch.setSetting(SETTING_OVERLAP_MS, 4); a->stretch.setSetting(SETTING_USE_AA_FILTER, 0);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "playback audio created start=%lld stereo=1 tempo=1.0..1.1", (long long)start);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "playback audio created start=%lld stereo=1 tempo=1.0..1.1 PCM_buffer=%ums/%u_frames",
(long long)start, bufferMs, a->bufferFrames);
return a;
}
@ -143,7 +148,7 @@ void stream_playback_audio_set_tempo(struct stream_playback_audio* a, int tempo)
}
int stream_playback_audio_needs_input(const struct stream_playback_audio* a) {
return a && !a->finished && !a->failed && a->stretch.numSamples() < framesPerPacket;
return a && !a->finished && !a->failed && a->stretch.numSamples() < a->bufferFrames;
}
/* Ограничение готовой границы применяется до выдачи PCM устройству, сохраняя общий A/V clock. */

4
src/stream/stream_playback_audio.h

@ -15,7 +15,9 @@ extern "C" {
#endif
struct stream_playback_audio;
struct stream_playback_audio *stream_playback_audio_create(int64_t start_pts_us);
/* buffer_ms (1..1000) — порог пополнения готового PCM; трансляции используют 40 мс.
* Это запас декодера, без аппаратной задержки и сетевого джиттер-буфера. */
struct stream_playback_audio *stream_playback_audio_create(int64_t start_pts_us, unsigned buffer_ms);
void stream_playback_audio_destroy(struct stream_playback_audio *audio);
int stream_playback_audio_feed(struct stream_playback_audio *audio, int64_t pts_us, const uint8_t *packet, size_t length);
void stream_playback_audio_set_tempo(struct stream_playback_audio *audio, int tempo_milli);

28
tools/chatgui-android/app/src/main/java/com/utun/chat/data/ConfigProvider.kt

@ -56,6 +56,14 @@ class ConfigProvider(private val context: Context) {
private val chatOpusCodecPreset = intPreferencesKey("chat_opus_codec_preset")
private val streamSubblocks = intPreferencesKey("stream_subblocks")
private val streamRsRedundant = intPreferencesKey("stream_rs_redundant")
private val streamCamera = stringPreferencesKey("stream_capture_camera")
private val streamWidth = intPreferencesKey("stream_capture_width")
private val streamHeight = intPreferencesKey("stream_capture_height")
private val streamFps = intPreferencesKey("stream_capture_fps")
private val streamCodec = intPreferencesKey("stream_capture_codec")
private val streamRotation = intPreferencesKey("stream_capture_rotation")
private val streamBitrate = intPreferencesKey("stream_capture_bitrate")
private val streamAudio = booleanPreferencesKey("stream_capture_audio")
private val callJitterMaxReserveMs = intPreferencesKey("call_jitter_max_reserve_ms")
private val callVideoBitrateRange = stringPreferencesKey("call_video_bitrate_range_kbps")
private val chatCompressorEnabled = intPreferencesKey("chat_compressor_enabled")
@ -166,6 +174,26 @@ class ConfigProvider(private val context: Context) {
LogManager.addLog("INFO", "STREAM", "Broadcast settings saved n=$subblocks r=$redundant; apply on next start")
}
suspend fun getStreamCaptureSettings(): StreamCaptureSettings {
val prefs = context.dataStore.data.first()
return StreamCaptureSettings(prefs[streamCamera] ?: "", prefs[streamWidth] ?: 1280, prefs[streamHeight] ?: 720,
prefs[streamFps] ?: 30, prefs[streamCodec] ?: StreamVideoCodec.H264.id, prefs[streamRotation] ?: StreamOrientation.LEFT.degrees,
prefs[streamBitrate], prefs[streamAudio] ?: true)
}
suspend fun saveStreamCaptureSettings(config: StreamCaptureConfig?, audio: Boolean?) {
context.dataStore.edit { prefs ->
config?.let {
prefs[streamCamera] = it.mode.camera
prefs[streamWidth] = it.mode.width; prefs[streamHeight] = it.mode.height; prefs[streamFps] = it.mode.fps
prefs[streamCodec] = it.mode.videoCodec.id; prefs[streamRotation] = it.orientation.degrees
if (it.videoBitrate == null) prefs.remove(streamBitrate) else prefs[streamBitrate] = it.videoBitrate
}
audio?.let { prefs[streamAudio] = it }
}
LogManager.addLog("INFO", "stream", "capture settings saved config=$config audio=$audio")
}
fun getString(key: String): String? = runBlocking {
when {
key == "node.private_key" -> readKeysFromFile()?.second ?: context.dataStore.data.first()[nodePrivateKey]

2
tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureConfig.kt

@ -8,7 +8,7 @@ enum class StreamOrientation(val degrees: Int) {
}
/** Режим камеры и ориентация подтверждаются и восстанавливаются одним значением. */
data class StreamCaptureConfig(val mode: StreamCaptureMode, val orientation: StreamOrientation = StreamOrientation.UPRIGHT,
data class StreamCaptureConfig(val mode: StreamCaptureMode, val orientation: StreamOrientation = StreamOrientation.LEFT,
val videoBitrate: Int? = null) {
val bitrate get() = videoBitrate ?: mode.bitrate
init { require(bitrate in mode.minimumBitrate..mode.maximumBitrate) }

29
tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamCaptureSettings.kt

@ -0,0 +1,29 @@
package com.utun.chat.data
/** Последний выбор пользователя; возможности encoder заново берём у устройства. */
data class StreamCaptureSettings(val camera: String = "", val width: Int = 1280, val height: Int = 720, val fps: Int = 30,
val codec: Int = StreamVideoCodec.H264.id, val rotation: Int = StreamOrientation.LEFT.degrees,
val bitrate: Int? = null, val audio: Boolean = true) {
fun resolve(cameras: List<StreamCamera>): StreamCaptureConfig? {
val selected = cameras.firstOrNull { it.id == camera }
?: cameras.firstOrNull { it.lens.type == LensType.MAIN } ?: cameras.firstOrNull()
if (selected == null) {
LogManager.addLog("WARN", "stream", "capture settings: no available camera"); return null
}
val modes = selected.modes.filter { it.videoCodec.id == codec }.ifEmpty { selected.modes }
val mode = modes.firstOrNull { it.width == width && it.height == height && it.fps == fps }
?: modes.filter { it.width == width && it.height == height && it.fps <= fps }.maxByOrNull { it.fps }
?: StreamCameraModes.choose(modes)
if (mode == null) {
LogManager.addLog("WARN", "stream", "capture settings: camera=${selected.id} has no available mode"); return null
}
val orientation = StreamOrientation.values().firstOrNull { it.degrees == rotation } ?: StreamOrientation.LEFT
val restored = StreamCaptureConfig(mode, orientation, bitrate?.coerceIn(mode.minimumBitrate, mode.maximumBitrate))
if (camera.isNotEmpty() && (camera != mode.camera || width != mode.width || height != mode.height || fps != mode.fps ||
codec != mode.videoCodec.id || rotation != orientation.degrees || bitrate != restored.videoBitrate))
LogManager.addLog("WARN", "stream", "capture settings adjusted: saved=$this restored=$restored")
LogManager.addLog("INFO", "stream", "capture settings restored camera=${mode.camera} mode=${mode.label} " +
"codec=${mode.videoCodec.label} rotation=${orientation.degrees} bitrate=${restored.videoBitrate ?: "auto"} audio=$audio")
return restored
}
}

17
tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamController.kt

@ -7,6 +7,11 @@ import android.widget.Toast
import androidx.core.content.ContextCompat
import com.utun.chat.ChatApplication
import com.utun.chat.R
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.launch
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import java.util.concurrent.atomic.AtomicLong
@ -34,6 +39,16 @@ object StreamController {
@Volatile var capturing = false
private set
private var service: StreamService? = null
private val settingsScope = CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate)
/** Запись переживает закрытие экрана; DataStore сохраняет выбор одним изменением. */
fun rememberSettings(config: StreamCaptureConfig? = null, audio: Boolean? = null) {
settingsScope.launch {
try { ChatApplication.instance.configProvider.saveStreamCaptureSettings(config, audio) }
catch (e: CancellationException) { throw e }
catch (e: Exception) { LogManager.addLog("ERROR", "stream", "capture settings save failed: $e") }
}
}
fun start(group: Long, config: StreamCaptureConfig, audio: Boolean) {
check(Looper.myLooper() == Looper.getMainLooper())
@ -45,6 +60,7 @@ object StreamController {
mutable.value = StreamState(id, group, StreamPhase.FAILED, R.string.stream_resources_busy); return
}
capturing = true
rememberSettings(config, audio)
mutable.value = StreamState(id, group, StreamPhase.STARTING, pendingConfig = config, audioAvailable = audio, audioEnabled = audio)
mutableTx.value = null
try {
@ -84,6 +100,7 @@ object StreamController {
val st = mutable.value ?: return@post
if (st.id != id || !st.ongoing || st.phase == StreamPhase.STOPPING) return@post
mutable.value = st.copy(config = config, pendingConfig = null, notice = if (restored) R.string.stream_mode_restored else null)
rememberSettings(config)
}
}

32
tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamScreen.kt

@ -25,7 +25,6 @@ import androidx.compose.material.icons.automirrored.filled.ArrowBack
import androidx.compose.material.icons.filled.Close
import androidx.compose.material3.*
import androidx.compose.runtime.*
import androidx.compose.runtime.saveable.rememberSaveable
import androidx.compose.ui.Modifier
import androidx.compose.ui.Alignment
import androidx.compose.ui.geometry.CornerRadius
@ -48,6 +47,7 @@ import androidx.core.content.ContextCompat
import androidx.lifecycle.Lifecycle
import androidx.lifecycle.LifecycleEventObserver
import com.utun.chat.R
import com.utun.chat.ChatApplication
import com.utun.chat.data.LogManager
import com.utun.chat.data.StreamController
import com.utun.chat.data.StreamPhase
@ -57,6 +57,7 @@ import com.utun.chat.data.StreamPreviewTarget
import com.utun.chat.data.StreamSource
import com.utun.chat.data.StreamCaptureMode
import com.utun.chat.data.StreamCaptureConfig
import com.utun.chat.data.StreamCaptureSettings
import com.utun.chat.data.StreamOrientation
import com.utun.chat.data.StreamCamera
import com.utun.chat.data.StreamCameraModes
@ -83,18 +84,22 @@ fun StreamScreen(channelId: String, channelName: String, sources: List<StreamSou
val local = state?.takeIf { it.group == group }
var previewedSession by remember(group) { mutableStateOf<Long?>(null) }
var audio by remember { mutableStateOf(true) }
var mode by remember { mutableStateOf<StreamCaptureMode?>(null) }
var orientation by rememberSaveable { mutableStateOf(StreamOrientation.UPRIGHT) }
var videoBitrate by rememberSaveable { mutableStateOf<Int?>(null) }
val config = mode?.let { StreamCaptureConfig(it, orientation, videoBitrate) }
var config by remember { mutableStateOf<StreamCaptureConfig?>(null) }
var cameras by remember { mutableStateOf<List<StreamCamera>>(emptyList()) }
var loadingModes by remember { mutableStateOf(true) }
LaunchedEffect(Unit) {
val saved = try { ChatApplication.instance.configProvider.getStreamCaptureSettings() }
catch (e: CancellationException) { throw e }
catch (e: Exception) {
LogManager.addLog("ERROR", "stream", "capture settings load failed: $e"); StreamCaptureSettings()
}
try { cameras = withContext(Dispatchers.Default) { StreamCameraModes.enumerate(context) } }
catch (e: CancellationException) { throw e }
catch (e: Exception) { LogManager.addLog("ERROR", "video", "capture modes failed: ${e.message}") }
val current = StreamController.state.value?.takeIf { it.group == group && it.ongoing }
config = current?.config ?: saved.resolve(cameras)
audio = current?.audioAvailable ?: saved.audio
loadingModes = false
mode = StreamCameraModes.choose((cameras.firstOrNull { it.lens.type == LensType.MAIN } ?: cameras.firstOrNull())?.modes.orEmpty())
}
var permissionVersion by remember { mutableStateOf(0) }
val permissions = rememberLauncherForActivityResult(ActivityResultContracts.RequestMultiplePermissions()) { permissionVersion++ }
@ -108,8 +113,8 @@ fun StreamScreen(channelId: String, channelName: String, sources: List<StreamSou
}
}
LaunchedEffect(local?.config?.orientation) {
local?.config?.orientation?.let { orientation = it }
LaunchedEffect(local?.config, local?.audioAvailable) {
local?.config?.let { config = it; audio = local.audioAvailable }
}
LaunchedEffect(local?.phase, local?.pendingConfig, selected?.id) {
@ -160,9 +165,16 @@ fun StreamScreen(channelId: String, channelName: String, sources: List<StreamSou
}))
Button(onClick = StreamController::stop, enabled = local.phase != StreamPhase.STOPPING) { Text(stringResource(R.string.stream_stop)) }
} else {
CaptureOptions(cameras, config, !loadingModes, displayDegrees) { mode = it.mode; orientation = it.orientation; videoBitrate = it.videoBitrate }
CaptureOptions(cameras, config, !loadingModes, displayDegrees) {
config = it; StreamController.rememberSettings(config = it)
}
if (!loadingModes && cameras.isEmpty()) Text(stringResource(R.string.stream_no_modes), color = MaterialTheme.colorScheme.error)
Row { Checkbox(checked = audio, onCheckedChange = { audio = it }); Text(stringResource(R.string.microphone), Modifier.padding(top = 12.dp)) }
Row {
Checkbox(checked = audio, enabled = !loadingModes, onCheckedChange = {
audio = it; StreamController.rememberSettings(audio = it)
})
Text(stringResource(R.string.microphone), Modifier.padding(top = 12.dp))
}
Button(onClick = {
if (!cameraGranted || (audio && !micGranted)) permissions.launch(
listOf(Manifest.permission.CAMERA).plus(if (audio) listOf(Manifest.permission.RECORD_AUDIO) else emptyList()).toTypedArray())

2
tools/chatgui-android/jni_bridge/android_stream_jni.c

@ -120,7 +120,7 @@ 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))) { u_free(p); return 0; }
if (audio && !(p->audio = stream_playback_audio_create(pts, 40))) { u_free(p); return 0; }
return (jlong)(intptr_t)p;
}

13
tools/chatgui/src/mainwindow.cpp

@ -83,13 +83,15 @@ static void unixSignalHandler(int) { g_signalFlag = 1; }
#endif
/* Сохранение key/value в БД (ui_state). Пишется из utun-потока через trampoline,
чтобы не трогать sqlite из GUI-потока. Буфер формата key\0value\0, ≤256 байт. */
чтобы не трогать sqlite из GUI-потока. Буфер формата key\0value\0. */
static void postUiStateKv(const char* key, const QString& value) {
QByteArray val = value.toUtf8();
size_t klen = strlen(key), vlen = (size_t)val.size(), total = klen + 1 + vlen + 1;
if (total > 256) return;
if (total > sizeof(save_ui_state_arg::data)) {
DEBUG_ERROR(DEBUG_CATEGORY_CONFIG, "UI setting too large key=%s bytes=%zu", key, total); return;
}
struct save_ui_state_arg* arg = (struct save_ui_state_arg*)u_malloc(sizeof(struct save_ui_state_arg));
if (!arg) return;
if (!arg) { DEBUG_ERROR(DEBUG_CATEGORY_CONFIG, "UI setting allocation failed key=%s", key); return; }
arg->inst = gui_bridge_get_inst();
memcpy(arg->data, key, klen + 1);
memcpy(arg->data + klen + 1, val.constData(), vlen + 1);
@ -628,7 +630,10 @@ void MainWindow::setupConnects() {
});
}
if (m_streamPublishWindow) delete m_streamPublishWindow.data();
m_streamPublishWindow = new StreamPublishWindow(m_streamPublisher, m_streamPublisher->active() ? m_streamPublisher->group() : group, this);
m_streamPublishWindow = new StreamPublishWindow(m_streamPublisher, m_streamPublisher->active() ? m_streamPublisher->group() : group,
this, m_db ? m_db->getUiState("stream_capture_settings") : QString());
connect(m_streamPublishWindow, &StreamPublishWindow::settingsChanged, this,
[](const QString& settings) { postUiStateKv("stream_capture_settings", settings); });
m_streamPublishWindow->show(); m_streamPublishWindow->raise(); m_streamPublishWindow->activateWindow();
});
connect(m_messageList, &MessageList::streamRequested, this,

43
tools/chatgui/src/streamplayer_window.cpp

@ -40,6 +40,13 @@ QString text(const char* value) { return QCoreApplication::translate("ChatGui",
struct StreamPlayerWindow::Playback {
struct Video { QImage image; int64_t pts; };
struct AudioGap {
uint64_t count = 0, silentFrames = 0;
unsigned requested = 0, received = 0;
int64_t reserve = 0, feedAge = -1, workerAge = -1, audioPts = -1;
size_t videoQueue = 0;
bool held = false;
} audioGap;
uint64_t id = ++nextId, group = 0, source = 0, epoch = 0;
std::atomic<bool> running{true}, audioAvailable{false};
std::atomic<float> volume{1.0f};
@ -56,6 +63,7 @@ struct StreamPlayerWindow::Playback {
uint64_t audioFrameStart = 0, shown = 0, underruns = 0, endFrames = UINT64_MAX;
bool started = false, presenting = false, videoReady = false, ended = false, videoEnded = false;
bool activeAudio = false, holdAudio = false, resync = false;
int64_t audioFedAt = 0, audioFedPts = -1, workerAt = 0;
~Playback() { stream_playback_audio_destroy(sound); }
uint64_t played(int64_t now) const {
@ -128,7 +136,13 @@ void StreamPlayerWindow::openAudio() {
if (count < 0) { p.resync = true; count = 0; }
if ((unsigned)count < frames) {
stream_playback_audio_silence(p.sound, frames - (unsigned)count);
if (p.presenting && !p.ended) p.underruns++;
if (p.presenting && !p.ended) {
p.underruns++; p.audioGap.count++; p.audioGap.silentFrames += frames - (unsigned)count;
p.audioGap.requested = frames; p.audioGap.received = (unsigned)count; p.audioGap.held = p.holdAudio;
p.audioGap.reserve = p.ready - p.position; p.audioGap.videoQueue = p.videos.size();
p.audioGap.feedAge = p.audioFedAt ? now - p.audioFedAt : -1; p.audioGap.audioPts = p.audioFedPts;
p.audioGap.workerAge = p.workerAt ? now - p.workerAt : -1;
}
}
float volume = p.volume.load();
for (int i = 0; i < count * 2; i++) pcm[i] = static_cast<int16_t>(pcm[i] * volume);
@ -200,6 +214,8 @@ void StreamPlayerWindow::decode() {
uint32_t revision = 0;
uint64_t decoded = 0, discarded = 0, reportedShown = 0;
int64_t deadline = nowUs() + 10000000, report = nowUs(), decoderProgress = -1, lastInputAt = -1;
uint64_t reportedGaps = 0;
int64_t gapReportAt = 0;
auto reset = [&](bool request) {
avcodec_free_context(&decoder); av_packet_unref(packet); waitKey = true; finishing = eosSent = false;
decoderProgress = lastInputAt = -1;
@ -210,6 +226,7 @@ void StreamPlayerWindow::decode() {
p.decoderLead = 0; p.framePeriod = 33333; p.lastInputPts = p.lastDecodedPts = -1;
p.ended = p.videoEnded = false;
p.audioDacAt = 0; p.audioFrameStart = 0; p.holdAudio = p.resync = false; p.endFrames = UINT64_MAX;
p.audioFedAt = 0; p.audioFedPts = -1;
}
if (request && !gui_stream_resync(p.id)) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop playback resync refused");
deadline = nowUs() + 10000000; emit statusChanged(text("Waiting for a key frame…"));
@ -226,7 +243,7 @@ void StreamPlayerWindow::decode() {
if (!status.initialized) return;
if (!p.started) {
p.activeAudio = status.audio_enabled && p.audioAvailable;
if (p.activeAudio) p.sound = stream_playback_audio_create(status.start_us);
if (p.activeAudio) p.sound = stream_playback_audio_create(status.start_us, 40);
if (p.activeAudio && !p.sound) { p.running = false; return; }
p.position = status.start_us; p.clockAt = nowUs(); p.started = true;
stream_playback_clock_reset(&p.adaptation, p.id);
@ -283,7 +300,18 @@ void StreamPlayerWindow::decode() {
else if (!gui_stream_open(p.id, p.group, p.source, p.epoch)) fail(text("Could not connect to the stream"));
else while (p.running) {
bool resync;
{ std::lock_guard<std::mutex> guard(p.lock); resync = p.resync; }
Playback::AudioGap gap;
{ std::lock_guard<std::mutex> guard(p.lock); resync = p.resync; p.workerAt = nowUs(); gap = p.audioGap; }
/* Callback сохраняет снимок; печать выполняется worker без блокировки аудиоустройства. */
if (gap.count != reportedGaps && nowUs() - gapReportAt >= 1000000) {
DEBUG_WARN(DEBUG_CATEGORY_DEBUG,
"desktop audio gap id=%llu count=%llu delta=%llu silence_frames=%llu cause=%s got=%u/%u "
"reserve_us=%lld feed_age_us=%lld worker_age_us=%lld audio_pts=%lld videoQ=%zu",
(unsigned long long)p.id, (unsigned long long)gap.count, (unsigned long long)(gap.count - reportedGaps),
(unsigned long long)gap.silentFrames, gap.held ? "video_hold" : "pcm_short", gap.received, gap.requested,
(long long)gap.reserve, (long long)gap.feedAge, (long long)gap.workerAge, (long long)gap.audioPts, gap.videoQueue);
reportedGaps = gap.count; gapReportAt = nowUs();
}
if (resync) reset(true);
observe();
drainVideo();
@ -359,10 +387,15 @@ void StreamPlayerWindow::decode() {
observe();
if (record.pts > INT64_MAX) { fail(text("Invalid stream timestamp")); break; }
if (record.kind == StreamRecordKind::Audio) {
int result = 0;
int result = 0, wanted = 0, stillWanted = 0;
{ std::lock_guard<std::mutex> guard(p.lock);
if (p.sound) wanted = stream_playback_audio_needs_input(p.sound);
if (p.sound) result = stream_playback_audio_feed(p.sound, (int64_t)record.pts,
reinterpret_cast<const uint8_t*>(record.data.constData()), (size_t)record.data.size()); }
reinterpret_cast<const uint8_t*>(record.data.constData()), (size_t)record.data.size());
if (p.sound && result == 0) { p.audioFedAt = nowUs(); p.audioFedPts = (int64_t)record.pts; }
if (p.sound) stillWanted = stream_playback_audio_needs_input(p.sound); }
DEBUG_TRACE(DEBUG_CATEGORY_DEBUG, "desktop audio feed id=%llu pts=%llu result=%d needs_input=%d->%d",
(unsigned long long)p.id, (unsigned long long)record.pts, result, wanted, stillWanted);
if (result == -2) reset(true);
else if (result < 0) { fail(text("Stream audio decoder failed")); break; }
continue;

16
tools/chatgui/src/voice_audio_io.cpp

@ -36,8 +36,13 @@ VoiceAudioIo::OutputLatency VoiceAudioIo::outputLatency() const {
return m_running && voice_time_us() - m_outputLatency.sampledUs <= 500000 ? m_outputLatency : OutputLatency{};
}
bool VoiceAudioIo::start(Kind kind, uint64_t session) {
bool VoiceAudioIo::start(Kind kind, uint64_t session, unsigned bufferMs) {
stop();
if (!bufferMs) bufferMs = kind == Kind::Call ? 20 : 40;
if (bufferMs < 20 || bufferMs > 80 || bufferMs % 20) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "voice-io: invalid PCM buffer=%ums expected=20/40/60/80ms", bufferMs); return false;
}
m_playbackBufferMs = bufferMs;
m_kind = kind; m_session = session;
m_captureEvents.reset(); m_references.reset(); m_playback.reset(); m_playProgress.reset();
m_progressDropped = 0;
@ -50,8 +55,9 @@ bool VoiceAudioIo::start(Kind kind, uint64_t session) {
m_running = false;
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "voice-io: RX worker start failed reason=%s", error.what()); return false;
}
DEBUG_INFO(DEBUG_CATEGORY_CALL, "voice-io: started session=%llu kind=%s PCM ready=40ms plus partial block; calls adaptive=20..80ms",
(unsigned long long)session, kind == Kind::Call ? "call" : "radio");
DEBUG_INFO(DEBUG_CATEGORY_CALL, "voice-io: started session=%llu kind=%s PCM_buffer=%ums plus partial block adaptive=%d max=%ums",
(unsigned long long)session, kind == Kind::Call ? "call" : "radio", bufferMs, kind == Kind::Call,
kind == Kind::Call ? 80 : bufferMs);
return true;
}
@ -175,7 +181,7 @@ void VoiceAudioIo::playbackLoop() {
int64_t lastClockReport = 0;
VoiceLatencyRange queueUs, pullUs;
unsigned missing = 0;
VoicePlaybackBuffer buffering;
VoicePlaybackBuffer buffering(m_playbackBufferMs);
PlaybackProgress lastProgress{};
int64_t lastProgressUs = 0;
call_clock_snapshot clock{};
@ -241,7 +247,7 @@ void VoiceAudioIo::playbackLoop() {
(unsigned long long)snapshot.played_samples, (unsigned long long)snapshot.written_samples,
(long long)snapshot.pts_us, snapshot.valid, snapshot.measured, snapshot.held_back_us);
}
if (m_playback.size() < (m_kind == Kind::Call ? buffering.blocks : 2)) {
if (m_playback.size() < buffering.blocks) {
Playback block{}; block.epoch = epoch; block.position = position;
int64_t pullAt = voice_time_us();
if (m_kind == Kind::Call) {

5
tools/chatgui/src/voice_audio_io.h

@ -17,7 +17,9 @@ public:
OutputLatency outputLatency() const;
VoiceAudioIo() = default;
~VoiceAudioIo();
bool start(Kind kind, uint64_t session);
/* bufferMs = 20..80, кратно 20; 0 выбирает звонок 20 мс / рацию 40 мс.
* Звонок увеличивает запас при starvation и возвращается к выбранному минимуму. */
bool start(Kind kind, uint64_t session, unsigned bufferMs = 0);
void stop();
bool startCapture(int channels, bool aecEnabled);
void stopCapture();
@ -59,6 +61,7 @@ private:
Kind m_kind = Kind::Call;
uint64_t m_session = 0;
int m_channels = 1;
unsigned m_playbackBufferMs = 20;
std::atomic<bool> m_aecRequested{false};
std::thread m_captureThread, m_playbackThread;
std::mutex m_waitMutex;

11
tools/chatgui/src/voice_playback_buffer.h

@ -2,17 +2,19 @@
#include <algorithm>
#include <cstdint>
/* Резерв готового PCM desktop. Размер callback задаёт нижнюю границу; недостача добавляет
/* Резерв готового PCM desktop: bufferMs = 20..80, кратно 20 (проверяет VoiceAudioIo::start).
* Размер callback и настроенный запас задают нижнюю границу; недостача добавляет
* один блок, 10 секунд стабильного вывода убирают один. Частично сыгранный блок — сверх резерва.
* Только RX worker; время монотонное. Четыре PCM-слота доступны в AudioEventQueue<Playback, 12>. */
class VoicePlaybackBuffer {
public:
unsigned blocks = 2;
explicit VoicePlaybackBuffer(unsigned bufferMs = 20) : minimumBlocks(bufferMs / 20) { reset(); }
unsigned blocks = 1;
unsigned callbackFrames = 960;
void reset() { blocks = 2; callbackFrames = 960; stableSince = 0; lastGrowth = 0; lastSample = 0; probeUs = 10000000; }
void reset() { blocks = minimumBlocks; callbackFrames = 960; stableSince = 0; lastGrowth = 0; lastSample = 0; probeUs = 10000000; }
void sample(int64_t nowUs, unsigned frames, bool missing) {
callbackFrames = frames;
unsigned minimum = std::clamp((frames + 959) / 960, 1u, 4u);
unsigned minimum = std::max(minimumBlocks, std::clamp((frames + 959) / 960, 1u, 4u));
if (!stableSince || nowUs - lastSample > 1000000) stableSince = nowUs;
lastSample = nowUs;
if (minimum > blocks) { blocks = minimum; stableSince = nowUs; }
@ -27,5 +29,6 @@ public:
}
}
private:
unsigned minimumBlocks;
int64_t stableSince = 0, lastGrowth = 0, lastSample = 0, probeUs = 10000000;
};

33
tools/chatgui/tests/test_stream_audio.cpp

@ -17,7 +17,7 @@ static void test_stereo_tempo() {
auto* encoder = opus_codec_encoder_create(48000, 2);
CHECK(encoder && opus_codec_encoder_application_set(encoder, OPUS_CODEC_APP_AUDIO) == 0);
CHECK(opus_codec_encoder_bitrate_set(encoder, 128000) == 0);
auto* audio = stream_playback_audio_create(0); CHECK(audio);
auto* audio = stream_playback_audio_create(0, 40); CHECK(audio);
stream_playback_audio_set_tempo(audio, 1100);
std::vector<int16_t> decoded;
int16_t pcm[1920], output[1920]; uint8_t packet[1275];
@ -61,7 +61,7 @@ static void test_stereo_tempo() {
}
static void test_silence_clock() {
auto* encoder = opus_codec_encoder_create(48000, 2); auto* audio = stream_playback_audio_create(1000000);
auto* encoder = opus_codec_encoder_create(48000, 2); auto* audio = stream_playback_audio_create(1000000, 40);
CHECK(encoder && audio);
int16_t pcm[1920] = {0}, out[1920]; uint8_t packet[1275];
for (unsigned i = 0; i < 4; i++) {
@ -82,9 +82,36 @@ static void test_silence_clock() {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] device silence freezes source time and preserves stereo decoder state");
}
/* Готовые пакеты ждут worker: больший запас покрывает паузу его планирования между callback устройства. */
static void test_pcm_buffer() {
CHECK(!stream_playback_audio_create(0, 0));
CHECK(!stream_playback_audio_create(0, 1001));
auto* encoder = opus_codec_encoder_create(48000, 2); CHECK(encoder);
int16_t pcm[1920] = {0}, output[960]; uint8_t packet[1275];
int length = opus_codec_encode(encoder, pcm, 960, packet, sizeof(packet)); CHECK(length > 0);
for (unsigned bufferMs : {20u, 40u}) {
auto* audio = stream_playback_audio_create(0, bufferMs); CHECK(audio);
unsigned packets = 0;
while (stream_playback_audio_needs_input(audio) && packets < 10) {
CHECK(stream_playback_audio_feed(audio, packets * 20000, packet, (size_t)length) == 0); ++packets;
}
CHECK(!stream_playback_audio_needs_input(audio));
unsigned played = 0;
for (unsigned callback = 0; callback < 4; callback++) {
int count = stream_playback_audio_pull(audio, output, 480, INT64_MAX); CHECK(count >= 0);
played += (unsigned)count;
}
CHECK(bufferMs == 40 ? played == 1920 : played < 1920);
CHECK(stream_playback_audio_needs_input(audio));
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] PCM buffer=%ums worker_gap=40ms played=%u/1920 frames", bufferMs, played);
stream_playback_audio_destroy(audio);
}
opus_codec_encoder_destroy(encoder);
}
int main() {
debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
size_t allocations = u_get_allocated_count();
test_stereo_tempo(); test_silence_clock(); CHECK(u_get_allocated_count() == allocations);
test_stereo_tempo(); test_silence_clock(); test_pcm_buffer(); CHECK(u_get_allocated_count() == allocations);
return 0;
}

15
tools/chatgui/tests/test_stream_view.cpp

@ -68,6 +68,8 @@ extern "C" void __wrap_avcodec_free_context(AVCodecContext** context) {
#ifdef TEST_NULL_AUDIO
static std::atomic<ma_device_data_proc> originalAudioCallback{nullptr};
static std::atomic<uint64_t> nonzeroSamples{0};
static std::atomic<uint64_t> silenceGaps{0};
static unsigned silentRun = 0; // Только callback; счётчик промежутков >=1мс между тестовыми тонами.
static std::atomic<uint64_t> leftEnergy{0}, rightEnergy{0};
static std::atomic<int64_t> playbackDelayUs{0};
extern "C" int64_t __real_audio_stream_frame_time(ma_device*, unsigned int, int*);
@ -90,6 +92,8 @@ static void observeAudio(ma_device* device, void* out, const void* in, ma_uint32
left += static_cast<int64_t>(pcm[2 * i]) * pcm[2 * i];
right += static_cast<int64_t>(pcm[2 * i + 1]) * pcm[2 * i + 1];
nonzero += pcm[2 * i] != 0; nonzero += pcm[2 * i + 1] != 0;
if (pcm[2 * i] || pcm[2 * i + 1]) silentRun = 0;
else if (++silentRun == 48) silenceGaps.fetch_add(1);
}
leftEnergy.fetch_add(left); rightEnergy.fetch_add(right); nonzeroSamples.fetch_add(nonzero);
}
@ -145,6 +149,10 @@ class StreamViewTest : public QObject {
private slots:
void initTestCase() {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
if (qEnvironmentVariableIsSet("UTUN_TEST_DEBUG")) {
debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_TRACE);
debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_TRACE);
}
debug_set_category_level(DEBUG_CATEGORY_SYS, DEBUG_LEVEL_ERROR);
allocations = u_get_allocated_count(); ua = uasync_create(); QVERIFY(ua);
inst = utun_instance_create_from_str(ua,
@ -266,8 +274,15 @@ private slots:
});
sender.start(100);
QTRY_VERIFY_WITH_TIMEOUT(shown.count() > 0, 5000);
QTest::qWait(400); // Пройти старт декодера и физическую задержку тестового устройства.
#ifdef TEST_NULL_AUDIO
uint64_t gapsBefore = silenceGaps.load();
#endif
int beforeSteady = shown.count(); QTest::qWait(1500);
QVERIFY2(shown.count() >= beforeSteady + 30, "30fps video must keep advancing with 200ms physical audio latency");
#ifdef TEST_NULL_AUDIO
QCOMPARE(silenceGaps.load(), gapsBefore);
#endif
int beforeSwitch = shown.count(); revision++; rotation = 270;
QTRY_VERIFY_WITH_TIMEOUT(shown.count() > beforeSwitch + 2, 5000);
stream_info switched{}; QCOMPARE(stream_get_info(publisher, &switched), STREAM_OK);

24
tools/chatgui/tests/test_voice_audio_io.cpp

@ -110,7 +110,7 @@ static void small_chunks() {
VoiceAudioIo voice;
REQUIRE(voice.start(VoiceAudioIo::Kind::Call, 46));
for (unsigned offset = 0; offset < total; offset += chunk) {
await([&] { return unsigned(pulls.load()) * 960 >= offset + chunk + 960; });
await([&] { return unsigned(pulls.load()) * 960 >= offset + chunk; });
float output[chunk * 2]{};
voice.mix(output, chunk, now_us() + 10000, true);
await([&] { return output_position.load() == offset + chunk; });
@ -161,9 +161,10 @@ int main() {
debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
// Резерв уменьшается только при стабильной работе; крупный callback и starvation повышают минимум.
VoicePlaybackBuffer buffering;
REQUIRE(buffering.blocks == 1);
buffering.sample(1, 512, false);
for (int64_t t = 1000000; t <= 10000000; t += 1000000) buffering.sample(t, 512, false);
REQUIRE(buffering.blocks == 2);
REQUIRE(buffering.blocks == 1);
buffering.sample(10000001, 512, false); REQUIRE(buffering.blocks == 1);
buffering.sample(10020001, 512, true); REQUIRE(buffering.blocks == 2);
buffering.sample(10020002, 512, true); REQUIRE(buffering.blocks == 2);
@ -173,9 +174,14 @@ int main() {
buffering.sample(50520002, 512, false); REQUIRE(buffering.blocks == 2);
for (int64_t t = 51520002; t <= 90520002; t += 1000000) buffering.sample(t, 512, false);
REQUIRE(buffering.blocks == 1);
buffering.reset(); REQUIRE(buffering.blocks == 2);
buffering.reset(); REQUIRE(buffering.blocks == 1);
buffering.sample(1, 512, true); REQUIRE(buffering.blocks == 2);
buffering.sample(1, 512, false);
buffering.sample(20000001, 512, false); REQUIRE(buffering.blocks == 2); // Пауза не означает стабильный вывод.
VoicePlaybackBuffer radioBuffer(40);
for (int64_t t = 1; t <= 20000001; t += 1000000) radioBuffer.sample(t, 512, false);
REQUIRE(radioBuffer.blocks == 2); // Настроенный минимум не уменьшается адаптацией.
radioBuffer.reset(); REQUIRE(radioBuffer.blocks == 2);
// Десять минут +/-300ppm дрейфа: связь с физическим временем сохраняется, шага >2мс нет.
for (double drift : {-0.0003, 0.0003}) {
VoiceCaptureClock clock;
@ -190,6 +196,16 @@ int main() {
clock.reset(); REQUIRE(clock.map(123, 900000) == 900000);
}
VoiceAudioIo voice;
REQUIRE(!voice.start(VoiceAudioIo::Kind::Call, 1, 15));
REQUIRE(!voice.start(VoiceAudioIo::Kind::Call, 1, 100));
for (auto kind : {VoiceAudioIo::Kind::Call, VoiceAudioIo::Kind::Radio}) {
pulls = 0; REQUIRE(voice.start(kind, 1));
unsigned expected = kind == VoiceAudioIo::Kind::Call ? 1 : 2;
await([&] { return unsigned(pulls.load()) >= expected; });
voice.stop(); REQUIRE(unsigned(pulls.load()) == expected);
}
pulls = 0; REQUIRE(voice.start(VoiceAudioIo::Kind::Call, 1, 40));
await([] { return pulls.load() >= 2; }); voice.stop(); REQUIRE(pulls == 2);
std::array<int16_t, 960> pcm{}; pcm.fill(1234);
// Отсутствие PCM должно попадать в лог без необходимости ждать следующего кадра.
debug_set_log_hook(capture_diagnostic);
@ -239,7 +255,7 @@ int main() {
voice.stop(); reset();
pulls = 0; output_position = 0;
REQUIRE(voice.start(VoiceAudioIo::Kind::Call, 42));
await([] { return pulls.load() >= 2; });
await([] { return pulls.load() >= 1; });
REQUIRE(output_position == 0);
voice.mix(partial, 240, now_us() + 500000, true);
await([] { return output_position.load() == 240; });

Loading…
Cancel
Save