jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Recover stream playback stalls with bounded proportional catchup

Browse Source
This commit is contained in:
evgeny committed 2026-10-21 00:56:47 +02:00
1 parent 58c0fc6bba
commit 6c619d5f2a
3 files changed
+58 -58
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/stream/stream_playback_clock.c
+17 -16
View File
@@ -3,13 +3,13 @@
#include "../../lib/debug_config.h"
#include "stream_playback_clock.h"
/* Новая временная шкала не наследует минутные минимумы прежней позиции. */
/* Новая временная шкала не наследует минимумы прежней позиции. */
void stream_playback_clock_reset(struct stream_playback_clock *c, uint64_t id) {
if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback clock reset: null clock id=%llu", (unsigned long long)id); return; }
memset(c, 0, sizeof(*c));
c->id = id; c->tempo_milli = 1000;
for (unsigned i = 0; i < STREAM_PLAYBACK_HISTORY; i++) c->history[i].second = -1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "playback clock reset id=%llu history=60s tempo=1.0..1.1", (unsigned long long)id);
for (unsigned i = 0; i < STREAM_PLAYBACK_HISTORY; i++) c->history[i].period = -1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "playback clock reset id=%llu history=200ms tempo=1.0..1.6", (unsigned long long)id);
}
/* Смещение обновляется по реально воспроизведённому материалу, а не по запрошенному темпу. */
@@ -35,7 +35,7 @@ int stream_playback_clock_update(struct stream_playback_clock *c, int64_t now, i
c->correction_us += elapsed - advanced;
if (elapsed > 1000000) {
c->last_bad_us = now;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "playback clock sampling gap id=%llu interval_us=%lld; restart healthy minute",
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "playback clock sampling gap id=%llu interval_us=%lld; restart observation window",
(unsigned long long)c->id, (long long)elapsed);
}
}
@@ -53,34 +53,35 @@ int stream_playback_clock_update(struct stream_playback_clock *c, int64_t now, i
(unsigned long long)c->id, (long long)(now - c->pause_us),
(long long)(ready - position), (long long)c->correction_us);
}
int64_t second = now / 1000000, reserve = ready - position;
struct stream_playback_bucket *bucket = &c->history[second % STREAM_PLAYBACK_HISTORY];
int64_t period = now / STREAM_PLAYBACK_BUCKET_US, reserve = ready - position;
struct stream_playback_bucket *bucket = &c->history[period % STREAM_PLAYBACK_HISTORY];
int64_t normalized = reserve - c->correction_us;
if (bucket->second != second) { bucket->second = second; bucket->minimum_us = normalized; }
if (bucket->period != period) { bucket->period = period; bucket->minimum_us = normalized; }
else if (normalized < bucket->minimum_us) bucket->minimum_us = normalized;
int64_t minimum = normalized;
unsigned samples = 0;
for (unsigned i = 0; i < STREAM_PLAYBACK_HISTORY; i++) {
const struct stream_playback_bucket *b = &c->history[i];
if (b->second < 0 || b->second > second || second - b->second >= STREAM_PLAYBACK_HISTORY) continue;
if (b->period < 0 || b->period > period || period - b->period >= STREAM_PLAYBACK_HISTORY) continue;
if (b->minimum_us < minimum) minimum = b->minimum_us;
samples++;
}
minimum += c->correction_us;
int64_t stable = now - c->last_bad_us;
int complete = samples == STREAM_PLAYBACK_HISTORY && now - c->first_us >= 60000000;
int complete = samples == STREAM_PLAYBACK_HISTORY && now - c->first_us >= STREAM_PLAYBACK_SETTLE_US;
int64_t removable = minimum - guard - pending_gain;
if (removable < 0) removable = 0;
/* Торможение начинается с учётом будущего расхода при плавном спаде 1.1 -> 1.0. */
/* Уже подготовленный PCM и тормозной путь не должны съесть защищённый запас. */
int excess = c->tempo_milli - 1000;
int64_t braking = (int64_t)excess * excess * STREAM_PLAYBACK_RAMP_US / 200000;
int range = STREAM_PLAYBACK_MAX_TEMPO - 1000;
int64_t braking = (int64_t)excess * excess * STREAM_PLAYBACK_RAMP_US / (2 * range * 1000);
int target = 1000;
if (complete && stable >= 60000000 && !c->waiting && removable > braking) {
if (complete && stable >= STREAM_PLAYBACK_SETTLE_US && !c->waiting && removable > braking) {
int64_t allowance = removable - braking;
int increase = allowance >= 50000 ? 100 : (int)(allowance * 100 / 50000);
int increase = allowance >= range * 2000 ? range : (int)(allowance / 2000);
target += increase;
}
c->ramp_remainder += (elapsed > STREAM_PLAYBACK_RAMP_US ? STREAM_PLAYBACK_RAMP_US : elapsed) * 100;
c->ramp_remainder += (elapsed > STREAM_PLAYBACK_RAMP_US ? STREAM_PLAYBACK_RAMP_US : elapsed) * range;
int step = (int)(c->ramp_remainder / STREAM_PLAYBACK_RAMP_US);
c->ramp_remainder %= STREAM_PLAYBACK_RAMP_US;
if (target > c->tempo_milli) c->tempo_milli += target - c->tempo_milli < step ? target - c->tempo_milli : step;
@@ -90,9 +91,9 @@ int stream_playback_clock_update(struct stream_playback_clock *c, int64_t now, i
.correction_us = c->correction_us, .stable_us = stable, .removable_us = removable,
.tempo_milli = c->tempo_milli, .waiting = c->waiting, .history_ready = complete};
if (!c->report_us || now - c->report_us >= 1000000) {
const char *reason = c->waiting ? "starved" : !complete ? "history" : stable < 60000000 ? "cooldown"
const char *reason = c->waiting ? "starved" : !complete ? "history" : stable < STREAM_PLAYBACK_SETTLE_US ? "cooldown"
: removable <= braking ? "reserve" : "catchup";
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
"playback reserve id=%llu pts=%lld ready=%lld reserve=%lld minimum=%lld correction=%lld "
"stable_ms=%lld history=%u tempo=%d pending_gain=%lld waiting=%d "
"guard_us=%lld removable_us=%lld braking_us=%lld reason=%s",
src/stream/stream_playback_clock.h
+7 -5
View File
@@ -1,9 +1,9 @@
/* Общая адаптация воспроизведения desktop/Android, без сетевых ресурсов и очередей медиа.
* Один поток либо внешняя блокировка. position_us — фактические общие часы исходного материала;
* ready_us — непрерывная готовая граница, включая данные в кольце/кодеке/устройстве.
* Шестьдесят секундных минимумов нормализованы по смещению wall-clock относительно media-clock:
* Минимумы короткого окна нормализованы по смещению wall-clock относительно media-clock:
* простой увеличивает все прежние запасы, догон уменьшает их без переписывания истории.
* Ускорение до 1.1 разрешено после полной минуты без нехватки/пропуска наблюдений.
* После 200 мс без фактической недостачи избыток погашается плавным ускорением до 1.6.
* guard_us и уже поставленный устройству pending_gain_us защищают остаток при снижении темпа.
* reset нужен при новой эпохе/реальном GAP/смене позиции; временная нехватка историю сохраняет. */
#ifndef STREAM_PLAYBACK_CLOCK_H
@@ -15,11 +15,13 @@
extern "C" {
#endif
#define STREAM_PLAYBACK_HISTORY 60
#define STREAM_PLAYBACK_MAX_TEMPO 1100
#define STREAM_PLAYBACK_HISTORY 4
#define STREAM_PLAYBACK_BUCKET_US 50000
#define STREAM_PLAYBACK_SETTLE_US 200000
#define STREAM_PLAYBACK_MAX_TEMPO 1600
#define STREAM_PLAYBACK_RAMP_US 500000
struct stream_playback_bucket { int64_t second, minimum_us; };
struct stream_playback_bucket { int64_t period, minimum_us; };
struct stream_playback_clock {
struct stream_playback_bucket history[STREAM_PLAYBACK_HISTORY];
uint64_t id, pauses;
tests/test_stream_playback.c
+34 -37
View File
@@ -13,49 +13,46 @@ static void test_history_correction(void) {
struct stream_playback_clock clock; struct stream_playback_state state;
stream_playback_clock_reset(&clock, 1);
CHECK(!stream_playback_clock_update(&clock, 0, 0, 120000, 20000, 0, 0, &state));
CHECK(!stream_playback_clock_update(&clock, 1000000, 1000000, 1300000, 20000, 0, 0, &state));
CHECK(!stream_playback_clock_update(&clock, 50000, 50000, 350000, 20000, 0, 0, &state));
CHECK(state.minimum_us == 120000 && state.correction_us == 0);
CHECK(!stream_playback_clock_update(&clock, 1080000, 1000000, 1300000, 20000, 0, 1, &state));
CHECK(!stream_playback_clock_update(&clock, 130000, 50000, 350000, 20000, 0, 1, &state));
CHECK(state.minimum_us == 200000 && state.correction_us == 80000 && state.waiting);
CHECK(!stream_playback_clock_update(&clock, 1090000, 1010000, 1310000, 20000, 0, 0, &state));
CHECK(!stream_playback_clock_update(&clock, 140000, 60000, 360000, 20000, 0, 0, &state));
CHECK(!state.waiting && state.tempo_milli == 1000);
CHECK(!stream_playback_clock_update(&clock, 2090000, 2110000, 2410000, 20000, 0, 0, &state));
CHECK(state.minimum_us == 100000 && state.correction_us == -20000);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] history: pause shifts all minima +80ms, actual x1.1 shifts -100ms");
CHECK(!stream_playback_clock_update(&clock, 150000, 90000, 390000, 20000, 0, 0, &state));
CHECK(state.minimum_us == 180000 && state.correction_us == 60000);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] history: pause shifts minima +80ms, actual media progress repays 20ms");
}
static void test_catchup_device_queue(void) {
struct stream_playback_clock clock; struct stream_playback_state state;
stream_playback_clock_reset(&clock, 2);
int queued[100]; for (unsigned i = 0; i < 100; i++) queued[i] = 1000;
int64_t position = 0;
int accelerated = 0, maximum = 1000;
for (int64_t tick = 0; tick <= 120000; tick++) {
unsigned slot = (unsigned)(tick % 100);
if (tick) position += queued[slot];
int64_t pending_gain = 0;
for (unsigned i = 0; i < 100; i++) pending_gain += queued[i] - 1000;
CHECK(!stream_playback_clock_update(&clock, tick * 1000, position, tick * 1000 + 300000,
30000, pending_gain, 0, &state));
CHECK(state.tempo_milli >= 1000 && state.tempo_milli <= 1100);
if (tick < 60000) CHECK(state.tempo_milli == 1000);
if (state.tempo_milli > 1000) accelerated = 1;
if (state.tempo_milli > maximum) maximum = state.tempo_milli;
CHECK(state.minimum_us >= 29000);
queued[slot] = state.tempo_milli;
for (unsigned pause_ms = 200; pause_ms <= 4000; pause_ms += 3800) {
struct stream_playback_clock clock; struct stream_playback_state state;
stream_playback_clock_reset(&clock, pause_ms);
/* 250ms prepared PCM + 100ms устройства: старый темп нельзя заменить задним числом. */
int queued[350]; for (unsigned i = 0; i < 350; i++) queued[i] = 1000;
int64_t position = 0, recovered_ms = -1;
int maximum = 1000;
for (int64_t tick = 0; tick <= 30000; tick++) {
unsigned slot = (unsigned)(tick % 350);
int starved = tick >= 1000 && tick < 1000 + pause_ms;
if (tick && !starved) position += queued[slot];
int64_t pending_gain = 0;
for (unsigned i = 0; i < 350; i++) pending_gain += queued[i] - 1000;
CHECK(!stream_playback_clock_update(&clock, tick * 1000, position, tick * 1000 + 60000,
60000, pending_gain, starved, &state));
CHECK(state.tempo_milli >= 1000 && state.tempo_milli <= STREAM_PLAYBACK_MAX_TEMPO);
if (starved) CHECK(state.tempo_milli == 1000);
if (state.tempo_milli > maximum) maximum = state.tempo_milli;
CHECK(state.reserve_us >= 59000);
if (tick > 1000 + pause_ms && state.reserve_us < 63000 && recovered_ms < 0) recovered_ms = tick;
queued[slot] = state.tempo_milli;
}
CHECK(maximum > 1040 && state.history_ready && recovered_ms > 0 && recovered_ms < 25000);
CHECK(state.reserve_us < 63000 && state.tempo_milli == 1000);
if (pause_ms == 4000) CHECK(maximum == 1600);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] catchup: pause=%ums queued=350ms recovered_at=%lldms max_tempo=%d reserve=%lldus",
pause_ms, (long long)recovered_ms, maximum, (long long)state.reserve_us);
}
CHECK(accelerated && maximum == 1100 && state.history_ready);
CHECK(state.reserve_us >= 29000 && state.reserve_us < 33000 && state.tempo_milli == 1000);
/* Короткая остановка, попавшая между секундными отсчётами, блокирует догон на полную минуту. */
int64_t now = 120000000;
CHECK(!stream_playback_clock_update(&clock, now + 10000, position, position, 30000, 0, 1, &state));
CHECK(!stream_playback_clock_update(&clock, now + 20000, position, position + 500000, 30000, 0, 0, &state));
for (int64_t tick = 1; tick < 60000; tick++) {
CHECK(!stream_playback_clock_update(&clock, now + 20000 + tick * 1000, position + tick * 1000,
position + tick * 1000 + 500000, 30000, 0, 0, &state));
CHECK(state.tempo_milli == 1000);
}
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] catchup: healthy minute, smooth x1.1, queued audio, safe residual and pause cooldown");
}
static void test_invalid_and_sampling_gap(void) {
@@ -68,7 +65,7 @@ static void test_invalid_and_sampling_gap(void) {
before = state;
CHECK(stream_playback_clock_update(&clock, 2000001, 1000000, 2200000, 20000, 0, 0, &state) < 0);
CHECK(!memcmp(&state, &before, sizeof(state)));
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] clock: invalid/backward positions preserve result, missed samples reset healthy minute");
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] clock: invalid/backward positions preserve result, missed samples restart observation");
}
static const struct stream_descriptor descriptor = {.group_id = 1, .source_node_id = 2, .source_epoch = 3,