From b1c5276ab4223098f45a7a22c54a709506cb7414 Mon Sep 17 00:00:00 2001 From: Maxim Devaev Date: Mon, 6 Jul 2026 09:22:19 +0300 Subject: [PATCH] memsink: caps protocol --- janus/src/plugin.c | 3 ++- python/src/ustreamer.c | 2 +- src/dump/main.c | 7 ++++--- src/libs/memsink.c | 41 ++++++++++++++++++++++++++++------------- src/libs/memsink.h | 12 ++++++++++-- src/libs/memsinksh.h | 13 +++++++++++-- src/ustreamer/stream.c | 4 +++- 7 files changed, 59 insertions(+), 23 deletions(-) diff --git a/janus/src/plugin.c b/janus/src/plugin.c index 5cf78fe..d76ef57 100644 --- a/janus/src/plugin.c +++ b/janus/src/plugin.c @@ -157,7 +157,8 @@ static void *_video_sink_thread(void *arg) { US_LOG_INFO("Memsink opened; reading frames ..."); while (!_STOP && _HAS_WATCHERS) { - const int got = us_memsink_client_get(sink, tmp, NULL, atomic_load(&_g_key_required)); + const us_memsink_wants_s w_put = {.key = atomic_load(&_g_key_required)}; + const int got = us_memsink_client_get(sink, tmp, NULL, &w_put); if (got == 0) { const int ri = us_ring_producer_acquire(_g_video_ring, 0); if (ri >= 0) { diff --git a/python/src/ustreamer.c b/python/src/ustreamer.c index 72b624c..e5fd8c4 100644 --- a/python/src/ustreamer.c +++ b/python/src/ustreamer.c @@ -230,7 +230,7 @@ static PyObject *_MemsinkObject_wait_frame(_MemsinkObject *self, PyObject *args, self->frame_id = mem->id; self->frame_ts = us_get_now_monotonic(); if (key_required) { - mem->key_requested = true; + mem->wants.key = true; } if (flock(self->fd, LOCK_UN) < 0) { diff --git a/src/dump/main.c b/src/dump/main.c index 142b00d..a2d9c5a 100644 --- a/src/dump/main.c +++ b/src/dump/main.c @@ -236,8 +236,9 @@ static int _dump_sink( long double last_ts = 0; while (!_g_stop) { - bool key_requested; - const int got = us_memsink_client_get(sink, frame, &key_requested, key_required); + us_memsink_wants_s w_get = {0}; + us_memsink_wants_s w_put = {.key = key_required}; + const int got = us_memsink_client_get(sink, frame, &w_get, (key_required ? &w_put : NULL)); if (got == 0) { key_required = false; @@ -252,7 +253,7 @@ static int _dump_sink( frame->height, frame->gop, frame->key, - key_requested, + w_get.key, frame->grab_end_ts - frame->grab_begin_ts, frame->encode_begin_ts - frame->grab_end_ts, frame->encode_end_ts - frame->encode_begin_ts, diff --git a/src/libs/memsink.c b/src/libs/memsink.c index 71ca94c..1150a1d 100644 --- a/src/libs/memsink.c +++ b/src/libs/memsink.c @@ -41,8 +41,8 @@ us_memsink_s *us_memsink_init_opened( const char *name, const char *obj, bool server, - mode_t mode, bool rm, uint client_ttl, uint timeout) { - + mode_t mode, bool rm, uint client_ttl, uint timeout +) { us_memsink_s *sink; US_CALLOC(sink, 1); sink->name = name; @@ -160,7 +160,11 @@ bool us_memsink_server_check(us_memsink_s *sink, const us_frame_s *frame) { return false; } -int us_memsink_server_put(us_memsink_s *sink, const us_frame_s *frame, bool *key_requested) { +int us_memsink_server_put( + us_memsink_s *sink, + const us_frame_s *frame, + us_memsink_wants_s *wants +) { US_A(sink->server); const ldf now = us_get_now_monotonic(); @@ -175,11 +179,11 @@ int us_memsink_server_put(us_memsink_s *sink, const us_frame_s *frame, bool *key US_LOG_VERBOSE("%s-sink: >>>>> Exposing new frame ...", sink->name); sink->mem->id = us_get_now_id(); - if (sink->mem->key_requested && frame->key) { - sink->mem->key_requested = false; + if (sink->mem->wants.key && frame->key) { + sink->mem->wants.key = false; } - if (key_requested != NULL) { // We don't need it for non-H264 sinks - *key_requested = sink->mem->key_requested; + if (wants != NULL) { + memcpy(wants, &sink->mem->wants, sizeof(us_memsink_wants_s)); } memcpy(us_memsink_get_data(sink->mem), frame->data, frame->used); @@ -189,7 +193,8 @@ int us_memsink_server_put(us_memsink_s *sink, const us_frame_s *frame, bool *key sink->mem->magic = US_MEMSINK_MAGIC; sink->mem->version = US_MEMSINK_VERSION; - atomic_store(&sink->has_clients, (sink->mem->last_client_ts + sink->client_ttl > us_get_now_monotonic())); + const bool has_clients = (sink->mem->last_client_ts + sink->client_ttl > us_get_now_monotonic()); + atomic_store(&sink->has_clients, has_clients); if (flock(sink->fd, LOCK_UN) < 0) { US_LOG_PERROR("%s-sink: Can't unlock memory", sink->name); @@ -208,7 +213,12 @@ int us_memsink_server_put(us_memsink_s *sink, const us_frame_s *frame, bool *key return 0; } -int us_memsink_client_get(us_memsink_s *sink, us_frame_s *frame, bool *key_requested, bool key_required) { +int us_memsink_client_get( + us_memsink_s *sink, + us_frame_s *frame, + us_memsink_wants_s *get, + const us_memsink_wants_s *put +) { US_A(!sink->server); // Client only if (us_flock_timedwait_monotonic(sink->fd, sink->timeout) < 0) { @@ -243,11 +253,16 @@ int us_memsink_client_get(us_memsink_s *sink, us_frame_s *frame, bool *key_reque sink->last_readed_id = sink->mem->id; us_frame_set_data(frame, us_memsink_get_data(sink->mem), sink->mem->used); US_FRAME_COPY_META(sink->mem, frame); - if (key_requested != NULL) { // We don't need it for non-H264 sinks - *key_requested = sink->mem->key_requested; + + if (get != NULL) { + memcpy(get, &sink->mem->wants, sizeof(us_memsink_wants_s)); } - if (key_required) { - sink->mem->key_requested = true; + if (put != NULL) { + const bool key = sink->mem->wants.key; + memcpy(&sink->mem->wants, put, sizeof(us_memsink_wants_s)); + if (key) { + sink->mem->wants.key = key; + } } done: diff --git a/src/libs/memsink.h b/src/libs/memsink.h index fbea888..224bfe6 100644 --- a/src/libs/memsink.h +++ b/src/libs/memsink.h @@ -57,6 +57,14 @@ us_memsink_s *us_memsink_init_opened( void us_memsink_destroy(us_memsink_s *sink); bool us_memsink_server_check(us_memsink_s *sink, const us_frame_s *frame); -int us_memsink_server_put(us_memsink_s *sink, const us_frame_s *frame, bool *key_requested); -int us_memsink_client_get(us_memsink_s *sink, us_frame_s *frame, bool *key_requested, bool key_required); +int us_memsink_server_put( + us_memsink_s *sink, + const us_frame_s *frame, + us_memsink_wants_s *wants); + +int us_memsink_client_get( + us_memsink_s *sink, + us_frame_s *frame, + us_memsink_wants_s *get, + const us_memsink_wants_s *put); diff --git a/src/libs/memsinksh.h b/src/libs/memsinksh.h index e65581c..2740327 100644 --- a/src/libs/memsinksh.h +++ b/src/libs/memsinksh.h @@ -27,9 +27,18 @@ #define US_MEMSINK_MAGIC ((u64)0xCAFEBABECAFEBABE) -#define US_MEMSINK_VERSION ((u32)8) +#define US_MEMSINK_VERSION ((u32)9) +typedef struct { + uint width; + uint height; + uint format; + uint stride; + uint fps; + bool key; +} us_memsink_wants_s; + typedef struct { u64 magic; u32 version; @@ -37,7 +46,7 @@ typedef struct { uz used; ldf last_client_ts; - bool key_requested; + us_memsink_wants_s wants; US_FRAME_META_DECLARE; } us_memsink_shared_s; diff --git a/src/ustreamer/stream.c b/src/ustreamer/stream.c index 19f6f83..5c1a5eb 100644 --- a/src/ustreamer/stream.c +++ b/src/ustreamer/stream.c @@ -758,7 +758,9 @@ static void _stream_encode_expose_h264(us_stream_s *stream, const us_frame_s *fr force_key = true; } if (!us_m2m_encoder_compress(run->h264_enc, frame, run->h264_dest, force_key)) { - meta.online = !us_memsink_server_put(stream->h264_sink, run->h264_dest, &run->h264_key_requested); + us_memsink_wants_s wants = {0}; + meta.online = !us_memsink_server_put(stream->h264_sink, run->h264_dest, &wants); + run->h264_key_requested = wants.key; } done: