diff --git a/src/stream/stream_viewer.c b/src/stream/stream_viewer.c index 3749ac91..65656a1a 100644 --- a/src/stream/stream_viewer.c +++ b/src/stream/stream_viewer.c @@ -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; diff --git a/src/stream/stream_viewer.h b/src/stream/stream_viewer.h index 6daf9b6f..9fc91d90 100644 --- a/src/stream/stream_viewer.h +++ b/src/stream/stream_viewer.h @@ -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 — отказ. */