updated donate links

after broken frame to avoid corrupted picture.

.bumpversion.cfg

@@ -1,39 +0,0 @@
-[bumpversion]
-commit = True
-tag = True
-current_version = 6.62
-parse = (?P<major>\d+)\.(?P<minor>\d+)
-serialize = 
-	{major}.{minor}
-
-[bumpversion:file:src/libs/const.h]
-parse = (?P<major>\d+)
-serialize = {major}
-search = VERSION_MAJOR {current_version}
-replace = VERSION_MAJOR {new_version}
-
-[bumpversion:file:./src/libs/const.h]
-parse = <major>\d+\.(?P<minor>\d+)
-serialize = {minor}
-search = VERSION_MINOR {current_version}
-replace = VERSION_MINOR {new_version}
-
-[bumpversion:file:python/setup.py]
-search = version="{current_version}"
-replace = version="{new_version}"
-
-[bumpversion:file:pkg/arch/PKGBUILD]
-search = pkgver={current_version}
-replace = pkgver={new_version}
-
-[bumpversion:file:pkg/openwrt/Makefile]
-search = PKG_VERSION:={current_version}
-replace = PKG_VERSION:={new_version}
-
-[bumpversion:file:man/ustreamer.1]
-search = "version {current_version}"
-replace = "version {new_version}"
-
-[bumpversion:file:man/ustreamer-dump.1]
-search = "version {current_version}"
-replace = "version {new_version}"

.github/FUNDING.yml

@@ -1,4 +1,4 @@
 # These are supported funding model platforms
 
 patreon: pikvm
-custom: https://paypal.me/pikvm
+ko_fi: pikvm

Makefile

@@ -132,7 +132,7 @@ linters:
 
 
 bump:
-	bumpversion $(if $(V),$(V),minor)
+	./bump.py $(if $(V),$(V),minor)
 
 
 push:

bump.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python3
+# ========================================================================== #
+#                                                                            #
+#    KVMD - The main PiKVM daemon.                                           #
+#                                                                            #
+#    Copyright (C) 2018-2024  Maxim Devaev <mdevaev@gmail.com>               #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+# ========================================================================== #
+
+
+import sys
+import os
+import re
+import subprocess
+
+
+# =====
+def _run(cmd: list[str]) -> str:
+    return str(subprocess.run(cmd, stdout=subprocess.PIPE, check=True).stdout.decode())
+
+
+def _get_version() -> tuple[int, int]:
+    ver = _run(["/usr/bin/git", "describe", "--tags", "--abbrev=0"]).strip()
+    assert (m := re.match(r"v(\d+)\.(\d+)", ver)), ver
+    return (int(m.group(1)), int(m.group(2)))
+
+
+def _make_version(major: int, minor: int) -> str:
+    return f"v{major}.{minor}"
+
+
+def _patch_file(path: str, replace: str, old: tuple[int, int], new: tuple[int, int]) -> None:
+    with open(path, "r+") as f:
+        old_text = f.read()
+        new_text = old_text.replace(
+            replace.format(major=old[0], minor=old[1]),
+            replace.format(major=new[0], minor=new[1]),
+        )
+        # assert old_text != new_text, path
+        f.seek(0, os.SEEK_SET)
+        f.write(new_text)
+
+
+def _bump(do_major: bool, items: list[tuple[str, str]]) -> None:
+    _run(["/usr/bin/git", "diff-index", "--quiet", "HEAD", "--"])
+
+    old = _get_version()
+    new = ((old[0] + 1, old[1]) if do_major else (old[0], old[1] + 1))
+
+    for (path, replace) in items:
+        _patch_file(path, replace, old, new)
+        _run(["/usr/bin/git", "add", path])
+
+    ver_new = _make_version(*new)
+    _run(["/usr/bin/git", "commit", "-m", f"Bump version: {_make_version(*old)} -> {ver_new}"])
+    _run(["/usr/bin/git", "tag", ver_new])
+
+
+# =====
+if __name__ == "__main__":
+    _bump(
+        do_major=(len(sys.argv) == 2 and sys.argv[1] == "major"),
+        items=[
+            ("src/libs/const.h",     "define US_VERSION_MAJOR {major}"),
+            ("src/libs/const.h",     "define US_VERSION_MINOR {minor}"),
+            ("python/setup.py",      "version=\"{major}.{minor}\""),
+            ("man/ustreamer.1",      "version {major}.{minor}"),
+            ("man/ustreamer-dump.1", "version {major}.{minor}"),
+            ("pkg/arch/PKGBUILD",    "pkgver={major}.{minor}"),
+            ("pkg/openwrt/Makefile", "PKG_VERSION:={major}.{minor}"),
+        ],
+    )

janus/src/client.c

@@ -38,11 +38,14 @@
 #include "uslibs/array.h"
 #include "uslibs/list.h"
 #include "uslibs/ring.h"
+#include "uslibs/frame.h"
 
 #include "au.h"
 #include "rtp.h"
 
 
+static us_frame_s *_add_vplay_frame_to_ring(us_frame_s *frame, void *v_client);
+
 static void *_video_thread(void *v_client);
 static void *_acap_thread(void *v_client);
 static void *_video_or_acap_thread(void *v_client, bool video);
@@ -60,6 +63,7 @@ us_janus_client_s *us_janus_client_init(janus_callbacks *gw, janus_plugin_sessio
 	atomic_init(&client->transmit, false);
 	atomic_init(&client->transmit_acap, false);
 	atomic_init(&client->transmit_aplay, false);
+	atomic_init(&client->transmit_vplay, false);
 	atomic_init(&client->video_orient, 0);
 
 	atomic_init(&client->stop, false);
@@ -74,6 +78,8 @@ us_janus_client_s *us_janus_client_init(janus_callbacks *gw, janus_plugin_sessio
 	US_RING_INIT_WITH_ITEMS(client->aplay_pcm_ring, 64, us_au_pcm_init);
 	US_THREAD_CREATE(client->aplay_tid, _aplay_thread, client);
 
+	client->rtpc = us_rtpc_init(_add_vplay_frame_to_ring, client);
+
 	return client;
 }
 
@@ -90,6 +96,9 @@ void us_janus_client_destroy(us_janus_client_s *client) {
 	US_RING_DELETE_WITH_ITEMS(client->aplay_enc_ring, us_au_encoded_destroy);
 	US_RING_DELETE_WITH_ITEMS(client->aplay_pcm_ring, us_au_pcm_destroy);
 
+	us_rtpc_destroy(client->rtpc);
+	// us_janus_client_stop_vplay(client);
+
 	free(client);
 }
 
@@ -110,51 +119,106 @@ void us_janus_client_send(us_janus_client_s *client, const us_rtp_s *rtp) {
 	}
 }
 
+static us_frame_s *_add_vplay_frame_to_ring(us_frame_s *ready_frame, void *v_client) {
+	us_janus_client_s *const client = v_client;
+
+	us_ring_s *const ring = client->vplay_enc_ring;
+	if (!ring) {
+		US_LOG_ERROR("Session %p has no vplay ring", client->session);
+		return ready_frame;
+	}
+
+	const int ri = us_ring_producer_acquire(ring, 0);
+	if (ri < 0) {
+		US_LOG_ERROR("Session %p vplay ring is full", client->session);
+		return ready_frame;
+	}
+
+	us_frame_s *const next_frame = ring->items[ri];
+	ring->items[ri] = ready_frame;
+	us_ring_producer_release(ring, ri);
+
+	return next_frame;
+}
+
 void us_janus_client_recv(us_janus_client_s *client, janus_plugin_rtp *packet) {
-	if (
-		packet->video
-		|| packet->length < sizeof(janus_rtp_header)
-		|| !atomic_load(&client->transmit)
-		|| !atomic_load(&client->transmit_aplay)
-	) {
+	if (packet->length < sizeof(janus_rtp_header) || !atomic_load(&client->transmit)) {
 		return;
 	}
 
+	int payload_size = 0;
+	const u8 *const payload = (u8 *)janus_rtp_payload(packet->buffer, packet->length, &payload_size);
+	if (payload == NULL || payload_size < 1) {
+		return;
+	}
 	const janus_rtp_header *const header = (janus_rtp_header*)packet->buffer;
-	if (header->type != US_RTP_OPUS_PAYLOAD) {
-		return;
-	}
-
 	const u16 seq = ntohs(header->seq_number);
-	if (
-		seq >= client->aplay_seq_next // In order or missing
-		|| (client->aplay_seq_next - seq) > 50 // In late sequence or sequence wrapped
-	) {
-		client->aplay_seq_next = seq + 1;
 
-		int size = 0;
-		const char *const data = janus_rtp_payload(packet->buffer, packet->length, &size);
-		if (data == NULL || size <= 0) {
-			return;
+	if (packet->video) {
+		if (header->type == US_RTP_H264_PAYLOAD && atomic_load(&client->transmit_vplay)) {
+			const u32 rts = ntohl(header->timestamp); // RTS == RTP TimeStamp
+			bool retry;
+			do {
+				retry = false;
+				switch (us_rtpc_unwrap(client->rtpc, payload, payload_size, seq, rts)) {
+					case US_RUR_SUCCESS:
+					case US_RUR_BAD_PACKET:
+					case US_RUR_INVALID_SEQ:
+					case US_RUR_UNSUPPORTED_UNIT_TYPE:
+					case US_RUR_DEPACKETIZATION_SKIPPED:
+						break;
+					case US_RUR_DEPACKETIZATION_FAILED_RETRY:
+						retry = true;
+						// fall through
+					case US_RUR_DEPACKETIZATION_FAILED:
+						client->last_key_req_ts = us_get_now_monotonic();
+						client->gw->send_pli(client->session);
+						break;
+				}
+			} while (retry);
 		}
-
-		us_ring_s *const ring = client->aplay_enc_ring;
-		const int ri = us_ring_producer_acquire(ring, 0);
-		if (ri < 0) {
-			// US_LOG_ERROR("Session %p aplay ring is full", client->session);
-			return;
+		const ldf now_ts = us_get_now_monotonic();
+		if (client->last_key_req_ts + 2 < now_ts) {
+			client->last_key_req_ts = now_ts;
+			client->gw->send_pli(client->session);
 		}
-		us_au_encoded_s *enc = ring->items[ri];
-		if ((uz)size < US_ARRAY_LEN(enc->data)) {
-			memcpy(enc->data, data, size);
-			enc->used = size;
-		} else {
-			enc->used = 0;
+	} else {
+		if (header->type == US_RTP_OPUS_PAYLOAD && atomic_load(&client->transmit_aplay)) {
+			if (seq >= client->aplay_seq_next || (client->aplay_seq_next - seq) > 50) {
+				// ^^^ In order or missing   |OR| ^^^ In late sequence or sequence wrapped
+				client->aplay_seq_next = seq + 1;
+				us_ring_s *const ring = client->aplay_enc_ring;
+				const int ri = us_ring_producer_acquire(ring, 0);
+				if (ri < 0) {
+					// US_LOG_ERROR("Session %p aplay ring is full", client->session);
+					return;
+				}
+				us_au_encoded_s *enc = ring->items[ri];
+				if ((uz)payload_size < US_ARRAY_LEN(enc->data)) {
+					memcpy(enc->data, payload, payload_size);
+					enc->used = payload_size;
+				} else {
+					enc->used = 0;
+				}
+				us_ring_producer_release(ring, ri);
+			}
 		}
-		us_ring_producer_release(ring, ri);
 	}
 }
 
+void us_janus_client_start_vplay(us_janus_client_s *client, us_ring_s* vplay_enc_ring) {
+	if (atomic_exchange(&client->transmit_vplay, true)) {
+		return;
+	}
+	US_A(client->vplay_enc_ring == NULL);
+	client->vplay_enc_ring = vplay_enc_ring;
+}
+
+/*void us_janus_client_stop_vplay(us_janus_client_s *client) {
+	atomic_store(&client->transmit_vplay, false);
+	client->vplay_enc_ring = NULL;
+}*/
+
 static void *_video_thread(void *v_client) {
 	US_THREAD_SETTLE("us_cx_vcap");
 	return _video_or_acap_thread(v_client, true);

janus/src/client.h

@@ -32,32 +32,41 @@
 #include "uslibs/ring.h"
 
 #include "rtp.h"
+#include "rtpv.h"
+#include "rtpc.h"
 
 
-typedef struct {
+typedef struct us_janus_client_sx {
 	janus_callbacks			*gw;
 	janus_plugin_session	*session;
-	u32						video_ssrc;
-	u32						audio_ssrc;
 
-	atomic_bool				transmit;
-	atomic_bool				transmit_acap;
-	atomic_bool				transmit_aplay;
-	atomic_uint				video_orient;
+	u32			video_ssrc;
+	u32			audio_ssrc;
 
-	pthread_t				video_tid;
-	pthread_t				acap_tid;
-	pthread_t				aplay_tid;
-	atomic_bool				stop;
+	atomic_bool	transmit;
+	atomic_bool	transmit_acap;
+	atomic_bool	transmit_vplay;
+	atomic_bool	transmit_aplay;
+	atomic_uint	video_orient;
 
-	us_ring_s				*video_ring;
-	us_ring_s				*acap_ring;
+	pthread_t	video_tid;
+	pthread_t	acap_tid;
+	pthread_t	aplay_tid;
+	atomic_bool	stop;
 
-	us_ring_s				*aplay_enc_ring;
-	u16						aplay_seq_next;
-	us_ring_s				*aplay_pcm_ring;
+	us_ring_s	*video_ring;
+	us_ring_s	*acap_ring;
 
-    US_LIST_DECLARE;
+	us_ring_s	*aplay_enc_ring;
+	u16			aplay_seq_next;
+	us_ring_s	*aplay_pcm_ring;
+
+	us_rtpc_s	*rtpc;
+	us_ring_s	*vplay_enc_ring;
+
+	ldf			last_key_req_ts;
+
+	US_LIST_DECLARE(struct us_janus_client_sx);
 } us_janus_client_s;
 
 
@@ -66,3 +75,6 @@ void us_janus_client_destroy(us_janus_client_s *client);
 
 void us_janus_client_send(us_janus_client_s *client, const us_rtp_s *rtp);
 void us_janus_client_recv(us_janus_client_s *client, janus_plugin_rtp *packet);
+
+void us_janus_client_start_vplay(us_janus_client_s *client, us_ring_s* vplay_enc_ring);
+// void us_janus_client_stop_vplay(us_janus_client_s *client);

janus/src/config.c

@@ -36,7 +36,7 @@
 
 
 static char *_get_value(janus_config *jcfg, const char *section, const char *option);
-static uint _get_uint(janus_config *jcfg, const char *section, const char *option, uint def);
+static uint _get_uint(janus_config *jcfg, const char *section, const char *option, uint def, uint base);
 // static bool _get_bool(janus_config *jcfg, const char *section, const char *option, bool def);
 
 
@@ -57,20 +57,27 @@ us_config_s *us_config_init(const char *config_dir_path) {
 	}
 	janus_config_print(jcfg);
 
-	if ((config->video_sink_name = _get_value(jcfg, "video", "sink")) == NULL) {
+	config->video_sink_name = _get_value(jcfg, "video", "sink");
+	if (!us_str_is_ok(config->video_sink_name)) {
 		US_LOG_ERROR("Missing config value: video.sink");
 		goto error;
 	}
-	if ((config->acap_dev_name = _get_value(jcfg, "acap", "device")) != NULL) {
-		config->acap_hz = _get_uint(jcfg, "acap", "sampling_rate", 0);
+
+	config->acap_dev_name = _get_value(jcfg, "acap", "device");
+	if (us_str_is_ok(config->acap_dev_name)) {
+		config->acap_hz = _get_uint(jcfg, "acap", "sampling_rate", 0, 10);
 		config->tc358743_dev_path = _get_value(jcfg, "acap", "tc358743");
 		if (config->acap_hz == 0 && !us_str_is_ok(config->tc358743_dev_path)) {
 			US_LOG_ERROR("Either acap.sampling_rate or acap.tc358743 required");
 			goto error;
 		}
 	}
+
 	config->aplay_dev_name = _get_value(jcfg, "aplay", "device");
-	// config->vplay_sink_name = _get_value(jcfg, "vplay", "sink");
+
+	config->vplay_sink_name = _get_value(jcfg, "vplay", "sink");
+	config->vplay_sink_mode = _get_uint(jcfg, "vplay", "sink_mode", 0660, 8);
+	config->vplay_dev_path = _get_value(jcfg, "vplay", "device");
 
 	goto ok;
 
@@ -89,6 +96,7 @@ void us_config_destroy(us_config_s *config) {
 	US_DELETE(config->tc358743_dev_path, free);
 	US_DELETE(config->aplay_dev_name, free);
 	US_DELETE(config->vplay_sink_name, free);
+	US_DELETE(config->vplay_dev_path, free);
 	free(config);
 }
 
@@ -101,12 +109,12 @@ static char *_get_value(janus_config *jcfg, const char *section, const char *opt
 	return us_strdup(option_obj->value);
 }
 
-static uint _get_uint(janus_config *jcfg, const char *section, const char *option, uint def) {
+static uint _get_uint(janus_config *jcfg, const char *section, const char *option, uint def, uint base) {
 	char *const tmp = _get_value(jcfg, section, option);
 	uint value = def;
 	if (tmp != NULL) {
 		errno = 0;
-		value = (uint)strtoul(tmp, NULL, 10);
+		value = (uint)strtoul(tmp, NULL, base);
 		if (errno != 0) {
 			value = def;
 		}

janus/src/config.h

@@ -34,7 +34,10 @@ typedef struct {
 	char	*tc358743_dev_path;
 
 	char	*aplay_dev_name;
+
 	char	*vplay_sink_name;
+	uint	vplay_sink_mode;
+	char	*vplay_dev_path;
 } us_config_s;
 
 

janus/src/h264.c

@@ -0,0 +1,270 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    This source file based on code of Janys-Gateway.                        #
+#                                                                            #
+#    Copyright (C)  Lorenzo Miniero <lorenzo@meetecho.com>                   #
+#    Copyright (C)  2026  Maxim Devaev <mdevaev@gmail.com>                   #
+#    Copyright (C)  2026  Sergey Radionov <rsatom@gmail.com>                 #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#include "h264.h"
+
+#include "uslibs/types.h"
+#include "uslibs/tools.h"
+
+
+static INLINE u32 _eg_get_bit(const u8 *buf, u32 bit_offset) {
+	return ((*(buf + (bit_offset >> 3))) >> (7 - (bit_offset & 0x07))) & 0x01;
+}
+
+static INLINE bool _is_valid_eg_offset(u32 bit_offset, uz size) {
+	return (bit_offset >> 3) + (bit_offset % 8 ? 1 : 0) < size;
+}
+
+static bool _eg_skip_u1(const u8 *buf, u32 *bit_offset, uz size) {
+	(void)buf;
+	if (!_is_valid_eg_offset(*bit_offset, size)) {
+		return false;
+	}
+	++(*bit_offset);
+	return true;
+}
+
+static bool _eg_decode_u1(const u8 *buf, u32 *bit_offset, uz size, bool *out) {
+	if (!_is_valid_eg_offset(*bit_offset, size)) {
+		return false;
+	}
+	*out = (_eg_get_bit(buf, *bit_offset) != 0);
+	++(*bit_offset);
+	return true;
+}
+
+static bool _eg_skip_ev(const u8 *buf, u32 *bit_offset, uz size) {
+	u32 zeros = 0;
+	while (true) {
+		if (!_is_valid_eg_offset(*bit_offset, size)) {
+			return false;
+		}
+		if (_eg_get_bit(buf, (*bit_offset)++) != 0) {
+			break;
+		}
+		++zeros;
+	}
+	for (; zeros; --zeros) {
+		if (!_is_valid_eg_offset((*bit_offset)++, size)) {
+			return false;
+		}
+	}
+	return true;
+}
+
+static bool _eg_decode_uev(const u8 *buf, u32 *bit_offset, uint size, u32 *out) {
+	u32 zeros = 0;
+	while (true) {
+		if (!_is_valid_eg_offset(*bit_offset, size)) {
+			return false;
+		}
+		if (_eg_get_bit(buf, (*bit_offset)++) != 0) {
+			break;
+		}
+		++zeros;
+	}
+	u32 res = (1 << zeros);
+	for (; zeros; --zeros) {
+		if (!_is_valid_eg_offset(*bit_offset, size)) {
+			return false;
+		}
+		res |= (_eg_get_bit(buf, *bit_offset) << (zeros - 1));
+		++(*bit_offset);
+	}
+	*out = res - 1;
+	return true;
+}
+
+int us_h264_parse_sps_nalu(const u8 *buf, uz size, uint *width, uint *height) {
+	// NAL header + profile_idc + constraint_setX_flag, reserved_zero_2bits + level_idc
+	if (size < 4) {
+		return -1;
+	}
+
+	// NAL header
+	++buf;
+	--size;
+
+	// profile_idc u(8) - Baseline profile
+	const u8 profile_idc = *buf;
+	if (profile_idc != 66) {
+		return -1;
+	}
+	++buf;
+	--size;
+
+	// constraint_set0_flag u(1)
+	// constraint_set1_flag u(1)
+	// constraint_set2_flag u(1)
+	// constraint_set3_flag u(1)
+	// constraint_set4_flag u(1)
+	// constraint_set5_flag u(1)
+	// reserved_zero_2bits u(2)
+	++buf;
+	--size;
+
+	// level_idc u(8)
+	++buf;
+	--size;
+
+	u32 bit_offset = 0;
+
+	// seq_parameter_set_id ue(v)
+	if (!_eg_skip_ev(buf, &bit_offset, size)) {
+		return -1;
+	}
+
+	// log2_max_frame_num_minus4 ue(v)
+	u32 log2_max_frame_num_minus4;
+	if (!_eg_decode_uev(buf, &bit_offset, size, &log2_max_frame_num_minus4)) {
+		return -1;
+	}
+
+	// pic_order_cnt_type ue(v)
+	u32 pic_order_cnt_type;
+	if (!_eg_decode_uev(buf, &bit_offset, size, &pic_order_cnt_type)) {
+		return -1;
+	}
+
+	if (pic_order_cnt_type == 0) {
+		// log2_max_pic_order_cnt_lsb_minus4 ue(v)
+		if (!_eg_skip_ev(buf, &bit_offset, size)) {
+			return -1;
+		}
+	} else if (pic_order_cnt_type == 1) {
+		// delta_pic_order_always_zero_flag u(1)
+		if (!_eg_skip_u1(buf, &bit_offset, size)) {
+			return -1;
+		}
+
+		// offset_for_non_ref_pic se(v)
+		if (!_eg_skip_ev(buf, &bit_offset, size)) {
+			return -1;
+		}
+
+		// offset_for_top_to_bottom_field se(v)
+		if (!_eg_skip_ev(buf, &bit_offset, size)) {
+			return -1;
+		}
+
+		// num_ref_frames_in_pic_order_cnt_cycle ue(v)
+		u32 num_ref_frames_in_pic_order_cnt_cycle;
+		if (!_eg_decode_uev(buf, &bit_offset, size, &num_ref_frames_in_pic_order_cnt_cycle)) {
+			return -1;
+		}
+
+		for (u32 i = 0; i < num_ref_frames_in_pic_order_cnt_cycle; ++i) {
+			// offset_for_ref_frame[i] se(v)
+			if (!_eg_skip_ev(buf, &bit_offset, size)) {
+				return -1;
+			}
+		}
+	}
+
+	// max_num_ref_frames ue(v)
+	if (!_eg_skip_ev(buf, &bit_offset, size)) {
+		return -1;
+	}
+
+	// gaps_in_frame_num_value_allowed_flag u(1)
+	if (!_eg_skip_u1(buf, &bit_offset, size)) {
+		return -1;
+	}
+
+	// pic_width_in_mbs_minus1 ue(v)
+	u32 pic_width_in_mbs_minus1;
+	if (!_eg_decode_uev(buf, &bit_offset, size, &pic_width_in_mbs_minus1)) {
+		return -1;
+	}
+
+	//pic_height_in_map_units_minus1 ue(v)
+	u32 pic_height_in_map_units_minus1;
+	if (!_eg_decode_uev(buf, &bit_offset, size, &pic_height_in_map_units_minus1)) {
+		return -1;
+	}
+
+	// frame_mbs_only_flag u(1)
+	bool frame_mbs_only_flag;
+	if (!_eg_decode_u1(buf, &bit_offset, size, &frame_mbs_only_flag)) {
+		return -1;
+	}
+
+	if (!frame_mbs_only_flag) {
+		// mb_adaptive_frame_field_flag u(1)
+		if (!_eg_skip_u1(buf, &bit_offset, size)) {
+			return -1;
+		}
+	}
+
+	//direct_8x8_inference_flag u(1)
+	if (!_eg_skip_u1(buf, &bit_offset, size)) {
+		return -1;
+	}
+
+	// frame_cropping_flag u(1)
+	bool frame_cropping_flag;
+	if (!_eg_decode_u1(buf, &bit_offset, size, &frame_cropping_flag)) {
+		return -1;
+	}
+
+	u32 frame_crop_left_offset = 0;
+	u32 frame_crop_right_offset = 0;
+	u32 frame_crop_top_offset = 0;
+	u32 frame_crop_bottom_offset = 0;
+	if (frame_cropping_flag) {
+		// frame_crop_left_offset ue(v)
+		if (!_eg_decode_uev(buf, &bit_offset, size, &frame_crop_left_offset)) {
+			return -1;
+		}
+
+		// frame_crop_right_offset ue(v)
+		if (!_eg_decode_uev(buf, &bit_offset, size, &frame_crop_right_offset)) {
+			return -1;
+		}
+
+		// frame_crop_top_offset ue(v)
+		if (!_eg_decode_uev(buf, &bit_offset, size, &frame_crop_top_offset)) {
+			return -1;
+		}
+
+		// frame_crop_bottom_offset ue(v)
+		if (!_eg_decode_uev(buf, &bit_offset, size, &frame_crop_bottom_offset)) {
+			return -1;
+		}
+	}
+
+	*width = (
+		(pic_width_in_mbs_minus1 + 1) * 16
+		- frame_crop_left_offset * 2
+		- frame_crop_right_offset * 2
+	);
+	*height = (
+		(pic_height_in_map_units_minus1 + 1) * (2 - frame_mbs_only_flag) * 16
+		- frame_crop_top_offset * 2
+		- frame_crop_bottom_offset * 2
+	);
+	return 0;
+}

janus/src/h264.h

@@ -0,0 +1,29 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C)  2026  Maxim Devaev <mdevaev@gmail.com>                   #
+#    Copyright (C)  2026  Sergey Radionov <rsatom@gmail.com>                 #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#pragma once
+
+#include "uslibs/types.h"
+
+
+int us_h264_parse_sps_nalu(const u8 *buf, uz size, uint *width, uint *height);

janus/src/plugin.c

@@ -26,15 +26,13 @@
 #include <fcntl.h>
 #include <errno.h>
 
-#include <sys/mman.h>
-#include <sys/stat.h>
-
 #include <pthread.h>
 #include <jansson.h>
 #include <janus/plugins/plugin.h>
 #include <janus/rtp.h>
 #include <janus/rtcp.h>
 #include <alsa/asoundlib.h>
+#include <linux/videodev2.h>
 
 #include "uslibs/types.h"
 #include "uslibs/const.h"
@@ -46,6 +44,7 @@
 #include "uslibs/ring.h"
 #include "uslibs/memsink.h"
 #include "uslibs/chip.h"
+#include "uslibs/memsink.h"
 
 #include "const.h"
 #include "client.h"
@@ -58,7 +57,7 @@
 #include "config.h"
 
 
-static const char *const	default_ice_url = "stun:stun.l.google.com:19302";
+static const char *const	_g_default_ice_url = "stun:stun.l.google.com:19302";
 
 static us_config_s		*_g_config = NULL;
 static const useconds_t	_g_watchers_polling = 100000;
@@ -66,6 +65,7 @@ static const useconds_t	_g_watchers_polling = 100000;
 static us_janus_client_s	*_g_clients = NULL;
 static janus_callbacks		*_g_gw = NULL;
 static us_ring_s			*_g_video_ring = NULL;
+static us_ring_s			*_g_vplay_ring = NULL;
 static us_rtpv_s			*_g_rtpv = NULL;
 static us_rtpa_s			*_g_rtpa = NULL;
 
@@ -73,12 +73,15 @@ static pthread_t		_g_video_rtp_tid;
 static atomic_bool		_g_video_rtp_tid_created = false;
 static pthread_t		_g_video_sink_tid;
 static atomic_bool		_g_video_sink_tid_created = false;
+static pthread_t		_g_vplay_tid;
+static atomic_bool		_g_vplay_tid_created = false;
 static pthread_t		_g_acap_tid;
 static atomic_bool		_g_acap_tid_created = false;
 static pthread_t		_g_aplay_tid;
 static atomic_bool		_g_aplay_tid_created = false;
 
 static pthread_mutex_t	_g_video_lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_mutex_t	_g_vplay_lock = PTHREAD_MUTEX_INITIALIZER;
 static pthread_mutex_t	_g_acap_lock = PTHREAD_MUTEX_INITIALIZER;
 static pthread_mutex_t	_g_aplay_lock = PTHREAD_MUTEX_INITIALIZER;
 static atomic_bool		_g_ready = false;
@@ -88,18 +91,29 @@ static atomic_bool		_g_has_listeners = false;
 static atomic_bool		_g_has_speakers = false;
 static atomic_bool		_g_key_required = false;
 
+static struct {
+	bool	active;
+	us_janus_client_s *client;
+	uint	width;
+	uint	height;
+	uint	fps;
+} _g_camera = {0}; // Access should be protected by _g_vplay_lock
+
 
 #define _LOCK_VIDEO		US_MUTEX_LOCK(_g_video_lock)
 #define _UNLOCK_VIDEO	US_MUTEX_UNLOCK(_g_video_lock)
 
+#define _LOCK_VPLAY		US_MUTEX_LOCK(_g_vplay_lock)
+#define _UNLOCK_VPLAY	US_MUTEX_UNLOCK(_g_vplay_lock)
+
 #define _LOCK_ACAP		US_MUTEX_LOCK(_g_acap_lock)
 #define _UNLOCK_ACAP	US_MUTEX_UNLOCK(_g_acap_lock)
 
 #define _LOCK_APLAY		US_MUTEX_LOCK(_g_aplay_lock)
 #define _UNLOCK_APLAY	US_MUTEX_UNLOCK(_g_aplay_lock)
 
-#define _LOCK_ALL		{ _LOCK_VIDEO; _LOCK_ACAP; _LOCK_APLAY; }
-#define _UNLOCK_ALL		{ _UNLOCK_APLAY; _UNLOCK_ACAP; _UNLOCK_VIDEO; }
+#define _LOCK_ALL		{ _LOCK_VIDEO; _LOCK_VPLAY; _LOCK_ACAP; _LOCK_APLAY; }
+#define _UNLOCK_ALL		{ _UNLOCK_APLAY; _UNLOCK_ACAP; _UNLOCK_VPLAY; _UNLOCK_VIDEO; }
 
 #define _READY			atomic_load(&_g_ready)
 #define _STOP			atomic_load(&_g_stop)
@@ -192,7 +206,7 @@ static void *_acap_thread(void *arg) {
 	US_THREAD_SETTLE("us_p_acap");
 	atomic_store(&_g_acap_tid_created, true);
 
-	US_A(_g_config->acap_dev_name != NULL);
+	US_A(us_str_is_ok(_g_config->acap_dev_name));
 	US_A(_g_rtpa != NULL);
 
 	int once = 0;
@@ -267,7 +281,7 @@ static void *_aplay_thread(void *arg) {
 	US_THREAD_SETTLE("us_p_aplay");
 	atomic_store(&_g_aplay_tid_created, true);
 
-	US_A(_g_config->aplay_dev_name != NULL);
+	US_A(us_str_is_ok(_g_config->aplay_dev_name));
 
 	int once = 0;
 
@@ -372,6 +386,189 @@ static void *_aplay_thread(void *arg) {
 	return NULL;
 }
 
+static void _unsafe_camera_push_event(const char *str) {
+	json_t *const json_event_type = json_string(str);
+
+	json_t *const camera = json_object();
+	json_object_set_new(camera, "action", json_event_type);
+
+	json_t *const result = json_object();
+	json_object_set_new(result, "ustreamer", json_string("event"));
+	json_object_set_new(result, "status", json_string("camera"));
+	json_object_set_new(result, "camera", camera);
+
+	json_t *const event = json_object();
+	json_object_set_new(event, "result", result);
+
+	US_LIST_ITERATE(_g_clients, client, {
+		_g_gw->push_event(client->session, create(), NULL, event, NULL);
+	});
+
+	json_decref(event);
+}
+
+static void _camera_push_online(uint width, uint height, uint fps) {
+	bool again = false;
+	_LOCK_VPLAY;
+	if (_g_camera.active) {
+		_unsafe_camera_push_event("offline");
+		again = true;
+	}
+	const uint old_width = _g_camera.width;
+	const uint old_height = _g_camera.height;
+	const uint old_fps = _g_camera.fps;
+	_g_camera.active = true;
+	_g_camera.width = width;
+	_g_camera.height = height;
+	_g_camera.fps = fps;
+	_unsafe_camera_push_event("online");
+	_UNLOCK_VPLAY;
+	if (again) {
+		US_LOG_INFO("Camera changed: %ux%u@%u -> %ux%u@%u",
+			old_width, old_height, old_fps, width, height, fps);
+	} else {
+		US_LOG_INFO("Camera online: %ux%u@%u", width, height, fps);
+	}
+}
+
+static void _camera_push_offline() {
+	_LOCK_VPLAY;
+	if (_g_camera.active) {
+		_g_camera.active = false;
+		_unsafe_camera_push_event("offline");
+		US_LOG_INFO("Camera offline");
+	}
+	_UNLOCK_VPLAY;
+}
+
+static void _camera_push_lost() {
+	_LOCK_VPLAY;
+	if (_g_camera.active) {
+		_unsafe_camera_push_event("lost");
+		US_LOG_INFO("Camera's source client is lost");
+	}
+	_UNLOCK_VPLAY;
+}
+
+static void *_vplay_thread(void *arg) {
+	(void)arg;
+	US_THREAD_SETTLE("us_p_vplay");
+	atomic_store(&_g_vplay_tid_created, true);
+
+	US_A(us_str_is_ok(_g_config->vplay_sink_name));
+
+	us_frame_s *frame = us_frame_init();
+
+	while (!_STOP) {
+		us_memsink_s* sink = us_memsink_init_opened(
+			"H264-CAM",
+			_g_config->vplay_sink_name,
+			true, // Server
+			_g_config->vplay_sink_mode,
+			false, // Don't remote sink on destroy
+			1, // Client TTL
+			1); // Timeout
+		if (sink == NULL) {
+			goto close_memsink;
+		}
+
+		const ldf memsink_check_interval = 1;
+		ldf last_memsink_check_time = 0;
+
+		int once = 0;
+		while (!_STOP) {
+			_LOCK_VPLAY;
+			const bool c_active = _g_camera.active;
+			const uint c_width = _g_camera.width;
+			const uint c_height = _g_camera.height;
+			_UNLOCK_VPLAY;
+
+			if (!c_active) {
+				frame->used = 0;
+			}
+
+			if (frame->used > 0 && (c_width != frame->width || c_height != frame->height)) {
+				US_ONCE({ US_LOG_INFO("Got WebRTC frame with wrong resolution"); });
+				frame->used = 0;
+			}
+
+			const ldf now_ts = us_get_now_monotonic();
+			if (frame->used > 0 || last_memsink_check_time + memsink_check_interval < now_ts) {
+				last_memsink_check_time = now_ts;
+
+				us_memsink_wants_s w_get = {0};
+				switch (us_memsink_server_x_lock(sink, &w_get)) {
+					case US_MSS_SUCCESS:
+						if (w_get.format == V4L2_PIX_FMT_H264) {
+							if (w_get.width == c_width && w_get.height == c_height) {
+								if (us_memsink_server_x_is_consumed(sink)) {
+									if (!c_active) {
+										_camera_push_online(w_get.width, w_get.height, w_get.fps);
+									} else if (frame->used > 0) {
+										US_ONCE({ US_LOG_INFO("Streaming to the camera ..."); });
+										us_memsink_server_x_put(sink, frame);
+										frame->used = 0;
+									}
+								} else {
+									if (!c_active) {
+										us_memsink_server_x_set_consumed(sink);
+									}
+								}
+							} else { // Notify to changed resolution
+								_camera_push_online(w_get.width, w_get.height, w_get.fps);
+							}
+						} else {
+							US_ONCE({ US_LOG_ERROR("Got invalid format from the camera: %u", w_get.format); });
+						}
+						if (us_memsink_server_x_unlock(sink) < 0) {
+							goto close_memsink;
+						}
+						break;
+
+					case US_MSS_NO_CLIENT:
+						_camera_push_offline();
+						break;
+
+					case US_MSS_BUSY: // Busy by some client
+						break;
+
+					case US_MSS_ERROR:
+					default:
+						goto close_memsink;
+				}
+			}
+
+			if (frame->used == 0) { // Frame sent (or discarded), get a new one from a client
+				const int in_ri = us_ring_consumer_acquire(_g_vplay_ring, 0.1);
+				if (in_ri >= 0) {
+					us_frame_s *tmp_frame = frame;
+					frame = _g_vplay_ring->items[in_ri];
+					_g_vplay_ring->items[in_ri] = tmp_frame;
+					us_ring_consumer_release(_g_vplay_ring, in_ri);
+				/*} else { // FIXME: Leads to spam with "Streaming to the camera ..."
+					US_ONCE({
+						_LOCK_VPLAY;
+						if (_g_camera.active) {
+							US_LOG_INFO("No frames have got from WebRTC");
+						}
+						_UNLOCK_VPLAY;
+					});*/
+				}
+			} else {
+				usleep(1000); // Don't iterate too frequently
+			}
+		}
+
+	close_memsink:
+		US_DELETE(sink, us_memsink_destroy);
+		US_LOG_INFO("Memsink closed");
+		sleep(1);
+	}
+
+	us_frame_destroy(frame);
+	return NULL;
+}
+
 static void _relay_rtp_clients(const us_rtp_s *rtp) {
 	US_LIST_ITERATE(_g_clients, client, {
 		us_janus_client_send(client, rtp);
@@ -408,13 +605,21 @@ static int _plugin_init(janus_callbacks *gw, const char *config_dir_path) {
 
 	US_RING_INIT_WITH_ITEMS(_g_video_ring, 64, us_frame_init);
 	_g_rtpv = us_rtpv_init(_relay_rtp_clients);
-	if (_g_config->acap_dev_name != NULL) {
+
+	if (us_str_is_ok(_g_config->vplay_sink_name)) {
+		US_RING_INIT_WITH_ITEMS(_g_vplay_ring, 15, us_frame_init);
+		US_THREAD_CREATE(_g_vplay_tid, _vplay_thread, NULL);
+	}
+
+	if (us_str_is_ok(_g_config->acap_dev_name)) {
 		_g_rtpa = us_rtpa_init(_relay_rtp_clients);
 		US_THREAD_CREATE(_g_acap_tid, _acap_thread, NULL);
 	}
-	if (_g_config->aplay_dev_name != NULL) {
+
+	if (us_str_is_ok(_g_config->aplay_dev_name)) {
 		US_THREAD_CREATE(_g_aplay_tid, _aplay_thread, NULL);
 	}
+
 	US_THREAD_CREATE(_g_video_rtp_tid, _video_rtp_thread, NULL);
 	US_THREAD_CREATE(_g_video_sink_tid, _video_sink_thread, NULL);
 
@@ -427,17 +632,22 @@ static void _plugin_destroy(void) {
 
 	atomic_store(&_g_stop, true);
 #	define JOIN(_tid) { if (atomic_load(&_tid##_created)) { US_THREAD_JOIN(_tid); } }
+	JOIN(_g_aplay_tid);
+	JOIN(_g_acap_tid);
+	JOIN(_g_vplay_tid);
 	JOIN(_g_video_sink_tid);
 	JOIN(_g_video_rtp_tid);
-	JOIN(_g_acap_tid);
-	JOIN(_g_aplay_tid);
 #	undef JOIN
 
 	US_LIST_ITERATE(_g_clients, client, {
+		if (_g_camera.client == client) {
+			_g_camera.client = NULL;
+		}
 		US_LIST_REMOVE(_g_clients, client);
 		us_janus_client_destroy(client);
 	});
 
+	US_RING_DELETE_WITH_ITEMS(_g_vplay_ring, us_frame_destroy);
 	US_RING_DELETE_WITH_ITEMS(_g_video_ring, us_frame_destroy);
 
 	US_DELETE(_g_rtpa, us_rtpa_destroy);
@@ -464,9 +674,14 @@ static void _plugin_destroy_session(janus_plugin_session* session, int *err) {
 	bool has_watchers = false;
 	bool has_listeners = false;
 	bool has_speakers = false;
+	bool camera_lost = false;
 	US_LIST_ITERATE(_g_clients, client, {
 		if (client->session == session) {
 			US_LOG_INFO("Removing session %p ...", session);
+			if (_g_camera.client == client) {
+				_g_camera.client = NULL;
+				camera_lost = true;
+			}
 			US_LIST_REMOVE(_g_clients, client);
 			us_janus_client_destroy(client);
 			found = true;
@@ -484,6 +699,9 @@ static void _plugin_destroy_session(janus_plugin_session* session, int *err) {
 	atomic_store(&_g_has_listeners, has_listeners);
 	atomic_store(&_g_has_speakers, has_speakers);
 	_UNLOCK_ALL;
+	if (camera_lost) {
+		_camera_push_lost();
+	}
 }
 
 static json_t *_plugin_query_session(janus_plugin_session *session) {
@@ -524,6 +742,16 @@ static void _set_transmit(janus_plugin_session *session, const char *msg, bool t
 static void _plugin_setup_media(janus_plugin_session *session) { _set_transmit(session, "Unmuted", true); }
 static void _plugin_hangup_media(janus_plugin_session *session) { _set_transmit(session, "Muted", false); }
 
+static bool _is_camera_enabled(void) {
+	bool enabled = us_str_is_ok(_g_config->vplay_sink_name);
+	if (enabled && us_str_is_ok(_g_config->vplay_dev_path)) {
+		if (access(_g_config->vplay_dev_path, F_OK) != 0) {
+			enabled = false;
+		}
+	}
+	return enabled;
+}
+
 static struct janus_plugin_result *_plugin_handle_message(
 	janus_plugin_session *session, char *transaction, json_t *msg, json_t *jsep) {
 
@@ -594,7 +822,7 @@ static struct janus_plugin_result *_plugin_handle_message(
 					}
 				READ_BOOL(with_acap, "audio", us_au_probe(_g_config->acap_dev_name));
 				READ_BOOL(with_aplay, "mic", us_au_probe(_g_config->aplay_dev_name));
-				READ_BOOL(with_vplay, "camera", (_g_config->vplay_sink_name != NULL));
+				READ_BOOL(with_vplay, "camera", _is_camera_enabled());
 #				undef READ_BOOL
 				{
 					json_t *const obj = json_object_get(params, "orientation");
@@ -611,6 +839,8 @@ static struct janus_plugin_result *_plugin_handle_message(
 
 		{
 			_LOCK_ALL;
+			with_vplay = (with_vplay && _g_camera.active && !_g_camera.client);
+
 			bool has_listeners = false;
 			bool has_speakers = false;
 			US_LIST_ITERATE(_g_clients, client, {
@@ -629,6 +859,10 @@ static struct janus_plugin_result *_plugin_handle_message(
 					atomic_store(&client->transmit_acap, with_acap);
 					atomic_store(&client->transmit_aplay, with_aplay);
 					atomic_store(&client->video_orient, video_orient);
+					if (with_vplay) {
+						us_janus_client_start_vplay(client, _g_vplay_ring);
+						_g_camera.client = client;
+					}
 				}
 				has_listeners = (has_listeners || atomic_load(&client->transmit_acap));
 				has_speakers = (has_speakers || atomic_load(&client->transmit_aplay));
@@ -640,12 +874,31 @@ static struct janus_plugin_result *_plugin_handle_message(
 
 	} else if (!strcmp(request_str, "features")) {
 		const char *const ice_url = getenv("JANUS_USTREAMER_WEB_ICE_URL");
+
+		_LOCK_VPLAY
+		json_t *camera_req = NULL;
+		if (_g_camera.active && _g_camera.client == NULL) {
+			json_t *resolution = json_object();
+			json_object_set_new(resolution, "width", json_integer(_g_camera.width));
+			json_object_set_new(resolution, "height", json_integer(_g_camera.height));
+
+			camera_req = json_object();
+			json_object_set_new(camera_req, "resolution", resolution);
+			json_object_set_new(camera_req, "fps", json_integer(_g_camera.fps));
+		}
+		_UNLOCK_VPLAY
+
 		json_t *const features = json_pack(
-			"{s:b, s:b, s:{s:s?}}",
+			"{s:b, s:b, s:{s:b, s:o*}, s:{s:s?}}",
 			"audio", us_au_probe(_g_config->acap_dev_name),
 			"mic", us_au_probe(_g_config->aplay_dev_name),
-			"ice", "url", (ice_url != NULL ? ice_url : default_ice_url)
+			"camera",
+				"enabled", _is_camera_enabled(),
+				"request", camera_req,
+			"ice",
+				"url", (ice_url != NULL ? ice_url : _g_default_ice_url)
 		);
+
 		PUSH_STATUS("features", features, NULL);
 		json_decref(features);
 
@@ -672,30 +925,60 @@ done:
 
 static void _plugin_incoming_rtp(janus_plugin_session *session, janus_plugin_rtp *packet) {
 	_IF_DISABLED({ return; });
-	if (session == NULL || packet == NULL || packet->video) {
-		return; // Accept only valid audio
+	if (session == NULL || packet == NULL) {
+		return; // Accept only valid packets
 	}
-	_LOCK_APLAY;
-	US_LIST_ITERATE(_g_clients, client, {
-		if (client->session == session) {
-			us_janus_client_recv(client, packet);
-			break;
+	if (packet->video) {
+		_LOCK_VPLAY;
+		if (_g_camera.client != NULL && _g_camera.client->session == session) {
+			us_janus_client_recv(_g_camera.client, packet);
 		}
-	});
-	_UNLOCK_APLAY;
+		_UNLOCK_VPLAY;
+	} else {
+		_LOCK_APLAY;
+		US_LIST_ITERATE(_g_clients, client, {
+			if (client->session == session) {
+				us_janus_client_recv(client, packet);
+				break;
+			}
+		});
+		_UNLOCK_APLAY;
+	}
 }
 
 static void _plugin_incoming_rtcp(janus_plugin_session *session, janus_plugin_rtcp *packet) {
 	_IF_DISABLED({ return; });
-	if (session == NULL || packet == NULL || !packet->video) {
-		return; // Accept only valid video
+	if (
+		session == NULL || packet == NULL || !packet->video
+		|| packet->length < sizeof(janus_rtcp_header)
+	) {
+		return;
 	}
+
 	if (
 		janus_rtcp_has_pli(packet->buffer, packet->length)
 		|| janus_rtcp_has_fir(packet->buffer, packet->length)
 	) {
 		atomic_store(&_g_key_required, true);
 	}
+
+	janus_rtcp_header *header = (janus_rtcp_header *)packet->buffer;
+
+	if (header->type == RTCP_SR && janus_rtcp_check_sr(header, packet->length)) {
+		const janus_rtcp_sr *const sr = (janus_rtcp_sr *)header;
+
+		if (sr->si.ntp_ts_msw && sr->si.ntp_ts_lsw) {
+			_LOCK_VPLAY;
+			US_LIST_ITERATE(_g_clients, client, {
+				if (client->session == session) {
+					const ldf ntp_ts = ntohl(sr->si.ntp_ts_msw) + ntohl(sr->si.ntp_ts_lsw) / 4294967296.L;
+					us_rtpc_sync_timestamp(client->rtpc, ntp_ts, ntohl(sr->si.rtp_ts));
+					break;
+				}
+			});
+			_UNLOCK_VPLAY;
+		}
+	}
 }
 
 

janus/src/rtpc.c

@@ -0,0 +1,334 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C)  2026  Maxim Devaev <mdevaev@gmail.com>                   #
+#    Copyright (C)  2026  Sergey Radionov <rsatom@gmail.com>                 #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#include "rtpc.h"
+
+#include <linux/videodev2.h>
+
+#include "uslibs/logging.h"
+
+#include "h264.h"
+
+
+// H.264 fragment types. Commented types are not used in WebRTC.
+enum {
+	_FR_REGULAR_NAL_MIN = 1,
+	_FR_REGULAR_NAL_MAX = 23,
+	_FR_STAP_A = 24,		// Single Time Aggregation Packet, Type A (STAP-A)
+	// _FR_STAP_B = 25,		// Single Time Aggregation Packet, Type B (STAP-B)
+	// _FR_MTAP_16 = 26,	// Multi-Time Aggregation Packet, Type A (MTAP-16)
+	// _FR_MTAP_24 = 27,	// Multi-Time Aggregation Packet, Type B (MTAP-24)
+	_FR_FU_A = 28,			// Fragmentation unit (FU-A)
+	// _FR_FU_B = 29,		// Fragmentation unit (FU-B)
+};
+
+enum {
+	_NALU_IDR = 5,
+	_NALU_SPS = 7,
+	_NALU_PPS = 8,
+};
+
+
+static int _update_sequence_for_new_packet(us_rtpc_s *rtpc, u16 seq);
+static ldf _rtp_to_local_ts(us_rtpc_s *rtpc, u32 rts);
+static void _update_timestamps_for_new_packet(us_rtpc_s *rtpc, u32 rts);
+
+static void _unwrapping_begin(us_rtpc_s *rtpc);
+static void _unwrapping_end(us_rtpc_s *rtpc);
+static void _unwrap_whole(us_rtpc_s *rtpc, const u8 *data, u16 size);
+
+
+us_rtpc_s *us_rtpc_init(us_frame_callback_f callback, void *user_data) {
+	us_rtpc_s *rtpc;
+	US_CALLOC(rtpc, 1);
+	rtpc->user_data = user_data;
+	rtpc->frame = us_frame_init();
+	rtpc->callback = callback;
+	return rtpc;
+}
+
+void us_rtpc_destroy(us_rtpc_s *rtpc) {
+	us_frame_destroy(rtpc->frame);
+	free(rtpc);
+}
+
+us_rtpc_unwrap_result_e us_rtpc_unwrap(
+	us_rtpc_s *rtpc,
+	const u8 *payload,
+	int payload_size,
+	u16 seq,
+	u32 rts
+) {
+	US_A(payload != NULL);
+	US_A(payload_size >= 1);
+
+	if (_update_sequence_for_new_packet(rtpc, seq) < 0) {
+		return US_RUR_INVALID_SEQ;
+	}
+	_update_timestamps_for_new_packet(rtpc, rts);
+
+	rtpc->had_packets = true;
+
+	const u8 fragment_type = payload[0] & 0x1F;
+
+	if (rtpc->fu && (rtpc->fu_rts != rts || fragment_type != _FR_FU_A)) {
+		rtpc->fu = false;
+		if (!rtpc->fu_is_bad) { // Not reported yet
+			rtpc->skip_until_key = true;
+			return US_RUR_DEPACKETIZATION_FAILED_RETRY;
+		}
+	}
+
+	if (fragment_type >= _FR_REGULAR_NAL_MIN && fragment_type <= _FR_REGULAR_NAL_MAX) {
+		_unwrap_whole(rtpc, payload, payload_size);
+
+	} else if (fragment_type == _FR_STAP_A) { // Single Time Aggregation Packet, Type A (STAP-A).
+		const u8 *aggregated_unit = payload + 1; // skip aggregation header (1 byte)
+		payload_size -= 1;
+
+		while (payload_size > 0) {
+			if (payload_size < 2) { // Aggregated NAL unit size (2 bytes) is missing
+				US_LOG_ERROR("Aggregated NALU size is missing: seq=%u, rts=%u", seq, rts);
+				return US_RUR_BAD_PACKET;
+			}
+
+			const u16 nalu_size = ntohs(*((const u16 *)aggregated_unit));
+			aggregated_unit	+= 2;
+			payload_size -= 2;
+
+			if (nalu_size == 0) {
+				US_LOG_ERROR("Aggregated NALU size is zero: seq=%u, rts=%u", seq, rts);
+				return US_RUR_BAD_PACKET;
+			}
+
+			if (payload_size < nalu_size) { // aggregated unit size declared as bigger than available data size
+				US_LOG_ERROR("Aggregated NALU size is too big: seq=%u, rts=%u", seq, rts);
+				return US_RUR_BAD_PACKET;
+			}
+
+			_unwrap_whole(rtpc, aggregated_unit, nalu_size);
+
+			aggregated_unit	+= nalu_size;
+			payload_size -= nalu_size;
+		}
+
+	} else if (fragment_type == _FR_FU_A) { // Fragmentation unit (FU-A).
+		if (payload_size < 2) { // FU header is missing (2 bytes)
+			US_LOG_ERROR("FU header is missing: seq=%u, rts=%u", seq, rts);
+			return US_RUR_BAD_PACKET;
+		}
+
+		const u8 fu_indicator = payload[0];
+		const u8 fu_header = payload[1];
+		const bool first_fragment = !!(fu_header & 0x80);
+		const bool last_fragment = !!(fu_header & 0x40);
+
+		if (first_fragment && last_fragment) { // should never happen according to RFC
+			return US_RUR_BAD_PACKET;
+		}
+
+		if (rtpc->fu && rtpc->fu_is_bad) {
+			return US_RUR_DEPACKETIZATION_SKIPPED;
+		}
+
+		if (!rtpc->fu) {
+			rtpc->fu = true;
+			rtpc->fu_seq = seq;
+			rtpc->fu_rts = rts;
+
+			if (!first_fragment) {
+				// US_LOG_INFO("Ignoring FU without a first fragment: seq=%u, rts=%u", seq, rts);
+				rtpc->fu_is_bad = true;
+				rtpc->skip_until_key = true;
+				return US_RUR_DEPACKETIZATION_FAILED;
+			}
+			rtpc->fu_is_bad = false;
+
+			const u8 header = (fu_indicator & 0xE0) | (fu_header & 0x1F);
+			_unwrapping_begin(rtpc);
+			us_frame_append_data(rtpc->frame, &header, sizeof(header));
+			// - FU indicator - FU header
+			us_frame_append_data(rtpc->frame, (const u8 *)payload + 2, payload_size - 2);
+
+		} else if (!first_fragment && (seq == rtpc->fu_seq + 1) && (rts == rtpc->fu_rts)) {
+			//       ^^^ Already had first fragment
+			rtpc->fu_seq = seq;
+
+			// - FU indicator - FU header
+			us_frame_append_data(rtpc->frame, (const u8 *)payload + 2, payload_size - 2);
+
+			if (last_fragment) {
+				_unwrapping_end(rtpc);
+				rtpc->fu = false;
+			}
+
+		} else {
+			// US_LOG_ERROR("Something went wrong with FU processing: seq=%u, rts=%u", seq, rts);
+			rtpc->fu_is_bad = true;
+			rtpc->skip_until_key = true;
+			return US_RUR_DEPACKETIZATION_FAILED;
+		}
+
+	} else {
+		US_LOG_ERROR("Unsupported video unit type=%x: seq=%u, rts=%u", fragment_type, seq, rts);
+		return US_RUR_UNSUPPORTED_UNIT_TYPE;
+	}
+	return US_RUR_SUCCESS;
+}
+
+void us_rtpc_sync_timestamp(us_rtpc_s *rtpc, ldf ntp_ts, u32 packet_rts) {
+	if (!rtpc->had_packets) {
+		return;
+	}
+	if (rtpc->reference_ntp_ts > ntp_ts) {
+		return;
+	}
+
+	if (rtpc->reference_ntp_ts == 0.L) {
+		if (rtpc->reference_packet_rts < packet_rts) {
+			const ldf diff = (packet_rts - rtpc->reference_packet_rts) / 90000.L;
+			rtpc->reference_local_ts += diff;
+			rtpc->reference_packet_rts = packet_rts;
+			rtpc->reference_ntp_ts = ntp_ts;
+		}
+	} else {
+		const ldf diff = ntp_ts - rtpc->reference_ntp_ts;
+		rtpc->reference_local_ts += diff;
+		rtpc->reference_packet_rts = packet_rts;
+		rtpc->reference_ntp_ts = ntp_ts;
+	}
+}
+
+static int _update_sequence_for_new_packet(us_rtpc_s *rtpc, u16 seq) {
+	if (
+		rtpc->had_packets
+		&& (
+			rtpc->last_packet_seq == seq
+			|| (rtpc->last_packet_seq < seq && seq - rtpc->last_packet_seq > (UINT16_MAX >> 1))
+			|| (seq < rtpc->last_packet_seq && rtpc->last_packet_seq - seq < (UINT16_MAX >> 1))
+		)
+	) {
+		return -1;
+	}
+	rtpc->last_packet_seq = seq;
+	return 0;
+}
+
+static ldf _rtp_to_local_ts(us_rtpc_s *rtpc, u32 rts) {
+	if (rtpc->reference_packet_rts <= rts) {
+		return rtpc->reference_local_ts + (rts - rtpc->reference_packet_rts) / 90000.L;
+	} else {
+		return rtpc->reference_local_ts - (rtpc->reference_packet_rts - rts) / 90000.L;
+	}
+}
+
+static void _update_timestamps_for_new_packet(us_rtpc_s *rtpc, u32 rts) {
+	const ldf now_ts = us_get_now_monotonic();
+	ldf ts; // Fill for any goto
+
+	if (!rtpc->had_packets) {
+		ts = _rtp_to_local_ts(rtpc, rts); // Used in packet_update
+		goto full_update;
+	}
+	if (rtpc->last_packet_rts != rts) { // If this is a new frame
+		ts = _rtp_to_local_ts(rtpc, rts);
+		if (rtpc->last_packet_rts > rts || fabsl(now_ts - ts) > .1L) {
+			// ^ RTP timestamp wrapped |OR| ^ Too significant mistake
+			goto full_update;
+		}
+		goto packet_update;
+	}
+
+	return;
+
+full_update:
+	rtpc->reference_local_ts = now_ts;
+	rtpc->reference_packet_rts = rts;
+	rtpc->reference_ntp_ts = 0.L; // Force resync
+	// US_LOG_INFO("Reference timestamps reseted");
+packet_update:
+	rtpc->last_packet_rts = rts;
+	rtpc->frame_ts = ts;
+}
+
+static void _unwrapping_begin(us_rtpc_s *rtpc) {
+	us_frame_s *const frame = rtpc->frame;
+	u8 *const data = frame->data;
+	const uz allocated = frame->allocated;
+
+	// Reset the content, start from scratch
+	memset(frame, 0, sizeof(us_frame_s));
+	frame->data = data;
+	frame->allocated = allocated;
+
+	static const u8 nalu_start_code[] = {0, 0, 0, 1};
+	us_frame_set_data(rtpc->frame, nalu_start_code, sizeof(nalu_start_code));
+}
+
+static void _unwrapping_end(us_rtpc_s *rtpc) {
+	us_frame_s *const frame = rtpc->frame;
+
+	US_A(frame->used > 4); // NAL unit start code
+
+	const u8 nalu_type = (frame->data[4] & 0x1F);
+	if (rtpc->skip_until_key && nalu_type != _NALU_SPS && nalu_type != _NALU_PPS) {
+		if (nalu_type != _NALU_IDR) {
+			frame->used = 0;
+			return;
+		} else {
+			rtpc->skip_until_key = false;
+		}
+	}
+
+	frame->format = V4L2_PIX_FMT_H264;
+	frame->width = rtpc->frame_width;
+	frame->height = rtpc->frame_height;
+	frame->grab_begin_ts = rtpc->frame_ts;
+	frame->grab_end_ts = rtpc->frame_ts;
+
+	// Exchange frames
+	rtpc->frame = rtpc->callback(frame, rtpc->user_data);
+	US_A(rtpc->frame != NULL);
+}
+
+static void _unwrap_whole(us_rtpc_s *rtpc, const u8 *data, u16 size) {
+	US_A(size >= 1);
+
+	const u8 nalu_type = (data[0] & 0x1F);
+	if (nalu_type == _NALU_SPS) {
+		uint width = 0;
+		uint height = 0;
+		if (
+			!us_h264_parse_sps_nalu(data, size, &width, &height)
+			&& (rtpc->frame_width != width || rtpc->frame_height != height)
+		) {
+			US_LOG_INFO("Frame size updated: width=%u, height=%u", width, height);
+			rtpc->frame_width = width;
+			rtpc->frame_height = height;
+		}
+	}
+
+	_unwrapping_begin(rtpc);
+	us_frame_append_data(rtpc->frame, data, size);
+	_unwrapping_end(rtpc);
+}

janus/src/rtpc.h

@@ -0,0 +1,85 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C)  2026  Maxim Devaev <mdevaev@gmail.com>                   #
+#    Copyright (C)  2026  Sergey Radionov <rsatom@gmail.com>                 #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#pragma once
+
+#include <janus/rtp.h>
+
+#include "uslibs/types.h"
+#include "uslibs/frame.h"
+
+
+// Should return:
+// * input frame
+// * replacement frame - to use it in following parsing
+typedef us_frame_s *(*us_frame_callback_f)(us_frame_s *frame, void *user_data);
+
+typedef struct { // *_rts == [R]TP [T]ime[S]tamp
+	us_frame_callback_f		callback;
+	void					*user_data;
+
+	bool		had_packets; // Были ли вообще какие-то пакеты когда-то, взводится один раз
+
+	ldf			reference_local_ts; // Локальное время
+	u32			reference_packet_rts;
+	ldf			reference_ntp_ts;
+
+	u16			last_packet_seq;
+	u32			last_packet_rts; // Время из последнего полученного пакета
+
+	bool		fu;
+	u16			fu_seq;
+	u32			fu_rts;
+	bool		fu_is_bad;
+
+	bool		skip_until_key;
+
+	us_frame_s	*frame;
+	uint		frame_width;
+	uint		frame_height;
+	ldf			frame_ts; // in case of fu, time of very first packet
+} us_rtpc_s;
+
+typedef enum {
+	US_RUR_SUCCESS = 0,
+	US_RUR_BAD_PACKET = -1,
+	US_RUR_INVALID_SEQ = -2,
+	US_RUR_DEPACKETIZATION_FAILED = -3,
+	// Incomplete FU found. Current packet not handled.
+	// It's required to call us_rtpc_unwrap one else time with the same packet.
+	US_RUR_DEPACKETIZATION_FAILED_RETRY = -4,
+	US_RUR_DEPACKETIZATION_SKIPPED = -5,
+	US_RUR_UNSUPPORTED_UNIT_TYPE = -6,
+} us_rtpc_unwrap_result_e;
+
+us_rtpc_s *us_rtpc_init(us_frame_callback_f callback, void *user_data);
+void us_rtpc_destroy(us_rtpc_s *rtpc);
+
+us_rtpc_unwrap_result_e us_rtpc_unwrap(
+	us_rtpc_s *rtpc,
+	const u8 *payload,
+	int payload_size,
+	u16 seq,
+	u32 rts);
+
+void us_rtpc_sync_timestamp(us_rtpc_s *rtpc, ldf ntp_ts, u32 packet_rts);

man/ustreamer-dump.1

@@ -1,6 +1,6 @@
 .\" Manpage for ustreamer-dump.
 .\" Open an issue or pull request to https://github.com/pikvm/ustreamer to correct errors or typos
-.TH USTREAMER-DUMP 1 "version 6.62" "January 2021"
+.TH USTREAMER-DUMP 1 "version 6.67" "January 2021"
 
 .SH NAME
 ustreamer-dump \- Dump uStreamer's memory sink to file

man/ustreamer.1

@@ -1,6 +1,6 @@
 .\" Manpage for ustreamer.
 .\" Open an issue or pull request to https://github.com/pikvm/ustreamer to correct errors or typos
-.TH USTREAMER 1 "version 6.62" "November 2020"
+.TH USTREAMER 1 "version 6.67" "November 2020"
 
 .SH NAME
 ustreamer \- stream MJPEG video from any V4L2 device to the network

pkg/arch/PKGBUILD

@@ -1,16 +1,15 @@
-# Contributor: Maxim Devaev <mdevaev@gmail.com>
-# Author: Maxim Devaev <mdevaev@gmail.com>
+# Maintainer: Maxim Devaev <mdevaev@gmail.com> [Software author]
 
 
 pkgname=ustreamer
-pkgver=6.62
+pkgver=6.67
 pkgrel=1
 pkgdesc="Lightweight and fast MJPEG-HTTP streamer"
 url="https://github.com/pikvm/ustreamer"
-license=(GPL)
+license=(GPL-3.0-or-later)
 arch=(i686 x86_64 armv6h armv7h aarch64)
 depends=(libjpeg-turbo libevent libbsd libgpiod systemd)
-makedepends=(gcc make pkgconf systemd)
+makedepends=(git)
 source=(${pkgname}::"git+https://github.com/pikvm/ustreamer#commit=v${pkgver}")
 md5sums=(SKIP)
 
@@ -19,7 +18,7 @@ _options="WITH_GPIO=1 WITH_SYSTEMD=1"
 if [ -e /usr/bin/python3 ]; then
 	_options="$_options WITH_PYTHON=1"
 	depends+=("python>=3.14" "python<3.15")
-	makedepends+=(python-setuptools python-pip python-build python-wheel python-installer)
+	makedepends+=(python-setuptools python-build python-wheel python-installer)
 fi
 if [ -e /usr/include/janus/plugins/plugin.h ];then
 	depends+=(janus-gateway alsa-lib opus)

pkg/openwrt/Makefile

@@ -9,7 +9,7 @@
 include $(TOPDIR)/rules.mk
 
 PKG_NAME:=ustreamer
-PKG_VERSION:=6.62
+PKG_VERSION:=6.67
 PKG_RELEASE:=1
 PKG_MAINTAINER:=Maxim Devaev <mdevaev@gmail.com>
 

python/setup.py

@@ -34,7 +34,7 @@ def main() -> None:
     flags = _find_flags()
     setup(
         name="ustreamer",
-        version="6.62",
+        version="6.67",
         description="uStreamer tools",
         author="Maxim Devaev",
         author_email="mdevaev@gmail.com",

python/src/ustreamer.c

@@ -63,7 +63,7 @@ typedef struct {
 
 static void _MemsinkObject_destroy_internals(_MemsinkObject *self) {
 	if (self->mem != NULL) {
-		us_memsink_shared_unmap(self->mem, self->data_size);
+		us_memsinksh_unmap(self->mem, self->data_size);
 		self->mem = NULL;
 	}
 	US_CLOSE_FD(self->fd);
@@ -94,7 +94,7 @@ static int _MemsinkObject_init(_MemsinkObject *self, PyObject *args, PyObject *k
 	SET_DOUBLE(drop_same_frames, >= 0);
 #	undef SET_DOUBLE
 
-	if ((self->data_size = us_memsink_calculate_size(self->obj)) == 0) {
+	if ((self->data_size = us_memsinksh_calculate_size(self->obj)) == 0) {
 		PyErr_SetString(PyExc_ValueError, "Invalid memsink object suffix");
 		return -1;
 	}
@@ -105,7 +105,7 @@ static int _MemsinkObject_init(_MemsinkObject *self, PyObject *args, PyObject *k
 		PyErr_SetFromErrno(PyExc_OSError);
 		goto error;
 	}
-	if ((self->mem = us_memsink_shared_map(self->fd, self->data_size)) == NULL) {
+	if ((self->mem = us_memsinksh_map(self->fd, self->data_size)) == NULL) {
 		PyErr_SetFromErrno(PyExc_OSError);
 		goto error;
 	}
@@ -165,6 +165,7 @@ static int _wait_frame(_MemsinkObject *self) {
 
 		// Let the sink know that the client is alive
 		mem->last_client_ts = now_ts;
+		mem->client_magic = US_MEMSINK_MAGIC;
 
 		if (mem->id == self->frame_id) {
 			goto retry;
@@ -174,7 +175,7 @@ static int _wait_frame(_MemsinkObject *self) {
 			if (
 				US_FRAME_COMPARE_GEOMETRY(self->mem, self->frame)
 				&& (self->frame_ts + self->drop_same_frames > now_ts)
-				&& !memcmp(self->frame->data, us_memsink_get_data(mem), mem->used)
+				&& !memcmp(self->frame->data, us_memsinksh_get_data(mem), mem->used)
 			) {
 				self->frame_id = mem->id;
 				goto retry;
@@ -225,7 +226,7 @@ static PyObject *_MemsinkObject_wait_frame(_MemsinkObject *self, PyObject *args,
 	}
 
 	us_memsink_shared_s *mem = self->mem;
-	us_frame_set_data(self->frame, us_memsink_get_data(mem), mem->used);
+	us_frame_set_data(self->frame, us_memsinksh_get_data(mem), mem->used);
 	US_FRAME_COPY_META(self->mem, self->frame);
 	self->frame_id = mem->id;
 	self->frame_ts = us_get_now_monotonic();

src/libs/const.h

@@ -26,7 +26,7 @@
 
 
 #define US_VERSION_MAJOR 6
-#define US_VERSION_MINOR 62
+#define US_VERSION_MINOR 67
 
 #define US_MAKE_VERSION2(_major, _minor) #_major "." #_minor
 #define US_MAKE_VERSION1(_major, _minor) US_MAKE_VERSION2(_major, _minor)

src/libs/list.h

@@ -25,9 +25,9 @@
 #include "tools.h"
 
 
-#define US_LIST_DECLARE \
-	void *prev; \
-	void *next;
+#define US_LIST_DECLARE(x_type) \
+	x_type *prev; \
+	x_type *next;
 
 #define US_LIST_ITERATE(x_first, x_item, ...) { \
 		for (__typeof__(x_first) x_item = x_first; x_item;) { \

src/libs/memsink.c

@@ -56,7 +56,7 @@ us_memsink_s *us_memsink_init_opened(
 
 	US_LOG_INFO("Using %s-sink: %s", name, obj);
 
-	if ((sink->data_size = us_memsink_calculate_size(obj)) == 0) {
+	if ((sink->data_size = us_memsinksh_calculate_size(obj)) == 0) {
 		US_LOG_ERROR("%s-sink: Invalid object suffix", name);
 		goto error;
 	}
@@ -75,7 +75,7 @@ us_memsink_s *us_memsink_init_opened(
 		goto error;
 	}
 
-	if ((sink->mem = us_memsink_shared_map(sink->fd, sink->data_size)) == NULL) {
+	if ((sink->mem = us_memsinksh_map(sink->fd, sink->data_size)) == NULL) {
 		US_LOG_PERROR("%s-sink: Can't mmap shared memory", name);
 		goto error;
 	}
@@ -88,7 +88,7 @@ error:
 
 void us_memsink_destroy(us_memsink_s *sink) {
 	if (sink->mem != NULL) {
-		if (us_memsink_shared_unmap(sink->mem, sink->data_size) < 0) {
+		if (us_memsinksh_unmap(sink->mem, sink->data_size) < 0) {
 			US_LOG_PERROR("%s-sink: Can't unmap shared memory", sink->name);
 		}
 	}
@@ -143,11 +143,10 @@ bool us_memsink_server_check(us_memsink_s *sink, const us_frame_s *frame) {
 	}
 
 	// Проверяем, есть ли у нас живой клиент по таймауту
-	const bool has_clients = (sink->mem->last_client_ts + sink->client_ttl > us_get_now_monotonic());
+	const bool has_clients = us_memsinksh_has_clients(sink->mem, sink->client_ttl);
 	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);
+	if (us_memsink_server_x_unlock(sink) < 0) {
 		return false;
 	}
 	if (has_clients) {
@@ -163,45 +162,34 @@ 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,
-	us_memsink_wants_s *wants
+	us_memsink_wants_s *w_get
 ) {
 	US_A(sink->server);
 
 	const ldf now = us_get_now_monotonic();
 
-	if (frame->used > sink->data_size) {
-		US_LOG_ERROR("%s-sink: Can't put frame: is too big (%zu > %zu)",
-			sink->name, frame->used, sink->data_size);
-		return 0;
-	}
-
 	if (us_flock_timedwait_monotonic(sink->fd, 1) == 0) {
-		US_LOG_VERBOSE("%s-sink: >>>>> Exposing new frame ...", sink->name);
-
-		sink->mem->id = us_get_now_id();
-		if (sink->mem->wants.key && frame->key) {
-			sink->mem->wants.key = false;
-		}
-		if (wants != NULL) {
-			memcpy(wants, &sink->mem->wants, sizeof(us_memsink_wants_s));
-		}
-
-		memcpy(us_memsink_get_data(sink->mem), frame->data, frame->used);
-		sink->mem->used = frame->used;
-		US_FRAME_COPY_META(frame, sink->mem);
-
-		sink->mem->magic = US_MEMSINK_MAGIC;
-		sink->mem->version = US_MEMSINK_VERSION;
-
-		const bool has_clients = (sink->mem->last_client_ts + sink->client_ttl > us_get_now_monotonic());
+		const bool has_clients = us_memsinksh_has_clients(sink->mem, sink->client_ttl);
 		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);
+		US_LOG_VERBOSE("%s-sink: >>>>> Exposing new frame ...", sink->name);
+		const bool exposed = !us_memsink_server_x_put(sink, frame);
+
+		if (w_get != NULL) {
+			if (has_clients) {
+				memcpy(w_get, &sink->mem->wants, sizeof(us_memsink_wants_s));
+			} else {
+				US_MEMSET_ZERO(*w_get);
+			}
+		}
+
+		if (us_memsink_server_x_unlock(sink) < 0) {
 			return -1;
 		}
-		US_LOG_VERBOSE("%s-sink: Exposed new frame; full exposition time = %.3Lf",
-			sink->name, us_get_now_monotonic() - now);
+		if (exposed) {
+			US_LOG_VERBOSE("%s-sink: Exposed new frame; full exposition time = %.3Lf",
+				sink->name, us_get_now_monotonic() - now);
+		}
 
 	} else if (errno == EWOULDBLOCK) {
 		US_LOG_VERBOSE("%s-sink: ===== Shared memory is busy now; frame skipped", sink->name);
@@ -216,8 +204,8 @@ int us_memsink_server_put(
 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_memsink_wants_s *w_get,
+	const us_memsink_wants_s *w_put
 ) {
 	US_A(!sink->server); // Client only
 
@@ -242,33 +230,121 @@ int us_memsink_client_get(
 		goto done;
 	}
 
+	const bool had_client_magic = (sink->mem->client_magic == US_MEMSINK_MAGIC);
 	// Let the sink know that the client is alive
 	sink->mem->last_client_ts = us_get_now_monotonic();
+	sink->mem->client_magic = US_MEMSINK_MAGIC;
 
-	if (sink->mem->id == sink->last_readed_id) {
-		retval = US_ERROR_NO_DATA; // Not updated
+	if (sink->mem->used == 0 || sink->mem->id == sink->last_readed_id) {
+		// ^ Zero frame     |OR| ^ Not updated
+		retval = US_ERROR_NO_DATA;
 		goto done;
 	}
 
 	sink->last_readed_id = sink->mem->id;
-	us_frame_set_data(frame, us_memsink_get_data(sink->mem), sink->mem->used);
+	us_frame_set_data(frame, us_memsinksh_get_data(sink->mem), sink->mem->used);
 	US_FRAME_COPY_META(sink->mem, frame);
 
-	if (get != NULL) {
-		memcpy(get, &sink->mem->wants, sizeof(us_memsink_wants_s));
-	}
-	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;
+	if (w_get != NULL) {
+		if (had_client_magic) { // Чтобы не прочитать мусор, если не было клиентов
+			memcpy(w_get, &sink->mem->wants, sizeof(us_memsink_wants_s));
+		} else {
+			US_MEMSET_ZERO(*w_get);
 		}
 	}
 
+	if (w_put != NULL) {
+		const bool key = sink->mem->wants.key;
+		memcpy(&sink->mem->wants, w_put, sizeof(us_memsink_wants_s));
+		if (key) {
+			sink->mem->wants.key = key;
+		}
+	} else {
+		US_MEMSET_ZERO(sink->mem->wants);
+	}
+
 done:
-	if (flock(sink->fd, LOCK_UN) < 0) {
-		US_LOG_PERROR("%s-sink: Can't unlock memory", sink->name);
+	if (us_memsink_server_x_unlock(sink) < 0) {
 		retval = -1;
 	}
 	return retval;
 }
+
+us_mss_lock_result_e us_memsink_server_x_lock(
+	us_memsink_s *sink,
+	us_memsink_wants_s *w_get
+) {
+	US_A(sink->server);
+
+	if (us_flock_timedwait_monotonic(sink->fd, 1) < 0) {
+		if (errno == EWOULDBLOCK) {
+			return US_MSS_BUSY;
+		}
+		US_LOG_PERROR("%s-sink: Can't lock memory", sink->name);
+		return US_MSS_ERROR;
+	}
+
+	if (sink->mem->magic == US_MEMSINK_MAGIC && sink->mem->version == US_MEMSINK_VERSION) {
+		// Если инициализировано - значит клиенту что-то уже могли написать
+		const bool has_clients = us_memsinksh_has_clients(sink->mem, sink->client_ttl);
+		atomic_store(&sink->has_clients, has_clients);
+		if (has_clients) {
+			if (w_get != NULL) {
+				memcpy(w_get, &sink->mem->wants, sizeof(us_memsink_wants_s));
+			}
+			// sink->unsafe_last_client_ts = sink->mem->last_client_ts; // _x_ functions don't use it
+			return US_MSS_SUCCESS;
+		}
+	} else {
+		// А иначе скормим клиентам инициализацию с пустым фреймом,
+		// в следующий раз прочтем от них что-то путное.
+		us_memsink_server_x_put(sink, NULL);
+	}
+
+	if (us_memsink_server_x_unlock(sink) < 0) {
+		return US_MSS_ERROR;
+	}
+	return US_MSS_NO_CLIENT;
+}
+
+bool us_memsink_server_x_is_consumed(us_memsink_s *sink) {
+	return (sink->mem->used == 0);
+}
+
+void us_memsink_server_x_set_consumed(us_memsink_s *sink) {
+	sink->mem->used = 0;
+}
+
+int us_memsink_server_x_put(us_memsink_s *sink, const us_frame_s *frame) {
+	if (frame != NULL) {
+		if (frame->used > sink->data_size) {
+			US_LOG_ERROR("%s-sink: Can't put frame: it's too big (%zu > %zu)",
+				sink->name, frame->used, sink->data_size);
+			return -1;
+		}
+
+		memcpy(us_memsinksh_get_data(sink->mem), frame->data, frame->used);
+		sink->mem->used = frame->used;
+		US_FRAME_COPY_META(frame, sink->mem);
+
+		if (sink->mem->wants.key && frame->key) {
+			// Можем писать в мусор, но пофигу
+			sink->mem->wants.key = false;
+		}
+	} else {
+		sink->mem->used = 0;
+	}
+
+	sink->mem->id = us_get_now_id();
+	sink->mem->version = US_MEMSINK_VERSION;
+	sink->mem->magic = US_MEMSINK_MAGIC;
+	return 0;
+}
+
+int us_memsink_server_x_unlock(us_memsink_s *sink) {
+	if (flock(sink->fd, LOCK_UN) < 0) {
+		US_LOG_PERROR("%s-sink: Can't unlock memory", sink->name);
+		return -1;
+	}
+	return 0;
+}

src/libs/memsink.h

@@ -61,10 +61,32 @@ 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,
-	us_memsink_wants_s *wants);
+	us_memsink_wants_s *w_get);
 
 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_memsink_wants_s *w_get,
+	const us_memsink_wants_s *w_put);
+
+
+// Low-level
+
+typedef enum {
+	US_MSS_SUCCESS = 0,
+	US_MSS_ERROR = -1,
+	US_MSS_BUSY = -2,
+	US_MSS_NO_CLIENT = -3,
+} us_mss_lock_result_e;
+
+
+us_mss_lock_result_e us_memsink_server_x_lock(
+	us_memsink_s *sink,
+	us_memsink_wants_s *w_get);
+
+bool us_memsink_server_x_is_consumed(us_memsink_s *sink);
+void us_memsink_server_x_set_consumed(us_memsink_s *sink);
+
+int us_memsink_server_x_put(us_memsink_s *sink, const us_frame_s *frame);
+
+int us_memsink_server_x_unlock(us_memsink_s *sink);

src/libs/memsinksh.c

@@ -31,7 +31,7 @@
 #include "tools.h"
 
 
-us_memsink_shared_s *us_memsink_shared_map(int fd, uz data_size) {
+us_memsink_shared_s *us_memsinksh_map(int fd, uz data_size) {
 	us_memsink_shared_s *mem = mmap(
 		NULL,
 		sizeof(us_memsink_shared_s) + data_size,
@@ -44,12 +44,12 @@ us_memsink_shared_s *us_memsink_shared_map(int fd, uz data_size) {
 	return mem;
 }
 
-int us_memsink_shared_unmap(us_memsink_shared_s *mem, uz data_size) {
+int us_memsinksh_unmap(us_memsink_shared_s *mem, uz data_size) {
 	US_A(mem != NULL);
 	return munmap(mem, sizeof(us_memsink_shared_s) + data_size);
 }
 
-uz us_memsink_calculate_size(const char *obj) {
+uz us_memsinksh_calculate_size(const char *obj) {
 	const char *ptr = strrchr(obj, ':');
 	if (ptr == NULL) {
 		ptr = strrchr(obj, '.');
@@ -67,6 +67,13 @@ uz us_memsink_calculate_size(const char *obj) {
 	return 0;
 }
 
-u8 *us_memsink_get_data(us_memsink_shared_s *mem) {
+u8 *us_memsinksh_get_data(us_memsink_shared_s *mem) {
 	return (u8*)(mem) + sizeof(us_memsink_shared_s);
 }
+
+bool us_memsinksh_has_clients(const us_memsink_shared_s *mem, u32 ttl) {
+	return (
+		mem->client_magic == US_MEMSINK_MAGIC
+		&& mem->last_client_ts + ttl > us_get_now_monotonic()
+	);
+}

src/libs/memsinksh.h

@@ -27,7 +27,7 @@
 
 
 #define US_MEMSINK_MAGIC	((u64)0xCAFEBABECAFEBABE)
-#define US_MEMSINK_VERSION	((u32)10)
+#define US_MEMSINK_VERSION	((u32)11)
 
 
 typedef struct {
@@ -44,6 +44,7 @@ typedef struct {
 	u64		id;
 	uz		used;
 
+	u64		client_magic;
 	ldf		last_client_ts;
 	us_memsink_wants_s	wants;
 
@@ -51,8 +52,9 @@ typedef struct {
 } us_memsink_shared_s;
 
 
-us_memsink_shared_s *us_memsink_shared_map(int fd, uz data_size);
-int us_memsink_shared_unmap(us_memsink_shared_s *mem, uz data_size);
+us_memsink_shared_s *us_memsinksh_map(int fd, uz data_size);
+int us_memsinksh_unmap(us_memsink_shared_s *mem, uz data_size);
 
-uz us_memsink_calculate_size(const char *obj);
-u8 *us_memsink_get_data(us_memsink_shared_s *mem);
+uz us_memsinksh_calculate_size(const char *obj);
+u8 *us_memsinksh_get_data(us_memsink_shared_s *mem);
+bool us_memsinksh_has_clients(const us_memsink_shared_s *mem, u32 ttl);

src/libs/signal.c

@@ -25,7 +25,7 @@
 #include <string.h>
 #include <signal.h>
 
-#if defined(__GLIBC__) && __GLIBC__ >= 2 && __GLIBC_MINOR__ >= 32
+#if defined(_GNU_SOURCE) && defined(__GLIBC__) && __GLIBC__ >= 2 && __GLIBC_MINOR__ >= 32
 #	define HAS_SIGABBREV_NP
 #endif
 

src/ustreamer/http/server.h

@@ -36,7 +36,7 @@
 #include "../stream.h"
 
 
-typedef struct {
+typedef struct us_stream_client_sx {
 	struct us_server_sx		*server;
 	struct evhttp_request	*req;
 
@@ -54,15 +54,15 @@ typedef struct {
 
 	us_fpsi_s *fpsi;
 
-	US_LIST_DECLARE;
+	US_LIST_DECLARE(struct us_stream_client_sx);
 } us_stream_client_s;
 
-typedef struct {
+typedef struct us_snapshot_client_sx {
 	struct us_server_sx		*server;
 	struct evhttp_request	*req;
 	ldf						req_ts;
 
-	US_LIST_DECLARE;
+	US_LIST_DECLARE(struct us_snapshot_client_sx);
 } us_snapshot_client_s;
 
 typedef struct {

src/ustreamer/options.c

@@ -156,6 +156,9 @@ enum _US_OPT_VALUES {
 #	endif
 	_O_NOTIFY_PARENT,
 
+	_O_CPU_SCALING_GOVERNOR_IDLE,
+	_O_CPU_SCALING_GOVERNOR_ACTIVE,
+
 	_O_LOG_LEVEL,
 	_O_PERF,
 	_O_VERBOSE,
@@ -270,6 +273,9 @@ static const struct option _LONG_OPTS[] = {
 #	endif
 	{"notify-parent",			no_argument,		NULL,	_O_NOTIFY_PARENT},
 
+	{"cpu-scaling-governor-idle",	required_argument,	NULL,	_O_CPU_SCALING_GOVERNOR_IDLE},
+	{"cpu-scaling-governor-active",	required_argument,	NULL,	_O_CPU_SCALING_GOVERNOR_ACTIVE},
+
 	{"log-level",				required_argument,	NULL,	_O_LOG_LEVEL},
 	{"perf",					no_argument,		NULL,	_O_PERF},
 	{"verbose",					no_argument,		NULL,	_O_VERBOSE},
@@ -550,6 +556,9 @@ int us_options_parse(
 #			endif
 			case _O_NOTIFY_PARENT:			OPT_SET(stream->notify_parent, true);
 
+			case _O_CPU_SCALING_GOVERNOR_IDLE:		OPT_SET(stream->governor_idle, optarg);
+			case _O_CPU_SCALING_GOVERNOR_ACTIVE:	OPT_SET(stream->governor_active, optarg);
+
 			case _O_LOG_LEVEL:			OPT_NUMBER("--log-level", us_g_log_level, US_LOG_LEVEL_INFO, US_LOG_LEVEL_DEBUG, 0);
 			case _O_PERF:				OPT_SET(us_g_log_level, US_LOG_LEVEL_PERF);
 			case _O_VERBOSE:			OPT_SET(us_g_log_level, US_LOG_LEVEL_VERBOSE);

src/ustreamer/stream.c

@@ -94,7 +94,8 @@ static void _stream_drm_ensure_no_signal(us_stream_s *stream);
 static void _stream_expose_jpeg(us_stream_s *stream, const us_frame_s *frame);
 static void _stream_expose_raw(us_stream_s *stream, const us_frame_s *frame);
 static void _stream_encode_expose_h264(us_stream_s *stream, const us_frame_s *frame, bool force_key);
-static void _stream_check_suicide(us_stream_s *stream);
+static void _stream_check_suicide(us_stream_s *stream, bool has_any_clients);
+static void _stream_set_governor(us_stream_s *stream, bool active);
 
 
 us_stream_s *us_stream_init(us_capture_s *cap, us_encoder_s *enc) {
@@ -204,7 +205,7 @@ void us_stream_loop(us_stream_s *stream) {
 
 		US_LOG_INFO("Capturing ...");
 
-		uint slowdown_count = 0;
+		int has_any_clients_prev = -1;
 		while (!atomic_load(&run->stop) && !atomic_load(&threads_stop)) {
 			us_capture_hwbuf_s *hw;
 			switch (us_capture_hwbuf_grab(cap, &hw)) {
@@ -233,17 +234,23 @@ void us_stream_loop(us_stream_s *stream) {
 			us_queue_put(releasers[hw->buf.index].q, hw, 0); // Plan to release
 
 			// Мы не обновляем здесь состояние синков, потому что это происходит внутри обслуживающих их потоков
-			_stream_check_suicide(stream);
-			if (stream->slowdown && !_stream_has_any_clients_cached(stream)) {
+			const bool has_any_clients = _stream_has_any_clients_cached(stream);
+
+			if (has_any_clients_prev != has_any_clients) {
+				_stream_set_governor(stream, has_any_clients);
+				has_any_clients_prev = has_any_clients;
+			}
+
+			_stream_check_suicide(stream, has_any_clients);
+
+			if (stream->slowdown && !has_any_clients) {
 				usleep(100 * 1000);
-				slowdown_count = (slowdown_count + 1) % 10;
-				if (slowdown_count > 0) {
-					continue;
-				}
 			}
 		}
 
 	close:
+		_stream_set_governor(stream, false);
+
 		atomic_store(&threads_stop, true);
 
 #		define DELETE_WORKER(x_ctx) if (x_ctx != NULL) { \
@@ -571,7 +578,7 @@ static int _stream_init_loop(us_stream_s *stream) {
 		UPDATE_SINK(stream->h264_sink);
 #		undef UPDATE_SINK
 
-		_stream_check_suicide(stream);
+		_stream_check_suicide(stream, _stream_has_any_clients_cached(stream));
 
 		stream->cap->dma_export = (
 			stream->enc->type == US_ENCODER_TYPE_M2M_VIDEO
@@ -767,7 +774,7 @@ done:
 	us_fpsi_update(run->http->h264_fpsi, meta.online, &meta);
 }
 
-static void _stream_check_suicide(us_stream_s *stream) {
+static void _stream_check_suicide(us_stream_s *stream, bool has_any_clients) {
 	if (stream->exit_on_no_clients == 0) {
 		return;
 	}
@@ -775,7 +782,7 @@ static void _stream_check_suicide(us_stream_s *stream) {
 	atomic_ullong *last_req_ts = &stream->run->http->last_req_ts;
 	const ldf now_ts = us_get_now_monotonic();
 
-	if (_stream_has_any_clients_cached(stream)) {
+	if (has_any_clients) {
 		atomic_store(last_req_ts, now_ts);
 	} else if (atomic_load(last_req_ts) + stream->exit_on_no_clients < now_ts) {
 		US_LOG_INFO("No requests or HTTP/sink clients found in last %u seconds, exiting ...",
@@ -784,3 +791,26 @@ static void _stream_check_suicide(us_stream_s *stream) {
 		us_process_suicide();
 	}
 }
+
+static void _stream_set_governor(us_stream_s *stream, bool active) {
+	if (!us_str_is_ok(stream->governor_idle) || !us_str_is_ok(stream->governor_active)) {
+		return;
+	}
+
+	const char *const path = "/sys/devices/system/cpu/cpufreq/policy0/scaling_governor";
+	const int fd = open(path, O_RDWR);
+	if (fd < 0) {
+		US_LOG_PERROR("Can't open CPU governor file");
+		return;
+	}
+
+	const char *const gov = (active ? stream->governor_active : stream->governor_idle);
+	const sz len = strlen(gov);
+	if (write(fd, gov, len) != len) {
+		US_LOG_PERROR("Can't set CPU governor to %s", gov);
+	} else {
+		US_LOG_INFO("CPU governor changed to %s", gov);
+	}
+
+	close(fd);
+}

src/ustreamer/stream.h

@@ -87,6 +87,9 @@ typedef struct {
 	bool			exit_on_device_error;
 	uint			exit_on_no_clients;
 
+	char			*governor_idle;
+	char			*governor_active;
+
 	us_memsink_s	*jpeg_sink;
 	us_memsink_s	*raw_sink;
 

src/ustreamer/workers.h

@@ -47,7 +47,7 @@ typedef struct us_worker_sx {
 
 	struct us_workers_pool_sx	*pool;
 
-	US_LIST_DECLARE;
+	US_LIST_DECLARE(struct us_worker_sx);
 } us_worker_s;
 
 typedef void *(*us_workers_pool_job_init_f)(void *arg);