Bump version: 6.21 → 6.22

Extension of c96559e4ac.
Some cheap Chinise cameras produces frames which are detected as 'broken'. However they
are later handled well.
Introduce an option which allows disable the check on demand.

.bumpversion.cfg

@@ -1,7 +1,7 @@
 [bumpversion]
 commit = True
 tag = True
-current_version = 6.10
+current_version = 6.22
 parse = (?P<major>\d+)\.(?P<minor>\d+)
 serialize = 
 	{major}.{minor}

Makefile

@@ -16,6 +16,12 @@ export
 _LINTERS_IMAGE ?= ustreamer-linters
 
 
+# =====
+ifeq (__not_found__,$(shell which pkg-config 2>/dev/null || echo "__not_found__"))
+$(error "No pkg-config found in $(PATH)")
+endif
+
+
 # =====
 define optbool
 $(filter $(shell echo $(1) | tr A-Z a-z), yes on 1)

README.md

@@ -23,7 +23,7 @@
 | Compatibility with mjpg-streamer's API | ✔ | :) |
 
 Footnotes:
-  * ```1``` Long before µStreamer, I made a [patch](https://github.com/jacksonliam/mjpg-streamer/pull/164) to add DV-timings support to mjpg-streamer and to keep it from hanging up no device disconnection. Alas, the patch is far from perfect and I can't guarantee it will work every time - mjpg-streamer's source code is very complicated and its structure is hard to understand. With this in mind, along with needing multithreading and JPEG hardware acceleration in the future, I decided to make my own stream server from scratch instead of supporting legacy code.
+  * ```1``` Long before µStreamer, I made a [patch](https://github.com/jacksonliam/mjpg-streamer/pull/164) to add DV-timings support to mjpg-streamer and to keep it from hanging up on device disconnection. Alas, the patch is far from perfect and I can't guarantee it will work every time - mjpg-streamer's source code is very complicated and its structure is hard to understand. With this in mind, along with needing multithreading and JPEG hardware acceleration in the future, I decided to make my own stream server from scratch instead of supporting legacy code.
 
   * ```2``` This feature allows to cut down outgoing traffic several-fold when streaming HDMI, but it increases CPU usage a little bit. The idea is that HDMI is a fully digital interface and each captured frame can be identical to the previous one byte-wise. There's no need to stream the same image over the net several times a second. With the `--drop-same-frames=20` option enabled, µStreamer will drop all the matching frames (with a limit of 20 in a row). Each new frame is matched with the previous one first by length, then using ```memcmp()```.
 

janus/src/audio.c

@@ -47,7 +47,8 @@
 
 // A number of frames per 1 channel:
 //   - https://github.com/xiph/opus/blob/7b05f44/src/opus_demo.c#L368
-#define _HZ_TO_FRAMES(_hz)	(6 * (_hz) / 50) // 120ms
+// #define _HZ_TO_FRAMES(_hz)	(6 * (_hz) / 50) // 120ms
+#define _HZ_TO_FRAMES(_hz)	((_hz) / 50) // 20ms
 #define _HZ_TO_BUF16(_hz)	(_HZ_TO_FRAMES(_hz) * 2) // One stereo frame = (16bit L) + (16bit R)
 #define _HZ_TO_BUF8(_hz)	(_HZ_TO_BUF16(_hz) * sizeof(s16))
 

janus/src/plugin.c

@@ -406,17 +406,13 @@ static void _plugin_hangup_media(janus_plugin_session *session) { _set_transmit(
 static struct janus_plugin_result *_plugin_handle_message(
 	janus_plugin_session *session, char *transaction, json_t *msg, json_t *jsep) {
 
-	assert(transaction != NULL);
-
-#	define FREE_MSG_JSEP { \
-			US_DELETE(msg, json_decref); \
-			US_DELETE(jsep, json_decref); \
-		}
+	janus_plugin_result_type result_type = JANUS_PLUGIN_OK;
+	char *result_msg = NULL;
 
 	if (session == NULL || msg == NULL) {
-		free(transaction);
-		FREE_MSG_JSEP;
-		return janus_plugin_result_new(JANUS_PLUGIN_ERROR, (msg ? "No session" : "No message"), NULL);
+		result_type = JANUS_PLUGIN_ERROR;
+		result_msg = (msg ? "No session" : "No message");
+		goto done;
 	}
 
 #	define PUSH_ERROR(x_error, x_reason) { \
@@ -425,20 +421,20 @@ static struct janus_plugin_result *_plugin_handle_message(
 			json_object_set_new(m_event, "ustreamer", json_string("event")); \
 			json_object_set_new(m_event, "error_code", json_integer(x_error)); \
 			json_object_set_new(m_event, "error", json_string(x_reason)); \
-			_g_gw->push_event(session, create(), transaction, m_event, NULL); \
+			_g_gw->push_event(session, create(), NULL, m_event, NULL); \
 			json_decref(m_event); \
 		}
 
 	json_t *const request = json_object_get(msg, "request");
 	if (request == NULL) {
 		PUSH_ERROR(400, "Request missing");
-		goto ok_wait;
+		goto done;
 	}
 
 	const char *const request_str = json_string_value(request);
 	if (request_str == NULL) {
 		PUSH_ERROR(400, "Request not a string");
-		goto ok_wait;
+		goto done;
 	}
 	// US_JLOG_INFO("main", "Message: %s", request_str);
 
@@ -448,10 +444,10 @@ static struct janus_plugin_result *_plugin_handle_message(
 			json_t *const m_result = json_object(); \
 			json_object_set_new(m_result, "status", json_string(x_status)); \
 			if (x_payload != NULL) { \
-				json_object_set_new(m_result, x_status, x_payload); \
+				json_object_set(m_result, x_status, x_payload); \
 			} \
 			json_object_set_new(m_event, "result", m_result); \
-			_g_gw->push_event(session, create(), transaction, m_event, x_jsep); \
+			_g_gw->push_event(session, create(), NULL, m_event, x_jsep); \
 			json_decref(m_event); \
 		}
 
@@ -462,8 +458,9 @@ static struct janus_plugin_result *_plugin_handle_message(
 		PUSH_STATUS("stopped", NULL, NULL);
 
 	} else if (!strcmp(request_str, "watch")) {
-		bool with_audio = false;
 		uint video_orient = 0;
+		bool with_audio = false;
+		bool with_mic = false;
 		{
 			json_t *const params = json_object_get(msg, "params");
 			if (params != NULL) {
@@ -473,6 +470,12 @@ static struct janus_plugin_result *_plugin_handle_message(
 						with_audio = (_g_rtpa != NULL && json_boolean_value(obj));
 					}
 				}
+				{
+					json_t *const obj = json_object_get(params, "microphone");
+					if (obj != NULL && json_is_boolean(obj)) {
+						with_mic = (with_audio && json_boolean_value(obj)); // FIXME: also check playback
+					}
+				}
 				{
 					json_t *const obj = json_object_get(params, "orientation");
 					if (obj != NULL && json_is_integer(obj)) {
@@ -489,7 +492,7 @@ static struct janus_plugin_result *_plugin_handle_message(
 		{
 			char *sdp;
 			char *const video_sdp = us_rtpv_make_sdp(_g_rtpv);
-			char *const audio_sdp = (with_audio ? us_rtpa_make_sdp(_g_rtpa) : us_strdup(""));
+			char *const audio_sdp = (with_audio ? us_rtpa_make_sdp(_g_rtpa, with_mic) : us_strdup(""));
 			US_ASPRINTF(sdp,
 				"v=0" RN
 				"o=- %" PRIu64 " 1 IN IP4 0.0.0.0" RN
@@ -542,13 +545,17 @@ static struct janus_plugin_result *_plugin_handle_message(
 		PUSH_ERROR(405, "Not implemented");
 	}
 
-ok_wait:
-	FREE_MSG_JSEP;
-	return janus_plugin_result_new(JANUS_PLUGIN_OK_WAIT, NULL, NULL);
+done:
+	US_DELETE(transaction, free);
+	US_DELETE(msg, json_decref);
+	US_DELETE(jsep, json_decref);
+
+	return janus_plugin_result_new(
+		result_type, result_msg,
+		(result_type == JANUS_PLUGIN_OK ? json_pack("{sb}", "ok", 1) : NULL));
 
 #	undef PUSH_STATUS
 #	undef PUSH_ERROR
-#	undef FREE_MSG_JSEP
 }
 
 static void _plugin_incoming_rtcp(janus_plugin_session *handle, janus_plugin_rtcp *packet) {

janus/src/rtpa.c

@@ -43,21 +43,22 @@ void us_rtpa_destroy(us_rtpa_s *rtpa) {
 	free(rtpa);
 }
 
-char *us_rtpa_make_sdp(us_rtpa_s *rtpa) {
+char *us_rtpa_make_sdp(us_rtpa_s *rtpa, bool mic) {
 	const uint pl = rtpa->rtp->payload;
 	char *sdp;
 	US_ASPRINTF(sdp,
 		"m=audio 1 RTP/SAVPF %u" RN
 		"c=IN IP4 0.0.0.0" RN
 		"a=rtpmap:%u OPUS/48000/2" RN
-		// "a=fmtp:%u useinbandfec=1" RN
+		"a=fmtp:%u sprop-stereo=1" RN // useinbandfec=1
 		"a=rtcp-fb:%u nack" RN
 		"a=rtcp-fb:%u nack pli" RN
 		"a=rtcp-fb:%u goog-remb" RN
 		"a=ssrc:%" PRIu32 " cname:ustreamer" RN
-		"a=sendonly" RN,
-		pl, pl, pl, pl, pl, // pl,
-		rtpa->rtp->ssrc
+		"a=%s" RN,
+		pl, pl, pl, pl, pl, pl,
+		rtpa->rtp->ssrc,
+		(mic ? "sendrecv" : "sendonly")
 	);
 	return sdp;
 }

janus/src/rtpa.h

@@ -36,5 +36,5 @@ typedef struct {
 us_rtpa_s *us_rtpa_init(us_rtp_callback_f callback);
 void us_rtpa_destroy(us_rtpa_s *rtpa);
 
-char *us_rtpa_make_sdp(us_rtpa_s *rtpa);
+char *us_rtpa_make_sdp(us_rtpa_s *rtpa, bool mic);
 void us_rtpa_wrap(us_rtpa_s *rtpa, const u8 *data, uz size, u32 pts);

linters/tox.ini

@@ -3,7 +3,7 @@ envlist = cppcheck, flake8, pylint, mypy, vulture, htmlhint
 skipsdist = true
 
 [testenv]
-basepython = python3.11
+basepython = python3.13
 changedir = /src
 
 [testenv:cppcheck]
@@ -13,8 +13,10 @@ commands = cppcheck \
 			--std=c17 \
 			--error-exitcode=1 \
 			--quiet \
+			--check-level=exhaustive \
 			--enable=warning,portability,performance,style \
 			--suppress=assignmentInAssert \
+			--suppress=assertWithSideEffect \
 			--suppress=variableScope \
 			--inline-suppr \
 			--library=python \
@@ -25,7 +27,7 @@ commands = cppcheck \
 allowlist_externals = bash
 commands = bash -c 'flake8 --config=linters/flake8.ini tools/*.py' python/*.py
 deps =
-	flake8==5.0.4
+	flake8
 	flake8-quotes
 
 [testenv:pylint]
@@ -33,6 +35,7 @@ allowlist_externals = bash
 commands = bash -c 'pylint --rcfile=linters/pylint.ini --output-format=colorized --reports=no tools/*.py python/*.py'
 deps =
 	pylint
+	setuptools
 
 [testenv:mypy]
 allowlist_externals = bash

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.10" "January 2021"
+.TH USTREAMER-DUMP 1 "version 6.22" "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.10" "November 2020"
+.TH USTREAMER 1 "version 6.22" "November 2020"
 
 .SH NAME
 ustreamer \- stream MJPEG video from any V4L2 device to the network
@@ -68,6 +68,9 @@ Desired FPS. Default: maximum possible.
 .BR \-z\ \fIN ", " \-\-min\-frame\-size\ \fIN
 Drop frames smaller then this limit. Useful if the device produces small\-sized garbage frames. Default: 128 bytes.
 .TP
+.BR \-T ", " \-\-allow\-truncated\-frames
+Allows to handle truncated frames. Useful if the device produces incorrect but still acceptable frames. Default: disabled.
+.TP
 .BR \-n ", " \-\-persistent
 Suppress repetitive signal source errors. Default: disabled.
 .TP
@@ -96,8 +99,6 @@ HW ─ Use pre-encoded MJPEG frames directly from camera hardware.
 M2M-VIDEO ─ GPU-accelerated MJPEG encoding.
 
 M2M-IMAGE ─ GPU-accelerated JPEG encoding.
-
-NOOP ─ Don't compress MJPEG stream (do nothing).
 .TP
 .BR \-g\ \fIWxH,... ", " \-\-glitched\-resolutions\ \fIWxH,...
 It doesn't do anything. Still here for compatibility.

pkg/arch/PKGBUILD

@@ -3,14 +3,14 @@
 
 
 pkgname=ustreamer
-pkgver=6.10
+pkgver=6.22
 pkgrel=1
 pkgdesc="Lightweight and fast MJPEG-HTTP streamer"
 url="https://github.com/pikvm/ustreamer"
 license=(GPL)
 arch=(i686 x86_64 armv6h armv7h aarch64)
 depends=(libjpeg libevent libbsd libgpiod systemd)
-makedepends=(gcc make systemd)
+makedepends=(gcc make pkgconf systemd)
 source=(${pkgname}::"git+https://github.com/pikvm/ustreamer#commit=v${pkgver}")
 md5sums=(SKIP)
 

pkg/docker/Dockerfile.alpine

@@ -24,7 +24,7 @@ RUN apk add --no-cache \
 WORKDIR /ustreamer
 COPY --from=build /build/ustreamer/src/ustreamer.bin ustreamer
 
-RUN wget https://raw.githubusercontent.com/pikvm/kvmd/master/configs/kvmd/edid/v3-hdmi.hex -O /edid.hex
+RUN wget https://raw.githubusercontent.com/pikvm/kvmd/master/configs/kvmd/edid/v2.hex -O /edid.hex
 COPY pkg/docker/entry.sh /
 
 EXPOSE 8080

pkg/openwrt/Makefile

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

python/setup.py

@@ -17,7 +17,7 @@ def _find_sources(suffix: str) -> list[str]:
 if __name__ == "__main__":
     setup(
         name="ustreamer",
-        version="6.10",
+        version="6.22",
         description="uStreamer tools",
         author="Maxim Devaev",
         author_email="mdevaev@gmail.com",

src/dump/main.c

@@ -111,9 +111,9 @@ int main(int argc, char *argv[]) {
 	US_LOGGING_INIT;
 	US_THREAD_RENAME("main");
 
-	char *sink_name = NULL;
+	const char *sink_name = NULL;
 	unsigned sink_timeout = 1;
-	char *output_path = NULL;
+	const char *output_path = NULL;
 	bool output_json = false;
 	long long count = 0;
 	long double interval = 0;

src/libs/capture.c

@@ -83,9 +83,9 @@ static const struct {
 };
 
 static int _capture_wait_buffer(us_capture_s *cap);
-static int _capture_consume_event(us_capture_s *cap);
+static int _capture_consume_event(const us_capture_s *cap);
 static void _v4l2_buffer_copy(const struct v4l2_buffer *src, struct v4l2_buffer *dest);
-static bool _capture_is_buffer_valid(us_capture_s *cap, const struct v4l2_buffer *buf, const u8 *data);
+static bool _capture_is_buffer_valid(const us_capture_s *cap, const struct v4l2_buffer *buf, const u8 *data);
 static int _capture_open_check_cap(us_capture_s *cap);
 static int _capture_open_dv_timings(us_capture_s *cap, bool apply);
 static int _capture_open_format(us_capture_s *cap, bool first);
@@ -98,12 +98,12 @@ static int _capture_open_queue_buffers(us_capture_s *cap);
 static int _capture_open_export_to_dma(us_capture_s *cap);
 static int _capture_apply_resolution(us_capture_s *cap, uint width, uint height, float hz);
 
-static void _capture_apply_controls(us_capture_s *cap);
+static void _capture_apply_controls(const us_capture_s *cap);
 static int _capture_query_control(
-	us_capture_s *cap, struct v4l2_queryctrl *query,
+	const us_capture_s *cap, struct v4l2_queryctrl *query,
 	const char *name, uint cid, bool quiet);
 static void _capture_set_control(
-	us_capture_s *cap, const struct v4l2_queryctrl *query,
+	const us_capture_s *cap, const struct v4l2_queryctrl *query,
 	const char *name, uint cid, int value, bool quiet);
 
 static const char *_format_to_string_nullable(uint format);
@@ -421,7 +421,7 @@ int us_capture_hwbuf_grab(us_capture_s *cap, us_capture_hwbuf_s **hw) {
 	return buf.index;
 }
 
-int us_capture_hwbuf_release(us_capture_s *cap, us_capture_hwbuf_s *hw) {
+int us_capture_hwbuf_release(const us_capture_s *cap, us_capture_hwbuf_s *hw) {
 	assert(atomic_load(&hw->refs) == 0);
 	const uint index = hw->buf.index;
 	_LOG_DEBUG("Releasing HW buffer=%u ...", index);
@@ -486,7 +486,7 @@ int _capture_wait_buffer(us_capture_s *cap) {
 	return 0;
 }
 
-static int _capture_consume_event(us_capture_s *cap) {
+static int _capture_consume_event(const us_capture_s *cap) {
 	struct v4l2_event event;
 	if (us_xioctl(cap->run->fd, VIDIOC_DQEVENT, &event) < 0) {
 		_LOG_PERROR("Can't consume V4L2 event");
@@ -513,7 +513,7 @@ static void _v4l2_buffer_copy(const struct v4l2_buffer *src, struct v4l2_buffer
 	}
 }
 
-bool _capture_is_buffer_valid(us_capture_s *cap, const struct v4l2_buffer *buf, const u8 *data) {
+bool _capture_is_buffer_valid(const us_capture_s *cap, const struct v4l2_buffer *buf, const u8 *data) {
 	// Workaround for broken, corrupted frames:
 	// Under low light conditions corrupted frames may get captured.
 	// The good thing is such frames are quite small compared to the regular frames.
@@ -544,8 +544,11 @@ bool _capture_is_buffer_valid(us_capture_s *cap, const struct v4l2_buffer *buf,
 		const u8 *const eoi_ptr = end_ptr - 2;
 		const u16 eoi_marker = (((u16)(eoi_ptr[0]) << 8) | eoi_ptr[1]);
 		if (eoi_marker != 0xFFD9 && eoi_marker != 0xD900 && eoi_marker != 0x0000) {
-			_LOG_DEBUG("Discarding truncated JPEG frame: eoi_marker=0x%04x, bytesused=%u", eoi_marker, buf->bytesused);
-			return false;
+			if (!cap->allow_truncated_frames) {
+				_LOG_DEBUG("Discarding truncated JPEG frame: eoi_marker=0x%04x, bytesused=%u", eoi_marker, buf->bytesused);
+				return false;
+			}
+			_LOG_DEBUG("Got truncated JPEG frame: eoi_marker=0x%04x, bytesused=%u", eoi_marker, buf->bytesused);
 		}
 	}
 
@@ -737,7 +740,7 @@ static int _capture_open_format(us_capture_s *cap, bool first) {
 			_format_to_string_supported(cap->format),
 			_format_to_string_supported(FMT(pixelformat)));
 
-		char *format_str;
+		const char *format_str;
 		if ((format_str = (char*)_format_to_string_nullable(FMT(pixelformat))) != NULL) {
 			_LOG_INFO("Falling back to format=%s", format_str);
 		} else {
@@ -1037,7 +1040,7 @@ static int _capture_apply_resolution(us_capture_s *cap, uint width, uint height,
 	return 0;
 }
 
-static void _capture_apply_controls(us_capture_s *cap) {
+static void _capture_apply_controls(const us_capture_s *cap) {
 #	define SET_CID_VALUE(x_cid, x_field, x_value, x_quiet) { \
 			struct v4l2_queryctrl m_query; \
 			if (_capture_query_control(cap, &m_query, #x_field, x_cid, x_quiet) == 0) { \
@@ -1094,7 +1097,7 @@ static void _capture_apply_controls(us_capture_s *cap) {
 }
 
 static int _capture_query_control(
-	us_capture_s *cap, struct v4l2_queryctrl *query,
+	const us_capture_s *cap, struct v4l2_queryctrl *query,
 	const char *name, uint cid, bool quiet) {
 
 	// cppcheck-suppress redundantPointerOp
@@ -1111,7 +1114,7 @@ static int _capture_query_control(
 }
 
 static void _capture_set_control(
-	us_capture_s *cap, const struct v4l2_queryctrl *query,
+	const us_capture_s *cap, const struct v4l2_queryctrl *query,
 	const char *name, uint cid, int value, bool quiet) {
 
 	if (value < query->minimum || value > query->maximum || value % query->step != 0) {

src/libs/capture.h

@@ -115,6 +115,7 @@ typedef struct {
 	bool				dma_required;
 	uint				desired_fps;
 	uz					min_frame_size;
+	bool				allow_truncated_frames;
 	bool				persistent;
 	uint				timeout;
 	us_controls_s 		ctl;
@@ -133,7 +134,7 @@ int us_capture_open(us_capture_s *cap);
 void us_capture_close(us_capture_s *cap);
 
 int us_capture_hwbuf_grab(us_capture_s *cap, us_capture_hwbuf_s **hw);
-int us_capture_hwbuf_release(us_capture_s *cap, us_capture_hwbuf_s *hw);
+int us_capture_hwbuf_release(const us_capture_s *cap, us_capture_hwbuf_s *hw);
 
 void us_capture_hwbuf_incref(us_capture_hwbuf_s *hw);
 void us_capture_hwbuf_decref(us_capture_hwbuf_s *hw);

src/libs/const.h

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

src/libs/drm/drm.c

@@ -614,7 +614,7 @@ static int _drm_find_sink(us_drm_s *drm, uint width, uint height, float hz) {
 			goto done;
 		}
 
-		drmModeModeInfo *best;
+		const drmModeModeInfo *best;
 		if ((best = _find_best_mode(conn, width, height, hz)) == NULL) {
 			_LOG_ERROR("Can't find any appropriate display modes");
 			drmModeFreeConnector(conn);

src/libs/list.h

@@ -46,6 +46,7 @@
 			x_item->prev = m_last; \
 			m_last->next = x_item; \
 		} \
+		x_item->next = NULL; \
 	}
 
 #define US_LIST_APPEND_C(x_first, x_item, x_count) { \
@@ -64,6 +65,8 @@
 			__typeof__(x_first) m_next = x_item->next; \
 			m_next->prev = x_item->prev; \
 		} \
+		x_item->prev = NULL; \
+		x_item->next = NULL; \
 	}
 
 #define US_LIST_REMOVE_C(x_first, x_item, x_count) { \

src/libs/threading.h

@@ -113,7 +113,9 @@ INLINE void us_thread_get_name(char *name) { // Always required for logging
 #if defined(__linux__)
 		const pid_t tid = syscall(SYS_gettid);
 #elif defined(__FreeBSD__)
-		const pid_t tid = syscall(SYS_thr_self);
+		long id;
+		assert(!syscall(SYS_thr_self, &id));
+		const pid_t tid = id;
 #elif defined(__OpenBSD__)
 		const pid_t tid = syscall(SYS_getthrid);
 #elif defined(__NetBSD__)

src/libs/types.h

@@ -25,6 +25,8 @@
 #include <stdbool.h>
 #include <stdint.h>
 
+#include <sys/types.h>
+
 
 typedef long long sll;
 typedef ssize_t sz;

src/ustreamer/encoder.c

@@ -22,6 +22,26 @@
 
 #include "encoder.h"
 
+#include <stdlib.h>
+#include <strings.h>
+#include <assert.h>
+
+#include <pthread.h>
+
+#include "../libs/types.h"
+#include "../libs/tools.h"
+#include "../libs/array.h"
+#include "../libs/threading.h"
+#include "../libs/logging.h"
+#include "../libs/frame.h"
+#include "../libs/capture.h"
+
+#include "workers.h"
+#include "m2m.h"
+
+#include "encoders/cpu/encoder.h"
+#include "encoders/hw/encoder.h"
+
 
 static const struct {
 	const char *name;
@@ -34,7 +54,7 @@ static const struct {
 	{"M2M-MJPEG",	US_ENCODER_TYPE_M2M_VIDEO},
 	{"M2M-JPEG",	US_ENCODER_TYPE_M2M_IMAGE},
 	{"OMX",			US_ENCODER_TYPE_M2M_IMAGE},
-	{"NOOP",		US_ENCODER_TYPE_NOOP},
+	{"NOOP",		US_ENCODER_TYPE_CPU},
 };
 
 
@@ -43,9 +63,6 @@ static void _worker_job_destroy(void *v_job);
 static bool _worker_run_job(us_worker_s *wr);
 
 
-#define _ER(x_next)	enc->run->x_next
-
-
 us_encoder_s *us_encoder_init(void) {
 	us_encoder_runtime_s *run;
 	US_CALLOC(run, 1);
@@ -62,14 +79,15 @@ us_encoder_s *us_encoder_init(void) {
 }
 
 void us_encoder_destroy(us_encoder_s *enc) {
-	if (_ER(m2ms) != NULL) {
-		for (unsigned index = 0; index < _ER(n_m2ms); ++index) {
-			US_DELETE(_ER(m2ms[index]), us_m2m_encoder_destroy)
+	us_encoder_runtime_s *const run = enc->run;
+	if (run->m2ms != NULL) {
+		for (uint index = 0; index < run->n_m2ms; ++index) {
+			US_DELETE(run->m2ms[index], us_m2m_encoder_destroy);
 		}
-		free(_ER(m2ms));
+		free(run->m2ms);
 	}
-	US_MUTEX_DESTROY(_ER(mutex));
-	free(enc->run);
+	US_MUTEX_DESTROY(run->mutex);
+	free(run);
 	free(enc);
 }
 
@@ -92,85 +110,69 @@ const char *us_encoder_type_to_string(us_encoder_type_e type) {
 }
 
 void us_encoder_open(us_encoder_s *enc, us_capture_s *cap) {
-	assert(enc->run->pool == NULL);
+	us_encoder_runtime_s *const run = enc->run;
+	us_capture_runtime_s *const cr = cap->run;
 
-#	define DR(x_next) cap->run->x_next
+	assert(run->pool == NULL);
 
-	us_encoder_type_e type = (_ER(cpu_forced) ? US_ENCODER_TYPE_CPU : enc->type);
-	unsigned quality = cap->jpeg_quality;
-	unsigned n_workers = US_MIN(enc->n_workers, DR(n_bufs));
-	bool cpu_forced = false;
+	us_encoder_type_e type = enc->type;
+	uint quality = cap->jpeg_quality;
+	uint n_workers = US_MIN(enc->n_workers, cr->n_bufs);
 
-	if (us_is_jpeg(DR(format)) && type != US_ENCODER_TYPE_HW) {
+	if (us_is_jpeg(cr->format) && type != US_ENCODER_TYPE_HW) {
 		US_LOG_INFO("Switching to HW encoder: the input is (M)JPEG ...");
 		type = US_ENCODER_TYPE_HW;
 	}
 
 	if (type == US_ENCODER_TYPE_HW) {
-		if (!us_is_jpeg(DR(format))) {
+		if (us_is_jpeg(cr->format)) {
+			quality = cr->jpeg_quality;
+			n_workers = 1;
+		} else {
 			US_LOG_INFO("Switching to CPU encoder: the input format is not (M)JPEG ...");
-			goto use_cpu;
+			type = US_ENCODER_TYPE_CPU;
+			quality = cap->jpeg_quality;
 		}
-		quality = DR(jpeg_quality);
-		n_workers = 1;
 
 	} else if (type == US_ENCODER_TYPE_M2M_VIDEO || type == US_ENCODER_TYPE_M2M_IMAGE) {
 		US_LOG_DEBUG("Preparing M2M-%s encoder ...", (type == US_ENCODER_TYPE_M2M_VIDEO ? "VIDEO" : "IMAGE"));
-		if (_ER(m2ms) == NULL) {
-			US_CALLOC(_ER(m2ms), n_workers);
+		if (run->m2ms == NULL) {
+			US_CALLOC(run->m2ms, n_workers);
 		}
-		for (; _ER(n_m2ms) < n_workers; ++_ER(n_m2ms)) {
+		for (; run->n_m2ms < n_workers; ++run->n_m2ms) {
 			// Начинаем с нуля и доинициализируем на следующих заходах при необходимости
 			char name[32];
-			US_SNPRINTF(name, 31, "JPEG-%u", _ER(n_m2ms));
+			US_SNPRINTF(name, 31, "JPEG-%u", run->n_m2ms);
 			if (type == US_ENCODER_TYPE_M2M_VIDEO) {
-				_ER(m2ms[_ER(n_m2ms)]) = us_m2m_mjpeg_encoder_init(name, enc->m2m_path, quality);
+				run->m2ms[run->n_m2ms] = us_m2m_mjpeg_encoder_init(name, enc->m2m_path, quality);
 			} else {
-				_ER(m2ms[_ER(n_m2ms)]) = us_m2m_jpeg_encoder_init(name, enc->m2m_path, quality);
+				run->m2ms[run->n_m2ms] = us_m2m_jpeg_encoder_init(name, enc->m2m_path, quality);
 			}
 		}
-
-	} else if (type == US_ENCODER_TYPE_NOOP) {
-		n_workers = 1;
-		quality = 0;
 	}
 
-	goto ok;
+	if (quality == 0) {
+		US_LOG_INFO("Using JPEG quality: encoder default");
+	} else {
+		US_LOG_INFO("Using JPEG quality: %u%%", quality);
+	}
 
-	use_cpu:
-		type = US_ENCODER_TYPE_CPU;
-		quality = cap->jpeg_quality;
+	US_MUTEX_LOCK(run->mutex);
+	run->type = type;
+	run->quality = quality;
+	US_MUTEX_UNLOCK(run->mutex);
 
-	ok:
-		if (type == US_ENCODER_TYPE_NOOP) {
-			US_LOG_INFO("Using JPEG NOOP encoder");
-		} else if (quality == 0) {
-			US_LOG_INFO("Using JPEG quality: encoder default");
-		} else {
-			US_LOG_INFO("Using JPEG quality: %u%%", quality);
-		}
+	const ldf desired_interval = (
+		cap->desired_fps > 0 && (cap->desired_fps < cap->run->hw_fps || cap->run->hw_fps == 0)
+		? (ldf)1 / cap->desired_fps
+		: 0
+	);
 
-		US_MUTEX_LOCK(_ER(mutex));
-		_ER(type) = type;
-		_ER(quality) = quality;
-		if (cpu_forced) {
-			_ER(cpu_forced) = true;
-		}
-		US_MUTEX_UNLOCK(_ER(mutex));
-
-		const long double desired_interval = (
-			cap->desired_fps > 0 && (cap->desired_fps < cap->run->hw_fps || cap->run->hw_fps == 0)
-			? (long double)1 / cap->desired_fps
-			: 0
-		);
-
-		enc->run->pool = us_workers_pool_init(
-			"JPEG", "jw", n_workers, desired_interval,
-			_worker_job_init, (void*)enc,
-			_worker_job_destroy,
-			_worker_run_job);
-
-#	undef DR
+	enc->run->pool = us_workers_pool_init(
+		"JPEG", "jw", n_workers, desired_interval,
+		_worker_job_init, (void*)enc,
+		_worker_job_destroy,
+		_worker_run_job);
 }
 
 void us_encoder_close(us_encoder_s *enc) {
@@ -178,11 +180,12 @@ void us_encoder_close(us_encoder_s *enc) {
 	US_DELETE(enc->run->pool, us_workers_pool_destroy);
 }
 
-void us_encoder_get_runtime_params(us_encoder_s *enc, us_encoder_type_e *type, unsigned *quality) {
-	US_MUTEX_LOCK(_ER(mutex));
-	*type = _ER(type);
-	*quality = _ER(quality);
-	US_MUTEX_UNLOCK(_ER(mutex));
+void us_encoder_get_runtime_params(us_encoder_s *enc, us_encoder_type_e *type, uint *quality) {
+	us_encoder_runtime_s *const run = enc->run;
+	US_MUTEX_LOCK(run->mutex);
+	*type = run->type;
+	*quality = run->quality;
+	US_MUTEX_UNLOCK(run->mutex);
 }
 
 static void *_worker_job_init(void *v_enc) {
@@ -200,35 +203,28 @@ static void _worker_job_destroy(void *v_job) {
 }
 
 static bool _worker_run_job(us_worker_s *wr) {
-	us_encoder_job_s *job = wr->job;
-	us_encoder_s *enc = job->enc; // Just for _ER()
-	const us_frame_s *src = &job->hw->raw;
-	us_frame_s *dest = job->dest;
+	us_encoder_job_s *const job = wr->job;
+	us_encoder_runtime_s *const run = job->enc->run;
+	const us_frame_s *const src = &job->hw->raw;
+	us_frame_s *const dest = job->dest;
 
-	if (_ER(type) == US_ENCODER_TYPE_CPU) {
+	if (run->type == US_ENCODER_TYPE_CPU) {
 		US_LOG_VERBOSE("Compressing JPEG using CPU: worker=%s, buffer=%u",
 			wr->name, job->hw->buf.index);
-		us_cpu_encoder_compress(src, dest, _ER(quality));
+		us_cpu_encoder_compress(src, dest, run->quality);
 
-	} else if (_ER(type) == US_ENCODER_TYPE_HW) {
+	} else if (run->type == US_ENCODER_TYPE_HW) {
 		US_LOG_VERBOSE("Compressing JPEG using HW (just copying): worker=%s, buffer=%u",
 			wr->name, job->hw->buf.index);
 		us_hw_encoder_compress(src, dest);
 
-	} else if (_ER(type) == US_ENCODER_TYPE_M2M_VIDEO || _ER(type) == US_ENCODER_TYPE_M2M_IMAGE) {
+	} else if (run->type == US_ENCODER_TYPE_M2M_VIDEO || run->type == US_ENCODER_TYPE_M2M_IMAGE) {
 		US_LOG_VERBOSE("Compressing JPEG using M2M-%s: worker=%s, buffer=%u",
-			(_ER(type) == US_ENCODER_TYPE_M2M_VIDEO ? "VIDEO" : "IMAGE"), wr->name, job->hw->buf.index);
-		if (us_m2m_encoder_compress(_ER(m2ms[wr->number]), src, dest, false) < 0) {
+			(run->type == US_ENCODER_TYPE_M2M_VIDEO ? "VIDEO" : "IMAGE"), wr->name, job->hw->buf.index);
+		if (us_m2m_encoder_compress(run->m2ms[wr->number], src, dest, false) < 0) {
 			goto error;
 		}
 
-	} else if (_ER(type) == US_ENCODER_TYPE_NOOP) {
-		US_LOG_VERBOSE("Compressing JPEG using NOOP (do nothing): worker=%s, buffer=%u",
-			wr->name, job->hw->buf.index);
-		us_frame_encoding_begin(src, dest, V4L2_PIX_FMT_JPEG);
-		usleep(5000); // Просто чтобы работала логика desired_fps
-		dest->encode_end_ts = us_get_now_monotonic(); // us_frame_encoding_end()
-
 	} else {
 		assert(0 && "Unknown encoder type");
 	}
@@ -238,14 +234,9 @@ static bool _worker_run_job(us_worker_s *wr) {
 		job->dest->encode_end_ts - job->dest->encode_begin_ts,
 		wr->name,
 		job->hw->buf.index);
-
 	return true;
 
-	error:
-		US_LOG_ERROR("Compression failed: worker=%s, buffer=%u", wr->name, job->hw->buf.index);
-		US_LOG_ERROR("Error while compressing buffer, falling back to CPU");
-		US_MUTEX_LOCK(_ER(mutex));
-		_ER(cpu_forced) = true;
-		US_MUTEX_UNLOCK(_ER(mutex));
-		return false;
+error:
+	US_LOG_ERROR("Compression failed: worker=%s, buffer=%u", wr->name, job->hw->buf.index);
+	return false;
 }

src/ustreamer/encoder.h

@@ -22,45 +22,32 @@
 
 #pragma once
 
-#include <stdlib.h>
-#include <stdbool.h>
-#include <strings.h>
-#include <assert.h>
-
 #include <pthread.h>
-#include <linux/videodev2.h>
 
-#include "../libs/tools.h"
-#include "../libs/array.h"
-#include "../libs/threading.h"
-#include "../libs/logging.h"
+#include "../libs/types.h"
 #include "../libs/frame.h"
 #include "../libs/capture.h"
 
 #include "workers.h"
 #include "m2m.h"
 
-#include "encoders/cpu/encoder.h"
-#include "encoders/hw/encoder.h"
 
+#define ENCODER_TYPES_STR "CPU, HW, M2M-VIDEO, M2M-IMAGE"
 
-#define ENCODER_TYPES_STR "CPU, HW, M2M-VIDEO, M2M-IMAGE, NOOP"
 
 typedef enum {
 	US_ENCODER_TYPE_CPU,
 	US_ENCODER_TYPE_HW,
 	US_ENCODER_TYPE_M2M_VIDEO,
 	US_ENCODER_TYPE_M2M_IMAGE,
-	US_ENCODER_TYPE_NOOP,
 } us_encoder_type_e;
 
 typedef struct {
 	us_encoder_type_e	type;
-	unsigned			quality;
-	bool				cpu_forced;
+	uint				quality;
 	pthread_mutex_t		mutex;
 
-	unsigned			n_m2ms;
+	uint				n_m2ms;
 	us_m2m_encoder_s	**m2ms;
 
 	us_workers_pool_s	*pool;
@@ -68,7 +55,7 @@ typedef struct {
 
 typedef struct {
 	us_encoder_type_e	type;
-	unsigned			n_workers;
+	uint				n_workers;
 	char				*m2m_path;
 
 	us_encoder_runtime_s *run;
@@ -90,4 +77,4 @@ const char *us_encoder_type_to_string(us_encoder_type_e type);
 void us_encoder_open(us_encoder_s *enc, us_capture_s *cap);
 void us_encoder_close(us_encoder_s *enc);
 
-void us_encoder_get_runtime_params(us_encoder_s *enc, us_encoder_type_e *type, unsigned *quality);
+void us_encoder_get_runtime_params(us_encoder_s *enc, us_encoder_type_e *type, uint *quality);

src/ustreamer/encoders/cpu/encoder.c

@@ -40,7 +40,10 @@ static void _jpeg_set_dest_frame(j_compress_ptr jpeg, us_frame_s *frame);
 static void _jpeg_write_scanlines_yuv(struct jpeg_compress_struct *jpeg, const us_frame_s *frame);
 static void _jpeg_write_scanlines_rgb565(struct jpeg_compress_struct *jpeg, const us_frame_s *frame);
 static void _jpeg_write_scanlines_rgb24(struct jpeg_compress_struct *jpeg, const us_frame_s *frame);
+#ifndef JCS_EXTENSIONS
+#warning JCS_EXT_BGR is not supported, please use libjpeg-turbo
 static void _jpeg_write_scanlines_bgr24(struct jpeg_compress_struct *jpeg, const us_frame_s *frame);
+#endif
 
 static void _jpeg_init_destination(j_compress_ptr jpeg);
 static boolean _jpeg_empty_output_buffer(j_compress_ptr jpeg);
@@ -67,6 +70,9 @@ void us_cpu_encoder_compress(const us_frame_s *src, us_frame_s *dest, unsigned q
 		case V4L2_PIX_FMT_YUYV:
 		case V4L2_PIX_FMT_YVYU:
 		case V4L2_PIX_FMT_UYVY: jpeg.in_color_space = JCS_YCbCr; break;
+#		ifdef JCS_EXTENSIONS
+		case V4L2_PIX_FMT_BGR24: jpeg.in_color_space = JCS_EXT_BGR; break;
+#		endif
 		default: jpeg.in_color_space = JCS_RGB; break;
 	}
 
@@ -82,7 +88,13 @@ void us_cpu_encoder_compress(const us_frame_s *src, us_frame_s *dest, unsigned q
 		case V4L2_PIX_FMT_UYVY:		_jpeg_write_scanlines_yuv(&jpeg, src); break;
 		case V4L2_PIX_FMT_RGB565:	_jpeg_write_scanlines_rgb565(&jpeg, src); break;
 		case V4L2_PIX_FMT_RGB24:	_jpeg_write_scanlines_rgb24(&jpeg, src); break;
-		case V4L2_PIX_FMT_BGR24:	_jpeg_write_scanlines_bgr24(&jpeg, src); break;
+		case V4L2_PIX_FMT_BGR24:
+#			ifdef JCS_EXTENSIONS
+			_jpeg_write_scanlines_rgb24(&jpeg, src); // Use native JCS_EXT_BGR
+#			else
+			_jpeg_write_scanlines_bgr24(&jpeg, src);
+#			endif
+			break;
 		default: assert(0 && "Unsupported input format for CPU encoder"); return;
 	}
 
@@ -196,6 +208,7 @@ static void _jpeg_write_scanlines_rgb24(struct jpeg_compress_struct *jpeg, const
 	}
 }
 
+#ifndef JCS_EXTENSIONS
 static void _jpeg_write_scanlines_bgr24(struct jpeg_compress_struct *jpeg, const us_frame_s *frame) {
 	uint8_t *line_buf;
 	US_CALLOC(line_buf, frame->width * 3);
@@ -222,6 +235,7 @@ static void _jpeg_write_scanlines_bgr24(struct jpeg_compress_struct *jpeg, const
 
 	free(line_buf);
 }
+#endif
 
 #define JPEG_OUTPUT_BUFFER_SIZE ((size_t)4096)
 

src/ustreamer/http/bev.c

@@ -1,65 +0,0 @@
-/*****************************************************************************
-#                                                                            #
-#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
-#                                                                            #
-#    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/>.  #
-#                                                                            #
-*****************************************************************************/
-
-
-#include "bev.h"
-
-#include <string.h>
-#include <errno.h>
-
-#include <event2/util.h>
-#include <event2/bufferevent.h>
-
-#include "../../libs/tools.h"
-
-
-char *us_bufferevent_format_reason(short what) {
-	char *reason;
-	US_CALLOC(reason, 2048);
-
-	// evutil_socket_error_to_string() is not thread-safe
-	char *const perror_str = us_errno_to_string(EVUTIL_SOCKET_ERROR());
-	bool first = true;
-
-	strncat(reason, perror_str, 1023);
-	free(perror_str);
-	strcat(reason, " (");
-
-#	define FILL_REASON(x_bev, x_name) { \
-			if (what & x_bev) { \
-				if (first) { \
-					first = false; \
-				} else { \
-					strcat(reason, ","); \
-				} \
-				strcat(reason, x_name); \
-			} \
-		}
-	FILL_REASON(BEV_EVENT_READING, "reading");
-	FILL_REASON(BEV_EVENT_WRITING, "writing");
-	FILL_REASON(BEV_EVENT_ERROR, "error");
-	FILL_REASON(BEV_EVENT_TIMEOUT, "timeout");
-	FILL_REASON(BEV_EVENT_EOF, "eof"); // cppcheck-suppress unreadVariable
-#	undef FILL_REASON
-
-	strcat(reason, ")");
-	return reason;
-}

src/ustreamer/http/bev.h

@@ -1,26 +0,0 @@
-/*****************************************************************************
-#                                                                            #
-#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
-#                                                                            #
-#    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/>.  #
-#                                                                            #
-*****************************************************************************/
-
-
-#pragma once
-
-
-char *us_bufferevent_format_reason(short what);

src/ustreamer/http/path.c

@@ -40,7 +40,7 @@ char *us_simplify_request_path(const char *str) {
 	char pre1; // The one before
 	char pre2; // The one before that
 	char *simplified;
-	char *start;
+	const char *start;
 	char *out;
 	char *slash;
 

src/ustreamer/http/server.c

@@ -56,7 +56,6 @@
 #include "../../libs/tools.h"
 #include "../../libs/threading.h"
 #include "../../libs/logging.h"
-#include "../../libs/process.h"
 #include "../../libs/frame.h"
 #include "../../libs/base64.h"
 #include "../../libs/list.h"
@@ -68,9 +67,7 @@
 #	include "../gpio/gpio.h"
 #endif
 
-#include "bev.h"
-#include "unix.h"
-#include "uri.h"
+#include "tools.h"
 #include "mime.h"
 #include "static.h"
 #ifdef WITH_SYSTEMD
@@ -98,9 +95,6 @@ static void _http_send_snapshot(us_server_s *server);
 
 static bool _expose_frame(us_server_s *server, const us_frame_s *frame);
 
-static const char *_http_get_header(struct evhttp_request *request, const char *key);
-static char *_http_get_client_hostport(struct evhttp_request *request);
-
 
 #define _LOG_ERROR(x_msg, ...)	US_LOG_ERROR("HTTP: " x_msg, ##__VA_ARGS__)
 #define _LOG_PERROR(x_msg, ...)	US_LOG_PERROR("HTTP: " x_msg, ##__VA_ARGS__)
@@ -203,8 +197,6 @@ int us_server_listen(us_server_s *server) {
 	}
 
 	us_frame_copy(stream->run->blank->jpeg, ex->frame);
-	ex->notify_last_width = ex->frame->width;
-	ex->notify_last_height = ex->frame->height;
 
 	{
 		struct timeval interval = {0};
@@ -282,8 +274,8 @@ static int _http_preprocess_request(struct evhttp_request *request, us_server_s
 	atomic_store(&server->stream->run->http->last_request_ts, us_get_now_monotonic());
 
 	if (server->allow_origin[0] != '\0') {
-		const char *const cors_headers = _http_get_header(request, "Access-Control-Request-Headers");
-		const char *const cors_method = _http_get_header(request, "Access-Control-Request-Method");
+		const char *const cors_headers = us_evhttp_get_header(request, "Access-Control-Request-Headers");
+		const char *const cors_method = us_evhttp_get_header(request, "Access-Control-Request-Method");
 
 		_A_ADD_HEADER(request, "Access-Control-Allow-Origin", server->allow_origin);
 		_A_ADD_HEADER(request, "Access-Control-Allow-Credentials", "true");
@@ -301,7 +293,7 @@ static int _http_preprocess_request(struct evhttp_request *request, us_server_s
 	}
 
 	if (run->auth_token != NULL) {
-		const char *const token = _http_get_header(request, "Authorization");
+		const char *const token = us_evhttp_get_header(request, "Authorization");
 		if (token == NULL || strcmp(token, run->auth_token) != 0) {
 			_A_ADD_HEADER(request, "WWW-Authenticate", "Basic realm=\"Restricted area\"");
 			evhttp_send_reply(request, 401, "Unauthorized", NULL);
@@ -593,7 +585,7 @@ static void _http_callback_stream(struct evhttp_request *request, void *v_server
 
 		struct evkeyvalq params;
 		evhttp_parse_query(evhttp_request_get_uri(request), &params);
-#		define PARSE_PARAM(x_type, x_name) client->x_name = us_uri_get_##x_type(&params, #x_name)
+#		define PARSE_PARAM(x_type, x_name) client->x_name = us_evkeyvalq_get_##x_type(&params, #x_name)
 		PARSE_PARAM(string, key);
 		PARSE_PARAM(true, extra_headers);
 		PARSE_PARAM(true, advance_headers);
@@ -602,7 +594,7 @@ static void _http_callback_stream(struct evhttp_request *request, void *v_server
 #		undef PARSE_PARAM
 		evhttp_clear_headers(&params);
 
-		client->hostport = _http_get_client_hostport(request);
+		client->hostport = us_evhttp_get_hostport(request);
 		client->id = us_get_now_id();
 
 		{
@@ -682,8 +674,8 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
 		_A_EVBUFFER_ADD_PRINTF(buf, "HTTP/1.0 200 OK" RN);
 		
 		if (client->server->allow_origin[0] != '\0') {
-			const char *const cors_headers = _http_get_header(client->request, "Access-Control-Request-Headers");
-			const char *const cors_method = _http_get_header(client->request, "Access-Control-Request-Method");
+			const char *const cors_headers = us_evhttp_get_header(client->request, "Access-Control-Request-Headers");
+			const char *const cors_method = us_evhttp_get_header(client->request, "Access-Control-Request-Method");
 
 			_A_EVBUFFER_ADD_PRINTF(buf,
 				"Access-Control-Allow-Origin: %s" RN
@@ -957,21 +949,6 @@ static void _http_refresher(int fd, short what, void *v_server) {
 
 	_http_send_stream(server, stream_updated, frame_updated);
 	_http_send_snapshot(server);
-
-	if (
-		frame_updated
-		&& server->notify_parent
-		&& (
-			ex->notify_last_online != ex->frame->online
-			|| ex->notify_last_width != ex->frame->width
-			|| ex->notify_last_height != ex->frame->height
-		)
-	) {
-		ex->notify_last_online = ex->frame->online;
-		ex->notify_last_width = ex->frame->width;
-		ex->notify_last_height = ex->frame->height;
-		us_process_notify_parent();
-	}
 }
 
 static bool _expose_frame(us_server_s *server, const us_frame_s *frame) {
@@ -1016,39 +993,3 @@ static bool _expose_frame(us_server_s *server, const us_frame_s *frame) {
 		 ex->frame->online, (ex->expose_end_ts - ex->expose_begin_ts));
 	return true; // Updated
 }
-
-static const char *_http_get_header(struct evhttp_request *request, const char *key) {
-	return evhttp_find_header(evhttp_request_get_input_headers(request), key);
-}
-
-static char *_http_get_client_hostport(struct evhttp_request *request) {
-	char *addr = NULL;
-	unsigned short port = 0;
-	struct evhttp_connection *conn = evhttp_request_get_connection(request);
-	if (conn != NULL) {
-		char *peer;
-		evhttp_connection_get_peer(conn, &peer, &port);
-		addr = us_strdup(peer);
-	}
-
-	const char *xff = _http_get_header(request, "X-Forwarded-For");
-	if (xff != NULL) {
-		US_DELETE(addr, free);
-		assert((addr = strndup(xff, 1024)) != NULL);
-		for (uint index = 0; addr[index]; ++index) {
-			if (addr[index] == ',') {
-				addr[index] = '\0';
-				break;
-			}
-		}
-	}
-
-	if (addr == NULL) {
-		addr = us_strdup("???");
-	}
-
-	char *hostport;
-	US_ASPRINTF(hostport, "[%s]:%u", addr, port);
-	free(addr);
-	return hostport;
-}

src/ustreamer/http/server.h

@@ -72,10 +72,6 @@ typedef struct {
 	ldf			expose_begin_ts;
 	ldf			expose_cmp_ts;
 	ldf			expose_end_ts;
-
-	bool		notify_last_online;
-	uint		notify_last_width;
-	uint		notify_last_height;
 } us_server_exposed_s;
 
 typedef struct {
@@ -121,8 +117,6 @@ typedef struct us_server_sx {
 	uint	fake_width;
 	uint	fake_height;
 
-	bool	notify_parent;
-
 	us_server_runtime_s *run;
 } us_server_s;
 

src/ustreamer/http/tools.c

@@ -20,7 +20,7 @@
 *****************************************************************************/
 
 
-#include "unix.h"
+#include "tools.h"
 
 #include <string.h>
 #include <unistd.h>
@@ -33,6 +33,8 @@
 
 #include <event2/http.h>
 #include <event2/util.h>
+#include <event2/keyvalq_struct.h>
+#include <event2/bufferevent.h>
 
 #include "../../libs/types.h"
 #include "../../libs/tools.h"
@@ -79,3 +81,94 @@ evutil_socket_t us_evhttp_bind_unix(struct evhttp *http, const char *path, bool
 	}
 	return fd;
 }
+
+const char *us_evhttp_get_header(struct evhttp_request *request, const char *key) {
+	return evhttp_find_header(evhttp_request_get_input_headers(request), key);
+}
+
+char *us_evhttp_get_hostport(struct evhttp_request *request) {
+	char *addr = NULL;
+	unsigned short port = 0;
+	struct evhttp_connection *conn = evhttp_request_get_connection(request);
+	if (conn != NULL) {
+		char *peer;
+		evhttp_connection_get_peer(conn, &peer, &port);
+		addr = us_strdup(peer);
+	}
+
+	const char *xff = us_evhttp_get_header(request, "X-Forwarded-For");
+	if (xff != NULL) {
+		US_DELETE(addr, free);
+		assert((addr = strndup(xff, 1024)) != NULL);
+		for (uint index = 0; addr[index]; ++index) {
+			if (addr[index] == ',') {
+				addr[index] = '\0';
+				break;
+			}
+		}
+	}
+
+	if (addr == NULL) {
+		addr = us_strdup("???");
+	}
+
+	char *hostport;
+	US_ASPRINTF(hostport, "[%s]:%u", addr, port);
+	free(addr);
+	return hostport;
+}
+
+bool us_evkeyvalq_get_true(struct evkeyvalq *params, const char *key) {
+	const char *value_str = evhttp_find_header(params, key);
+	if (value_str != NULL) {
+		if (
+			value_str[0] == '1'
+			|| !evutil_ascii_strcasecmp(value_str, "true")
+			|| !evutil_ascii_strcasecmp(value_str, "yes")
+		) {
+			return true;
+		}
+	}
+	return false;
+}
+
+char *us_evkeyvalq_get_string(struct evkeyvalq *params, const char *key) {
+	const char *const value_str = evhttp_find_header(params, key);
+	if (value_str != NULL) {
+		return evhttp_encode_uri(value_str);
+	}
+	return NULL;
+}
+
+char *us_bufferevent_format_reason(short what) {
+	char *reason;
+	US_CALLOC(reason, 2048);
+
+	// evutil_socket_error_to_string() is not thread-safe
+	char *const perror_str = us_errno_to_string(EVUTIL_SOCKET_ERROR());
+	bool first = true;
+
+	strncat(reason, perror_str, 1023);
+	free(perror_str);
+	strcat(reason, " (");
+
+#	define FILL_REASON(x_bev, x_name) { \
+			if (what & x_bev) { \
+				if (first) { \
+					first = false; \
+				} else { \
+					strcat(reason, ","); \
+				} \
+				strcat(reason, x_name); \
+			} \
+		}
+	FILL_REASON(BEV_EVENT_READING, "reading");
+	FILL_REASON(BEV_EVENT_WRITING, "writing");
+	FILL_REASON(BEV_EVENT_ERROR, "error");
+	FILL_REASON(BEV_EVENT_TIMEOUT, "timeout");
+	FILL_REASON(BEV_EVENT_EOF, "eof"); // cppcheck-suppress unreadVariable
+#	undef FILL_REASON
+
+	strcat(reason, ")");
+	return reason;
+}

src/ustreamer/http/tools.h

@@ -25,9 +25,17 @@
 #include <sys/stat.h>
 
 #include <event2/http.h>
-#include <event2/util.h>
+#include <event2/keyvalq_struct.h>
 
 #include "../../libs/types.h"
 
 
 evutil_socket_t us_evhttp_bind_unix(struct evhttp *http, const char *path, bool rm, mode_t mode);
+
+const char *us_evhttp_get_header(struct evhttp_request *request, const char *key);
+char *us_evhttp_get_hostport(struct evhttp_request *request);
+
+bool us_evkeyvalq_get_true(struct evkeyvalq *params, const char *key);
+char *us_evkeyvalq_get_string(struct evkeyvalq *params, const char *key);
+
+char *us_bufferevent_format_reason(short what);

src/ustreamer/http/uri.c

@@ -1,52 +0,0 @@
-/*****************************************************************************
-#                                                                            #
-#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
-#                                                                            #
-#    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/>.  #
-#                                                                            #
-*****************************************************************************/
-
-
-#include "uri.h"
-
-#include <event2/util.h>
-#include <event2/http.h>
-#include <event2/keyvalq_struct.h>
-
-#include "../../libs/types.h"
-
-
-bool us_uri_get_true(struct evkeyvalq *params, const char *key) {
-	const char *value_str = evhttp_find_header(params, key);
-	if (value_str != NULL) {
-		if (
-			value_str[0] == '1'
-			|| !evutil_ascii_strcasecmp(value_str, "true")
-			|| !evutil_ascii_strcasecmp(value_str, "yes")
-		) {
-			return true;
-		}
-	}
-	return false;
-}
-
-char *us_uri_get_string(struct evkeyvalq *params, const char *key) {
-	const char *const value_str = evhttp_find_header(params, key);
-	if (value_str != NULL) {
-		return evhttp_encode_uri(value_str);
-	}
-	return NULL;
-}

src/ustreamer/http/uri.h

@@ -1,31 +0,0 @@
-/*****************************************************************************
-#                                                                            #
-#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
-#                                                                            #
-#    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/>.  #
-#                                                                            #
-*****************************************************************************/
-
-
-#pragma once
-
-#include <event2/keyvalq_struct.h>
-
-#include "../../libs/types.h"
-
-
-bool us_uri_get_true(struct evkeyvalq *params, const char *key);
-char *us_uri_get_string(struct evkeyvalq *params, const char *key);

src/ustreamer/main.c

@@ -90,6 +90,7 @@ int main(int argc, char *argv[]) {
 	_g_server = us_server_init(_g_stream);
 
 	if ((exit_code = options_parse(options, cap, enc, _g_stream, _g_server)) == 0) {
+		us_stream_update_blank(_g_stream, cap);
 #		ifdef WITH_GPIO
 		us_gpio_init();
 #		endif

src/ustreamer/options.c

@@ -32,6 +32,7 @@ enum _US_OPT_VALUES {
 	_O_IO_METHOD = 'I',
 	_O_DESIRED_FPS = 'f',
 	_O_MIN_FRAME_SIZE = 'z',
+	_O_ALLOW_TRUNCATED_FRAMES = 'T',
 	_O_PERSISTENT = 'n',
 	_O_DV_TIMINGS = 't',
 	_O_BUFFERS = 'b',
@@ -142,6 +143,7 @@ static const struct option _LONG_OPTS[] = {
 	{"io-method",				required_argument,	NULL,	_O_IO_METHOD},
 	{"desired-fps",				required_argument,	NULL,	_O_DESIRED_FPS},
 	{"min-frame-size",			required_argument,	NULL,	_O_MIN_FRAME_SIZE},
+	{"allow-truncated-frames",	no_argument,		NULL,	_O_ALLOW_TRUNCATED_FRAMES},
 	{"persistent",				no_argument,		NULL,	_O_PERSISTENT},
 	{"dv-timings",				no_argument,		NULL,	_O_DV_TIMINGS},
 	{"buffers",					required_argument,	NULL,	_O_BUFFERS},
@@ -353,7 +355,7 @@ int options_parse(us_options_s *options, us_capture_s *cap, us_encoder_s *enc, u
 		}
 
 #	define ADD_SINK(x_prefix) \
-		char *x_prefix##_name = NULL; \
+		const char *x_prefix##_name = NULL; \
 		mode_t x_prefix##_mode = 0660; \
 		bool x_prefix##_rm = false; \
 		unsigned x_prefix##_client_ttl = 10; \
@@ -364,7 +366,7 @@ int options_parse(us_options_s *options, us_capture_s *cap, us_encoder_s *enc, u
 #	undef ADD_SINK
 
 #	ifdef WITH_SETPROCTITLE
-	char *process_name_prefix = NULL;
+	const char *process_name_prefix = NULL;
 #	endif
 
 	char short_opts[128];
@@ -384,6 +386,7 @@ int options_parse(us_options_s *options, us_capture_s *cap, us_encoder_s *enc, u
 			case _O_IO_METHOD:			OPT_PARSE_ENUM("IO method", cap->io_method, us_capture_parse_io_method, US_IO_METHODS_STR);
 			case _O_DESIRED_FPS:		OPT_NUMBER("--desired-fps", cap->desired_fps, 0, US_VIDEO_MAX_FPS, 0);
 			case _O_MIN_FRAME_SIZE:		OPT_NUMBER("--min-frame-size", cap->min_frame_size, 1, 8192, 0);
+			case _O_ALLOW_TRUNCATED_FRAMES:	OPT_SET(cap->allow_truncated_frames, true);
 			case _O_PERSISTENT:			OPT_SET(cap->persistent, true);
 			case _O_DV_TIMINGS:			OPT_SET(cap->dv_timings, true);
 			case _O_BUFFERS:			OPT_NUMBER("--buffers", cap->n_bufs, 1, 32, 0);
@@ -491,7 +494,7 @@ int options_parse(us_options_s *options, us_capture_s *cap, us_encoder_s *enc, u
 #			ifdef WITH_SETPROCTITLE
 			case _O_PROCESS_NAME_PREFIX:	OPT_SET(process_name_prefix, optarg);
 #			endif
-			case _O_NOTIFY_PARENT:			OPT_SET(server->notify_parent, true);
+			case _O_NOTIFY_PARENT:			OPT_SET(stream->notify_parent, true);
 
 			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);
@@ -632,6 +635,8 @@ static void _help(FILE *fp, const us_capture_s *cap, const us_encoder_s *enc, co
 	SAY("    -f|--desired-fps <N>  ──────────────── Desired FPS. Default: maximum possible.\n");
 	SAY("    -z|--min-frame-size <N>  ───────────── Drop frames smaller then this limit. Useful if the device");
 	SAY("                                           produces small-sized garbage frames. Default: %zu bytes.\n", cap->min_frame_size);
+	SAY("    -T|--allow-truncated-frames  ───────── Allows to handle truncated frames. Useful if the device");
+	SAY("                                           produces incorrect but still acceptable frames. Default: disabled.\n");
 	SAY("    -n|--persistent  ───────────────────── Don't re-initialize device on timeout. Default: disabled.\n");
 	SAY("    -t|--dv-timings  ───────────────────── Enable DV-timings querying and events processing");
 	SAY("                                           to automatic resolution change. Default: disabled.\n");
@@ -651,8 +656,7 @@ static void _help(FILE *fp, const us_capture_s *cap, const us_encoder_s *enc, co
 	SAY("                                             * CPU  ──────── Software MJPEG encoding (default);");
 	SAY("                                             * HW  ───────── Use pre-encoded MJPEG frames directly from camera hardware;");
 	SAY("                                             * M2M-VIDEO  ── GPU-accelerated MJPEG encoding using V4L2 M2M video interface;");
-	SAY("                                             * M2M-IMAGE  ── GPU-accelerated JPEG encoding using V4L2 M2M image interface;");
-	SAY("                                             * NOOP  ─────── Don't compress MJPEG stream (do nothing).\n");
+	SAY("                                             * M2M-IMAGE  ── GPU-accelerated JPEG encoding using V4L2 M2M image interface.\n");
 	SAY("    -g|--glitched-resolutions <WxH,...>  ─ It doesn't do anything. Still here for compatibility.\n");
 	SAY("    -k|--blank <path>  ─────────────────── It doesn't do anything. Still here for compatibility.\n");
 	SAY("    -K|--last-as-blank <sec>  ──────────── It doesn't do anything. Still here for compatibility.\n");

src/ustreamer/stream.c

@@ -85,6 +85,7 @@ static us_capture_hwbuf_s *_get_latest_hw(us_queue_s *queue);
 static bool _stream_has_jpeg_clients_cached(us_stream_s *stream);
 static bool _stream_has_any_clients_cached(us_stream_s *stream);
 static int _stream_init_loop(us_stream_s *stream);
+static void _stream_update_captured_fpsi(us_stream_s *stream, const us_frame_s *frame, bool bump);
 #ifdef WITH_V4P
 static void _stream_drm_ensure_no_signal(us_stream_s *stream);
 #endif
@@ -122,13 +123,17 @@ us_stream_s *us_stream_init(us_capture_s *cap, us_encoder_s *enc) {
 	stream->h264_gop = 30;
 	stream->run = run;
 
-	us_blank_draw(run->blank, "< NO SIGNAL >", cap->width, cap->height);
-	us_fpsi_meta_s meta = {0};
-	us_fpsi_frame_to_meta(run->blank->raw, &meta);
-	us_fpsi_update(http->captured_fpsi, false, &meta);
+	us_stream_update_blank(stream, cap); // Init blank
 	return stream;
 }
 
+void us_stream_update_blank(us_stream_s *stream, const us_capture_s *cap) {
+	us_stream_runtime_s *const run = stream->run;
+	us_blank_draw(run->blank, "< NO SIGNAL >", cap->width, cap->height);
+	us_fpsi_frame_to_meta(run->blank->raw, &run->notify_meta); // Initial "unchanged" meta
+	_stream_update_captured_fpsi(stream, run->blank->raw, false);
+}
+
 void us_stream_destroy(us_stream_s *stream) {
 	us_fpsi_destroy(stream->run->http->captured_fpsi);
 	US_RING_DELETE_WITH_ITEMS(stream->run->http->jpeg_ring, us_frame_destroy);
@@ -200,9 +205,7 @@ void us_stream_loop(us_stream_s *stream) {
 				default: goto close; // Any error
 			}
 
-			us_fpsi_meta_s meta = {0};
-			us_fpsi_frame_to_meta(&hw->raw, &meta);
-			us_fpsi_update(run->http->captured_fpsi, true, &meta);
+			_stream_update_captured_fpsi(stream, &hw->raw, true);
 
 #			ifdef WITH_GPIO
 			us_gpio_set_stream_online(true);
@@ -313,23 +316,23 @@ static void *_jpeg_thread(void *v_ctx) {
 	uint fluency_passed = 0;
 
 	while (!atomic_load(ctx->stop)) {
-		us_worker_s *const ready_wr = us_workers_pool_wait(stream->enc->run->pool);
-		us_encoder_job_s *const ready_job = ready_wr->job;
+		us_worker_s *const wr = us_workers_pool_wait(stream->enc->run->pool);
+		us_encoder_job_s *const job = wr->job;
 
-		if (ready_job->hw != NULL) {
-			us_capture_hwbuf_decref(ready_job->hw);
-			ready_job->hw = NULL;
-			if (ready_wr->job_failed) {
+		if (job->hw != NULL) {
+			us_capture_hwbuf_decref(job->hw);
+			job->hw = NULL;
+			if (wr->job_failed) {
 				// pass
-			} else if (ready_wr->job_timely) {
-				_stream_expose_jpeg(stream, ready_job->dest);
+			} else if (wr->job_timely) {
+				_stream_expose_jpeg(stream, job->dest);
 				if (atomic_load(&stream->run->http->snapshot_requested) > 0) { // Process real snapshots
 					atomic_fetch_sub(&stream->run->http->snapshot_requested, 1);
 				}
 				US_LOG_PERF("JPEG: ##### Encoded JPEG exposed; worker=%s, latency=%.3Lf",
-					ready_wr->name, us_get_now_monotonic() - ready_job->dest->grab_ts);
+					wr->name, us_get_now_monotonic() - job->dest->grab_ts);
 			} else {
-				US_LOG_PERF("JPEG: ----- Encoded JPEG dropped; worker=%s", ready_wr->name);
+				US_LOG_PERF("JPEG: ----- Encoded JPEG dropped; worker=%s", wr->name);
 			}
 		}
 
@@ -355,13 +358,13 @@ static void *_jpeg_thread(void *v_ctx) {
 		}
 		fluency_passed = 0;
 
-		const ldf fluency_delay = us_workers_pool_get_fluency_delay(stream->enc->run->pool, ready_wr);
+		const ldf fluency_delay = us_workers_pool_get_fluency_delay(stream->enc->run->pool, wr);
 		grab_after_ts = now_ts + fluency_delay;
 		US_LOG_VERBOSE("JPEG: Fluency: delay=%.03Lf, grab_after=%.03Lf", fluency_delay, grab_after_ts);
 
-		ready_job->hw = hw;
-		us_workers_pool_assign(stream->enc->run->pool, ready_wr);
-		US_LOG_DEBUG("JPEG: Assigned new frame in buffer=%d to worker=%s", hw->buf.index, ready_wr->name);
+		job->hw = hw;
+		us_workers_pool_assign(stream->enc->run->pool, wr);
+		US_LOG_DEBUG("JPEG: Assigned new frame in buffer=%d to worker=%s", hw->buf.index, wr->name);
 	}
 	return NULL;
 }
@@ -576,10 +579,7 @@ static int _stream_init_loop(us_stream_s *stream) {
 				}
 				us_blank_draw(run->blank, "< NO SIGNAL >", width, height);
 
-				us_fpsi_meta_s meta = {0};
-				us_fpsi_frame_to_meta(run->blank->raw, &meta);
-				us_fpsi_update(run->http->captured_fpsi, false, &meta);
-
+				_stream_update_captured_fpsi(stream, run->blank->raw, false);
 				_stream_expose_jpeg(stream, run->blank->jpeg);
 				_stream_expose_raw(stream, run->blank->raw);
 				_stream_encode_expose_h264(stream, run->blank->raw, true);
@@ -594,6 +594,19 @@ static int _stream_init_loop(us_stream_s *stream) {
 	return -1;
 }
 
+static void _stream_update_captured_fpsi(us_stream_s *stream, const us_frame_s *frame, bool bump) {
+	us_stream_runtime_s *const run = stream->run;
+
+	us_fpsi_meta_s meta = {0};
+	us_fpsi_frame_to_meta(frame, &meta);
+	us_fpsi_update(run->http->captured_fpsi, bump, &meta);
+
+	if (stream->notify_parent && memcmp(&run->notify_meta, &meta, sizeof(us_fpsi_meta_s))) {
+		memcpy(&run->notify_meta, &meta, sizeof(us_fpsi_meta_s));
+		us_process_notify_parent();
+	}
+}
+
 #ifdef WITH_V4P
 static void _stream_drm_ensure_no_signal(us_stream_s *stream) {
 	if (stream->drm == NULL) {

src/ustreamer/stream.h

@@ -68,6 +68,8 @@ typedef struct {
 
 	us_blank_s			*blank;
 
+	us_fpsi_meta_s		notify_meta;
+
 	atomic_bool			stop;
 } us_stream_runtime_s;
 
@@ -75,6 +77,7 @@ typedef struct {
 	us_capture_s	*cap;
 	us_encoder_s	*enc;
 
+	bool			notify_parent;
 	bool			slowdown;
 	uint			error_delay;
 	uint			exit_on_no_clients;
@@ -96,6 +99,7 @@ typedef struct {
 
 
 us_stream_s *us_stream_init(us_capture_s *cap, us_encoder_s *enc);
+void us_stream_update_blank(us_stream_s *stream, const us_capture_s *cap);
 void us_stream_destroy(us_stream_s *stream);
 
 void us_stream_loop(us_stream_s *stream);

src/ustreamer/workers.c

@@ -22,12 +22,22 @@
 
 #include "workers.h"
 
+#include <stdatomic.h>
+
+#include <pthread.h>
+
+#include "../libs/types.h"
+#include "../libs/tools.h"
+#include "../libs/threading.h"
+#include "../libs/logging.h"
+#include "../libs/list.h"
+
 
 static void *_worker_thread(void *v_worker);
 
 
 us_workers_pool_s *us_workers_pool_init(
-	const char *name, const char *wr_prefix, unsigned n_workers, long double desired_interval,
+	const char *name, const char *wr_prefix, uint n_workers, ldf desired_interval,
 	us_workers_pool_job_init_f job_init, void *job_init_arg,
 	us_workers_pool_job_destroy_f job_destroy,
 	us_workers_pool_run_job_f run_job) {
@@ -44,28 +54,28 @@ us_workers_pool_s *us_workers_pool_init(
 	atomic_init(&pool->stop, false);
 
 	pool->n_workers = n_workers;
-	US_CALLOC(pool->workers, pool->n_workers);
 
 	US_MUTEX_INIT(pool->free_workers_mutex);
 	US_COND_INIT(pool->free_workers_cond);
 
-	for (unsigned number = 0; number < pool->n_workers; ++number) {
-#		define WR(x_next) pool->workers[number].x_next
+	for (uint index = 0; index < pool->n_workers; ++index) {
+		us_worker_s *wr;
+		US_CALLOC(wr, 1);
 
-		WR(number) = number;
-		US_ASPRINTF(WR(name), "%s-%u", wr_prefix, number);
+		wr->number = index;
+		US_ASPRINTF(wr->name, "%s-%u", wr_prefix, index);
 
-		US_MUTEX_INIT(WR(has_job_mutex));
-		atomic_init(&WR(has_job), false);
-		US_COND_INIT(WR(has_job_cond));
+		US_MUTEX_INIT(wr->has_job_mutex);
+		atomic_init(&wr->has_job, false);
+		US_COND_INIT(wr->has_job_cond);
 
-		WR(pool) = pool;
-		WR(job) = job_init(job_init_arg);
+		wr->pool = pool;
+		wr->job = job_init(job_init_arg);
 
-		US_THREAD_CREATE(WR(tid), _worker_thread, (void*)&(pool->workers[number]));
+		US_THREAD_CREATE(wr->tid, _worker_thread, (void*)wr);
 		pool->free_workers += 1;
 
-#		undef WR
+		US_LIST_APPEND(pool->workers, wr);
 	}
 	return pool;
 }
@@ -74,98 +84,70 @@ void us_workers_pool_destroy(us_workers_pool_s *pool) {
 	US_LOG_INFO("Destroying workers pool %s ...", pool->name);
 
 	atomic_store(&pool->stop, true);
-	for (unsigned number = 0; number < pool->n_workers; ++number) {
-#		define WR(x_next) pool->workers[number].x_next
+	US_LIST_ITERATE(pool->workers, wr, { // cppcheck-suppress constStatement
+		US_MUTEX_LOCK(wr->has_job_mutex);
+		atomic_store(&wr->has_job, true); // Final job: die
+		US_MUTEX_UNLOCK(wr->has_job_mutex);
+		US_COND_SIGNAL(wr->has_job_cond);
 
-		US_MUTEX_LOCK(WR(has_job_mutex));
-		atomic_store(&WR(has_job), true); // Final job: die
-		US_MUTEX_UNLOCK(WR(has_job_mutex));
-		US_COND_SIGNAL(WR(has_job_cond));
+		US_THREAD_JOIN(wr->tid);
+		US_MUTEX_DESTROY(wr->has_job_mutex);
+		US_COND_DESTROY(wr->has_job_cond);
 
-		US_THREAD_JOIN(WR(tid));
-		US_MUTEX_DESTROY(WR(has_job_mutex));
-		US_COND_DESTROY(WR(has_job_cond));
+		pool->job_destroy(wr->job);
 
-		free(WR(name));
-
-		pool->job_destroy(WR(job));
-
-#		undef WR
-	}
+		free(wr->name);
+		free(wr);
+	});
 
 	US_MUTEX_DESTROY(pool->free_workers_mutex);
 	US_COND_DESTROY(pool->free_workers_cond);
 
-	free(pool->workers);
 	free(pool);
 }
 
 us_worker_s *us_workers_pool_wait(us_workers_pool_s *pool) {
-	us_worker_s *ready_wr = NULL;
-
 	US_MUTEX_LOCK(pool->free_workers_mutex);
 	US_COND_WAIT_FOR(pool->free_workers, pool->free_workers_cond, pool->free_workers_mutex);
 	US_MUTEX_UNLOCK(pool->free_workers_mutex);
 
-	if (pool->oldest_wr && !atomic_load(&pool->oldest_wr->has_job)) {
-		ready_wr = pool->oldest_wr;
-		ready_wr->job_timely = true;
-		pool->oldest_wr = pool->oldest_wr->next_wr;
-	} else {
-		for (unsigned number = 0; number < pool->n_workers; ++number) {
-			if (
-				!atomic_load(&pool->workers[number].has_job) && (
-					ready_wr == NULL
-					|| ready_wr->job_start_ts < pool->workers[number].job_start_ts
-				)
-			) {
-				ready_wr = &pool->workers[number];
-				break;
-			}
+	us_worker_s *found = NULL;
+	US_LIST_ITERATE(pool->workers, wr, { // cppcheck-suppress constStatement
+		if (!atomic_load(&wr->has_job) && (found == NULL || found->job_start_ts <= wr->job_start_ts)) {
+			found = wr;
 		}
-		assert(ready_wr != NULL);
-		ready_wr->job_timely = false; // Освободился воркер, получивший задание позже (или самый первый при самом первом захвате)
+	});
+	assert(found != NULL);
+	US_LIST_REMOVE(pool->workers, found);
+	US_LIST_APPEND(pool->workers, found); // Перемещаем в конец списка
+
+	found->job_timely = (found->job_start_ts > pool->job_timely_ts);
+	if (found->job_timely) {
+		pool->job_timely_ts = found->job_start_ts;
 	}
-	return ready_wr;
+	return found;
 }
 
-void us_workers_pool_assign(us_workers_pool_s *pool, us_worker_s *ready_wr/*, void *job*/) {
-	if (pool->oldest_wr == NULL) {
-		pool->oldest_wr = ready_wr;
-		pool->latest_wr = pool->oldest_wr;
-	} else {
-		if (ready_wr->next_wr != NULL) {
-			ready_wr->next_wr->prev_wr = ready_wr->prev_wr;
-		}
-		if (ready_wr->prev_wr != NULL) {
-			ready_wr->prev_wr->next_wr = ready_wr->next_wr;
-		}
-		ready_wr->prev_wr = pool->latest_wr;
-		pool->latest_wr->next_wr = ready_wr;
-		pool->latest_wr = ready_wr;
-	}
-	pool->latest_wr->next_wr = NULL;
-
-	US_MUTEX_LOCK(ready_wr->has_job_mutex);
-	//ready_wr->job = job;
-	atomic_store(&ready_wr->has_job, true);
-	US_MUTEX_UNLOCK(ready_wr->has_job_mutex);
-	US_COND_SIGNAL(ready_wr->has_job_cond);
+void us_workers_pool_assign(us_workers_pool_s *pool, us_worker_s *wr) {
+	US_MUTEX_LOCK(wr->has_job_mutex);
+	atomic_store(&wr->has_job, true);
+	US_MUTEX_UNLOCK(wr->has_job_mutex);
+	US_COND_SIGNAL(wr->has_job_cond);
 
 	US_MUTEX_LOCK(pool->free_workers_mutex);
 	pool->free_workers -= 1;
 	US_MUTEX_UNLOCK(pool->free_workers_mutex);
 }
 
-long double us_workers_pool_get_fluency_delay(us_workers_pool_s *pool, const us_worker_s *ready_wr) {
-	const long double approx_job_time = pool->approx_job_time * 0.9 + ready_wr->last_job_time * 0.1;
+ldf us_workers_pool_get_fluency_delay(us_workers_pool_s *pool, const us_worker_s *wr) {
+	const ldf approx_job_time = pool->approx_job_time * 0.9 + wr->last_job_time * 0.1;
 
 	US_LOG_VERBOSE("Correcting pool's %s approx_job_time: %.3Lf -> %.3Lf (last_job_time=%.3Lf)",
-		pool->name, pool->approx_job_time, approx_job_time, ready_wr->last_job_time);
+		pool->name, pool->approx_job_time, approx_job_time, wr->last_job_time);
 
 	pool->approx_job_time = approx_job_time;
 
-	const long double min_delay = pool->approx_job_time / pool->n_workers; // Среднее время работы размазывается на N воркеров
+	const ldf min_delay = pool->approx_job_time / pool->n_workers; // Среднее время работы размазывается на N воркеров
 
 	if (pool->desired_interval > 0 && min_delay > 0 && pool->desired_interval > min_delay) {
 		// Искусственное время задержки на основе желаемого FPS, если включен --desired-fps
@@ -176,7 +158,7 @@ long double us_workers_pool_get_fluency_delay(us_workers_pool_s *pool, const us_
 }
 
 static void *_worker_thread(void *v_worker) {
-	us_worker_s *wr = v_worker;
+	us_worker_s *const wr = v_worker;
 
 	US_THREAD_SETTLE("%s", wr->name);
 	US_LOG_DEBUG("Hello! I am a worker %s ^_^", wr->name);
@@ -189,13 +171,12 @@ static void *_worker_thread(void *v_worker) {
 		US_MUTEX_UNLOCK(wr->has_job_mutex);
 
 		if (!atomic_load(&wr->pool->stop)) {
-			const long double job_start_ts = us_get_now_monotonic();
+			const ldf job_start_ts = us_get_now_monotonic();
 			wr->job_failed = !wr->pool->run_job(wr);
 			if (!wr->job_failed) {
 				wr->job_start_ts = job_start_ts;
 				wr->last_job_time = us_get_now_monotonic() - wr->job_start_ts;
 			}
-			//wr->job = NULL;
 			atomic_store(&wr->has_job, false);
 		}
 

src/ustreamer/workers.h

@@ -22,37 +22,32 @@
 
 #pragma once
 
-#include <stdbool.h>
 #include <stdatomic.h>
 
-#include <sys/types.h>
-
 #include <pthread.h>
 
-#include "../libs/tools.h"
-#include "../libs/threading.h"
-#include "../libs/logging.h"
+#include "../libs/types.h"
+#include "../libs/list.h"
 
 
 typedef struct us_worker_sx {
-	pthread_t		tid;
-	unsigned		number;
-	char			*name;
+	pthread_t	tid;
+	uint		number;
+	char		*name;
 
-	long double		last_job_time;
+	ldf			last_job_time;
 
 	pthread_mutex_t	has_job_mutex;
 	void			*job;
 	atomic_bool		has_job;
 	bool			job_timely;
 	bool			job_failed;
-	long double		job_start_ts;
+	ldf				job_start_ts;
 	pthread_cond_t	has_job_cond;
 
-	struct us_worker_sx		*prev_wr;
-	struct us_worker_sx		*next_wr;
-
 	struct us_workers_pool_sx	*pool;
+
+	US_LIST_DECLARE;
 } us_worker_s;
 
 typedef void *(*us_workers_pool_job_init_f)(void *arg);
@@ -61,20 +56,19 @@ typedef bool (*us_workers_pool_run_job_f)(us_worker_s *wr);
 
 typedef struct us_workers_pool_sx {
 	const char		*name;
-	long double		desired_interval;
+	ldf				desired_interval;
 
 	us_workers_pool_job_destroy_f	job_destroy;
 	us_workers_pool_run_job_f		run_job;
 
-	unsigned		n_workers;
+	uint			n_workers;
 	us_worker_s		*workers;
-	us_worker_s		*oldest_wr;
-	us_worker_s		*latest_wr;
+	ldf				job_timely_ts;
 
-	long double		approx_job_time;
+	ldf				approx_job_time;
 
 	pthread_mutex_t	free_workers_mutex;
-	unsigned		free_workers;
+	uint			free_workers;
 	pthread_cond_t	free_workers_cond;
 
 	atomic_bool		stop;
@@ -82,7 +76,7 @@ typedef struct us_workers_pool_sx {
 
 
 us_workers_pool_s *us_workers_pool_init(
-	const char *name, const char *wr_prefix, unsigned n_workers, long double desired_interval,
+	const char *name, const char *wr_prefix, uint n_workers, ldf desired_interval,
 	us_workers_pool_job_init_f job_init, void *job_init_arg,
 	us_workers_pool_job_destroy_f job_destroy,
 	us_workers_pool_run_job_f run_job);
@@ -90,6 +84,6 @@ us_workers_pool_s *us_workers_pool_init(
 void us_workers_pool_destroy(us_workers_pool_s *pool);
 
 us_worker_s *us_workers_pool_wait(us_workers_pool_s *pool);
-void us_workers_pool_assign(us_workers_pool_s *pool, us_worker_s *ready_wr/*, void *job*/);
+void us_workers_pool_assign(us_workers_pool_s *pool, us_worker_s *ready_wr);
 
-long double us_workers_pool_get_fluency_delay(us_workers_pool_s *pool, const us_worker_s *ready_wr);
+ldf us_workers_pool_get_fluency_delay(us_workers_pool_s *pool, const us_worker_s *ready_wr);