From 36fceec6b2d71cfafea693b2ae7bb74dc4ed8364 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 01:49:07 +0200 Subject: [PATCH] Bridge Android live streams through bounded session queues --- .../libutun_lite/stream_bridge.c | 332 ++++++++++++++++++ .../libutun_lite/stream_bridge.h | 25 ++ 2 files changed, 357 insertions(+) create mode 100644 tools/chatgui-android/libutun_lite/stream_bridge.c create mode 100644 tools/chatgui-android/libutun_lite/stream_bridge.h diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c new file mode 100644 index 00000000..791c36a1 --- /dev/null +++ b/tools/chatgui-android/libutun_lite/stream_bridge.c @@ -0,0 +1,332 @@ +#include +#include + +#include "../../../lib/debug_config.h" +#include "../../../lib/mem.h" +#include "../../../lib/u_async.h" +#include "../../../src/utun_instance.h" +#include "../../../src/stream/stream.h" +#include "../../../src/stream/stream_media.h" +#include "stream_bridge.h" + +#define TX_LIMIT (2u * 1024u * 1024u) +#define RX_LIMIT (4u * 1024u * 1024u) +#define TX_RECORDS 128 +#define RX_RECORDS 256 +#define MEDIA_AGE_US 250000u + +struct media_record { + struct media_record *next; + uint64_t id, pts, frame_id, created; + size_t length, offset; + int key; + uint8_t data[]; +}; +struct media_queue { struct media_record *head, *tail; size_t bytes; unsigned count; }; +struct android_media_session { + uint64_t id, group, source, next_frame, last_pts[2], drain_until; + struct STREAM_HANDLE *handle; + struct stream_media_receiver *receiver; + struct stream_media_format format; + int opening, stopping, finished, need_key, seen[2], rx_overflow; +}; + +static pthread_mutex_t media_lock = PTHREAD_MUTEX_INITIALIZER; +static struct UTUN_INSTANCE *media_inst; +static struct posted_task *media_post; +static void *media_timer; +static struct android_media_session publisher, viewer; +static struct media_queue video_queue, audio_queue, output_queue; +static void media_process(void *arg); + +/* Очереди обмена pthread: LL_QUEUE принадлежит uasync и сюда не подходит. */ +static void queue_append(struct media_queue *q, struct media_record *record) { + record->next = NULL; + if (q->tail) q->tail->next = record; else q->head = record; + q->tail = record; q->bytes += record->length; q->count++; +} +static struct media_record *queue_take(struct media_queue *q) { + struct media_record *record = q->head; + if (!record) return NULL; + q->head = record->next; if (!q->head) q->tail = NULL; + q->bytes -= record->length; q->count--; record->next = NULL; + return record; +} +static void queue_clear(struct media_queue *q) { + struct media_record *record; + while ((record = queue_take(q))) u_free(record); +} + +/* Все обращения к ua, включая post, защищены тем же mutex, что и detach перед destroy. */ +static int media_schedule(void) { + if (media_post || media_timer) return 0; + if (!media_inst || !uasync_is_running(media_inst->ua)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream schedule: core detached/stopped"); return -1; + } + struct posted_task *task = u_calloc(1, sizeof(*task)); + if (!task) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream post allocation failed"); return -1; } + task->callback = media_process; + media_post = task; uasync_post_reserved(media_inst->ua, task); + return 0; +} + +/* Выдача владеет копией: lifetime scratch stream callback не выходит в Kotlin. */ +static void media_output(uint64_t id, int kind, const struct stream_media_format *format, uint64_t pts, + int flags, const uint8_t *data, size_t size) { + if (!id || size > STREAM_MEDIA_MAX_FRAME) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android output invalid id/size=%zu", size); return; } + if (output_queue.count >= RX_RECORDS || size + 20 > RX_LIMIT - output_queue.bytes) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream RX full id=%llu bytes=%zu count=%u", + (unsigned long long)id, output_queue.bytes, output_queue.count); + viewer.rx_overflow = 1; + queue_clear(&output_queue); + if (kind == 1 || kind == 2) { media_output(id, 3, NULL, 0, 0, NULL, 0); return; } + } + struct media_record *record = u_calloc(1, sizeof(*record) + size + 20); + if (!record) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream RX allocation failed bytes=%zu", size); viewer.rx_overflow = 1; return; + } + record->id = id; record->length = size + 20; record->created = get_time_us(); + uint8_t *p = record->data; p[0] = (uint8_t)kind; p[1] = (uint8_t)flags; + if (format) { + stream_write_be(p + 2, format->width, 2); stream_write_be(p + 4, format->height, 2); + stream_write_be(p + 6, format->rotation, 2); + } + stream_write_be(p + 8, pts, 8); stream_write_be(p + 16, size, 4); + if (size) memcpy(p + 20, data, size); + queue_append(&output_queue, record); +} + +/* Callback receiver работает под media_lock, декодирование видео/аудио выполняет внешний worker. */ +static void media_frame(void *arg, const struct stream_media_format *format, + const struct stream_media_video *video, const struct stream_media_audio *audio) { + struct android_media_session *s = arg; + if (s->rx_overflow) return; + if (video) media_output(s->id, 1, format, video->pts_us, video->flags, video->data, video->length); + if (audio) media_output(s->id, 2, format, audio->pts_us, 0, audio->data, audio->length); +} + +/* Состояния потока не передаются через chat SharedFlow; здесь только владельцы выбранного viewer. */ +static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { + (void)arg; + pthread_mutex_lock(&media_lock); + if (inst == media_inst && viewer.handle && event->descriptor.group_id == viewer.group + && event->descriptor.source_node_id == viewer.source) { + if (event->type == STREAM_EVENT_DATA) { + int result = stream_media_receiver_feed(viewer.receiver, event->data, event->length); + if (result != STREAM_OK || viewer.rx_overflow) { + stream_media_receiver_gap(viewer.receiver); viewer.rx_overflow = 0; + media_output(viewer.id, 3, NULL, 0, 0, NULL, 0); + } + } else if (event->type == STREAM_EVENT_GAP) { + stream_media_receiver_gap(viewer.receiver); media_output(viewer.id, 3, NULL, 0, 0, NULL, 0); + } else { + media_output(viewer.id, event->type == STREAM_EVENT_END ? 5 : 6, NULL, 0, event->reason, NULL, 0); + stream_close(viewer.handle); viewer.handle = NULL; + stream_media_receiver_destroy(viewer.receiver); viewer.receiver = NULL; viewer.stopping = 1; + } + } + pthread_mutex_unlock(&media_lock); +} + +/* Publisher держит конечное окно выдачи после FIN; abrupt detach ничего не обещает получателям. */ +static void media_close(struct android_media_session *s, int kind) { + media_output(s->id, kind, NULL, 0, 0, NULL, 0); + if (s->handle) stream_close(s->handle); + stream_media_receiver_destroy(s->receiver); + if (s == &publisher) { queue_clear(&video_queue); queue_clear(&audio_queue); } + memset(s, 0, sizeof(*s)); +} + +/* Сформировать ограниченную пачку: аудио целиком, видео с продолжениями и несколькими началами. */ +static int media_publish(void) { + struct stream_media_block block = {.format = publisher.format}; + uint8_t wire[7080]; size_t capacity = stream_publish_capacity(publisher.handle), used = STREAM_MEDIA_HEADER_SIZE; + if (capacity > sizeof(wire) || capacity < 256) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android publisher invalid capacity=%zu", capacity); return STREAM_ERR_STATE; + } + struct media_record *a = audio_queue.head; + while (a && block.audio_count < 8 && used + STREAM_MEDIA_AUDIO_SIZE + a->length + STREAM_MEDIA_VIDEO_SIZE < capacity) { + block.audio[block.audio_count++] = (struct stream_media_audio){a->pts, (uint32_t)a->length, 960, a->data}; + used += STREAM_MEDIA_AUDIO_SIZE + a->length; a = a->next; + } + struct media_record *v = video_queue.head; + while (v && block.video_count < STREAM_MEDIA_MAX_ENTRIES && used + STREAM_MEDIA_VIDEO_SIZE < capacity) { + size_t take = capacity - used - STREAM_MEDIA_VIDEO_SIZE; + if (take > v->length - v->offset) take = v->length - v->offset; + block.video[block.video_count++] = (struct stream_media_video){v->frame_id, v->pts, (uint32_t)v->length, + (uint32_t)v->offset, (uint32_t)take, (uint8_t)v->key, v->data + v->offset}; + used += STREAM_MEDIA_VIDEO_SIZE + take; + if (take < v->length - v->offset) break; + v = v->next; + } + 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(publisher.handle, wire, length); + if (result != STREAM_OK) return result; + for (unsigned i = 0; i < block.audio_count; i++) u_free(queue_take(&audio_queue)); + for (unsigned i = 0; i < block.video_count; i++) { + video_queue.head->offset += block.video[i].length; + if (video_queue.head->offset == video_queue.head->length) u_free(queue_take(&video_queue)); + } + return STREAM_OK; +} + +/* Один actor обслуживает команды и не больше двух FEC-пачек за вызов; idle не будит мобильное ядро. */ +static void media_process(void *arg) { + (void)arg; + pthread_mutex_lock(&media_lock); + media_post = NULL; media_timer = NULL; + if (!media_inst) { pthread_mutex_unlock(&media_lock); return; } + struct android_media_session *sessions[] = {&publisher, &viewer}; + for (unsigned i = 0; i < 2; i++) { + struct android_media_session *s = sessions[i]; + if (s->opening) { + s->opening = 0; + int result = s->source ? stream_subscribe_open(media_inst, s->group, s->source, &s->handle) + : stream_publish_open(media_inst, s->group, NULL, &s->handle); + if (result == STREAM_OK && s->source) result = stream_media_receiver_create(&s->receiver, media_frame, s); + if (result != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream 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); + media_close(s, 6); + } else { + media_output(s->id, 4, NULL, 0, 0, NULL, 0); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream opened id=%llu publisher=%d group=%llx source=%llx", + (unsigned long long)s->id, !s->source, (unsigned long long)s->group, (unsigned long long)s->source); + } + } + } + if (viewer.stopping && viewer.id) media_close(&viewer, 5); + if (publisher.handle && !stream_handle_active(publisher.handle)) media_close(&publisher, 6); + uint64_t now = get_time_us(); + if (video_queue.head && now - video_queue.head->created > MEDIA_AGE_US) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android TX video expired queued=%u bytes=%zu; require KEY", video_queue.count, video_queue.bytes); + queue_clear(&video_queue); publisher.need_key = 1; + } + while (audio_queue.head && now - audio_queue.head->created > MEDIA_AGE_US) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android TX audio expired pts=%llu", (unsigned long long)audio_queue.head->pts); + u_free(queue_take(&audio_queue)); + } + if (publisher.handle && !publisher.finished) { + for (unsigned i = 0; i < 2 && (video_queue.head || audio_queue.head); i++) { + int result = media_publish(); + if (result < 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media publish failed result=%d", result); media_close(&publisher, 6); break; } + } + if (publisher.handle && publisher.stopping && !video_queue.head && !audio_queue.head) { + if (stream_publish_finish(publisher.handle) != STREAM_OK) media_close(&publisher, 6); + else { publisher.finished = 1; publisher.drain_until = now + 2000000; } + } + } + if (publisher.finished && now >= publisher.drain_until) media_close(&publisher, 5); + if (video_queue.head || audio_queue.head || publisher.finished) { + media_timer = uasync_set_timeout(media_inst->ua, publisher.finished ? 200 : 50, NULL, media_process, "android_stream"); + if (!media_timer) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media timer allocation failed"); media_close(&publisher, 6); } + } + pthread_mutex_unlock(&media_lock); +} + +/* Attach/detach — обязательная граница lifetime ядра. */ +void android_stream_attach(struct UTUN_INSTANCE *inst) { + pthread_mutex_lock(&media_lock); + media_inst = inst; stream_set_event_callback(inst, media_event, NULL); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge attached"); + pthread_mutex_unlock(&media_lock); +} +void android_stream_detach(struct UTUN_INSTANCE *inst) { + pthread_mutex_lock(&media_lock); + if (media_inst == inst) { + if (media_post) uasync_cancel_post(inst->ua, media_post); + if (media_timer) uasync_cancel_timeout(inst->ua, media_timer); + media_post = NULL; media_timer = NULL; + stream_set_event_callback(inst, NULL, NULL); + if (publisher.id) media_close(&publisher, 6); + if (viewer.id) media_close(&viewer, 6); + media_inst = NULL; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge detached; producers rejected"); + } + pthread_mutex_unlock(&media_lock); +} + +/* Команда содержит идентификаторы, а не borrowed pointer ядра. */ +int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height, + unsigned rotation, int audio) { + if (!id || !group || (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid arguments"); return -1; + } + pthread_mutex_lock(&media_lock); + struct android_media_session *s = source ? &viewer : &publisher; + int result = -1; + if (s->id || id == publisher.id || id == viewer.id || !media_inst || !uasync_is_running(media_inst->ua)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open busy/detached id=%llu", (unsigned long long)id); + } else if (media_schedule() == 0) { + *s = (struct android_media_session){.id = id, .group = group, .source = source, .opening = 1, .need_key = 1, + .format = {(uint16_t)width, (uint16_t)height, (uint16_t)rotation, STREAM_MEDIA_H264, audio ? STREAM_MEDIA_OPUS : 0}}; + result = 0; + } + pthread_mutex_unlock(&media_lock); return result; +} + +/* Перегрузка удаляет целые видео-кадры и блокирует P-кадры до нового KEY. */ +int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length) { + if (!data || !length || length > (video ? STREAM_MEDIA_MAX_FRAME : STREAM_MEDIA_MAX_OPUS) || pts > INT64_MAX) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream feed invalid length=%zu", length); return -1; + } + pthread_mutex_lock(&media_lock); + int result = -1, track = video ? 0 : 1; + if (!media_inst || publisher.id != id || publisher.stopping || publisher.finished) goto done; + if ((!video && !publisher.format.audio_codec) || (publisher.seen[track] && pts <= publisher.last_pts[track])) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream feed invalid track/PTS id=%llu pts=%llu", + (unsigned long long)id, (unsigned long long)pts); goto done; + } + if (video && (video_queue.bytes + audio_queue.bytes + length > TX_LIMIT + || video_queue.count + audio_queue.count >= TX_RECORDS)) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android TX overload bytes=%zu count=%u; require KEY", + video_queue.bytes + audio_queue.bytes, video_queue.count + audio_queue.count); + queue_clear(&video_queue); publisher.need_key = 1; + } + if (video && publisher.need_key && !key) { result = 0; goto done; } + if (video_queue.bytes + audio_queue.bytes + length > TX_LIMIT || video_queue.count + audio_queue.count >= TX_RECORDS) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android TX admission refused length=%zu video=%d", length, video); result = 0; goto done; + } + struct media_record *record = u_calloc(1, sizeof(*record) + length); + if (!record) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android TX allocation failed bytes=%zu", length); goto done; } + if (media_schedule() < 0) { u_free(record); goto done; } + record->length = length; record->pts = pts; record->created = get_time_us(); record->key = key ? STREAM_MEDIA_KEY : 0; + if (video) { record->frame_id = ++publisher.next_frame; publisher.need_key = 0; } + memcpy(record->data, data, length); queue_append(video ? &video_queue : &audio_queue, record); + publisher.seen[track] = 1; publisher.last_pts[track] = pts; result = 1; +done: + pthread_mutex_unlock(&media_lock); return result; +} + +/* Stop идемпотентен; publisher дренирует принятую очередь, viewer закрывается сразу в actor. */ +void android_stream_stop(uint64_t id) { + pthread_mutex_lock(&media_lock); + struct android_media_session *s = publisher.id == id ? &publisher : viewer.id == id ? &viewer : NULL; + if (s) { s->stopping = 1; if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream stop schedule failed"); } + pthread_mutex_unlock(&media_lock); +} + +/* Выбрать только события текущего owner; старые owner-события имеют ограниченный TTL. */ +uint8_t *android_stream_pull(uint64_t id, size_t *length) { + if (!length) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream pull missing length"); return NULL; } + *length = 0; pthread_mutex_lock(&media_lock); + struct media_record **link = &output_queue.head, *previous = NULL; + uint8_t *data = NULL; uint64_t now = get_time_us(); + while (*link) { + struct media_record *r = *link; + if (r->id == id || now - r->created > 5000000) { + *link = r->next; if (output_queue.tail == r) output_queue.tail = previous; + output_queue.bytes -= r->length; output_queue.count--; + if (r->id == id) { + data = u_malloc((uint32_t)r->length); + if (data) { memcpy(data, r->data, r->length); *length = r->length; } + else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream pull allocation failed"); + u_free(r); break; + } + u_free(r); + } else { previous = r; link = &r->next; } + } + pthread_mutex_unlock(&media_lock); return data; +} diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.h b/tools/chatgui-android/libutun_lite/stream_bridge.h new file mode 100644 index 00000000..4232af5b --- /dev/null +++ b/tools/chatgui-android/libutun_lite/stream_bridge.h @@ -0,0 +1,25 @@ +/* Android-очереди live media: attach/detach только в uasync, вне callbacks. + * UI/кодировщики используют только session_id, никогда не сохраняют inst/ua. + * Один publisher и один viewer, TX <=2MB/128 records, RX <=4MB/256 records. + * Data принимает полные H.264 Annex B кадры (KEY содержит SPS/PPS/IDR) и Opus 48k mono/20ms. + * pull передаёт владение u_malloc-буфером: kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,data. + * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6. Числа big-endian. + * Медиаданные передаются через pull, не через общий чатовый event bus. */ +#ifndef ANDROID_STREAM_BRIDGE_H +#define ANDROID_STREAM_BRIDGE_H + +#include +#include + +struct UTUN_INSTANCE; +void android_stream_attach(struct UTUN_INSTANCE *inst); +void android_stream_detach(struct UTUN_INSTANCE *inst); +int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height, + unsigned rotation, int audio); +/* source=0 — publisher; source!=0 — viewer. 0 означает, что команда принята, результат приходит через pull. */ +int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length); +/* feed: 1 accepted, 0 dropped (video: запросить KEY), -1 detached/stopped/invalid. */ +void android_stream_stop(uint64_t id); +uint8_t *android_stream_pull(uint64_t id, size_t *length); + +#endif