From 9fb4823df0f98f18dd7159155ad095ed7f0a7ca8 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 21 Oct 2026 00:13:01 +0200 Subject: [PATCH] Batch Android viewer records and wake readers on data or stop --- tests/test_android_stream.c | 69 ++++++++-- .../libutun_lite/stream_bridge.c | 121 +++++++++++++++--- .../libutun_lite/stream_bridge.h | 21 ++- 3 files changed, 174 insertions(+), 37 deletions(-) diff --git a/tests/test_android_stream.c b/tests/test_android_stream.c index cd910c4a..79f9dfa6 100644 --- a/tests/test_android_stream.c +++ b/tests/test_android_stream.c @@ -29,6 +29,12 @@ static int producer_started, producer_stop; static unsigned source_events; static uint64_t source_group; static char *source_snapshot; +struct signal_waiter { struct android_stream_signal *signal; int started, result; }; +static void *wait_signal(void *arg) { + struct signal_waiter *w = arg; + __atomic_store_n(&w->started, 1, __ATOMIC_RELEASE); + w->result = android_stream_signal_wait(w->signal, 50); return NULL; +} static void chat_event(struct UTUN_INSTANCE *inst, int type, const uint8_t *data, int length) { (void)inst; @@ -41,7 +47,7 @@ static void chat_event(struct UTUN_INSTANCE *inst, int type, const uint8_t *data static uint8_t *wait_event(uint64_t id, int kind, size_t *size) { for (unsigned i = 0; i < 1000; i++) { - if (id == 2 || id == 7) CHECK(android_stream_read_request(id, kind == 5 ? 1000000 : -1, 1, 80000) == 0); + if (id == 2 || id == 7) CHECK(android_stream_read_request(id, kind == 5 ? 1000000 : -1, 1, 80000, 1) == 0); uint8_t *data = android_stream_pull(id, size); if (data && data[0] == kind) return data; u_free(data); test_clock_poll(ua, 100); @@ -98,6 +104,10 @@ int main(void) { uint8_t *error = wait_event(5, 6, &size); CHECK(error[1] == STREAM_CLOSE_SOURCE_CHANGED); u_free(error); CHECK(android_stream_open(2, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0); u_free(wait_event(2, 4, &size)); + struct android_stream_signal *signal = android_stream_signal_create(2); CHECK(signal); + CHECK(android_stream_signal_wait(signal, 0) == 0); + android_stream_signal_notify(signal); android_stream_signal_notify(signal); + CHECK(android_stream_signal_wait(signal, 0) == 1 && android_stream_signal_wait(signal, 0) == 0); CHECK(android_stream_open(7, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0); u_free(wait_event(7, 4, &size)); CHECK(android_stream_open(8, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0); @@ -109,7 +119,7 @@ int main(void) { CHECK(feed(1, 0, 100000, 0, (const uint8_t *)"ops", 3) == 1); uint8_t *output = NULL; int video_seen = 0, audio_seen = 0; for (unsigned i = 0; i < 1000 && (!video_seen || !audio_seen); i++) { - CHECK(android_stream_read_request(2, 100000, 1, 80000) == 0); + CHECK(android_stream_read_request(2, 100000, 1, 80000, 1) == 0); output = android_stream_pull(2, &size); if (output && output[0] == 1) { CHECK(size == sizeof(video) + ANDROID_STREAM_HEADER_SIZE && stream_read_be(output + 8, 8) == 100000 && !memcmp(output + ANDROID_STREAM_HEADER_SIZE, video, sizeof(video))); @@ -137,7 +147,7 @@ int main(void) { next_format.video_codec = STREAM_MEDIA_H265; video_seen = audio_seen = 0; for (unsigned i = 0; i < 1000 && (!video_seen || !audio_seen); i++) { - CHECK(android_stream_read_request(2, 300000, 1, 80000) == 0); + CHECK(android_stream_read_request(2, 300000, 1, 80000, 1) == 0); output = android_stream_pull(2, &size); if (output && output[0] == 1) { CHECK(stream_read_be(output + 8, 8) == 300000 && stream_read_be(output + 20, 4) == 2 && output[24] == STREAM_MEDIA_H265); @@ -157,7 +167,7 @@ int main(void) { CHECK(feed(1, 0, audio_pts, 0, (const uint8_t *)"ops", 3) == 1); video_seen = audio_seen = 0; for (unsigned i = 0; i < 1000 && (!video_seen || !audio_seen); i++) { - CHECK(android_stream_read_request(2, (int64_t)video_pts, 1, 80000) == 0); + CHECK(android_stream_read_request(2, (int64_t)video_pts, 1, 80000, 1) == 0); output = android_stream_pull(2, &size); if (output && output[0] == 1) { CHECK(stream_read_be(output + 8, 8) == video_pts && stream_read_be(output + 20, 4) == next_format.revision); @@ -176,7 +186,7 @@ int main(void) { debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG); for (unsigned i = 0; i < 5; i++) CHECK(android_stream_feed(1, 1, 800000 + i * 40000, 0, video, sizeof(video), &next_format) == 1); - for (unsigned i = 0; i < 50; i++) { CHECK(android_stream_read_request(2, 700000, 1, -1) == 0); test_clock_poll(ua, 100); } + for (unsigned i = 0; i < 50; i++) { CHECK(android_stream_read_request(2, 700000, 1, -1, 1) == 0); test_clock_poll(ua, 100); } output = android_stream_pull(2, &size); CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 800000); u_free(output); /* AVAILABLE между pull и отменой demand раньше выдавал второй кадр при занятом decoder input. */ CHECK(android_stream_feed(1, 1, 1000000, 0, video, sizeof(video), &next_format) == 1); @@ -185,37 +195,70 @@ int main(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "demand regression: new data after reply, unsolicited_kind=%d", output ? output[0] : 0); CHECK(!output); test_clock_run_for(ua, 1000); CHECK(!android_stream_pull(2, &size)); - CHECK(android_stream_read_request(2, 700000, 0, -1) == 0); test_clock_run_for(ua, 1000); + CHECK(android_stream_read_request(2, 700000, 0, -1, 1) == 0); test_clock_run_for(ua, 1000); CHECK(!android_stream_pull(2, &size)); for (unsigned frame = 1; frame < 6; frame++) { output = NULL; for (unsigned i = 0; i < 1000 && !output; i++) { - CHECK(android_stream_read_request(2, 700000, 1, -1) == 0); test_clock_poll(ua, 100); output = android_stream_pull(2, &size); + CHECK(android_stream_read_request(2, 700000, 1, -1, 1) == 0); test_clock_poll(ua, 100); output = android_stream_pull(2, &size); } CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 800000 + frame * 40000); u_free(output); } /* Пустой pull не отменяет WAIT: уже запрошенный кадр приходит после появления данных. */ - CHECK(android_stream_read_request(2, 1000000, 1, -1) == 0); test_clock_run_for(ua, 1000); + CHECK(android_stream_read_request(2, 1000000, 1, -1, 1) == 0); test_clock_run_for(ua, 1000); CHECK(!android_stream_pull(2, &size)); CHECK(android_stream_feed(1, 1, 1040000, 0, video, sizeof(video), &next_format) == 1); test_clock_run_for(ua, 1000); output = android_stream_pull(2, &size); CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 1040000); u_free(output); /* Отмена WAIT сохраняет следующий кадр, а запрос только аудио не забирает видео. */ - CHECK(android_stream_read_request(2, 1040000, 1, -1) == 0); test_clock_run_for(ua, 1000); - CHECK(!android_stream_pull(2, &size)); CHECK(android_stream_read_request(2, 1040000, 0, -1) == 0); + CHECK(android_stream_read_request(2, 1040000, 1, -1, 1) == 0); test_clock_run_for(ua, 1000); + CHECK(!android_stream_pull(2, &size)); CHECK(android_stream_read_request(2, 1040000, 0, -1, 1) == 0); CHECK(android_stream_feed(1, 1, 1080000, 0, video, sizeof(video), &next_format) == 1); CHECK(feed(1, 0, 220000, 0, (const uint8_t *)"ops", 3) == 1); test_clock_run_for(ua, 1000); CHECK(!android_stream_pull(2, &size)); - CHECK(android_stream_read_request(2, 1040000, 0, 80000) == 0); test_clock_run_for(ua, 1000); + CHECK(android_stream_read_request(2, 1040000, 0, 80000, 1) == 0); test_clock_run_for(ua, 1000); output = android_stream_pull(2, &size); CHECK(output && output[0] == 2 && stream_read_be(output + 8, 8) == 220000); u_free(output); - CHECK(android_stream_read_request(2, 1040000, 1, -1) == 0); test_clock_run_for(ua, 1000); + CHECK(android_stream_read_request(2, 1040000, 1, -1, 1) == 0); test_clock_run_for(ua, 1000); output = android_stream_pull(2, &size); CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 1080000); u_free(output); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS demand regression: one reply, WAIT survives empty pull, cancellation and audio-only preserve video"); + for (unsigned i = 0; i < 3; i++) { + CHECK(android_stream_feed(1, 1, 1120000 + i * 40000, 0, video, sizeof(video), &next_format) == 1); + CHECK(feed(1, 0, 240000 + i * 20000, 0, (const uint8_t *)"ops", 3) == 1); + } + test_clock_run_for(ua, 1000); + CHECK(android_stream_signal_wait(signal, 0) >= 0); + unsigned batch_video = 0, batch_audio = 0, largest = 0; + for (unsigned attempt = 0; attempt < 20 && (batch_video < 3 || batch_audio < 3); attempt++) { + CHECK(android_stream_read_request(2, 1080000, 3 - batch_video, 80000, 16) == 0); + test_clock_run_for(ua, 1000); + output = android_stream_pull(2, &size); CHECK(output); + CHECK(android_stream_signal_wait(signal, 0) == 1); + unsigned records = 0; + for (size_t offset = 0; offset < size;) { + CHECK(size - offset >= ANDROID_STREAM_HEADER_SIZE); + size_t length = (size_t)stream_read_be(output + offset + 16, 4); + CHECK(length <= size - offset - ANDROID_STREAM_HEADER_SIZE); + uint64_t pts = stream_read_be(output + offset + 8, 8); + if (output[offset] == 1) { + CHECK(pts == 1120000 + batch_video * 40000 && length == sizeof(video)); batch_video++; + } else { CHECK(output[offset] == 2 && pts == 240000 + batch_audio * 20000 && length == 3); batch_audio++; } + records++; offset += ANDROID_STREAM_HEADER_SIZE + length; + } + CHECK(records <= 16); if (records > largest) largest = records; u_free(output); + } + CHECK(batch_video == 3 && batch_audio == 3 && largest > 1); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS Android batches: video=%u audio=%u largest=%u notified=1", batch_video, batch_audio, largest); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO); - android_stream_stop(1); u_free(wait_event(2, 5, &size)); android_stream_stop(2); u_free(wait_event(1, 5, &size)); + android_stream_stop(1); u_free(wait_event(2, 5, &size)); + CHECK(android_stream_signal_wait(signal, 0) >= 0); + struct signal_waiter waiter = {.signal = signal}; pthread_t waiter_thread; + CHECK(pthread_create(&waiter_thread, NULL, wait_signal, &waiter) == 0); + while (!__atomic_load_n(&waiter.started, __ATOMIC_ACQUIRE)) {} + android_stream_stop(2); CHECK(pthread_join(waiter_thread, NULL) == 0 && waiter.result == 1); + android_stream_signal_destroy(signal); u_free(wait_event(1, 5, &size)); while ((output = android_stream_pull(2, &size))) u_free(output); CHECK(!strcmp(source_snapshot, "[]")); current_format.rotation = 0; current_format.audio_codec = 0; diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c index 500db089..dafc4125 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.c +++ b/tools/chatgui-android/libutun_lite/stream_bridge.c @@ -1,4 +1,6 @@ #include +#include +#include #include #include "../../../lib/debug_config.h" #include "../../../lib/mem.h" @@ -126,11 +128,9 @@ void android_stream_stop(uint64_t id) { pthread_mutex_unlock(&media_lock); } -/* Kotlin получает owned wire-копию только ближайшей записи своего reader. */ -static uint8_t *encode_record(int kind, int flags, const struct stream_media_format *format, uint64_t pts, - const uint8_t *data, size_t bytes, size_t *length) { - uint8_t *p = u_calloc(1, ANDROID_STREAM_HEADER_SIZE + bytes); - if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android output allocation failed bytes=%zu", bytes); return NULL; } +/* Запись в заранее выделенный нулевой буфер; записи батча следуют друг за другом без дополнительной копии. */ +static void encode_record(uint8_t *p, int kind, int flags, const struct stream_media_format *format, uint64_t pts, + const uint8_t *data, size_t bytes) { 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); @@ -138,7 +138,6 @@ static uint8_t *encode_record(int kind, int flags, const struct stream_media_for } stream_write_be(p + 8, pts, 8); stream_write_be(p + 16, bytes, 4); if (bytes) memcpy(p + ANDROID_STREAM_HEADER_SIZE, data, bytes); - *length = ANDROID_STREAM_HEADER_SIZE + bytes; return p; } uint8_t *android_stream_pull(uint64_t id, size_t *length) { @@ -149,32 +148,49 @@ uint8_t *android_stream_pull(uint64_t id, size_t *length) { struct stream_publisher_status status; if (stream_publisher_snapshot(transmitter, id, &status) == 0) { if (status.state == STREAM_PUBLISHER_ENDED || status.state == STREAM_PUBLISHER_FAILED) { - data = encode_record(status.state == STREAM_PUBLISHER_ENDED ? 5 : 6, status.reason, NULL, 0, NULL, 0, length); + data = u_calloc(1, ANDROID_STREAM_HEADER_SIZE); + if (data) encode_record(data, status.state == STREAM_PUBLISHER_ENDED ? 5 : 6, status.reason, NULL, 0, NULL, 0); + else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android publisher completion allocation failed id=%llu", (unsigned long long)id); if (data) { publisher_id = 0; publisher_opening = 0; } } else if (publisher_opening && status.state >= STREAM_PUBLISHER_LIVE && status.state <= STREAM_PUBLISHER_DRAINING) { - data = encode_record(4, 0, NULL, 0, NULL, 0, length); + data = u_calloc(1, ANDROID_STREAM_HEADER_SIZE); + if (data) encode_record(data, 4, 0, NULL, 0, NULL, 0); + else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android publisher open event allocation failed id=%llu", (unsigned long long)id); if (data) publisher_opening = 0; } + if (data) *length = ANDROID_STREAM_HEADER_SIZE; } } else { - struct stream_viewer_record *r = stream_viewer_pull(viewers, id); - if (r) { - data = encode_record(r->kind, r->kind == STREAM_VIEWER_CLOSED ? r->reason : r->flags, - &r->format, r->pts, r->data, r->length, length); - if (!data) stream_viewer_resync(viewers, id); - u_free(r); + struct stream_viewer_record *head = stream_viewer_pull(viewers, id); + if (head) { + size_t bytes = 0; + for (struct stream_viewer_record *r = head; r; r = r->next) bytes += ANDROID_STREAM_HEADER_SIZE + r->length; + data = u_calloc(1, bytes); + if (data) { + uint8_t *p = data; + for (struct stream_viewer_record *r = head; r; r = r->next) { + encode_record(p, r->kind, r->kind == STREAM_VIEWER_CLOSED ? r->reason : r->flags, + &r->format, r->pts, r->data, r->length); + p += ANDROID_STREAM_HEADER_SIZE + r->length; + } + *length = bytes; + } else { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android batch allocation failed id=%llu bytes=%zu", (unsigned long long)id, bytes); + stream_viewer_resync(viewers, id); + } + stream_viewer_records_free(head); } } pthread_mutex_unlock(&media_lock); return data; } -int android_stream_read_request(uint64_t id, int64_t position, int want_video, int64_t audio_ahead) { - if ((want_video != 0 && want_video != 1) || audio_ahead < -1 || audio_ahead > 60000000) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media demand invalid video=%d audio_ahead=%lld", - want_video, (long long)audio_ahead); return -1; +int android_stream_read_request(uint64_t id, int64_t position, unsigned video, int64_t audio_ahead, unsigned records) { + if (!records || records > STREAM_VIEWER_BATCH_RECORDS || video > records || audio_ahead < -1 || audio_ahead > 60000000) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media demand invalid video=%u records=%u audio_ahead=%lld", + video, records, (long long)audio_ahead); return -1; } pthread_mutex_lock(&media_lock); - if (viewers && id != publisher_id) stream_viewer_request(viewers, id, position, want_video, audio_ahead, 1); + if (viewers && id != publisher_id) stream_viewer_request(viewers, id, position, video, audio_ahead, records); /* CLOSED/END остаются доступны pull даже после закрытия cursor; неизвестный ID даёт пустой pull. */ pthread_mutex_unlock(&media_lock); return 0; } @@ -191,3 +207,70 @@ int android_stream_resync(uint64_t id) { int result = stream_viewer_resync(viewers, id); pthread_mutex_unlock(&media_lock); return result; } + +/* Lifetime принадлежит worker, а не manager: stop/detach снимает callback до освобождения signal. */ +struct android_stream_signal { + pthread_mutex_t lock; + pthread_cond_t ready; + uint64_t id, notifications, waits; + int pending; +}; + +void android_stream_signal_notify(struct android_stream_signal *s) { + if (!s) return; + pthread_mutex_lock(&s->lock); s->pending = 1; s->notifications++; + pthread_cond_signal(&s->ready); pthread_mutex_unlock(&s->lock); +} + +static void signal_ready(void *arg) { android_stream_signal_notify(arg); } + +struct android_stream_signal *android_stream_signal_create(uint64_t id) { + struct android_stream_signal *s = u_calloc(1, sizeof(*s)); + if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal allocation failed"); return NULL; } + int result = pthread_mutex_init(&s->lock, NULL); + if (result) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal mutex failed=%d", result); u_free(s); return NULL; } + pthread_condattr_t attr; + result = pthread_condattr_init(&attr); + if (!result) { + result = pthread_condattr_setclock(&attr, CLOCK_MONOTONIC); + if (!result) result = pthread_cond_init(&s->ready, &attr); + pthread_condattr_destroy(&attr); + } + if (result) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal condition failed=%d", result); + pthread_mutex_destroy(&s->lock); u_free(s); return NULL; + } + s->id = id; + pthread_mutex_lock(&media_lock); + result = stream_viewer_set_notify(viewers, id, signal_ready, s); + pthread_mutex_unlock(&media_lock); + if (result < 0) { android_stream_signal_destroy(s); return NULL; } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android viewer signal attached id=%llu", (unsigned long long)id); + return s; +} + +int android_stream_signal_wait(struct android_stream_signal *s, unsigned timeout_ms) { + if (!s || timeout_ms > 50) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal wait invalid timeout=%u", timeout_ms); return -1; + } + struct timespec until; + if (clock_gettime(CLOCK_MONOTONIC, &until) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal clock failed errno=%d", errno); return -1; + } + until.tv_nsec += timeout_ms * 1000000L; + until.tv_sec += until.tv_nsec / 1000000000L; until.tv_nsec %= 1000000000L; + pthread_mutex_lock(&s->lock); s->waits++; + int result = 0; + while (!s->pending && !result) result = pthread_cond_timedwait(&s->ready, &s->lock, &until); + if (result && result != ETIMEDOUT) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal wait failed=%d", result); + int signaled = s->pending; s->pending = 0; + pthread_mutex_unlock(&s->lock); + return result && result != ETIMEDOUT ? -1 : signaled; +} + +void android_stream_signal_destroy(struct android_stream_signal *s) { + if (!s) return; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android viewer signal released id=%llu notifications=%llu waits=%llu", + (unsigned long long)s->id, (unsigned long long)s->notifications, (unsigned long long)s->waits); + pthread_cond_destroy(&s->ready); pthread_mutex_destroy(&s->lock); u_free(s); +} diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.h b/tools/chatgui-android/libutun_lite/stream_bridge.h index 06b4f8c5..18398cad 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.h +++ b/tools/chatgui-android/libutun_lite/stream_bridge.h @@ -1,9 +1,11 @@ /* Android-очереди live media: attach/detach только в uasync, вне callbacks. * UI/кодировщики используют только session_id, никогда не сохраняют inst/ua. - * Один publisher и независимые viewers, TX <=2MB/128 records, RX — одна ближайшая запись на viewer. + * Один publisher и независимые viewers, TX <=2MB/128 records, RX — ограниченный батч на viewer. * Data принимает полные access units пяти кодеков uAV2 (KEY: config_length:4/config/frame) и Opus 48k mono/stereo/20ms. - * pull передаёт u_malloc-буфер: kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,revision:4,codec:1,reserved:3,data. - * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6; error.flags = stream_close_reason. Числа big-endian. + * pull передаёт u_malloc-буфер последовательных записей: + * kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,revision:4,codec:1,reserved:3,data. + * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6,video_end=7,audio_end=8; error.flags = stream_close_reason. + * Конец трека следует за его media; GAP/error отменяют старые записи. Числа big-endian. * Медиаданные передаются через pull, не через общий чатовый event bus. */ #ifndef ANDROID_STREAM_BRIDGE_H #define ANDROID_STREAM_BRIDGE_H @@ -14,6 +16,7 @@ struct UTUN_INSTANCE; struct stream_media_format; struct stream_media_status; +struct android_stream_signal; #define ANDROID_STREAM_HEADER_SIZE 28 struct android_stream_tx_stats { uint64_t id, sample_us, video_bytes, audio_bytes, stream_bytes, transport_bytes, inflight_bytes, acked_1s, total_bytes; @@ -36,10 +39,18 @@ int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uin /* 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); -/* Один ответ на запрос: следующий кадр при want_video=1, звук до position_us+audio_ahead_us. +/* Один батч на запрос: до video кадров и records записей, звук до position_us+audio_ahead_us. * Повторные запросы до получения ответа не создают дополнительные записи. Указатели кольца остаются в uasync. */ -int android_stream_read_request(uint64_t id, int64_t position_us, int want_video, int64_t audio_ahead_us); +int android_stream_read_request(uint64_t id, int64_t position_us, unsigned video, int64_t audio_ahead_us, unsigned records); int android_stream_get_status(uint64_t id, struct stream_media_status *status); int android_stream_resync(uint64_t id); +/* Worker владеет signal; create привязывает его к открытому viewer. Готовность/stop/detach будят wait. + * signal также будит worker после расходования PCM. Уведомления сливаются, но не теряются до wait. + * destroy — только после stream_stop и остановки всех внешних signal/wait callers. */ +struct android_stream_signal *android_stream_signal_create(uint64_t id); +void android_stream_signal_notify(struct android_stream_signal *signal); +/* 1 — уведомление, 0 — deadline, -1 — ошибка; timeout_ms 0..50, ожидание не держит mutex ядра/bridge. */ +int android_stream_signal_wait(struct android_stream_signal *signal, unsigned timeout_ms); +void android_stream_signal_destroy(struct android_stream_signal *signal); #endif