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

Add bounded viewer batches and readiness notifications

Browse Source
This commit is contained in:
evgeny committed 2026-10-20 23:29:23 +02:00
1 parent 6997a82af1
commit 589cdcc81f
2 files changed
+108 -43
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/stream/stream_viewer.c
+93 -39
View File
@@ -12,15 +12,18 @@ struct viewer {
uint64_t id, group, source, epoch;
unsigned tracks;
struct STREAM_HANDLE *handle;
struct stream_viewer_record *record;
struct stream_viewer_record *record, *tail;
void (*notify)(void *);
void *notify_arg;
struct stream_media_status status;
int opening, stopping, resync, requested, waiting;
int video;
unsigned video, records;
int64_t position, audio_ahead;
uint64_t replies, fulfilled_repeats; /* Выданные ответы и погашенные ими повторные запросы. */
uint64_t created_us, prepared_us, report_us;
/* Время подготовки ответа и его ожидания до pull; статистика за отчётный интервал. */
uint64_t demand_us, delivered, prepare_total_us, prepare_max_us, handoff_total_us, handoff_max_us;
uint64_t batches, batch_records_max, batch_bytes_max;
unsigned first_records; /* Диагностика первой выдачи VIDEO/AUDIO; не влияет на cursor. */
};
struct stream_viewers {
@@ -46,7 +49,11 @@ static int schedule(struct stream_viewers *v) {
return 0;
}
/* Единственный слот ответа: память ограничена одним access unit на viewer, без общей очереди давления. */
void stream_viewer_records_free(struct stream_viewer_record *records) {
while (records) { struct stream_viewer_record *next = records->next; u_free(records); records = next; }
}
/* Батч копирует только разрешённые ближайшие записи; оставшийся поток хранится в общем кольце. */
static int output(struct viewer *s, enum stream_viewer_kind kind, int reason, const struct stream_media_packet *p) {
size_t length = !p ? 0 : kind == STREAM_VIEWER_VIDEO ? p->video.length : p->audio.length;
struct stream_viewer_record *r = u_calloc(1, sizeof(*r) + length);
@@ -62,7 +69,10 @@ static int output(struct viewer *s, enum stream_viewer_kind kind, int reason, co
r->flags = kind == STREAM_VIEWER_VIDEO ? p->video.flags : 0;
memcpy(r->data, kind == STREAM_VIEWER_VIDEO ? p->video.data : p->audio.data, length);
}
u_free(s->record); s->record = r; s->prepared_us = get_time_us();
if (s->tail) s->tail->next = r;
else s->record = r;
s->tail = r; s->prepared_us = get_time_us();
if (s->record == r && s->notify) s->notify(s->notify_arg);
return 0;
}
@@ -70,6 +80,9 @@ static int output(struct viewer *s, enum stream_viewer_kind kind, int reason, co
static void close_viewer(struct viewer *s, enum stream_viewer_kind kind, int reason) {
if (s->handle) stream_close(s->handle);
s->handle = NULL; s->opening = s->requested = s->waiting = s->resync = 0;
if (kind == STREAM_VIEWER_CLOSED) {
stream_viewer_records_free(s->record); s->record = s->tail = NULL;
}
if (kind) output(s, kind, reason, NULL);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer released id=%llu group=%llx source=%llx kind=%d reason=%d replies=%llu repeats=%llu",
(unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, kind, reason,
@@ -92,7 +105,7 @@ static void event(struct UTUN_INSTANCE *inst, const struct stream_event *e, void
pthread_mutex_unlock(&v->lock);
}
/* Не более одного read на reader за проход; MORE откладывает продолжение, WAIT не создаёт busy loop. */
/* Ограниченный батч на reader за проход; MORE/WAIT возвращают уже собранное без ожидания остальных записей. */
static void process(void *arg) {
struct stream_viewers *v = arg;
pthread_mutex_lock(&v->lock);
@@ -102,7 +115,7 @@ static void process(void *arg) {
struct viewer *s = *link;
if (s->stopping) {
close_viewer(s, 0, STREAM_CLOSE_STOPPED); *link = s->next;
u_free(s->record); u_free(s); continue;
stream_viewer_records_free(s->record); u_free(s); continue;
}
link = &s->next;
if (s->opening) {
@@ -134,7 +147,8 @@ static void process(void *arg) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
"viewer buffer id=%llu source=%llx epoch=%llx elapsed_us=%llu initialized=%d generation=%llu "
"start_us=%lld position_us=%lld ready_us=%lld video_us=%lld audio_us=%lld reserve_us=%lld "
"requested=%d waiting=%d slot=%d slot_age_us=%llu records=%llu prepare_us=%llu/%llu handoff_us=%llu/%llu timing=mean/max",
"requested=%d waiting=%d slot=%d slot_age_us=%llu records=%llu prepare_us=%llu/%llu handoff_us=%llu/%llu "
"batches=%llu batch_records_max=%llu batch_bytes_max=%llu timing=mean/max",
(unsigned long long)s->id, (unsigned long long)s->source, (unsigned long long)s->epoch,
(unsigned long long)(now - s->created_us), s->status.initialized, (unsigned long long)s->status.generation,
(long long)s->status.start_us, (long long)s->position, (long long)s->status.ready_us,
@@ -143,29 +157,45 @@ static void process(void *arg) {
s->requested, s->waiting, s->record ? s->record->kind : 0,
(unsigned long long)(s->record ? now - s->prepared_us : 0), (unsigned long long)s->delivered,
(unsigned long long)(s->delivered ? s->prepare_total_us / s->delivered : 0), (unsigned long long)s->prepare_max_us,
(unsigned long long)(s->delivered ? s->handoff_total_us / s->delivered : 0), (unsigned long long)s->handoff_max_us);
(unsigned long long)(s->delivered ? s->handoff_total_us / s->delivered : 0), (unsigned long long)s->handoff_max_us,
(unsigned long long)s->batches, (unsigned long long)s->batch_records_max, (unsigned long long)s->batch_bytes_max);
s->delivered = s->prepare_total_us = s->prepare_max_us = s->handoff_total_us = s->handoff_max_us = 0;
s->batches = s->batch_records_max = s->batch_bytes_max = 0;
}
if (!s->requested || s->waiting || s->record) continue;
s->requested = 0;
struct stream_media_packet p;
int result = stream_media_read(s->handle, s->position, s->video, s->audio_ahead, &p);
if (stream_media_get_status(s->handle, &s->status) < 0) result = STREAM_ERR_FORMAT;
if (result == STREAM_OK) {
if (p.kind == STREAM_MEDIA_READ_END) close_viewer(s, STREAM_VIEWER_END, 0);
else if (output(s, p.kind == STREAM_MEDIA_READ_VIDEO ? STREAM_VIEWER_VIDEO : STREAM_VIEWER_AUDIO, 0, &p) < 0)
close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED);
} else if (result == STREAM_MORE) {
s->requested = 1;
if (schedule(v) < 0) close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED);
} else if (result == STREAM_WAIT) s->requested = s->waiting = 1;
else if (result == STREAM_GAP || result == STREAM_EXPIRED) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer resync id=%llu result=%d", (unsigned long long)s->id, result);
s->position = -1; memset(&s->status, 0, sizeof(s->status));
if (stream_media_resync(s->handle) < 0 || output(s, STREAM_VIEWER_GAP, 0, NULL) < 0)
close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED);
} else {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer read failed id=%llu result=%d", (unsigned long long)s->id, result);
unsigned count = 0, video = s->video;
size_t bytes = 0;
uint64_t began = get_time_us();
do {
struct stream_media_packet p;
int result = stream_media_read(s->handle, s->position, video > 0, s->audio_ahead, &p);
if (result == STREAM_OK) {
if (p.kind == STREAM_MEDIA_READ_END) { close_viewer(s, STREAM_VIEWER_END, 0); break; }
int kind = p.kind == STREAM_MEDIA_READ_VIDEO ? STREAM_VIEWER_VIDEO : STREAM_VIEWER_AUDIO;
if (output(s, kind, 0, &p) < 0) { close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); break; }
count++; bytes += s->tail->length;
if (kind == STREAM_VIEWER_VIDEO) video--;
} else if (result == STREAM_MORE || result == STREAM_WAIT) {
if (!s->record) {
s->requested = 1; s->waiting = result == STREAM_WAIT;
if (!s->waiting && schedule(v) < 0) close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED);
}
break;
} else if (result == STREAM_GAP || result == STREAM_EXPIRED) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer resync id=%llu result=%d", (unsigned long long)s->id, result);
s->position = -1; memset(&s->status, 0, sizeof(s->status));
if (stream_media_resync(s->handle) < 0 || output(s, STREAM_VIEWER_GAP, 0, NULL) < 0)
close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED);
break;
} else {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer read failed id=%llu result=%d", (unsigned long long)s->id, result);
close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); break;
}
} while (count < s->records && bytes < 256u * 1024u && get_time_us() - began < 2000 && (video || s->audio_ahead >= 0));
if (!s->handle) continue;
if (stream_media_get_status(s->handle, &s->status) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer batch status failed id=%llu", (unsigned long long)s->id);
close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED);
}
}
@@ -188,7 +218,8 @@ void stream_viewers_destroy(struct stream_viewers *v) {
if (v->inst->stream) stream_set_event_callback(v->inst, NULL, NULL);
while (v->head) {
struct viewer *s = v->head; v->head = s->next;
close_viewer(s, 0, STREAM_CLOSE_SERVICE); u_free(s->record); u_free(s);
if (s->notify) s->notify(s->notify_arg);
close_viewer(s, 0, STREAM_CLOSE_SERVICE); stream_viewer_records_free(s->record); u_free(s);
}
pthread_mutex_destroy(&v->lock); u_free(v);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer manager detached");
@@ -208,7 +239,7 @@ int stream_viewer_open(struct stream_viewers *v, uint64_t id, uint64_t group, ui
else if (schedule(v) < 0) u_free(s);
else {
*s = (struct viewer){.next = v->head, .id = id, .group = group, .source = source, .epoch = epoch,
.tracks = tracks, .opening = 1, .position = -1, .audio_ahead = STREAM_MEDIA_READ_AHEAD_US,
.tracks = tracks, .opening = 1, .records = 1, .position = -1, .audio_ahead = STREAM_MEDIA_READ_AHEAD_US,
.created_us = get_time_us()};
v->head = s; result = 0;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer requested id=%llu group=%llx source=%llx epoch=%llx tracks=%u",
@@ -223,6 +254,8 @@ void stream_viewer_stop(struct stream_viewers *v, uint64_t id) {
pthread_mutex_lock(&v->lock);
struct viewer *s = find(v, id);
if (s) {
if (s->notify) s->notify(s->notify_arg);
s->notify = NULL; s->notify_arg = NULL;
s->stopping = 1;
if (schedule(v) < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer stop scheduling failed id=%llu", (unsigned long long)id);
}
@@ -233,8 +266,10 @@ struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *v, uint64
if (!v) return NULL;
pthread_mutex_lock(&v->lock);
struct viewer *s = find(v, id);
struct stream_viewer_record *r = s && !s->stopping ? s->record : NULL;
if (r) {
struct stream_viewer_record *head = s && !s->stopping ? s->record : NULL;
unsigned count = 0; size_t bytes = 0;
for (struct stream_viewer_record *r = head; r; r = r->next) {
count++; bytes += r->length;
unsigned bit = r->kind == STREAM_VIEWER_VIDEO ? 1 : r->kind == STREAM_VIEWER_AUDIO ? 2 : 0;
if (bit) {
uint64_t handoff = get_time_us() - s->prepared_us;
@@ -254,35 +289,54 @@ struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *v, uint64
(long long)s->status.start_us, (long long)s->status.ready_us,
(long long)(s->status.initialized ? s->status.ready_us - (int64_t)r->pts : -1));
}
/* Ответ обслуживает и повторный request: AVAILABLE не должен прочитать следующий кадр до нового demand. */
}
if (head) {
s->batches++;
if (count > s->batch_records_max) s->batch_records_max = count;
if (bytes > s->batch_bytes_max) s->batch_bytes_max = bytes;
/* Батч обслуживает и повторный request: AVAILABLE не должен читать сверх выданного бюджета. */
int repeated = s->requested;
s->record = NULL; s->requested = 0; s->replies++;
s->record = s->tail = NULL; s->requested = 0; s->replies++;
if (repeated) s->fulfilled_repeats++;
if (s->replies == 1 || (repeated && (s->fulfilled_repeats == 1 || s->fulfilled_repeats % 256 == 0)))
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "viewer reply id=%llu kind=%d pts=%llu repeated=%d replies=%llu repeats=%llu pending=0",
(unsigned long long)s->id, r->kind, (unsigned long long)r->pts, repeated,
(unsigned long long)s->id, head->kind, (unsigned long long)head->pts, repeated,
(unsigned long long)s->replies, (unsigned long long)s->fulfilled_repeats);
}
pthread_mutex_unlock(&v->lock); return r;
pthread_mutex_unlock(&v->lock); return head;
}
int stream_viewer_request(struct stream_viewers *v, uint64_t id, int64_t position, int video, int64_t audio_ahead) {
if (!v || (video != 0 && video != 1) || audio_ahead < -1 || audio_ahead > 60000000) {
int stream_viewer_request(struct stream_viewers *v, uint64_t id, int64_t position, unsigned video,
int64_t audio_ahead, unsigned records) {
if (!v || !records || records > STREAM_VIEWER_BATCH_RECORDS || video > records || audio_ahead < -1 || audio_ahead > 60000000) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer demand invalid"); return -1;
}
pthread_mutex_lock(&v->lock);
struct viewer *s = find(v, id);
int result = -1;
if (s && !s->stopping && (s->opening || s->handle)) {
int changed = s->position != position || s->video != video || s->audio_ahead != audio_ahead;
int changed = s->position != position || s->video != video || s->audio_ahead != audio_ahead || s->records != records;
if (!s->requested && !s->record) s->demand_us = get_time_us();
s->position = position; s->video = video; s->audio_ahead = audio_ahead; s->requested = video || audio_ahead >= 0;
s->position = position; s->video = video; s->audio_ahead = audio_ahead; s->records = records;
s->requested = video || audio_ahead >= 0;
result = 0;
if (s->requested && !s->record && (!s->waiting || changed)) { s->waiting = 0; result = schedule(v); }
}
pthread_mutex_unlock(&v->lock); return result;
}
int stream_viewer_set_notify(struct stream_viewers *v, uint64_t id, void (*notify)(void *), void *arg) {
if (!v) return -1;
pthread_mutex_lock(&v->lock);
struct viewer *s = find(v, id);
int result = s && !s->stopping ? 0 : -1;
if (!result) {
s->notify = notify; s->notify_arg = arg;
if (s->record && notify) notify(arg);
} else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "viewer notify unavailable id=%llu", (unsigned long long)id);
pthread_mutex_unlock(&v->lock); return result;
}
int stream_viewer_status(struct stream_viewers *v, uint64_t id, struct stream_media_status *status) {
if (!v || !status) return -1;
pthread_mutex_lock(&v->lock);
@@ -298,7 +352,7 @@ int stream_viewer_resync(struct stream_viewers *v, uint64_t id) {
struct viewer *s = find(v, id);
int result = -1;
if (s && s->handle && !s->stopping && schedule(v) == 0) {
u_free(s->record); s->record = NULL;
stream_viewer_records_free(s->record); s->record = s->tail = NULL;
memset(&s->status, 0, sizeof(s->status));
s->demand_us = get_time_us();
s->resync = s->requested = 1; s->waiting = 0; s->position = -1; result = 0;
src/stream/stream_viewer.h
+15 -4
View File
@@ -1,8 +1,8 @@
/* Общие viewers desktop/Android: независимый media cursor и один ближайший ответ на session ID.
/* Общие viewers desktop/Android: независимый media cursor и ограниченный батч на session ID.
* create/destroy — uasync вне callbacks; внешний bridge защищает destroy от параллельных callers.
* Остальные API потокобезопасны. Один manager занимает stream event callback ядра.
* stop снимает только своего reader; медленные/остановленные viewers не задерживают остальных.
* pull передаёт владение record вызывающему: освобождать u_free(). Кольца RS остаются общими. */
* pull передаёт владение цепочкой records вызывающему: освобождать records_free(). Кольца RS остаются общими. */
#ifndef STREAM_VIEWER_H
#define STREAM_VIEWER_H
#include "stream_media_buffer.h"
@@ -11,11 +11,13 @@ extern "C" {
#endif
struct UTUN_INSTANCE;
struct stream_viewers;
#define STREAM_VIEWER_BATCH_RECORDS 16u
enum stream_viewer_kind {
STREAM_VIEWER_VIDEO = 1, STREAM_VIEWER_AUDIO, STREAM_VIEWER_GAP,
STREAM_VIEWER_OPENED, STREAM_VIEWER_END, STREAM_VIEWER_CLOSED
};
struct stream_viewer_record {
struct stream_viewer_record *next; /* Следующая запись батча; NULL завершает цепочку. */
enum stream_viewer_kind kind;
struct stream_media_format format;
uint64_t pts; /* PTS только этого источника, микросекунды. */
@@ -30,12 +32,21 @@ void stream_viewers_destroy(struct stream_viewers *viewers);
* Открытие асинхронно: OPENED/CLOSED возвращается через pull. tracks — VIDEO или VIDEO|AUDIO. */
int stream_viewer_open(struct stream_viewers *viewers, uint64_t id, uint64_t group, uint64_t source, uint64_t epoch, unsigned tracks);
void stream_viewer_stop(struct stream_viewers *viewers, uint64_t id);
/* Один owned ответ или NULL. Выдача ответа погашает demand, включая повторный request до pull.
/* Owned батч или NULL. Выдача батча погашает demand, включая повторный request до pull.
* NULL сохраняет ожидающий запрос. Следующие данные требуют нового request; idle не копирует кольцо. */
struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *viewers, uint64_t id);
void stream_viewer_records_free(struct stream_viewer_record *records);
/* Позиция и audio_ahead относятся только к этому reader; повторный demand заменяет прежний.
* video — максимум видеокадров; records — максимум всех записей, 1..STREAM_VIEWER_BATCH_RECORDS.
* Батч отдаётся сразу при WAIT/MORE или после бюджета 2ms/256KiB, без ожидания заполнения.
* Последняя запись может пересечь порог байтов, но не превышает STREAM_MEDIA_MAX_FRAME.
* 0 — принято, -1 — неизвестный/закрытый ID или ошибка scheduling. */
int stream_viewer_request(struct stream_viewers *viewers, uint64_t id, int64_t position, int video, int64_t audio_ahead);
int stream_viewer_request(struct stream_viewers *viewers, uint64_t id, int64_t position, unsigned video,
int64_t audio_ahead, unsigned records);
/* Уведомление о готовом батче/закрытии, под lock manager: только разбудить worker, API ядра не вызывать.
* arg живёт до stop/destroy или снятия callback (notify=NULL); после возврата старый callback уже не выполняется.
* Уже готовый батч уведомляет сразу. 0 — установлен, -1 — ID отсутствует/остановлен. */
int stream_viewer_set_notify(struct stream_viewers *viewers, uint64_t id, void (*notify)(void *), void *arg);
/* Последний снимок сохраняется после END до stop. 0 — получен, -1 — ID отсутствует. */
int stream_viewer_status(struct stream_viewers *viewers, uint64_t id, struct stream_media_status *status);
/* Очистить ответ и искать ближайший KEY только этого reader. 0 — принято, -1 — отказ. */