diff --git a/src/Makefile.am b/src/Makefile.am index 967fea4a..88037b49 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -21,6 +21,8 @@ stream_core_files = \ stream/stream_media.c \ stream/stream_publisher.h \ stream/stream_publisher.c \ + stream/stream_viewer.h \ + stream/stream_viewer.c \ stream/stream_media_buffer.h \ stream/stream_media_buffer.c \ stream/stream_playback_clock.h \ diff --git a/src/stream/stream_viewer.c b/src/stream/stream_viewer.c new file mode 100644 index 00000000..72ba7ecb --- /dev/null +++ b/src/stream/stream_viewer.c @@ -0,0 +1,250 @@ +#include +#include +#include "../../lib/debug_config.h" +#include "../../lib/mem.h" +#include "../../lib/u_async.h" +#include "../utun_instance.h" +#include "stream.h" +#include "stream_viewer.h" + +struct viewer { + struct viewer *next; + uint64_t id, group, source, epoch; + unsigned tracks; + struct STREAM_HANDLE *handle; + struct stream_viewer_record *record; + struct stream_media_status status; + int opening, stopping, resync, requested, waiting; + int video; + int64_t position, audio_ahead; +}; +struct stream_viewers { + pthread_mutex_t lock; + struct UTUN_INSTANCE *inst; + struct posted_task *pending; + struct viewer *head; +}; +static void process(void *arg); + +static struct viewer *find(struct stream_viewers *v, uint64_t id) { + for (struct viewer *s = v->head; s; s = s->next) if (s->id == id) return s; + return NULL; +} + +/* Резервируем continuation до изменения команды; все обращения к ядру ограничены lifetime manager. */ +static int schedule(struct stream_viewers *v) { + if (v->pending) return 0; + struct posted_task *task = u_calloc(1, sizeof(*task)); + if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer task allocation failed"); return -1; } + task->callback = process; task->arg = v; + v->pending = task; uasync_post_reserved(v->inst->ua, task); + return 0; +} + +/* Единственный слот ответа: память ограничена одним access unit на viewer, без общей очереди давления. */ +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); + if (!r) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer output allocation failed id=%llu bytes=%zu", + (unsigned long long)s->id, length); + return -1; + } + r->kind = kind; r->reason = reason; r->length = length; + if (p) { + r->format = p->format; + r->pts = kind == STREAM_VIEWER_VIDEO ? p->video.pts_us : p->audio.pts_us; + 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; + return 0; +} + +/* Закрываем handle, сохраняя последний status/END до чтения и явного stop владельца. */ +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) output(s, kind, reason, NULL); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer released id=%llu group=%llx source=%llx kind=%d reason=%d", + (unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, kind, reason); +} + +/* Событие будит всех readers этой эпохи, не меняя их независимые позиции и запросы. */ +static void event(struct UTUN_INSTANCE *inst, const struct stream_event *e, void *arg) { + struct stream_viewers *v = arg; + pthread_mutex_lock(&v->lock); + for (struct viewer *s = v->head; s; s = s->next) { + if (inst != v->inst || !s->handle || s->stopping || s->group != e->descriptor.group_id + || s->source != e->descriptor.source_node_id || s->epoch != e->descriptor.source_epoch) continue; + if (!stream_handle_active(s->handle)) close_viewer(s, STREAM_VIEWER_CLOSED, stream_handle_reason(s->handle)); + else if (e->type == STREAM_EVENT_AVAILABLE) { + s->waiting = 0; + if (schedule(v) < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer ready scheduling failed id=%llu", (unsigned long long)s->id); + } + } + pthread_mutex_unlock(&v->lock); +} + +/* Не более одного read на reader за проход; MORE откладывает продолжение, WAIT не создаёт busy loop. */ +static void process(void *arg) { + struct stream_viewers *v = arg; + pthread_mutex_lock(&v->lock); + v->pending = NULL; + struct viewer **link = &v->head; + while (*link) { + 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; + } + link = &s->next; + if (s->opening) { + s->opening = 0; + int result = stream_media_open(v->inst, s->group, s->source, s->epoch, s->tracks, &s->handle); + if (result != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer open failed id=%llu group=%llx source=%llx result=%d", + (unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, result); + close_viewer(s, STREAM_VIEWER_CLOSED, result == STREAM_ERR_CONFLICT ? STREAM_CLOSE_SOURCE_CHANGED : STREAM_CLOSE_FAILED); + continue; + } + if (output(s, STREAM_VIEWER_OPENED, 0, NULL) < 0) { close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); continue; } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer opened id=%llu group=%llx source=%llx epoch=%llx tracks=%u", + (unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, + (unsigned long long)s->epoch, s->tracks); + } + if (!s->handle) continue; + if (s->resync) { + s->resync = s->waiting = 0; s->position = -1; + if (stream_media_resync(s->handle) < 0) { close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); continue; } + } + if (stream_media_get_status(s->handle, &s->status) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer index failed id=%llu", (unsigned long long)s->id); + close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); continue; + } + 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); + close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); + } + } + pthread_mutex_unlock(&v->lock); +} + +int stream_viewers_create(struct stream_viewers **out, struct UTUN_INSTANCE *inst) { + if (!out || *out || !inst || !inst->stream) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer manager invalid create"); return -1; } + struct stream_viewers *v = u_calloc(1, sizeof(*v)); + if (!v) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer manager allocation failed"); return -1; } + int result = pthread_mutex_init(&v->lock, NULL); + if (result) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer mutex init failed result=%d", result); u_free(v); return -1; } + v->inst = inst; *out = v; stream_set_event_callback(inst, event, v); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer manager attached"); return 0; +} + +void stream_viewers_destroy(struct stream_viewers *v) { + if (!v) return; + if (v->pending) uasync_cancel_post(v->inst->ua, v->pending); + 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); + } + pthread_mutex_destroy(&v->lock); u_free(v); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer manager detached"); +} + +int stream_viewer_open(struct stream_viewers *v, uint64_t id, uint64_t group, uint64_t source, uint64_t epoch, unsigned tracks) { + if (!v || !id || !group || !source || !epoch || !(tracks & STREAM_MEDIA_TRACK_VIDEO) + || (tracks & ~(STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO))) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer open invalid identity/tracks"); return -1; + } + pthread_mutex_lock(&v->lock); + int result = -1; + if (find(v, id)) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer duplicate id=%llu", (unsigned long long)id); + else { + struct viewer *s = u_calloc(1, sizeof(*s)); + if (!s) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer allocation failed id=%llu", (unsigned long long)id); + 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}; + v->head = s; result = 0; + } + } + pthread_mutex_unlock(&v->lock); return result; +} + +void stream_viewer_stop(struct stream_viewers *v, uint64_t id) { + if (!v) return; + pthread_mutex_lock(&v->lock); + struct viewer *s = find(v, id); + if (s) { + s->stopping = 1; + if (schedule(v) < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer stop scheduling failed id=%llu", (unsigned long long)id); + } + pthread_mutex_unlock(&v->lock); +} + +struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *v, uint64_t id) { + 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) s->record = NULL; + pthread_mutex_unlock(&v->lock); return r; +} + +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) { + 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; + s->position = position; s->video = video; s->audio_ahead = audio_ahead; 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_status(struct stream_viewers *v, uint64_t id, struct stream_media_status *status) { + if (!v || !status) return -1; + pthread_mutex_lock(&v->lock); + struct viewer *s = find(v, id); + int result = s && !s->stopping ? 0 : -1; + if (!result) *status = s->status; + pthread_mutex_unlock(&v->lock); return result; +} + +int stream_viewer_resync(struct stream_viewers *v, uint64_t id) { + if (!v) return -1; + pthread_mutex_lock(&v->lock); + 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; + memset(&s->status, 0, sizeof(s->status)); + s->resync = s->requested = 1; s->waiting = 0; s->position = -1; result = 0; + } + pthread_mutex_unlock(&v->lock); return result; +} diff --git a/src/stream/stream_viewer.h b/src/stream/stream_viewer.h new file mode 100644 index 00000000..ca44e584 --- /dev/null +++ b/src/stream/stream_viewer.h @@ -0,0 +1,45 @@ +/* Общие 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 остаются общими. */ +#ifndef STREAM_VIEWER_H +#define STREAM_VIEWER_H +#include "stream_media_buffer.h" +#ifdef __cplusplus +extern "C" { +#endif +struct UTUN_INSTANCE; +struct stream_viewers; +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 { + enum stream_viewer_kind kind; + struct stream_media_format format; + uint64_t pts; /* PTS только этого источника, микросекунды. */ + size_t length; /* Длина owned payload после структуры. */ + int flags, reason; /* VIDEO: KEY; CLOSED: stream_close_reason. */ + uint8_t data[]; +}; +/* 0 — успех, -1 — ошибка; out заранее NULL. */ +int stream_viewers_create(struct stream_viewers **out, struct UTUN_INSTANCE *inst); +void stream_viewers_destroy(struct stream_viewers *viewers); +/* Уникальный ненулевой ID принадлежит caller до stop, включая завершённый/ошибочный сеанс. + * Открытие асинхронно: 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. Новые данные запрашиваются request; idle не копирует кольцо. */ +struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *viewers, uint64_t id); +/* Позиция и audio_ahead относятся только к этому reader; повторный demand заменяет прежний. + * 0 — принято, -1 — неизвестный/закрытый ID или ошибка scheduling. */ +int stream_viewer_request(struct stream_viewers *viewers, uint64_t id, int64_t position, int video, int64_t audio_ahead); +/* Последний снимок сохраняется после 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 — отказ. */ +int stream_viewer_resync(struct stream_viewers *viewers, uint64_t id); +#ifdef __cplusplus +} +#endif +#endif