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

Share bounded media publication and send telemetry across frontends

master
evgeny 6 days ago
parent
a5f8b8431d
commit
f5d112a712
3 changed files with 335 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      src/Makefile.am
  2. 301
      src/stream/stream_publisher.c
  3. 32
      src/stream/stream_publisher.h

2
src/Makefile.am
Unescape View File

@ -19,6 +19,8 @@ stream_core_files = \
stream/stream_proto.h \
stream/stream_media.h \
stream/stream_media.c \
stream/stream_publisher.h \
stream/stream_publisher.c \
stream/stream_media_buffer.h \
stream/stream_media_buffer.c \
stream/stream_playback_clock.h \

301
src/stream/stream_publisher.c
Unescape View File

@ -0,0 +1,301 @@
#include <pthread.h>
#include <string.h>
#include "../../lib/mem.h"
#include "../../lib/debug_config.h"
#include "../../lib/u_async.h"
#include "../utun_instance.h"
#include "stream.h"
#include "stream_network.h"
#include "stream_publisher.h"
#define TX_LIMIT (2u * 1024u * 1024u)
#define TX_RECORDS 128
#define MEDIA_AGE_US 250000u
struct publisher_record {
struct publisher_record *next;
uint64_t pts, frame_id, created;
size_t length, offset;
int key;
uint8_t data[];
};
struct publisher_queue { struct publisher_record *head, *tail; size_t bytes; unsigned count; };
struct stream_publisher {
pthread_mutex_t lock;
struct UTUN_INSTANCE *inst;
struct STREAM_HANDLE *handle;
struct posted_task *post;
void *timer, *stats_timer;
struct stream_media_format format;
struct stream_publisher_status status;
struct publisher_queue video, audio;
uint64_t group, next_frame, last_pts[2], drain_until, report_us;
int seen[2], stopping;
};
static void publisher_process(void *arg);
/* Очередь producer/actor защищена mutex; LL_QUEUE между потоками не применяется. */
static void record_append(struct publisher_queue *q, struct publisher_record *r) {
r->next = NULL;
if (q->tail) q->tail->next = r; else q->head = r;
q->tail = r; q->bytes += r->length; q->count++;
}
static struct publisher_record *record_take(struct publisher_queue *q) {
struct publisher_record *r = q->head;
if (!r) return NULL;
q->head = r->next; if (!q->head) q->tail = NULL;
q->bytes -= r->length; q->count--; return r;
}
static void records_clear(struct publisher_queue *q) {
struct publisher_record *r;
while ((r = record_take(q))) u_free(r);
}
/* Под lock; резервированный post не может потерять принятую команду. */
static int publisher_schedule(struct stream_publisher *p) {
if (p->post || p->timer) return 0;
if (!p->inst || !uasync_is_running(p->inst->ua)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher schedule: core stopped"); return -1;
}
p->post = u_calloc(1, sizeof(*p->post));
if (!p->post) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher post allocation failed"); return -1; }
p->post->callback = publisher_process; p->post->arg = p;
uasync_post_reserved(p->inst->ua, p->post); return 0;
}
/* Только uasync под lock; терминальный снимок остаётся доступен до следующего open. */
static void publisher_close(struct stream_publisher *p, int reason) {
if (p->timer) uasync_cancel_timeout(p->inst->ua, p->timer);
if (p->stats_timer) uasync_cancel_timeout(p->inst->ua, p->stats_timer);
p->timer = p->stats_timer = NULL;
if (p->handle) stream_close(p->handle);
p->handle = NULL; records_clear(&p->video); records_clear(&p->audio);
p->status.state = reason ? STREAM_PUBLISHER_FAILED : STREAM_PUBLISHER_ENDED; p->status.reason = reason;
p->status.video_bytes = p->status.audio_bytes = p->status.total_bytes = 0;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "publisher closed id=%llu epoch=%llx reason=%d",
(unsigned long long)p->status.id, (unsigned long long)p->status.epoch, reason);
}
/* Один снимок всех соединений, без повторного учёта input_send_q внутри inflight. */
static void publisher_stats(void *arg) {
struct stream_publisher *p = arg;
pthread_mutex_lock(&p->lock); p->stats_timer = NULL;
if (p->handle && stream_handle_active(p->handle)) {
uint64_t now = get_time_us(); struct etcp_tx_snapshot transport; struct stream_info info;
if (etcp_tx_get_snapshot(p->inst, now, &transport) == 0 && stream_get_info(p->handle, &info) == STREAM_OK) {
p->status.sample_us = now; p->status.video_bytes = p->video.bytes - (p->video.head ? p->video.head->offset : 0);
p->status.audio_bytes = p->audio.bytes; p->status.stream_bytes = stream_network_pending_bytes(p->inst, &info.descriptor);
p->status.transport_bytes = transport.queued_bytes; p->status.inflight_bytes = transport.inflight_bytes;
p->status.acked_1s = transport.acked_1s; p->status.connections = transport.connections;
p->status.total_bytes = p->status.video_bytes + p->status.audio_bytes + p->status.stream_bytes
+ p->status.transport_bytes + p->status.inflight_bytes;
if (now - p->report_us >= 1000000) {
p->report_us = now;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
"publisher TX id=%llu video=%llu audio=%llu RS=%llu ETCP=%llu inflight=%llu total=%llu ACK_1s=%llu conns=%u KEY=%d",
(unsigned long long)p->status.id, (unsigned long long)p->status.video_bytes, (unsigned long long)p->status.audio_bytes,
(unsigned long long)p->status.stream_bytes, (unsigned long long)p->status.transport_bytes,
(unsigned long long)p->status.inflight_bytes, (unsigned long long)p->status.total_bytes,
(unsigned long long)p->status.acked_1s, p->status.connections, p->status.need_key);
}
}
p->stats_timer = uasync_set_timeout(p->inst->ua, 1000, p, publisher_stats, "publisher_stats");
if (!p->stats_timer) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher stats timer allocation failed"); publisher_close(p, STREAM_CLOSE_FAILED); }
} else if (p->handle) publisher_close(p, stream_handle_reason(p->handle));
pthread_mutex_unlock(&p->lock);
}
/* Упаковка общая для обеих платформ: Opus целиком, видео с ограниченными продолжениями. */
static int publisher_block(struct stream_publisher *p) {
struct stream_media_block block = {.format = p->format};
uint8_t wire[STREAM_DEFAULT_INPUT_CAPACITY];
size_t capacity = stream_publish_capacity(p->handle), used = STREAM_MEDIA_HEADER_SIZE;
if (capacity > sizeof(wire) || capacity < 256) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher invalid capacity=%zu", capacity); return STREAM_ERR_STATE;
}
for (struct publisher_record *r = p->audio.head; r && block.audio_count < 8; r = r->next) {
if (used + STREAM_MEDIA_AUDIO_SIZE + r->length + STREAM_MEDIA_VIDEO_SIZE >= capacity) break;
block.audio[block.audio_count++] = (struct stream_media_audio){r->pts, (uint32_t)r->length, 960, r->data};
used += STREAM_MEDIA_AUDIO_SIZE + r->length;
}
for (struct publisher_record *r = p->video.head; r && block.video_count < STREAM_MEDIA_MAX_ENTRIES; r = r->next) {
if (used + STREAM_MEDIA_VIDEO_SIZE >= capacity) break;
size_t take = capacity - used - STREAM_MEDIA_VIDEO_SIZE;
if (take > r->length - r->offset) take = r->length - r->offset;
block.video[block.video_count++] = (struct stream_media_video){r->frame_id, r->pts, (uint32_t)r->length,
(uint32_t)r->offset, (uint32_t)take, (uint8_t)r->key, r->data + r->offset};
used += STREAM_MEDIA_VIDEO_SIZE + take;
if (take < r->length - r->offset) break;
}
if (!block.video_count && !block.audio_count) return STREAM_WAIT;
size_t length = 0;
int result = stream_media_encode(&block, wire, capacity, &length);
if (result == STREAM_OK) result = stream_publish_block(p->handle, wire, length);
if (result != STREAM_OK) return result;
for (unsigned i = 0; i < block.audio_count; i++) u_free(record_take(&p->audio));
for (unsigned i = 0; i < block.video_count; i++) {
p->video.head->offset += block.video[i].length;
if (p->video.head->offset == p->video.head->length) u_free(record_take(&p->video));
}
return STREAM_OK;
}
/* Не более двух RS-пачек за проход; таймер продолжает работу, не занимая сетевой поток. */
static void publisher_process(void *arg) {
struct stream_publisher *p = arg;
pthread_mutex_lock(&p->lock); p->post = NULL; p->timer = NULL;
if (p->status.state == STREAM_PUBLISHER_OPENING) {
int result = p->stopping ? STREAM_ERR_STATE : stream_publish_open(p->inst, p->group, NULL, &p->handle);
if (result == STREAM_OK) result = stream_media_enable(p->handle, 0);
struct stream_info info;
if (result == STREAM_OK) result = stream_get_info(p->handle, &info);
if (result != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher open failed id=%llu group=%llx result=%d",
(unsigned long long)p->status.id, (unsigned long long)p->group, result);
publisher_close(p, p->stopping ? 0 : STREAM_CLOSE_FAILED);
} else {
p->status.epoch = info.descriptor.source_epoch; p->status.state = STREAM_PUBLISHER_LIVE;
p->stats_timer = uasync_set_timeout(p->inst->ua, 1000, p, publisher_stats, "publisher_stats");
if (!p->stats_timer) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher stats start failed"); publisher_close(p, STREAM_CLOSE_FAILED); }
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "publisher opened id=%llu group=%llx epoch=%llx codec=%u size=%ux%u audio=%u",
(unsigned long long)p->status.id, (unsigned long long)p->group, (unsigned long long)p->status.epoch,
p->format.video_codec, p->format.width, p->format.height, p->format.audio_codec);
}
}
uint64_t now = get_time_us();
if (p->handle && !stream_handle_active(p->handle)) publisher_close(p, stream_handle_reason(p->handle));
if (p->video.head && now - p->video.head->created > MEDIA_AGE_US) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher video expired bytes=%zu records=%u; require KEY", p->video.bytes, p->video.count);
records_clear(&p->video); p->status.need_key = 1;
}
while (p->audio.head && now - p->audio.head->created > MEDIA_AGE_US) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher audio expired pts=%llu", (unsigned long long)p->audio.head->pts);
u_free(record_take(&p->audio));
}
if (p->handle && p->status.state != STREAM_PUBLISHER_DRAINING) {
for (unsigned i = 0; i < 2 && (p->video.head || p->audio.head); i++) {
size_t pending = p->video.bytes + p->audio.bytes - (p->video.head ? p->video.head->offset : 0);
uint64_t oldest = p->video.head ? p->video.head->created : p->audio.head->created;
if (p->audio.head && p->audio.head->created < oldest) oldest = p->audio.head->created;
int immediate = p->video.head && p->video.head->key;
if (!p->stopping && !immediate && pending + STREAM_MEDIA_HEADER_SIZE + STREAM_MEDIA_VIDEO_SIZE < stream_publish_capacity(p->handle)
&& now - oldest < 20000) break;
int result = publisher_block(p);
if (result < 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher block failed result=%d", result); publisher_close(p, STREAM_CLOSE_FAILED); break; }
}
if (p->handle && p->stopping && !p->video.head && !p->audio.head) {
if (stream_publish_finish(p->handle) != STREAM_OK) publisher_close(p, STREAM_CLOSE_FAILED);
else { p->status.state = STREAM_PUBLISHER_DRAINING; p->drain_until = now + 2000000; }
}
}
if (p->handle && p->status.state == STREAM_PUBLISHER_DRAINING && now >= p->drain_until) publisher_close(p, 0);
if (p->handle && (p->video.head || p->audio.head || p->status.state == STREAM_PUBLISHER_DRAINING)) {
p->timer = uasync_set_timeout(p->inst->ua, p->status.state == STREAM_PUBLISHER_DRAINING ? 200 : 50,
p, publisher_process, "publisher_media");
if (!p->timer) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher media timer allocation failed"); publisher_close(p, STREAM_CLOSE_FAILED); }
}
pthread_mutex_unlock(&p->lock);
}
/* Контекст принадлежит bridge, который снимает producers перед destroy. */
int stream_publisher_create(struct stream_publisher **out, struct UTUN_INSTANCE *inst) {
if (!out || *out || !inst || !inst->stream) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher create invalid owner"); return -1; }
struct stream_publisher *p = u_calloc(1, sizeof(*p));
if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher allocation failed"); return -1; }
if (pthread_mutex_init(&p->lock, NULL) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher mutex failed"); u_free(p); return -1; }
p->inst = inst; *out = p; return 0;
}
void stream_publisher_destroy(struct stream_publisher *p) {
if (!p) return;
pthread_mutex_lock(&p->lock);
if (p->post) uasync_cancel_post(p->inst->ua, p->post);
publisher_close(p, STREAM_CLOSE_SERVICE); p->inst = NULL;
pthread_mutex_unlock(&p->lock); pthread_mutex_destroy(&p->lock); u_free(p);
}
/* Wire формат не допускает смену audio codec внутри одной эпохи. */
static int format_valid(const struct stream_media_format *f) {
return f && f->revision && f->width && f->width <= 4096 && f->height && f->height <= 4096
&& f->rotation <= 270 && f->rotation % 90 == 0 && f->video_codec >= STREAM_MEDIA_H264
&& f->video_codec <= STREAM_MEDIA_H266 && f->audio_codec <= STREAM_MEDIA_OPUS_STEREO;
}
int stream_publisher_open(struct stream_publisher *p, uint64_t id, uint64_t group, const struct stream_media_format *f) {
if (!p || !id || !group || !format_valid(f)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher open invalid configuration"); return -1; }
pthread_mutex_lock(&p->lock); int result = -1;
if (p->handle || p->post || p->status.state == STREAM_PUBLISHER_OPENING) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher open busy id=%llu owner=%llu", (unsigned long long)id, (unsigned long long)p->status.id);
} else if (publisher_schedule(p) == 0) {
p->status = (struct stream_publisher_status){.id = id, .state = STREAM_PUBLISHER_OPENING, .need_key = 1};
p->group = group; p->format = *f; p->next_frame = 0; p->stopping = 0; memset(p->seen, 0, sizeof(p->seen)); result = 0;
}
pthread_mutex_unlock(&p->lock); return result;
}
/* Admission сохраняет PTS только принятого кадра; новая revision коммитится атомарно с KEY. */
int stream_publisher_feed(struct stream_publisher *p, uint64_t id, int video, uint64_t pts, int key,
const uint8_t *data, size_t length, const struct stream_media_format *f) {
if (!p || !data || !length || pts > INT64_MAX || length > (video ? STREAM_MEDIA_MAX_FRAME : STREAM_MEDIA_MAX_OPUS)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher feed invalid data length=%zu", length); return -1;
}
pthread_mutex_lock(&p->lock); int result = -1, track = video ? 0 : 1, change = 0;
if (p->status.id != id || p->stopping || (p->status.state != STREAM_PUBLISHER_OPENING && p->status.state != STREAM_PUBLISHER_LIVE)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher feed stale/stopped id=%llu owner=%llu state=%d",
(unsigned long long)id, (unsigned long long)p->status.id, p->status.state); goto done;
}
if (video) {
if (!format_valid(f) || f->audio_codec != p->format.audio_codec) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher invalid video format/audio codec"); goto done;
}
change = f->revision > p->format.revision;
if (f->revision < p->format.revision || (change && !key)) { result = 0; goto done; }
if (!change && (f->width != p->format.width || f->height != p->format.height
|| f->rotation != p->format.rotation || f->video_codec != p->format.video_codec)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher changed format without revision=%u", f->revision); goto done;
}
}
if ((!video && !p->format.audio_codec) || (p->seen[track] && pts <= p->last_pts[track])) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher invalid track/PTS pts=%llu", (unsigned long long)pts); goto done;
}
if (video && !change && (p->video.bytes + p->audio.bytes + length > TX_LIMIT || p->video.count + p->audio.count >= TX_RECORDS)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher TX overload video=%zu audio=%zu; require KEY", p->video.bytes, p->audio.bytes);
records_clear(&p->video); p->status.need_key = 1;
}
if (video && p->status.need_key && !key) { result = 0; goto done; }
if ((change ? 0 : p->video.bytes) + p->audio.bytes + length > TX_LIMIT
|| (change ? 0 : p->video.count) + p->audio.count >= TX_RECORDS) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "publisher admission refused video=%d bytes=%zu", video, length); result = 0; goto done;
}
struct publisher_record *r = u_calloc(1, sizeof(*r) + length);
if (!r) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher record allocation failed bytes=%zu", length); goto done; }
if (publisher_schedule(p) < 0) { u_free(r); goto done; }
if (change) {
records_clear(&p->video);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "publisher revision committed id=%llu %u->%u codec=%u size=%ux%u rotation=%u",
(unsigned long long)id, p->format.revision, f->revision, f->video_codec, f->width, f->height, f->rotation);
p->format = *f;
}
r->length = length; r->pts = pts; r->created = get_time_us(); r->key = key ? STREAM_MEDIA_KEY : 0;
if (video) { r->frame_id = ++p->next_frame; p->status.need_key = 0; }
memcpy(r->data, data, length); record_append(video ? &p->video : &p->audio, r);
p->seen[track] = 1; p->last_pts[track] = pts; result = 1;
done:
pthread_mutex_unlock(&p->lock); return result;
}
/* Stop идемпотентен; принятые кадры дренируются actor перед FIN. */
void stream_publisher_stop(struct stream_publisher *p, uint64_t id) {
if (!p) return;
pthread_mutex_lock(&p->lock);
if (p->status.id == id && p->status.state < STREAM_PUBLISHER_DRAINING) {
p->stopping = 1;
if (p->status.state != STREAM_PUBLISHER_OPENING) p->status.state = STREAM_PUBLISHER_STOPPING;
if (publisher_schedule(p) < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher stop scheduling failed");
}
pthread_mutex_unlock(&p->lock);
}
int stream_publisher_snapshot(struct stream_publisher *p, uint64_t id, struct stream_publisher_status *out) {
if (!p || !out) return -1;
pthread_mutex_lock(&p->lock); int result = p->status.id == id && id ? 0 : -1;
if (!result) *out = p->status;
pthread_mutex_unlock(&p->lock); return result;
}

32
src/stream/stream_publisher.h
Unescape View File

@ -0,0 +1,32 @@
/* Общий publisher Android/desktop: принимает полные access units и Opus, публикует uAV2/RS.
* create/destroy выполняются в uasync вне callbacks, внешний bridge сериализует destroy с producers.
* open/feed/stop/snapshot потокобезопасны; вызывающие используют session ID, не pointers ядра.
* Один publisher на контекст. Snapshot содержит состояние и 100ms телеметрию, callbacks не устанавливаются.
* TX ограничен 2MiB/128 records/250ms; потеря видео требует KEY. Stop завершает очередь и держит FIN 2s. */
#ifndef STREAM_PUBLISHER_H
#define STREAM_PUBLISHER_H
#include "stream_media.h"
#ifdef __cplusplus
extern "C" {
#endif
struct UTUN_INSTANCE;
struct stream_publisher;
enum stream_publisher_state { STREAM_PUBLISHER_OPENING = 1, STREAM_PUBLISHER_LIVE, STREAM_PUBLISHER_STOPPING,
STREAM_PUBLISHER_DRAINING, STREAM_PUBLISHER_ENDED, STREAM_PUBLISHER_FAILED };
struct stream_publisher_status {
uint64_t id, epoch, sample_us, video_bytes, audio_bytes, stream_bytes, transport_bytes, inflight_bytes, acked_1s, total_bytes;
unsigned connections;
int state, reason, need_key;
};
int stream_publisher_create(struct stream_publisher **out, struct UTUN_INSTANCE *inst);
void stream_publisher_destroy(struct stream_publisher *publisher);
int stream_publisher_open(struct stream_publisher *publisher, uint64_t id, uint64_t group, const struct stream_media_format *format);
/* 1 принято, 0 требуется KEY/перегрузка, -1 ошибка или старый/закрытый session ID. */
int stream_publisher_feed(struct stream_publisher *publisher, uint64_t id, int video, uint64_t pts, int key,
const uint8_t *data, size_t length, const struct stream_media_format *format);
void stream_publisher_stop(struct stream_publisher *publisher, uint64_t id);
int stream_publisher_snapshot(struct stream_publisher *publisher, uint64_t id, struct stream_publisher_status *status);
#ifdef __cplusplus
}
#endif
#endif
Write Preview
Loading…
Cancel
Save