Add independent live stream viewer sessions in the shared core
This commit is contained in:
evgeny committed
1 parent
9c7b49e91f
commit
0bc3b62f46
3 files changed
+297
No files matched your search
@@ -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 \
|
||||
|
||||
@@ -0,0 +1,250 @@
|
||||
#include <pthread.h>
|
||||
#include <string.h>
|
||||
#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;
|
||||
}
|
||||
@@ -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
|
||||
Reference in new issue
Block a user