Bump version: 6.12 → 6.13

The correct signature is:
int thr_self(long *id);

It was called as thr_self() which caused memory corruption.

.bumpversion.cfg

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

Makefile

@@ -16,6 +16,12 @@ export
 _LINTERS_IMAGE ?= ustreamer-linters
 
 
+# =====
+ifeq (, $(shell which pkg-config))
+	$(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

@@ -32,6 +32,7 @@
 #include <opus/opus.h>
 
 #include "uslibs/types.h"
+#include "uslibs/errors.h"
 #include "uslibs/tools.h"
 #include "uslibs/array.h"
 #include "uslibs/ring.h"
@@ -185,12 +186,12 @@ int us_audio_get_encoded(us_audio_s *audio, u8 *data, uz *size, u64 *pts) {
 	}
 	const int ri = us_ring_consumer_acquire(audio->enc_ring, 0.1);
 	if (ri < 0) {
-		return -2;
+		return US_ERROR_NO_DATA;
 	}
 	const _enc_buffer_s *const buf = audio->enc_ring->items[ri];
 	if (*size < buf->used) {
 		us_ring_consumer_release(audio->enc_ring, ri);
-		return -3;
+		return US_ERROR_NO_DATA;
 	}
 	memcpy(data, buf->data, buf->used);
 	*size = buf->used;

janus/src/client.h

@@ -34,7 +34,7 @@
 #include "rtp.h"
 
 
-typedef struct us_janus_client_sx {
+typedef struct {
 	janus_callbacks			*gw;
 	janus_plugin_session	*session;
 	atomic_bool				transmit;
@@ -48,7 +48,7 @@ typedef struct us_janus_client_sx {
 	us_ring_s				*video_ring;
 	us_ring_s				*audio_ring;
 
-    US_LIST_STRUCT(struct us_janus_client_sx);
+    US_LIST_DECLARE;
 } us_janus_client_s;
 
 

janus/src/memsinkfd.c

@@ -27,6 +27,7 @@
 #include <linux/videodev2.h>
 
 #include "uslibs/types.h"
+#include "uslibs/errors.h"
 #include "uslibs/tools.h"
 #include "uslibs/frame.h"
 #include "uslibs/memsinksh.h"
@@ -54,7 +55,7 @@ int us_memsink_fd_wait_frame(int fd, us_memsink_shared_s *mem, u64 last_id) {
 		}
 		usleep(1000); // lock_polling
 	} while (now_ts < deadline_ts);
-	return -2;
+	return US_ERROR_NO_DATA;
 }
 
 int us_memsink_fd_get_frame(int fd, us_memsink_shared_s *mem, us_frame_s *frame, u64 *frame_id, bool key_required) {

janus/src/plugin.c

@@ -37,6 +37,7 @@
 
 #include "uslibs/types.h"
 #include "uslibs/const.h"
+#include "uslibs/errors.h"
 #include "uslibs/tools.h"
 #include "uslibs/threading.h"
 #include "uslibs/list.h"
@@ -178,7 +179,7 @@ static void *_video_sink_thread(void *arg) {
 				if (ri >= 0 && frame->key) {
 					atomic_store(&_g_key_required, false);
 				}
-			} else if (waited != -2) {
+			} else if (waited != US_ERROR_NO_DATA) {
 				goto close_memsink;
 			}
 		}

janus/src/uslibs/errors.h

@@ -0,0 +1 @@
+../../../src/libs/errors.h

linters/tox.ini

@@ -3,7 +3,7 @@ envlist = cppcheck, flake8, pylint, mypy, vulture, htmlhint
 skipsdist = true
 
 [testenv]
-basepython = python3.11
+basepython = python3.12
 changedir = /src
 
 [testenv:cppcheck]
@@ -13,6 +13,7 @@ commands = cppcheck \
 			--std=c17 \
 			--error-exitcode=1 \
 			--quiet \
+			--check-level=exhaustive \
 			--enable=warning,portability,performance,style \
 			--suppress=assignmentInAssert \
 			--suppress=variableScope \
@@ -25,7 +26,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 +34,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.9" "January 2021"
+.TH USTREAMER-DUMP 1 "version 6.13" "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.9" "November 2020"
+.TH USTREAMER 1 "version 6.13" "November 2020"
 
 .SH NAME
 ustreamer \- stream MJPEG video from any V4L2 device to the network
@@ -96,8 +96,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,7 +3,7 @@
 
 
 pkgname=ustreamer
-pkgver=6.9
+pkgver=6.13
 pkgrel=1
 pkgdesc="Lightweight and fast MJPEG-HTTP streamer"
 url="https://github.com/pikvm/ustreamer"

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.9
+PKG_VERSION:=6.13
 PKG_RELEASE:=1
 PKG_MAINTAINER:=Maxim Devaev <mdevaev@gmail.com>
 

python/Makefile

@@ -7,7 +7,8 @@ PY ?= python3
 
 
 # =====
-all:
+all: root
+root: $(shell find src -type f,l)
 	$(info == PY_BUILD ustreamer-*.so)
 	rm -rf root
 	$(ECHO) $(PY) -m build --skip-dependency-check --no-isolation

python/setup.py

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

python/src/uslibs/errors.h

@@ -0,0 +1 @@
+../../../src/libs/errors.h

python/src/ustreamer.c

@@ -14,6 +14,7 @@
 #include <Python.h>
 
 #include "uslibs/types.h"
+#include "uslibs/errors.h"
 #include "uslibs/tools.h"
 #include "uslibs/frame.h"
 #include "uslibs/memsinksh.h"
@@ -175,9 +176,9 @@ static int _wait_frame(_MemsinkObject *self) {
 		if (PyErr_CheckSignals() < 0) {
 			return -1;
 		}
-
 	} while (now_ts < deadline_ts);
-	return -2;
+
+	return US_ERROR_NO_DATA;
 }
 
 static PyObject *_MemsinkObject_wait_frame(_MemsinkObject *self, PyObject *args, PyObject *kwargs) {
@@ -194,7 +195,7 @@ static PyObject *_MemsinkObject_wait_frame(_MemsinkObject *self, PyObject *args,
 
 	switch (_wait_frame(self)) {
 		case 0: break;
-		case -2: Py_RETURN_NONE;
+		case US_ERROR_NO_DATA: Py_RETURN_NONE;
 		default: return NULL;
 	}
 

src/dump/main.c

@@ -31,10 +31,12 @@
 #include <assert.h>
 
 #include "../libs/const.h"
+#include "../libs/errors.h"
 #include "../libs/tools.h"
 #include "../libs/logging.h"
 #include "../libs/frame.h"
 #include "../libs/memsink.h"
+#include "../libs/fpsi.h"
 #include "../libs/signal.h"
 #include "../libs/options.h"
 
@@ -109,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;
@@ -220,26 +222,22 @@ static int _dump_sink(
 	const useconds_t interval_us = interval * 1000000;
 
 	us_frame_s *frame = us_frame_init();
+	us_fpsi_s *fpsi = us_fpsi_init("SINK", false);
 	us_memsink_s *sink = NULL;
 
-	if ((sink = us_memsink_init("input", sink_name, false, 0, false, 0, sink_timeout)) == NULL) {
+	if ((sink = us_memsink_init_opened("input", sink_name, false, 0, false, 0, sink_timeout)) == NULL) {
 		goto error;
 	}
 
-	unsigned fps = 0;
-	unsigned fps_accum = 0;
-	long long fps_second = 0;
-
 	long double last_ts = 0;
 
 	while (!_g_stop) {
 		bool key_requested;
-		const int error = us_memsink_client_get(sink, frame, &key_requested, key_required);
+		const int got = us_memsink_client_get(sink, frame, &key_requested, key_required);
-		if (error == 0) {
+		if (got == 0) {
 			key_required = false;
 
 			const long double now = us_get_now_monotonic();
-			const long long now_second = us_floor_ms(now);
 
 			char fourcc_str[8];
 			US_LOG_VERBOSE("Frame: %s - %ux%u -- online=%d, key=%d, kr=%d, gop=%u, latency=%.3Lf, backlog=%.3Lf, size=%zu",
@@ -253,13 +251,7 @@ static int _dump_sink(
 			US_LOG_DEBUG("       stride=%u, grab_ts=%.3Lf, encode_begin_ts=%.3Lf, encode_end_ts=%.3Lf",
 				frame->stride, frame->grab_ts, frame->encode_begin_ts, frame->encode_end_ts);
 
-			if (now_second != fps_second) {
+			us_fpsi_update(fpsi, true, NULL);
-				fps = fps_accum;
-				fps_accum = 0;
-				fps_second = now_second;
-				US_LOG_PERF_FPS("A new second has come; captured_fps=%u", fps);
-			}
-			fps_accum += 1;
 
 			if (ctx->v_output != NULL) {
 				ctx->write(ctx->v_output, frame);
@@ -275,7 +267,7 @@ static int _dump_sink(
 			if (interval_us > 0) {
 				usleep(interval_us);
 			}
-		} else if (error == -2) {
+		} else if (got == US_ERROR_NO_DATA) {
 			usleep(1000);
 		} else {
 			goto error;
@@ -286,6 +278,7 @@ static int _dump_sink(
 
 error:
 	US_DELETE(sink, us_memsink_destroy);
+	us_fpsi_destroy(fpsi);
 	us_frame_destroy(frame);
 	US_LOG_INFO("Bye-bye");
 	return retval;

src/libs/capture.c

@@ -41,6 +41,7 @@
 #include <linux/v4l2-controls.h>
 
 #include "types.h"
+#include "errors.h"
 #include "tools.h"
 #include "array.h"
 #include "logging.h"
@@ -82,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);
@@ -97,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);
@@ -175,16 +176,15 @@ int us_capture_open(us_capture_s *cap) {
 	us_capture_runtime_s *const run = cap->run;
 
 	if (access(cap->path, R_OK | W_OK) < 0) {
-		if (run->open_error_reported != -errno) {
+		US_ONCE_FOR(run->open_error_once, -errno, {
-			run->open_error_reported = -errno; // Don't confuse it with __LINE__
 			US_LOG_PERROR("No access to capture device");
-		}
+		});
-		goto tmp_error;
+		goto error_no_device;
 	}
 
 	_LOG_DEBUG("Opening capture device ...");
 	if ((run->fd = open(cap->path, O_RDWR | O_NONBLOCK)) < 0) {
-		_LOG_PERROR("Can't capture open device");
+		_LOG_PERROR("Can't open capture device");
 		goto error;
 	}
 	_LOG_DEBUG("Capture device fd=%d opened", run->fd);
@@ -192,15 +192,15 @@ int us_capture_open(us_capture_s *cap) {
 	if (cap->dv_timings && cap->persistent) {
 		_LOG_DEBUG("Probing DV-timings or QuerySTD ...");
 		if (_capture_open_dv_timings(cap, false) < 0) {
-			const int line = __LINE__;
+			US_ONCE_FOR(run->open_error_once, __LINE__, {
-			if (run->open_error_reported != line) {
-				run->open_error_reported = line;
 				_LOG_ERROR("No signal from source");
-			}
+			});
-			goto tmp_error;
+			goto error_no_signal;
 		}
 	}
 
+	US_LOG_INFO("Using V4L2 device: %s", cap->path);
+
 	if (_capture_open_check_cap(cap) < 0) {
 		goto error;
 	}
@@ -237,16 +237,20 @@ int us_capture_open(us_capture_s *cap) {
 	}
 	run->streamon = true;
 
-	run->open_error_reported = 0;
+	run->open_error_once = 0;
 	_LOG_INFO("Capturing started");
 	return 0;
 
-tmp_error:
+error_no_device:
 	us_capture_close(cap);
-	return -2;
+	return US_ERROR_NO_DEVICE;
+
+error_no_signal:
+	us_capture_close(cap);
+	return US_ERROR_NO_DATA;
 
 error:
-	run->open_error_reported = 0;
+	run->open_error_once = 0;
 	us_capture_close(cap);
 	return -1;
 }
@@ -305,7 +309,7 @@ int us_capture_hwbuf_grab(us_capture_s *cap, us_capture_hwbuf_s **hw) {
 	//     или эвент V4L2. Обработка эвентов более приоритетна, чем кадров.
 	//   - Если есть новые фреймы, то пропустить их все, пока не закончатся и вернуть
 	//     самый-самый свежий, содержащий при этом валидные данные.
-	//   - Если таковых не нашлось, вернуть -2.
+	//   - Если таковых не нашлось, вернуть US_ERROR_NO_DATA.
 	//   - Ошибка -1 возвращается при любых сбоях.
 
 	if (_capture_wait_buffer(cap) < 0) {
@@ -392,7 +396,7 @@ int us_capture_hwbuf_grab(us_capture_s *cap, us_capture_hwbuf_s **hw) {
 				if (buf_got) {
 					break; // Process any latest valid frame
 				} else if (broken) {
-					return -2; // If we have only broken frames on this capture session
+					return US_ERROR_NO_DATA; // If we have only broken frames on this capture session
 				}
 			}
 			_LOG_PERROR("Can't grab HW buffer");
@@ -417,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);
@@ -482,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");
@@ -509,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.
@@ -733,7 +737,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 {
@@ -1033,7 +1037,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) { \
@@ -1090,7 +1094,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
@@ -1107,7 +1111,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

@@ -67,7 +67,7 @@ typedef struct {
 	enum v4l2_buf_type	capture_type;
 	bool				capture_mplane;
 	bool				streamon;
-	int					open_error_reported;
+	int					open_error_once;
 } us_capture_runtime_s;
 
 typedef enum {
@@ -133,7 +133,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 9
+#define US_VERSION_MINOR 13
 
 #define US_MAKE_VERSION2(_major, _minor) #_major "." #_minor
 #define US_MAKE_VERSION1(_major, _minor) US_MAKE_VERSION2(_major, _minor)

src/libs/drm/drm.c

@@ -38,6 +38,7 @@
 #include <libdrm/drm.h>
 
 #include "../types.h"
+#include "../errors.h"
 #include "../tools.h"
 #include "../logging.h"
 #include "../frame.h"
@@ -71,6 +72,7 @@ us_drm_s *us_drm_init(void) {
 	run->fd = -1;
 	run->status_fd = -1;
 	run->dpms_state = -1;
+	run->opened = -1;
 	run->has_vsync = true;
 	run->exposing_dma_fd = -1;
 	run->ft = us_frametext_init();
@@ -81,6 +83,7 @@ us_drm_s *us_drm_init(void) {
 	drm->path = "/dev/dri/by-path/platform-gpu-card";
 	drm->port = "HDMI-A-2"; // OUT2 on PiKVM V4 Plus
 	drm->timeout = 5;
+	drm->blank_after = 5;
 	drm->run = run;
 	return drm;
 }
@@ -98,10 +101,11 @@ int us_drm_open(us_drm_s *drm, const us_capture_s *cap) {
 
 	switch (_drm_check_status(drm)) {
 		case 0: break;
-		case -2: goto unplugged;
+		case US_ERROR_NO_DEVICE: goto unplugged;
 		default: goto error;
 	}
 
+	_LOG_INFO("Using passthrough: %s[%s]", drm->path, drm->port);
 	_LOG_INFO("Configuring DRM device for %s ...", (cap == NULL ? "STUB" : "DMA"));
 
 	if ((run->fd = open(drm->path, O_RDWR | O_CLOEXEC | O_NONBLOCK)) < 0) {
@@ -143,7 +147,7 @@ int us_drm_open(us_drm_s *drm, const us_capture_s *cap) {
 	const uint hz = (stub > 0 ? 0 : cap->run->hz);
 	switch (_drm_find_sink(drm, width, height, hz)) {
 		case 0: break;
-		case -2: goto unplugged;
+		case US_ERROR_NO_DEVICE: goto unplugged;
 		default: goto error;
 	}
 	if ((stub == 0) && (width != run->mode.hdisplay || height < run->mode.vdisplay)) {
@@ -163,23 +167,24 @@ int us_drm_open(us_drm_s *drm, const us_capture_s *cap) {
 		goto error;
 	}
 
-	run->opened_for_stub = (stub > 0);
+	_LOG_INFO("Opened for %s ...", (stub > 0 ? "STUB" : "DMA"));
 	run->exposing_dma_fd = -1;
-	run->unplugged_reported = false;
+	run->blank_at_ts = 0;
-	_LOG_INFO("Opened for %s ...", (run->opened_for_stub ? "STUB" : "DMA"));
+	run->opened = stub;
-	return stub;
+	run->once = 0;
+	return run->opened;
 
 error:
 	us_drm_close(drm);
-	return -1;
+	return run->opened; // -1 after us_drm_close()
 
 unplugged:
-	if (!run->unplugged_reported) {
+	US_ONCE_FOR(run->once, __LINE__, {
 		_LOG_ERROR("Display is not plugged");
-		run->unplugged_reported = true;
+	});
-	}
 	us_drm_close(drm);
-	return -2;
+	run->opened = US_ERROR_NO_DEVICE;
+	return run->opened;
 }
 
 void us_drm_close(us_drm_s *drm) {
@@ -233,6 +238,7 @@ void us_drm_close(us_drm_s *drm) {
 
 	run->crtc_id = 0;
 	run->dpms_state = -1;
+	run->opened = -1;
 	run->has_vsync = true;
 	run->stub_n_buf = 0;
 
@@ -241,11 +247,39 @@ void us_drm_close(us_drm_s *drm) {
 	}
 }
 
+int us_drm_ensure_no_signal(us_drm_s *drm) {
+	us_drm_runtime_s *const run = drm->run;
+
+	assert(run->fd >= 0);
+	assert(run->opened > 0);
+
+	const ldf now_ts = us_get_now_monotonic();
+	if (run->blank_at_ts == 0) {
+		run->blank_at_ts = now_ts + drm->blank_after;
+	}
+	const ldf saved_ts = run->blank_at_ts; // us_drm*() rewrites it to 0
+
+	int retval;
+	if (now_ts <= run->blank_at_ts) {
+		retval = us_drm_wait_for_vsync(drm);
+		if (retval == 0) {
+			retval = us_drm_expose_stub(drm, US_DRM_STUB_NO_SIGNAL, NULL);
+		}
+	} else {
+		US_ONCE_FOR(run->once, __LINE__, {
+			_LOG_INFO("Turning off the display by timeout ...");
+		});
+		retval = us_drm_dpms_power_off(drm);
+	}
+	run->blank_at_ts = saved_ts;
+	return retval;
+}
+
 int us_drm_dpms_power_off(us_drm_s *drm) {
 	assert(drm->run->fd >= 0);
 	switch (_drm_check_status(drm)) {
 		case 0: break;
-		case -2: return 0; // Unplugged, nice
+		case US_ERROR_NO_DEVICE: return 0; // Unplugged, nice
 		// Во время переключения DPMS монитор моргает один раз состоянием disconnected,
 		// а потом почему-то снова оказывается connected. Так что просто считаем,
 		// что отсоединенный монитор на этом этапе - это нормально.
@@ -259,10 +293,11 @@ int us_drm_wait_for_vsync(us_drm_s *drm) {
 	us_drm_runtime_s *const run = drm->run;
 
 	assert(run->fd >= 0);
+	run->blank_at_ts = 0;
 
 	switch (_drm_check_status(drm)) {
 		case 0: break;
-		case -2: return -2;
+		case US_ERROR_NO_DEVICE: return US_ERROR_NO_DEVICE;
 		default: return -1;
 	}
 	_drm_ensure_dpms_power(drm, true);
@@ -313,11 +348,12 @@ int us_drm_expose_stub(us_drm_s *drm, us_drm_stub_e stub, const us_capture_s *ca
 	us_drm_runtime_s *const run = drm->run;
 
 	assert(run->fd >= 0);
-	assert(run->opened_for_stub);
+	assert(run->opened > 0);
+	run->blank_at_ts = 0;
 
 	switch (_drm_check_status(drm)) {
 		case 0: break;
-		case -2: return -2;
+		case US_ERROR_NO_DEVICE: return US_ERROR_NO_DEVICE;
 		default: return -1;
 	}
 	_drm_ensure_dpms_power(drm, true);
@@ -377,11 +413,12 @@ int us_drm_expose_dma(us_drm_s *drm, const us_capture_hwbuf_s *hw) {
 	us_drm_buffer_s *const buf = &run->bufs[hw->buf.index];
 
 	assert(run->fd >= 0);
-	assert(!run->opened_for_stub);
+	assert(run->opened == 0);
+	run->blank_at_ts = 0;
 
 	switch (_drm_check_status(drm)) {
 		case 0: break;
-		case -2: return -2;
+		case US_ERROR_NO_DEVICE: return US_ERROR_NO_DEVICE;
 		default: return -1;
 	}
 	_drm_ensure_dpms_power(drm, true);
@@ -434,7 +471,7 @@ static int _drm_check_status(us_drm_s *drm) {
 		goto error;
 	}
 	_LOG_DEBUG("Current display status: %c", status_ch);
-	return (status_ch == 'd' ? -2 : 0);
+	return (status_ch == 'd' ? US_ERROR_NO_DEVICE : 0);
 
 error:
 	US_CLOSE_FD(run->status_fd);
@@ -577,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);
@@ -613,7 +650,7 @@ done:
 
 unplugged:
 	drmModeFreeResources(res);
-	return -2;
+	return US_ERROR_NO_DEVICE;
 }
 
 static drmModeModeInfo *_find_best_mode(drmModeConnector *conn, uint width, uint height, float hz) {

src/libs/drm/drm.h

@@ -63,11 +63,14 @@ typedef struct {
 	uint			n_bufs;
 	drmModeCrtc		*saved_crtc;
 	int				dpms_state;
-	bool			opened_for_stub;
+	int				opened;
+
 	bool			has_vsync;
 	int				exposing_dma_fd;
 	uint			stub_n_buf;
-	bool			unplugged_reported;
+	ldf				blank_at_ts;
+
+	int				once;
 	us_frametext_s	*ft;
 } us_drm_runtime_s;
 
@@ -75,6 +78,7 @@ typedef struct {
 	char	*path;
 	char	*port;
 	uint	timeout;
+	uint	blank_after;
 
 	us_drm_runtime_s *run;
 } us_drm_s;
@@ -90,3 +94,4 @@ int us_drm_dpms_power_off(us_drm_s *drm);
 int us_drm_wait_for_vsync(us_drm_s *drm);
 int us_drm_expose_stub(us_drm_s *drm, us_drm_stub_e stub, const us_capture_s *cap);
 int us_drm_expose_dma(us_drm_s *drm, const us_capture_hwbuf_s *hw);
+int us_drm_ensure_no_signal(us_drm_s *drm);

src/libs/errors.h

@@ -20,59 +20,8 @@
 *****************************************************************************/
 
 
-#include "h264.h"
+#pragma once
 
-#include <stdatomic.h>
+#define US_ERROR_COMMON		-1
-
+#define US_ERROR_NO_DEVICE	-2
-#include "../libs/types.h"
+#define US_ERROR_NO_DATA	-3
-#include "../libs/tools.h"
-#include "../libs/logging.h"
-#include "../libs/frame.h"
-#include "../libs/memsink.h"
-#include "../libs/unjpeg.h"
-
-#include "m2m.h"
-
-
-us_h264_stream_s *us_h264_stream_init(us_memsink_s *sink, const char *path, uint bitrate, uint gop) {
-	us_h264_stream_s *h264;
-	US_CALLOC(h264, 1);
-	h264->sink = sink;
-	h264->tmp_src = us_frame_init();
-	h264->dest = us_frame_init();
-	atomic_init(&h264->online, false);
-	h264->enc = us_m2m_h264_encoder_init("H264", path, bitrate, gop);
-	return h264;
-}
-
-void us_h264_stream_destroy(us_h264_stream_s *h264) {
-	us_m2m_encoder_destroy(h264->enc);
-	us_frame_destroy(h264->dest);
-	us_frame_destroy(h264->tmp_src);
-	free(h264);
-}
-
-void us_h264_stream_process(us_h264_stream_s *h264, const us_frame_s *frame, bool force_key) {
-	if (us_is_jpeg(frame->format)) {
-		const ldf now_ts = us_get_now_monotonic();
-		US_LOG_DEBUG("H264: Input frame is JPEG; decoding ...");
-		if (us_unjpeg(frame, h264->tmp_src, true) < 0) {
-			atomic_store(&h264->online, false);
-			return;
-		}
-		frame = h264->tmp_src;
-		US_LOG_VERBOSE("H264: JPEG decoded; time=%.3Lf", us_get_now_monotonic() - now_ts);
-	}
-
-	if (h264->key_requested) {
-		US_LOG_INFO("H264: Requested keyframe by a sink client");
-		h264->key_requested = false;
-		force_key = true;
-	}
-
-	bool online = false;
-	if (!us_m2m_encoder_compress(h264->enc, frame, h264->dest, force_key)) {
-		online = !us_memsink_server_put(h264->sink, h264->dest, &h264->key_requested);
-	}
-	atomic_store(&h264->online, online);
-}

src/libs/fpsi.c

@@ -0,0 +1,112 @@
+/*****************************************************************************
+#                                                                            #
+#    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 "fpsi.h"
+
+#include <stdatomic.h>
+
+#include <pthread.h>
+
+#include "types.h"
+#include "tools.h"
+#include "threading.h"
+#include "logging.h"
+#include "frame.h"
+
+
+us_fpsi_s *us_fpsi_init(const char *name, bool with_meta) {
+	us_fpsi_s *fpsi;
+	US_CALLOC(fpsi, 1);
+	fpsi->name = us_strdup(name);
+	fpsi->with_meta = with_meta;
+	atomic_init(&fpsi->state_sec_ts, 0);
+	atomic_init(&fpsi->state, 0);
+	return fpsi;
+}
+
+void us_fpsi_destroy(us_fpsi_s *fpsi) {
+	free(fpsi->name);
+	free(fpsi);
+}
+
+void us_fpsi_frame_to_meta(const us_frame_s *frame, us_fpsi_meta_s *meta) {
+	meta->width = frame->width;
+	meta->height = frame->height;
+	meta->online = frame->online;
+}
+
+void us_fpsi_update(us_fpsi_s *fpsi, bool bump, const us_fpsi_meta_s *meta) {
+	if (meta != NULL) {
+		assert(fpsi->with_meta);
+	} else {
+		assert(!fpsi->with_meta);
+	}
+
+	const sll now_sec_ts = us_floor_ms(us_get_now_monotonic());
+	if (atomic_load(&fpsi->state_sec_ts) != now_sec_ts) {
+		US_LOG_PERF_FPS("FPS: %s: %u", fpsi->name, fpsi->accum);
+
+		// Fast mutex-less store method
+		ull state = (ull)fpsi->accum & 0xFFFF;
+		if (fpsi->with_meta) {
+			assert(meta != NULL);
+			state |= (ull)(meta->width & 0xFFFF) << 16;
+			state |= (ull)(meta->height & 0xFFFF) << 32;
+			state |= (ull)(meta->online ? 1 : 0) << 48;
+		}
+		atomic_store(&fpsi->state, state); // Сначала инфа
+		atomic_store(&fpsi->state_sec_ts, now_sec_ts); // Потом время, это важно
+		fpsi->accum = 0;
+	}
+	if (bump) {
+		++fpsi->accum;
+	}
+}
+
+uint us_fpsi_get(us_fpsi_s *fpsi, us_fpsi_meta_s *meta) {
+	if (meta != NULL) {
+		assert(fpsi->with_meta);
+	} else {
+		assert(!fpsi->with_meta);
+	}
+
+	// Между чтением инфы и времени может быть гонка,
+	// но это неважно. Если время свежее, до данные тоже
+	// будут свежмими, обратный случай не так важен.
+	const sll now_sec_ts = us_floor_ms(us_get_now_monotonic());
+	const sll state_sec_ts = atomic_load(&fpsi->state_sec_ts); // Сначала время
+	const ull state = atomic_load(&fpsi->state); // Потом инфа
+
+	uint current = state & 0xFFFF;
+	if (fpsi->with_meta) {
+		assert(meta != NULL);
+		meta->width = (state >> 16) & 0xFFFF;
+		meta->height = (state >> 32) & 0xFFFF;
+		meta->online = (state >> 48) & 1;
+	}
+
+	if (state_sec_ts != now_sec_ts && (state_sec_ts + 1) != now_sec_ts) {
+		// Только текущая или прошлая секунда
+		current = 0;
+	}
+	return current;
+}

src/libs/fpsi.h

@@ -24,23 +24,28 @@
 
 #include <stdatomic.h>
 
-#include "../libs/types.h"
+#include "types.h"
-#include "../libs/frame.h"
+#include "frame.h"
-#include "../libs/memsink.h"
-
-#include "m2m.h"
 
 
 typedef struct {
-	us_memsink_s		*sink;
+	uint	width;
-	bool				key_requested;
+	uint	height;
-	us_frame_s			*tmp_src;
+	bool	online;
-	us_frame_s			*dest;
+} us_fpsi_meta_s;
-	us_m2m_encoder_s	*enc;
+
-	atomic_bool			online;
+typedef struct {
-} us_h264_stream_s;
+	char			*name;
+	bool			with_meta;
+	uint			accum;
+	atomic_llong	state_sec_ts;
+	atomic_ullong	state;
+} us_fpsi_s;
 
 
-us_h264_stream_s *us_h264_stream_init(us_memsink_s *sink, const char *path, uint bitrate, uint gop);
+us_fpsi_s *us_fpsi_init(const char *name, bool with_meta);
-void us_h264_stream_destroy(us_h264_stream_s *h264);
+void us_fpsi_destroy(us_fpsi_s *fpsi);
-void us_h264_stream_process(us_h264_stream_s *h264, const us_frame_s *frame, bool force_key);
+
+void us_fpsi_frame_to_meta(const us_frame_s *frame, us_fpsi_meta_s *meta);
+void us_fpsi_update(us_fpsi_s *fpsi, bool bump, const us_fpsi_meta_s *meta);
+uint us_fpsi_get(us_fpsi_s *fpsi, us_fpsi_meta_s *meta);

src/libs/frame.h

@@ -26,54 +26,58 @@
 #include "tools.h"
 
 
+#define US_FRAME_META_DECLARE \
+	uint	width; \
+	uint	height; \
+	uint	format; \
+	uint	stride; \
+	/* Stride is a bytesperline in V4L2 */ \
+	/* https://www.kernel.org/doc/html/v4.14/media/uapi/v4l/pixfmt-v4l2.html */ \
+	/* https://medium.com/@oleg.shipitko/what-does-stride-mean-in-image-processing-bba158a72bcd */ \
+	bool	online; \
+	bool	key; \
+	uint	gop; \
+	\
+	ldf		grab_ts; \
+	ldf		encode_begin_ts; \
+	ldf		encode_end_ts;
+
+
 typedef struct {
 	u8		*data;
 	uz		used;
 	uz		allocated;
 	int		dma_fd;
 
-	uint	width;
+	US_FRAME_META_DECLARE;
-	uint	height;
-	uint	format;
-	uint	stride;
-	// Stride is a bytesperline in V4L2
-	// https://www.kernel.org/doc/html/v4.14/media/uapi/v4l/pixfmt-v4l2.html
-	// https://medium.com/@oleg.shipitko/what-does-stride-mean-in-image-processing-bba158a72bcd
-	bool	online;
-	bool	key;
-	uint	gop;
-
-	ldf		grab_ts;
-	ldf		encode_begin_ts;
-	ldf		encode_end_ts;
 } us_frame_s;
 
 
 #define US_FRAME_COPY_META(x_src, x_dest) { \
-		x_dest->width = x_src->width; \
+		(x_dest)->width = (x_src)->width; \
-		x_dest->height = x_src->height; \
+		(x_dest)->height = (x_src)->height; \
-		x_dest->format = x_src->format; \
+		(x_dest)->format = (x_src)->format; \
-		x_dest->stride = x_src->stride; \
+		(x_dest)->stride = (x_src)->stride; \
-		x_dest->online = x_src->online; \
+		(x_dest)->online = (x_src)->online; \
-		x_dest->key = x_src->key; \
+		(x_dest)->key = (x_src)->key; \
-		x_dest->gop = x_src->gop; \
+		(x_dest)->gop = (x_src)->gop; \
 		\
-		x_dest->grab_ts = x_src->grab_ts; \
+		(x_dest)->grab_ts = (x_src)->grab_ts; \
-		x_dest->encode_begin_ts = x_src->encode_begin_ts; \
+		(x_dest)->encode_begin_ts = (x_src)->encode_begin_ts; \
-		x_dest->encode_end_ts = x_src->encode_end_ts; \
+		(x_dest)->encode_end_ts = (x_src)->encode_end_ts; \
 	}
 
 #define US_FRAME_COMPARE_GEOMETRY(x_a, x_b) ( \
 		/* Compare the used size and significant meta (no timings) */ \
-		x_a->used == x_b->used \
+		(x_a)->used == (x_b)->used \
 		\
-		&& x_a->width == x_b->width \
+		&& (x_a)->width == (x_b)->width \
-		&& x_a->height == x_b->height \
+		&& (x_a)->height == (x_b)->height \
-		&& x_a->format == x_b->format \
+		&& (x_a)->format == (x_b)->format \
-		&& x_a->stride == x_b->stride \
+		&& (x_a)->stride == (x_b)->stride \
-		&& x_a->online == x_b->online \
+		&& (x_a)->online == (x_b)->online \
-		&& x_a->key == x_b->key \
+		&& (x_a)->key == (x_b)->key \
-		&& x_a->gop == x_b->gop \
+		&& (x_a)->gop == (x_b)->gop \
 	)
 
 

src/libs/list.h

@@ -25,9 +25,9 @@
 #include <assert.h>
 
 
-#define US_LIST_STRUCT(...) \
+#define US_LIST_DECLARE \
-	__VA_ARGS__ *prev; \
+	void *prev; \
-	__VA_ARGS__ *next;
+	void *next;
 
 #define US_LIST_ITERATE(x_first, x_item, ...) { \
 		for (__typeof__(x_first) x_item = x_first; x_item;) { \
@@ -42,10 +42,11 @@
 			x_first = x_item; \
 		} else { \
 			__typeof__(x_first) m_last = x_first; \
-			for (; m_last->next; m_last = m_last->next); \
+			for (; m_last->next != NULL; m_last = m_last->next); \
 			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) { \
@@ -57,11 +58,15 @@
 		if (x_item->prev == NULL) { \
 			x_first = x_item->next; \
 		} else { \
-			x_item->prev->next = x_item->next; \
+			__typeof__(x_first) m_prev = x_item->prev; \
+			m_prev->next = x_item->next; \
 		} \
 		if (x_item->next != NULL) { \
-			x_item->next->prev = x_item->prev; \
+			__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/memsink.c

@@ -33,13 +33,14 @@
 #include <sys/mman.h>
 
 #include "types.h"
+#include "errors.h"
 #include "tools.h"
 #include "logging.h"
 #include "frame.h"
 #include "memsinksh.h"
 
 
-us_memsink_s *us_memsink_init(
+us_memsink_s *us_memsink_init_opened(
 	const char *name, const char *obj, bool server,
 	mode_t mode, bool rm, uint client_ttl, uint timeout) {
 
@@ -168,7 +169,7 @@ int us_memsink_server_put(us_memsink_s *sink, const us_frame_s *frame, bool *key
 	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; // -2
+		return 0;
 	}
 
 	if (us_flock_timedwait_monotonic(sink->fd, 1) == 0) {
@@ -213,7 +214,7 @@ int us_memsink_client_get(us_memsink_s *sink, us_frame_s *frame, bool *key_reque
 
 	if (us_flock_timedwait_monotonic(sink->fd, sink->timeout) < 0) {
 		if (errno == EWOULDBLOCK) {
-			return -2;
+			return US_ERROR_NO_DATA;
 		}
 		US_LOG_PERROR("%s-sink: Can't lock memory", sink->name);
 		return -1;
@@ -222,7 +223,7 @@ int us_memsink_client_get(us_memsink_s *sink, us_frame_s *frame, bool *key_reque
 	int retval = 0;
 
 	if (sink->mem->magic != US_MEMSINK_MAGIC) {
-		retval = -2; // Not updated
+		retval = US_ERROR_NO_DATA; // Not updated
 		goto done;
 	}
 	if (sink->mem->version != US_MEMSINK_VERSION) {
@@ -236,7 +237,7 @@ int us_memsink_client_get(us_memsink_s *sink, us_frame_s *frame, bool *key_reque
 	sink->mem->last_client_ts = us_get_now_monotonic();
 
 	if (sink->mem->id == sink->last_readed_id) {
-		retval = -2; // Not updated
+		retval = US_ERROR_NO_DATA; // Not updated
 		goto done;
 	}
 

src/libs/memsink.h

@@ -50,7 +50,7 @@ typedef struct {
 } us_memsink_s;
 
 
-us_memsink_s *us_memsink_init(
+us_memsink_s *us_memsink_init_opened(
 	const char *name, const char *obj, bool server,
 	mode_t mode, bool rm, uint client_ttl, uint timeout);
 

src/libs/memsinksh.h

@@ -23,33 +23,23 @@
 #pragma once
 
 #include "types.h"
+#include "frame.h"
 
 
 #define US_MEMSINK_MAGIC	((u64)0xCAFEBABECAFEBABE)
-#define US_MEMSINK_VERSION	((u32)6)
+#define US_MEMSINK_VERSION	((u32)7)
 
 
 typedef struct {
 	u64		magic;
 	u32		version;
-
 	u64		id;
-
 	uz		used;
-	uint	width;
-	uint	height;
-	uint	format;
-	uint	stride;
-	bool	online;
-	bool	key;
-	uint	gop;
-
-	ldf		grab_ts;
-	ldf		encode_begin_ts;
-	ldf		encode_end_ts;
 
 	ldf		last_client_ts;
 	bool	key_requested;
+
+	US_FRAME_META_DECLARE;
 } us_memsink_shared_s;
 
 

src/libs/tc358743.c

@@ -58,7 +58,7 @@ int us_tc358743_xioctl_get_audio_hz(int fd, uint *audio_hz) {
 	US_MEMSET_ZERO(ctl);
 	ctl.id = TC358743_CID_AUDIO_SAMPLING_RATE;
 	if (us_xioctl(fd, VIDIOC_G_CTRL, &ctl) < 0) {
-		return -2;
+		return -1;
 	}
 	*audio_hz = ctl.value;
 	return 0;

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/tools.h

@@ -72,14 +72,16 @@
 		(m_a > m_b ? m_a : m_b); \
 	})
 
-#define US_ONCE(...) { \
+#define US_ONCE_FOR(x_once, x_value, ...) { \
-		const int m_reported = __LINE__; \
+		const int m_reported = (x_value); \
-		if (m_reported != once) { \
+		if (m_reported != (x_once)) { \
 			__VA_ARGS__; \
-			once = m_reported; \
+			(x_once) = m_reported; \
 		} \
 	}
 
+#define US_ONCE(...) US_ONCE_FOR(once, __LINE__, ##__VA_ARGS__)
+
 
 INLINE char *us_strdup(const char *str) {
 	char *const new = strdup(str);

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) {
+	us_encoder_runtime_s *const run = enc->run;
-		for (unsigned index = 0; index < _ER(n_m2ms); ++index) {
+	if (run->m2ms != NULL) {
-			US_DELETE(_ER(m2ms[index]), us_m2m_encoder_destroy)
+		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));
+	US_MUTEX_DESTROY(run->mutex);
-	free(enc->run);
+	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);
+	us_encoder_type_e type = enc->type;
-	unsigned quality = cap->jpeg_quality;
+	uint quality = cap->jpeg_quality;
-	unsigned n_workers = US_MIN(enc->n_workers, DR(n_bufs));
+	uint n_workers = US_MIN(enc->n_workers, cr->n_bufs);
-	bool cpu_forced = false;
 
-	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) {
+		if (run->m2ms == NULL) {
-			US_CALLOC(_ER(m2ms), n_workers);
+			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:
+	US_MUTEX_LOCK(run->mutex);
-		type = US_ENCODER_TYPE_CPU;
+	run->type = type;
-		quality = cap->jpeg_quality;
+	run->quality = quality;
+	US_MUTEX_UNLOCK(run->mutex);
 
-	ok:
+	const ldf desired_interval = (
-		if (type == US_ENCODER_TYPE_NOOP) {
+		cap->desired_fps > 0 && (cap->desired_fps < cap->run->hw_fps || cap->run->hw_fps == 0)
-			US_LOG_INFO("Using JPEG NOOP encoder");
+		? (ldf)1 / cap->desired_fps
-		} else if (quality == 0) {
+		: 0
-			US_LOG_INFO("Using JPEG quality: encoder default");
+	);
-		} else {
-			US_LOG_INFO("Using JPEG quality: %u%%", quality);
-		}
 
-		US_MUTEX_LOCK(_ER(mutex));
+	enc->run->pool = us_workers_pool_init(
-		_ER(type) = type;
+		"JPEG", "jw", n_workers, desired_interval,
-		_ER(quality) = quality;
+		_worker_job_init, (void*)enc,
-		if (cpu_forced) {
+		_worker_job_destroy,
-			_ER(cpu_forced) = true;
+		_worker_run_job);
-		}
-		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
 }
 
 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) {
+void us_encoder_get_runtime_params(us_encoder_s *enc, us_encoder_type_e *type, uint *quality) {
-	US_MUTEX_LOCK(_ER(mutex));
+	us_encoder_runtime_s *const run = enc->run;
-	*type = _ER(type);
+	US_MUTEX_LOCK(run->mutex);
-	*quality = _ER(quality);
+	*type = run->type;
-	US_MUTEX_UNLOCK(_ER(mutex));
+	*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_job_s *const job = wr->job;
-	us_encoder_s *enc = job->enc; // Just for _ER()
+	us_encoder_runtime_s *const run = job->enc->run;
-	const us_frame_s *src = &job->hw->raw;
+	const us_frame_s *const src = &job->hw->raw;
-	us_frame_s *dest = job->dest;
+	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);
+			(run->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) {
+		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:
+error:
-		US_LOG_ERROR("Compression failed: worker=%s, buffer=%u", wr->name, job->hw->buf.index);
+	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");
+	return false;
-		US_MUTEX_LOCK(_ER(mutex));
-		_ER(cpu_forced) = true;
-		US_MUTEX_UNLOCK(_ER(mutex));
-		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/types.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"
 
+#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;
+	uint				quality;
-	bool				cpu_forced;
 	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/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

@@ -120,6 +120,7 @@ us_server_s *us_server_init(us_stream_s *stream) {
 	us_server_exposed_s *exposed;
 	US_CALLOC(exposed, 1);
 	exposed->frame = us_frame_init();
+	exposed->queued_fpsi = us_fpsi_init("MJPEG-QUEUED", false);
 
 	us_server_runtime_s *run;
 	US_CALLOC(run, 1);
@@ -168,6 +169,7 @@ void us_server_destroy(us_server_s *server) {
 	});
 
 	US_LIST_ITERATE(run->stream_clients, client, { // cppcheck-suppress constStatement
+		us_fpsi_destroy(client->fpsi);
 		free(client->key);
 		free(client->hostport);
 		free(client);
@@ -175,6 +177,7 @@ void us_server_destroy(us_server_s *server) {
 
 	US_DELETE(run->auth_token, free);
 
+	us_fpsi_destroy(run->exposed->queued_fpsi);
 	us_frame_destroy(run->exposed->frame);
 	free(run->exposed);
 	free(server->run);
@@ -276,7 +279,7 @@ void us_server_loop_break(us_server_s *server) {
 static int _http_preprocess_request(struct evhttp_request *request, us_server_s *server) {
 	const us_server_runtime_s *const run = server->run;
 
-	atomic_store(&server->stream->run->http_last_request_ts, us_get_now_monotonic());
+	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");
@@ -473,12 +476,27 @@ static void _http_callback_state(struct evhttp_request *request, void *v_server)
 		enc_quality
 	);
 
-	if (stream->run->h264 != NULL) {
+#	ifdef WITH_V4P
+	if (stream->drm != NULL) {
+		us_fpsi_meta_s meta;
+		const uint fps = us_fpsi_get(stream->run->http->drm_fpsi, &meta);
 		_A_EVBUFFER_ADD_PRINTF(buf,
-			" \"h264\": {\"bitrate\": %u, \"gop\": %u, \"online\": %s},",
+			" \"drm\": {\"live\": %s, \"fps\": %u},",
+			us_bool_to_string(meta.online),
+			fps
+		);
+	}
+#	endif
+
+	if (stream->h264_sink != NULL) {
+		us_fpsi_meta_s meta;
+		const uint fps = us_fpsi_get(stream->run->http->h264_fpsi, &meta);
+		_A_EVBUFFER_ADD_PRINTF(buf,
+			" \"h264\": {\"bitrate\": %u, \"gop\": %u, \"online\": %s, \"fps\": %u},",
 			stream->h264_bitrate,
 			stream->h264_gop,
-			us_bool_to_string(atomic_load(&stream->run->h264->online))
+			us_bool_to_string(meta.online),
+			fps
 		);
 	}
 
@@ -500,21 +518,18 @@ static void _http_callback_state(struct evhttp_request *request, void *v_server)
 		_A_EVBUFFER_ADD_PRINTF(buf, "},");
 	}
 
-	uint width;
+	us_fpsi_meta_s captured_meta;
-	uint height;
+	const uint captured_fps = us_fpsi_get(stream->run->http->captured_fpsi, &captured_meta);
-	bool online;
-	uint captured_fps;
-	us_stream_get_capture_state(stream, &width, &height, &online, &captured_fps);
 	_A_EVBUFFER_ADD_PRINTF(buf,
 		" \"source\": {\"resolution\": {\"width\": %u, \"height\": %u},"
 		" \"online\": %s, \"desired_fps\": %u, \"captured_fps\": %u},"
 		" \"stream\": {\"queued_fps\": %u, \"clients\": %u, \"clients_stat\": {",
-		(server->fake_width ? server->fake_width : width),
+		(server->fake_width ? server->fake_width : captured_meta.width),
-		(server->fake_height ? server->fake_height : height),
+		(server->fake_height ? server->fake_height : captured_meta.height),
-		us_bool_to_string(online),
+		us_bool_to_string(captured_meta.online),
 		stream->cap->desired_fps,
 		captured_fps,
-		ex->queued_fps,
+		us_fpsi_get(ex->queued_fpsi, NULL),
 		run->stream_clients_count
 	);
 
@@ -523,7 +538,7 @@ static void _http_callback_state(struct evhttp_request *request, void *v_server)
 			"\"%" PRIx64 "\": {\"fps\": %u, \"extra_headers\": %s, \"advance_headers\": %s,"
 			" \"dual_final_frames\": %s, \"zero_data\": %s, \"key\": \"%s\"}%s",
 			client->id,
-			client->fps,
+			us_fpsi_get(client->fpsi, NULL),
 			us_bool_to_string(client->extra_headers),
 			us_bool_to_string(client->advance_headers),
 			us_bool_to_string(client->dual_final_frames),
@@ -551,7 +566,7 @@ static void _http_callback_snapshot(struct evhttp_request *request, void *v_serv
 	client->request = request;
 	client->request_ts = us_get_now_monotonic();
 
-	atomic_fetch_add(&server->stream->run->http_snapshot_requested, 1);
+	atomic_fetch_add(&server->stream->run->http->snapshot_requested, 1);
 	US_LIST_APPEND(server->run->snapshot_clients, client);
 }
 
@@ -590,10 +605,17 @@ static void _http_callback_stream(struct evhttp_request *request, void *v_server
 		client->hostport = _http_get_client_hostport(request);
 		client->id = us_get_now_id();
 
+		{
+			char *name;
+			US_ASPRINTF(name, "MJPEG-CLIENT-%" PRIx64, client->id);
+			client->fpsi = us_fpsi_init(name, false);
+			free(name);
+		}
+
 		US_LIST_APPEND_C(run->stream_clients, client, run->stream_clients_count);
 
 		if (run->stream_clients_count == 1) {
-			atomic_store(&server->stream->run->http_has_clients, true);
+			atomic_store(&server->stream->run->http->has_clients, true);
 #			ifdef WITH_GPIO
 			us_gpio_set_has_http_clients(true);
 #			endif
@@ -626,15 +648,7 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
 	us_server_s *const server = client->server;
 	us_server_exposed_s *const ex = server->run->exposed;
 
-	const ldf now_ts = us_get_now_monotonic();
+	us_fpsi_update(client->fpsi, true, NULL);
-	const sll now_sec_ts = us_floor_ms(now_ts);
-
-	if (now_sec_ts != client->fps_ts) {
-		client->fps = client->fps_accum;
-		client->fps_accum = 0;
-		client->fps_ts = now_sec_ts;
-	}
-	client->fps_accum += 1;
 
 	struct evbuffer *buf;
 	_A_EVBUFFER_NEW(buf);
@@ -716,6 +730,7 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
 			us_get_now_real(),
 			(client->extra_headers ? "" : RN)
 		);
+		const ldf now_ts = us_get_now_monotonic();
 		if (client->extra_headers) {
 			_A_EVBUFFER_ADD_PRINTF(buf,
 				"X-UStreamer-Online: %s" RN
@@ -736,7 +751,7 @@ static void _http_callback_stream_write(struct bufferevent *buf_event, void *v_c
 				ex->dropped,
 				ex->frame->width,
 				ex->frame->height,
-				client->fps,
+				us_fpsi_get(client->fpsi, NULL),
 				ex->frame->grab_ts,
 				ex->frame->encode_begin_ts,
 				ex->frame->encode_end_ts,
@@ -779,7 +794,7 @@ static void _http_callback_stream_error(struct bufferevent *buf_event, short wha
 	US_LIST_REMOVE_C(run->stream_clients, client, run->stream_clients_count);
 
 	if (run->stream_clients_count == 0) {
-		atomic_store(&server->stream->run->http_has_clients, false);
+		atomic_store(&server->stream->run->http->has_clients, false);
 #		ifdef WITH_GPIO
 		us_gpio_set_has_http_clients(false);
 #		endif
@@ -793,6 +808,7 @@ static void _http_callback_stream_error(struct bufferevent *buf_event, short wha
 	struct evhttp_connection *conn = evhttp_request_get_connection(client->request);
 	US_DELETE(conn, evhttp_connection_free);
 
+	us_fpsi_destroy(client->fpsi);
 	free(client->key);
 	free(client->hostport);
 	free(client);
@@ -802,8 +818,8 @@ static void _http_send_stream(us_server_s *server, bool stream_updated, bool fra
 	us_server_runtime_s *const run = server->run;
 	us_server_exposed_s *const ex = run->exposed;
 
-	bool has_clients = false;
 	bool queued = false;
+	bool has_clients = true;
 
 	US_LIST_ITERATE(run->stream_clients, client, { // cppcheck-suppress constStatement
 		struct evhttp_connection *const conn = evhttp_request_get_connection(client->request);
@@ -833,23 +849,14 @@ static void _http_send_stream(us_server_s *server, bool stream_updated, bool fra
 			} else if (stream_updated) { // Для dual
 				client->updated_prev = false;
 			}
-
 			has_clients = true;
 		}
 	});
 
 	if (queued) {
-		static uint queued_fps_accum = 0;
+		us_fpsi_update(ex->queued_fpsi, true, NULL);
-		static sll queued_fps_ts = 0;
-		const sll now_sec_ts = us_floor_ms(us_get_now_monotonic());
-		if (now_sec_ts != queued_fps_ts) {
-			ex->queued_fps = queued_fps_accum;
-			queued_fps_accum = 0;
-			queued_fps_ts = now_sec_ts;
-		}
-		queued_fps_accum += 1;
 	} else if (!has_clients) {
-		ex->queued_fps = 0;
+		us_fpsi_update(ex->queued_fpsi, false, NULL);
 	}
 }
 
@@ -866,24 +873,22 @@ static void _http_send_snapshot(us_server_s *server) {
 			US_SNPRINTF(header_buf, 255, "%u", x_value); \
 			_A_ADD_HEADER(request, x_key, header_buf); \
 		}
-	uint width;
+
-	uint height;
+	us_fpsi_meta_s captured_meta;
-	uint captured_fps; // Unused
+	us_fpsi_get(server->stream->run->http->captured_fpsi, &captured_meta);
-	bool online;
-	us_stream_get_capture_state(server->stream, &width, &height, &online, &captured_fps);
 
 	US_LIST_ITERATE(server->run->snapshot_clients, client, { // cppcheck-suppress constStatement
 		struct evhttp_request *request = client->request;
 
-		const bool has_fresh_snapshot = (atomic_load(&server->stream->run->http_snapshot_requested) == 0);
+		const bool has_fresh_snapshot = (atomic_load(&server->stream->run->http->snapshot_requested) == 0);
 		const bool timed_out = (client->request_ts + US_MAX((uint)1, server->stream->error_delay * 3) < us_get_now_monotonic());
 
 		if (has_fresh_snapshot || timed_out) {
 			us_frame_s *frame = ex->frame;
-			if (!online) {
+			if (!captured_meta.online) {
 				if (blank == NULL) {
 					blank = us_blank_init();
-					us_blank_draw(blank, "< NO SIGNAL >", width, height);
+					us_blank_draw(blank, "< NO SIGNAL >", captured_meta.width, captured_meta.height);
 				}
 				frame = blank->jpeg;
 			}
@@ -930,7 +935,7 @@ static void _http_refresher(int fd, short what, void *v_server) {
 
 	us_server_s *server = v_server;
 	us_server_exposed_s *ex = server->run->exposed;
-	us_ring_s *const ring = server->stream->run->http_jpeg_ring;
+	us_ring_s *const ring = server->stream->run->http->jpeg_ring;
 
 	bool stream_updated = false;
 	bool frame_updated = false;

src/ustreamer/http/server.h

@@ -31,11 +31,12 @@
 #include "../../libs/types.h"
 #include "../../libs/frame.h"
 #include "../../libs/list.h"
+#include "../../libs/fpsi.h"
 #include "../encoder.h"
 #include "../stream.h"
 
 
-typedef struct us_stream_client_sx {
+typedef struct {
 	struct us_server_sx		*server;
 	struct evhttp_request	*request;
 
@@ -50,25 +51,23 @@ typedef struct us_stream_client_sx {
 	bool	need_initial;
 	bool	need_first_frame;
 	bool	updated_prev;
-	uint	fps_accum;
-	sll		fps_ts;
-	uint	fps;
 
-	US_LIST_STRUCT(struct us_stream_client_sx);
+	us_fpsi_s *fpsi;
+
+	US_LIST_DECLARE;
 } us_stream_client_s;
 
-typedef struct us_snapshot_client_sx {
+typedef struct {
 	struct us_server_sx		*server;
 	struct evhttp_request	*request;
 	ldf						request_ts;
 
-	US_LIST_STRUCT(struct us_snapshot_client_sx);
+	US_LIST_DECLARE;
 } us_snapshot_client_s;
 
 typedef struct {
 	us_frame_s	*frame;
-	uint		captured_fps;
+	us_fpsi_s	*queued_fpsi;
-	uint		queued_fps;
 	uint		dropped;
 	ldf			expose_begin_ts;
 	ldf			expose_cmp_ts;

src/ustreamer/m2m.c

@@ -95,15 +95,30 @@ void us_m2m_encoder_destroy(us_m2m_encoder_s *enc) {
 int us_m2m_encoder_compress(us_m2m_encoder_s *enc, const us_frame_s *src, us_frame_s *dest, bool force_key) {
 	us_m2m_encoder_runtime_s *const run = enc->run;
 
-	us_frame_encoding_begin(src, dest, (enc->output_format == V4L2_PIX_FMT_MJPEG ? V4L2_PIX_FMT_JPEG : enc->output_format));
+	uint dest_format = enc->output_format;
+	switch (enc->output_format) {
+		case V4L2_PIX_FMT_JPEG:
+			force_key = false;
+			// fall through
+		case V4L2_PIX_FMT_MJPEG:
+			dest_format = V4L2_PIX_FMT_JPEG;
+			break;
+		case V4L2_PIX_FMT_H264:
+			force_key = (
+				force_key
+				|| run->last_online != src->online
+				|| run->last_encode_ts + 0.5 < us_get_now_monotonic()
+			);
+			break;
+	}
+
+	us_frame_encoding_begin(src, dest, dest_format);
 
 	_m2m_encoder_ensure(enc, src);
 	if (!run->ready) { // Already prepared but failed
 		return -1;
 	}
 
-	force_key = (enc->output_format == V4L2_PIX_FMT_H264 && (force_key || run->last_online != src->online));
-
 	_LOG_DEBUG("Compressing new frame; force_key=%d ...", force_key);
 
 	if (_m2m_encoder_compress_raw(enc, src, dest, force_key) < 0) {
@@ -118,6 +133,7 @@ int us_m2m_encoder_compress(us_m2m_encoder_s *enc, const us_frame_s *src, us_fra
 		dest->used, dest->encode_end_ts - dest->encode_begin_ts, force_key);
 
 	run->last_online = src->online;
+	run->last_encode_ts = dest->encode_end_ts;
 	return 0;
 }
 

src/ustreamer/m2m.h

@@ -47,6 +47,7 @@ typedef struct {
 
 	bool	ready;
 	int		last_online;
+	ldf		last_encode_ts;
 } us_m2m_encoder_runtime_s;
 
 typedef struct {

src/ustreamer/options.c

@@ -270,6 +270,9 @@ void us_options_destroy(us_options_s *options) {
 	US_DELETE(options->jpeg_sink, us_memsink_destroy);
 	US_DELETE(options->raw_sink, us_memsink_destroy);
 	US_DELETE(options->h264_sink, us_memsink_destroy);
+#	ifdef WITH_V4P
+	US_DELETE(options->drm, us_drm_destroy);
+#	endif
 
 	for (unsigned index = 0; index < options->argc; ++index) {
 		free(options->argv_copy[index]);
@@ -350,7 +353,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; \
@@ -361,7 +364,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];
@@ -463,7 +466,10 @@ int options_parse(us_options_s *options, us_capture_s *cap, us_encoder_s *enc, u
 			case _O_H264_M2M_DEVICE:		OPT_SET(stream->h264_m2m_path, optarg);
 
 #			ifdef WITH_V4P
-			case _O_V4P:					OPT_SET(stream->v4p, true);
+			case _O_V4P:
+				options->drm = us_drm_init();
+				stream->drm = options->drm;
+				break;
 #			endif
 
 #			ifdef WITH_GPIO
@@ -507,7 +513,7 @@ int options_parse(us_options_s *options, us_capture_s *cap, us_encoder_s *enc, u
 
 #	define ADD_SINK(x_label, x_prefix) { \
 			if (x_prefix##_name && x_prefix##_name[0] != '\0') { \
-				options->x_prefix = us_memsink_init( \
+				options->x_prefix = us_memsink_init_opened( \
 					x_label, \
 					x_prefix##_name, \
 					true, \
@@ -645,8 +651,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("                                             * M2M-IMAGE  ── GPU-accelerated JPEG encoding using V4L2 M2M image interface.\n");
-	SAY("                                             * NOOP  ─────── Don't compress MJPEG stream (do nothing).\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/options.h

@@ -40,6 +40,9 @@
 #include "../libs/memsink.h"
 #include "../libs/options.h"
 #include "../libs/capture.h"
+#ifdef WITH_V4P
+#	include "../libs/drm/drm.h"
+#endif
 
 #include "encoder.h"
 #include "stream.h"
@@ -56,6 +59,9 @@ typedef struct {
 	us_memsink_s	*jpeg_sink;
 	us_memsink_s	*raw_sink;
 	us_memsink_s	*h264_sink;
+#	ifdef WITH_V4P
+	us_drm_s		*drm;
+#	endif
 } us_options_s;
 
 

src/ustreamer/stream.c

@@ -24,6 +24,7 @@
 
 #include <stdlib.h>
 #include <stdatomic.h>
+#include <limits.h>
 #include <unistd.h>
 #include <errno.h>
 #include <assert.h>
@@ -31,6 +32,7 @@
 #include <pthread.h>
 
 #include "../libs/types.h"
+#include "../libs/errors.h"
 #include "../libs/tools.h"
 #include "../libs/threading.h"
 #include "../libs/process.h"
@@ -39,6 +41,8 @@
 #include "../libs/frame.h"
 #include "../libs/memsink.h"
 #include "../libs/capture.h"
+#include "../libs/unjpeg.h"
+#include "../libs/fpsi.h"
 #ifdef WITH_V4P
 #	include "../libs/drm/drm.h"
 #endif
@@ -46,7 +50,7 @@
 #include "blank.h"
 #include "encoder.h"
 #include "workers.h"
-#include "h264.h"
+#include "m2m.h"
 #ifdef WITH_GPIO
 #	include "gpio/gpio.h"
 #endif
@@ -68,12 +72,10 @@ typedef struct {
 } _worker_context_s;
 
 
-static void _stream_set_capture_state(us_stream_s *stream, uint width, uint height, bool online, uint captured_fps);
-
 static void *_releaser_thread(void *v_ctx);
 static void *_jpeg_thread(void *v_ctx);
-static void *_h264_thread(void *v_ctx);
 static void *_raw_thread(void *v_ctx);
+static void *_h264_thread(void *v_ctx);
 #ifdef WITH_V4P
 static void *_drm_thread(void *v_ctx);
 #endif
@@ -88,19 +90,28 @@ static void _stream_drm_ensure_no_signal(us_stream_s *stream);
 #endif
 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);
 
 
 us_stream_s *us_stream_init(us_capture_s *cap, us_encoder_s *enc) {
+	us_stream_http_s *http;
+	US_CALLOC(http, 1);
+#	ifdef WITH_V4P
+	http->drm_fpsi = us_fpsi_init("DRM", true);
+#	endif
+	http->h264_fpsi = us_fpsi_init("H264", true);
+	US_RING_INIT_WITH_ITEMS(http->jpeg_ring, 4, us_frame_init);
+	atomic_init(&http->has_clients, false);
+	atomic_init(&http->snapshot_requested, 0);
+	atomic_init(&http->last_request_ts, 0);
+	http->captured_fpsi = us_fpsi_init("STREAM-CAPTURED", true);
+
 	us_stream_runtime_s *run;
 	US_CALLOC(run, 1);
-	US_RING_INIT_WITH_ITEMS(run->http_jpeg_ring, 4, us_frame_init);
-	atomic_init(&run->http_has_clients, false);
-	atomic_init(&run->http_snapshot_requested, 0);
-	atomic_init(&run->http_last_request_ts, 0);
-	atomic_init(&run->http_capture_state, 0);
 	atomic_init(&run->stop, false);
 	run->blank = us_blank_init();
+	run->http = http;
 
 	us_stream_s *stream;
 	US_CALLOC(stream, 1);
@@ -112,13 +123,21 @@ us_stream_s *us_stream_init(us_capture_s *cap, us_encoder_s *enc) {
 	stream->run = run;
 
 	us_blank_draw(run->blank, "< NO SIGNAL >", cap->width, cap->height);
-	_stream_set_capture_state(stream, cap->width, cap->height, false, 0);
+	us_fpsi_meta_s meta = {0};
+	us_fpsi_frame_to_meta(run->blank->raw, &meta);
+	us_fpsi_update(http->captured_fpsi, false, &meta);
 	return stream;
 }
 
 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);
+	us_fpsi_destroy(stream->run->http->h264_fpsi);
+#	ifdef WITH_V4P
+	us_fpsi_destroy(stream->run->http->drm_fpsi);
+#	endif
 	us_blank_destroy(stream->run->blank);
-	US_RING_DELETE_WITH_ITEMS(stream->run->http_jpeg_ring, us_frame_destroy);
+	free(stream->run->http);
 	free(stream->run);
 	free(stream);
 }
@@ -127,23 +146,14 @@ void us_stream_loop(us_stream_s *stream) {
 	us_stream_runtime_s *const run = stream->run;
 	us_capture_s *const cap = stream->cap;
 
-	US_LOG_INFO("Using V4L2 device: %s", cap->path);
+	atomic_store(&run->http->last_request_ts, us_get_now_monotonic());
-	US_LOG_INFO("Using desired FPS: %u", cap->desired_fps);
-
-	atomic_store(&run->http_last_request_ts, us_get_now_monotonic());
 
 	if (stream->h264_sink != NULL) {
-		run->h264 = us_h264_stream_init(stream->h264_sink, stream->h264_m2m_path, stream->h264_bitrate, stream->h264_gop);
+		run->h264_enc = us_m2m_h264_encoder_init("H264", stream->h264_m2m_path, stream->h264_bitrate, stream->h264_gop);
+		run->h264_tmp_src = us_frame_init();
+		run->h264_dest = us_frame_init();
 	}
 
-#	ifdef WITH_V4P
-	if (stream->v4p) {
-		run->drm = us_drm_init();
-		run->drm_opened = -1;
-		US_LOG_INFO("Using passthrough: %s[%s]", run->drm->path, run->drm->port);
-	}
-#	endif
-
 	while (!_stream_init_loop(stream)) {
 		atomic_bool threads_stop;
 		atomic_init(&threads_stop, false);
@@ -162,42 +172,22 @@ void us_stream_loop(us_stream_s *stream) {
 			US_THREAD_CREATE(ctx->tid, _releaser_thread, ctx);
 		}
 
-		_worker_context_s jpeg_ctx = {
+#		define CREATE_WORKER(x_cond, x_ctx, x_thread, x_capacity) \
-			.queue = us_queue_init(cap->run->n_bufs),
+			_worker_context_s *x_ctx = NULL; \
-			.stream = stream,
+			if (x_cond) { \
-			.stop = &threads_stop,
+				US_CALLOC(x_ctx, 1); \
-		};
+				x_ctx->queue = us_queue_init(x_capacity); \
-		US_THREAD_CREATE(jpeg_ctx.tid, _jpeg_thread, &jpeg_ctx);
+				x_ctx->stream = stream; \
-
+				x_ctx->stop = &threads_stop; \
-		_worker_context_s h264_ctx;
+				US_THREAD_CREATE(x_ctx->tid, (x_thread), x_ctx); \
-		if (run->h264 != NULL) {
+			}
-			h264_ctx.queue = us_queue_init(cap->run->n_bufs);
+		CREATE_WORKER(true, jpeg_ctx, _jpeg_thread, cap->run->n_bufs);
-			h264_ctx.stream = stream;
+		CREATE_WORKER((stream->raw_sink != NULL), raw_ctx, _raw_thread, 2);
-			h264_ctx.stop = &threads_stop;
+		CREATE_WORKER((stream->h264_sink != NULL), h264_ctx, _h264_thread, cap->run->n_bufs);
-			US_THREAD_CREATE(h264_ctx.tid, _h264_thread, &h264_ctx);
-		}
-
-		_worker_context_s raw_ctx;
-		if (stream->raw_sink != NULL) {
-			raw_ctx.queue = us_queue_init(2);
-			raw_ctx.stream = stream;
-			raw_ctx.stop = &threads_stop;
-			US_THREAD_CREATE(raw_ctx.tid, _raw_thread, &raw_ctx);
-		}
-
 #		ifdef WITH_V4P
-		_worker_context_s drm_ctx;
+		CREATE_WORKER((stream->drm != NULL), drm_ctx, _drm_thread, cap->run->n_bufs); // cppcheck-suppress assertWithSideEffect
-		if (stream->v4p) {
-			drm_ctx.queue = us_queue_init(cap->run->n_bufs);
-			drm_ctx.stream = stream;
-			drm_ctx.stop = &threads_stop;
-			US_THREAD_CREATE(drm_ctx.tid, _drm_thread, &drm_ctx); // cppcheck-suppress assertWithSideEffect
-		}
 #		endif
-
+#		undef CREATE_WORKER
-		uint captured_fps_accum = 0;
-		sll captured_fps_ts = 0;
-		uint captured_fps = 0;
 
 		US_LOG_INFO("Capturing ...");
 
@@ -205,41 +195,30 @@ void us_stream_loop(us_stream_s *stream) {
 		while (!atomic_load(&run->stop) && !atomic_load(&threads_stop)) {
 			us_capture_hwbuf_s *hw;
 			switch (us_capture_hwbuf_grab(cap, &hw)) {
-				case -2: continue; // Broken frame
+				case 0 ... INT_MAX: break; // Grabbed buffer number
-				case -1: goto close; // Error
+				case US_ERROR_NO_DATA: continue; // Broken frame
-				default: break; // Grabbed on >= 0
+				default: goto close; // Any error
 			}
 
-			const sll now_sec_ts = us_floor_ms(us_get_now_monotonic());
+			us_fpsi_meta_s meta = {0};
-			if (now_sec_ts != captured_fps_ts) {
+			us_fpsi_frame_to_meta(&hw->raw, &meta);
-				captured_fps = captured_fps_accum;
+			us_fpsi_update(run->http->captured_fpsi, true, &meta);
-				captured_fps_accum = 0;
-				captured_fps_ts = now_sec_ts;
-				US_LOG_PERF_FPS("A new second has come; captured_fps=%u", captured_fps);
-			}
-			captured_fps_accum += 1;
 
-			_stream_set_capture_state(stream, cap->run->width, cap->run->height, true, captured_fps);
 #			ifdef WITH_GPIO
 			us_gpio_set_stream_online(true);
 #			endif
 
-			us_capture_hwbuf_incref(hw); // JPEG
+#			define QUEUE_HW(x_ctx) if (x_ctx != NULL) { \
-			us_queue_put(jpeg_ctx.queue, hw, 0);
+					us_capture_hwbuf_incref(hw); \
-			if (run->h264 != NULL) {
+					us_queue_put(x_ctx->queue, hw, 0); \
-				us_capture_hwbuf_incref(hw); // H264
+				}
-				us_queue_put(h264_ctx.queue, hw, 0);
+			QUEUE_HW(jpeg_ctx);
-			}
+			QUEUE_HW(raw_ctx);
-			if (stream->raw_sink != NULL) {
+			QUEUE_HW(h264_ctx);
-				us_capture_hwbuf_incref(hw); // RAW
-				us_queue_put(raw_ctx.queue, hw, 0);
-			}
 #			ifdef WITH_V4P
-			if (stream->v4p) {
+			QUEUE_HW(drm_ctx);
-				us_capture_hwbuf_incref(hw); // DRM
-				us_queue_put(drm_ctx.queue, hw, 0);
-			}
 #			endif
+#			undef QUEUE_HW
 			us_queue_put(releasers[hw->buf.index].queue, hw, 0); // Plan to release
 
 			// Мы не обновляем здесь состояние синков, потому что это происходит внутри обслуживающих их потоков
@@ -256,25 +235,18 @@ void us_stream_loop(us_stream_s *stream) {
 	close:
 		atomic_store(&threads_stop, true);
 
+#		define DELETE_WORKER(x_ctx) if (x_ctx != NULL) { \
+				US_THREAD_JOIN(x_ctx->tid); \
+				us_queue_destroy(x_ctx->queue); \
+				free(x_ctx); \
+			}
 #		ifdef WITH_V4P
-		if (stream->v4p) {
+		DELETE_WORKER(drm_ctx);
-			US_THREAD_JOIN(drm_ctx.tid);
-			us_queue_destroy(drm_ctx.queue);
-		}
 #		endif
-
+		DELETE_WORKER(h264_ctx);
-		if (stream->raw_sink != NULL) {
+		DELETE_WORKER(raw_ctx);
-			US_THREAD_JOIN(raw_ctx.tid);
+		DELETE_WORKER(jpeg_ctx);
-			us_queue_destroy(raw_ctx.queue);
+#		undef DELETE_WORKER
-		}
-
-		if (run->h264 != NULL) {
-			US_THREAD_JOIN(h264_ctx.tid);
-			us_queue_destroy(h264_ctx.queue);
-		}
-
-		US_THREAD_JOIN(jpeg_ctx.tid);
-		us_queue_destroy(jpeg_ctx.queue);
 
 		for (uint index = 0; index < n_releasers; ++index) {
 			US_THREAD_JOIN(releasers[index].tid);
@@ -293,34 +265,15 @@ void us_stream_loop(us_stream_s *stream) {
 		}
 	}
 
-#	ifdef WITH_V4P
+	US_DELETE(run->h264_enc, us_m2m_encoder_destroy);
-	US_DELETE(run->drm, us_drm_destroy);
+	US_DELETE(run->h264_tmp_src, us_frame_destroy);
-#	endif
+	US_DELETE(run->h264_dest, us_frame_destroy);
-	US_DELETE(run->h264, us_h264_stream_destroy);
 }
 
 void us_stream_loop_break(us_stream_s *stream) {
 	atomic_store(&stream->run->stop, true);
 }
 
-void us_stream_get_capture_state(us_stream_s *stream, uint *width, uint *height, bool *online, uint *captured_fps) {
-	const u64 state = atomic_load(&stream->run->http_capture_state);
-	*width = state & 0xFFFF;
-	*height = (state >> 16) & 0xFFFF;
-	*captured_fps = (state >> 32) & 0xFFFF;
-	*online = (state >> 48) & 1;
-}
-
-void _stream_set_capture_state(us_stream_s *stream, uint width, uint height, bool online, uint captured_fps) {
-	const u64 state = (
-		(u64)(width & 0xFFFF)
-		| ((u64)(height & 0xFFFF) << 16)
-		| ((u64)(captured_fps & 0xFFFF) << 32)
-		| ((u64)(online ? 1 : 0) << 48)
-	);
-	atomic_store(&stream->run->http_capture_state, state);
-}
-
 static void *_releaser_thread(void *v_ctx) {
 	US_THREAD_SETTLE("str_rel")
 	_releaser_context_s *ctx = v_ctx;
@@ -360,23 +313,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_worker_s *const wr = us_workers_pool_wait(stream->enc->run->pool);
-		us_encoder_job_s *const ready_job = ready_wr->job;
+		us_encoder_job_s *const job = wr->job;
 
-		if (ready_job->hw != NULL) {
+		if (job->hw != NULL) {
-			us_capture_hwbuf_decref(ready_job->hw);
+			us_capture_hwbuf_decref(job->hw);
-			ready_job->hw = NULL;
+			job->hw = NULL;
-			if (ready_wr->job_failed) {
+			if (wr->job_failed) {
 				// pass
-			} else if (ready_wr->job_timely) {
+			} else if (wr->job_timely) {
-				_stream_expose_jpeg(stream, ready_job->dest);
+				_stream_expose_jpeg(stream, job->dest);
-				if (atomic_load(&stream->run->http_snapshot_requested) > 0) { // Process real snapshots
+				if (atomic_load(&stream->run->http->snapshot_requested) > 0) { // Process real snapshots
-					atomic_fetch_sub(&stream->run->http_snapshot_requested, 1);
+					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);
 			}
 		}
 
@@ -402,57 +355,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;
+		job->hw = hw;
-		us_workers_pool_assign(stream->enc->run->pool, ready_wr);
+		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, ready_wr->name);
+		US_LOG_DEBUG("JPEG: Assigned new frame in buffer=%d to worker=%s", hw->buf.index, wr->name);
-	}
-	return NULL;
-}
-
-static void *_h264_thread(void *v_ctx) {
-	US_THREAD_SETTLE("str_h264");
-	_worker_context_s *ctx = v_ctx;
-	us_h264_stream_s *h264 = ctx->stream->run->h264;
-
-	ldf grab_after_ts = 0;
-	ldf last_encode_ts = us_get_now_monotonic();
-
-	while (!atomic_load(ctx->stop)) {
-		us_capture_hwbuf_s *hw = _get_latest_hw(ctx->queue);
-		if (hw == NULL) {
-			continue;
-		}
-
-		if (!us_memsink_server_check(h264->sink, NULL)) {
-			us_capture_hwbuf_decref(hw);
-			US_LOG_VERBOSE("H264: Passed encoding because nobody is watching");
-			continue;
-		}
-
-		if (hw->raw.grab_ts < grab_after_ts) {
-			us_capture_hwbuf_decref(hw);
-			US_LOG_VERBOSE("H264: Passed encoding for FPS limit: %u", h264->enc->run->fps_limit);
-			continue;
-		}
-
-		// Форсим кейфрейм, если от захвата давно не было фреймов
-		const ldf now_ts = us_get_now_monotonic();
-		const bool force_key = (last_encode_ts + 0.5 < now_ts);
-		us_h264_stream_process(h264, &hw->raw, force_key);
-		last_encode_ts = now_ts;
-
-		// M2M-енкодер увеличивает задержку на 100 милисекунд при 1080p, если скормить ему больше 30 FPS.
-		// Поэтому у нас есть два режима: 60 FPS для маленьких видео и 30 для 1920x1080(1200).
-		// Следующй фрейм захватывается не раньше, чем это требуется по FPS, минус небольшая
-		// погрешность (если захват неравномерный) - немного меньше 1/60, и примерно треть от 1/30.
-		const ldf frame_interval = (ldf)1 / h264->enc->run->fps_limit;
-		grab_after_ts = hw->raw.grab_ts + frame_interval - 0.01;
-
-		us_capture_hwbuf_decref(hw);
 	}
 	return NULL;
 }
@@ -467,13 +376,50 @@ static void *_raw_thread(void *v_ctx) {
 			continue;
 		}
 
-		if (!us_memsink_server_check(ctx->stream->raw_sink, NULL)) {
+		if (us_memsink_server_check(ctx->stream->raw_sink, NULL)) {
-			us_capture_hwbuf_decref(hw);
+			us_memsink_server_put(ctx->stream->raw_sink, &hw->raw, false);
+		} else {
 			US_LOG_VERBOSE("RAW: Passed publishing because nobody is watching");
+		}
+		us_capture_hwbuf_decref(hw);
+	}
+	return NULL;
+}
+
+static void *_h264_thread(void *v_ctx) {
+	US_THREAD_SETTLE("str_h264");
+	_worker_context_s *ctx = v_ctx;
+	us_stream_s *stream = ctx->stream;
+
+	ldf grab_after_ts = 0;
+	while (!atomic_load(ctx->stop)) {
+		us_capture_hwbuf_s *hw = _get_latest_hw(ctx->queue);
+		if (hw == NULL) {
 			continue;
 		}
 
-		us_memsink_server_put(ctx->stream->raw_sink, &hw->raw, false);
+		if (!us_memsink_server_check(stream->h264_sink, NULL)) {
+			US_LOG_VERBOSE("H264: Passed encoding because nobody is watching");
+			goto decref;
+		}
+		if (hw->raw.grab_ts < grab_after_ts) {
+			US_LOG_DEBUG("H264: Passed encoding for FPS limit");
+			goto decref;
+		}
+
+		_stream_encode_expose_h264(ctx->stream, &hw->raw, false);
+
+		// M2M-енкодер увеличивает задержку на 100 милисекунд при 1080p, если скормить ему больше 30 FPS.
+		// Поэтому у нас есть два режима: 60 FPS для маленьких видео и 30 для 1920x1080(1200).
+		// Следующй фрейм захватывается не раньше, чем это требуется по FPS, минус небольшая
+		// погрешность (если захват неравномерный) - немного меньше 1/60, и примерно треть от 1/30.
+		const uint fps_limit = stream->run->h264_enc->run->fps_limit;
+		if (fps_limit > 0) {
+			const ldf frame_interval = (ldf)1 / fps_limit;
+			grab_after_ts = hw->raw.grab_ts + frame_interval - 0.01;
+		}
+
+	decref:
 		us_capture_hwbuf_decref(hw);
 	}
 	return NULL;
@@ -483,17 +429,16 @@ static void *_raw_thread(void *v_ctx) {
 static void *_drm_thread(void *v_ctx) {
 	US_THREAD_SETTLE("str_drm");
 	_worker_context_s *ctx = v_ctx;
-	us_stream_runtime_s *run = ctx->stream->run;
+	us_stream_s *stream = ctx->stream;
 
 	// Close previously opened DRM for a stub
-	us_drm_close(run->drm);
+	us_drm_close(stream->drm);
-	run->drm_opened = -1;
 
 	us_capture_hwbuf_s *prev_hw = NULL;
 	while (!atomic_load(ctx->stop)) {
 #		define CHECK(x_arg) if ((x_arg) < 0) { goto close; }
 #		define SLOWDOWN { \
-				ldf m_next_ts = us_get_now_monotonic() + 1; \
+				const ldf m_next_ts = us_get_now_monotonic() + 1; \
 				while (!atomic_load(ctx->stop) && us_get_now_monotonic() < m_next_ts) { \
 					us_capture_hwbuf_s *m_pass_hw = _get_latest_hw(ctx->queue); \
 					if (m_pass_hw != NULL) { \
@@ -502,10 +447,10 @@ static void *_drm_thread(void *v_ctx) {
 				} \
 			}
 
-		CHECK(run->drm_opened = us_drm_open(run->drm, ctx->stream->cap));
+		CHECK(us_drm_open(stream->drm, ctx->stream->cap));
 
 		while (!atomic_load(ctx->stop)) {
-			CHECK(us_drm_wait_for_vsync(run->drm));
+			CHECK(us_drm_wait_for_vsync(stream->drm));
 			US_DELETE(prev_hw, us_capture_hwbuf_decref);
 
 			us_capture_hwbuf_s *hw = _get_latest_hw(ctx->queue);
@@ -513,22 +458,28 @@ static void *_drm_thread(void *v_ctx) {
 				continue;
 			}
 
-			if (run->drm_opened == 0) {
+			if (stream->drm->run->opened == 0) {
-				CHECK(us_drm_expose_dma(run->drm, hw));
+				CHECK(us_drm_expose_dma(stream->drm, hw));
 				prev_hw = hw;
+				us_fpsi_meta_s meta = {.online = true}; // Online means live video
+				us_fpsi_update(stream->run->http->drm_fpsi, true, &meta);
 				continue;
 			}
 
-			CHECK(us_drm_expose_stub(run->drm, run->drm_opened, ctx->stream->cap));
+			CHECK(us_drm_expose_stub(stream->drm, stream->drm->run->opened, ctx->stream->cap));
 			us_capture_hwbuf_decref(hw);
 
+			us_fpsi_meta_s meta = {.online = false};
+			us_fpsi_update(stream->run->http->drm_fpsi, true, &meta);
+
 			SLOWDOWN;
 		}
 
 	close:
-		us_drm_close(run->drm);
+		us_drm_close(stream->drm);
-		run->drm_opened = -1;
 		US_DELETE(prev_hw, us_capture_hwbuf_decref);
+		us_fpsi_meta_s meta = {.online = false};
+		us_fpsi_update(stream->run->http->drm_fpsi, false, &meta);
 		SLOWDOWN;
 
 #		undef SLOWDOWN
@@ -553,28 +504,27 @@ static us_capture_hwbuf_s *_get_latest_hw(us_queue_s *queue) {
 static bool _stream_has_jpeg_clients_cached(us_stream_s *stream) {
 	const us_stream_runtime_s *const run = stream->run;
 	return (
-		atomic_load(&run->http_has_clients)
+		atomic_load(&run->http->has_clients)
-		|| (atomic_load(&run->http_snapshot_requested) > 0)
+		|| (atomic_load(&run->http->snapshot_requested) > 0)
 		|| (stream->jpeg_sink != NULL && atomic_load(&stream->jpeg_sink->has_clients))
 	);
 }
 
 static bool _stream_has_any_clients_cached(us_stream_s *stream) {
-	const us_stream_runtime_s *const run = stream->run;
 	return (
-#		ifdef WITH_V4P
-		stream->v4p ||
-#		endif
 		_stream_has_jpeg_clients_cached(stream)
-		|| (run->h264 != NULL && atomic_load(&run->h264->sink->has_clients))
+		|| (stream->h264_sink != NULL && atomic_load(&stream->h264_sink->has_clients))
 		|| (stream->raw_sink != NULL && atomic_load(&stream->raw_sink->has_clients))
+#		ifdef WITH_V4P
+		|| (stream->drm != NULL)
+#		endif
 	);
 }
 
 static int _stream_init_loop(us_stream_s *stream) {
 	us_stream_runtime_s *const run = stream->run;
 
-	bool waiting_reported = false;
+	int once = 0;
 	while (!atomic_load(&stream->run->stop)) {
 #		ifdef WITH_GPIO
 		us_gpio_set_stream_online(false);
@@ -582,40 +532,31 @@ static int _stream_init_loop(us_stream_s *stream) {
 
 		// Флаги has_clients у синков не обновляются сами по себе, поэтому обновим их
 		// на каждой итерации старта стрима. После старта этим будут заниматься воркеры.
-		if (stream->jpeg_sink != NULL) {
+#		define UPDATE_SINK(x_sink) if (x_sink != NULL) { us_memsink_server_check(x_sink, NULL); }
-			us_memsink_server_check(stream->jpeg_sink, NULL);
+		UPDATE_SINK(stream->jpeg_sink);
-		}
+		UPDATE_SINK(stream->raw_sink);
-		if (stream->run->h264 != NULL) {
+		UPDATE_SINK(stream->h264_sink);
-			us_memsink_server_check(stream->run->h264->sink, NULL);
+#		undef UPDATE_SINK
-		}
-		if (stream->raw_sink != NULL) {
-			us_memsink_server_check(stream->raw_sink, NULL);
-		}
 
 		_stream_check_suicide(stream);
 
 		stream->cap->dma_export = (
 			stream->enc->type == US_ENCODER_TYPE_M2M_VIDEO
 			|| stream->enc->type == US_ENCODER_TYPE_M2M_IMAGE
-			|| run->h264 != NULL
+			|| stream->h264_sink != NULL
+#			ifdef WITH_V4P
+			|| stream->drm != NULL
+#			endif
 		);
 		switch (us_capture_open(stream->cap)) {
-			case -2:
+			case 0: break;
-				if (!waiting_reported) {
+			case US_ERROR_NO_DEVICE:
-					waiting_reported = true;
+			case US_ERROR_NO_DATA:
-					US_LOG_INFO("Waiting for the capture device ...");
+				US_ONCE({ US_LOG_INFO("Waiting for the capture device ..."); });
-				}
-#				ifdef WITH_V4P
-				_stream_drm_ensure_no_signal(stream);
-#				endif
 				goto offline_and_retry;
-			case -1:
+			default:
-				waiting_reported = false;
+				once = 0;
-#				ifdef WITH_V4P
-				_stream_drm_ensure_no_signal(stream);
-#				endif
 				goto offline_and_retry;
-			default: break;
 		}
 		us_encoder_open(stream->enc, stream->cap);
 		return 0;
@@ -635,13 +576,17 @@ static int _stream_init_loop(us_stream_s *stream) {
 				}
 				us_blank_draw(run->blank, "< NO SIGNAL >", width, height);
 
-				_stream_set_capture_state(stream, width, height, false, 0);
+				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_expose_jpeg(stream, run->blank->jpeg);
-				if (run->h264 != NULL) {
-					us_h264_stream_process(run->h264, run->blank->raw, true);
-				}
 				_stream_expose_raw(stream, run->blank->raw);
+				_stream_encode_expose_h264(stream, run->blank->raw, true);
+
+#				ifdef WITH_V4P
+				_stream_drm_ensure_no_signal(stream);
+#				endif
 			}
 			usleep(100 * 1000);
 		}
@@ -651,51 +596,40 @@ static int _stream_init_loop(us_stream_s *stream) {
 
 #ifdef WITH_V4P
 static void _stream_drm_ensure_no_signal(us_stream_s *stream) {
-	us_stream_runtime_s *const run = stream->run;
+	if (stream->drm == NULL) {
-
-	if (!stream->v4p) {
 		return;
 	}
 
-#	define CHECK(x_arg) if ((x_arg) < 0) { goto close; }
+	const us_fpsi_meta_s meta = {.online = false};
-	if (run->drm_opened <= 0) {
+	if (stream->drm->run->opened <= 0) {
-		us_drm_close(run->drm);
+		us_drm_close(stream->drm);
-		run->drm_blank_at_ts = 0;
+		if (us_drm_open(stream->drm, NULL) < 0) {
-		CHECK(run->drm_opened = us_drm_open(run->drm, NULL));
+			goto close;
+		}
 	}
-
+	if (us_drm_ensure_no_signal(stream->drm) < 0) {
-	ldf now_ts = us_get_now_monotonic();
+		goto close;
-	if (run->drm_blank_at_ts == 0) {
-		run->drm_blank_at_ts = now_ts + 5;
-	}
-
-	if (now_ts <= run->drm_blank_at_ts) {
-		CHECK(us_drm_wait_for_vsync(run->drm));
-		CHECK(us_drm_expose_stub(run->drm, US_DRM_STUB_NO_SIGNAL, NULL));
-	} else {
-		// US_ONCE({ US_LOG_INFO("DRM: Turning off the display by timeout ..."); });
-		CHECK(us_drm_dpms_power_off(run->drm));
 	}
+	us_fpsi_update(stream->run->http->drm_fpsi, true, &meta);
 	return;
-#	undef CHECK
 
 close:
-	us_drm_close(run->drm);
+	us_fpsi_update(stream->run->http->drm_fpsi, false, &meta);
-	run->drm_opened = -1;
+	us_drm_close(stream->drm);
 }
 #endif
 
 static void _stream_expose_jpeg(us_stream_s *stream, const us_frame_s *frame) {
 	us_stream_runtime_s *const run = stream->run;
 	int ri;
-	while ((ri = us_ring_producer_acquire(run->http_jpeg_ring, 0)) < 0) {
+	while ((ri = us_ring_producer_acquire(run->http->jpeg_ring, 0)) < 0) {
 		if (atomic_load(&run->stop)) {
 			return;
 		}
 	}
-	us_frame_s *const dest = run->http_jpeg_ring->items[ri];
+	us_frame_s *const dest = run->http->jpeg_ring->items[ri];
 	us_frame_copy(frame, dest);
-	us_ring_producer_release(run->http_jpeg_ring, ri);
+	us_ring_producer_release(run->http->jpeg_ring, ri);
 	if (stream->jpeg_sink != NULL) {
 		us_memsink_server_put(stream->jpeg_sink, dest, NULL);
 	}
@@ -707,19 +641,46 @@ 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) {
+	if (stream->h264_sink == NULL) {
+		return;
+	}
+	us_stream_runtime_s *run = stream->run;
+
+	us_fpsi_meta_s meta = {.online = false};
+	if (us_is_jpeg(frame->format)) {
+		if (us_unjpeg(frame, run->h264_tmp_src, true) < 0) {
+			goto done;
+		}
+		frame = run->h264_tmp_src;
+	}
+	if (run->h264_key_requested) {
+		US_LOG_INFO("H264: Requested keyframe by a sink client");
+		run->h264_key_requested = false;
+		force_key = true;
+	}
+	if (!us_m2m_encoder_compress(run->h264_enc, frame, run->h264_dest, force_key)) {
+		meta.online = !us_memsink_server_put(stream->h264_sink, run->h264_dest, &run->h264_key_requested);
+	}
+
+done:
+	us_fpsi_update(run->http->h264_fpsi, meta.online, &meta);
+}
+
 static void _stream_check_suicide(us_stream_s *stream) {
 	if (stream->exit_on_no_clients == 0) {
 		return;
 	}
 	us_stream_runtime_s *const run = stream->run;
+
 	const ldf now_ts = us_get_now_monotonic();
-	const ull http_last_request_ts = atomic_load(&run->http_last_request_ts); // Seconds
+	const ull http_last_request_ts = atomic_load(&run->http->last_request_ts); // Seconds
 	if (_stream_has_any_clients_cached(stream)) {
-		atomic_store(&run->http_last_request_ts, now_ts);
+		atomic_store(&run->http->last_request_ts, now_ts);
 	} else if (http_last_request_ts + stream->exit_on_no_clients < now_ts) {
 		US_LOG_INFO("No requests or HTTP/sink clients found in last %u seconds, exiting ...",
 			stream->exit_on_no_clients);
 		us_process_suicide();
-		atomic_store(&run->http_last_request_ts, now_ts);
+		atomic_store(&run->http->last_request_ts, now_ts);
 	}
 }

src/ustreamer/stream.h

@@ -29,42 +29,52 @@
 #include "../libs/types.h"
 #include "../libs/queue.h"
 #include "../libs/ring.h"
+#include "../libs/frame.h"
 #include "../libs/memsink.h"
 #include "../libs/capture.h"
+#include "../libs/fpsi.h"
 #ifdef WITH_V4P
 #	include "../libs/drm/drm.h"
 #endif
 
 #include "blank.h"
 #include "encoder.h"
-#include "h264.h"
+#include "m2m.h"
 
 
 typedef struct {
-	us_h264_stream_s *h264;
-
 #	ifdef WITH_V4P
-	us_drm_s		*drm;
+	atomic_bool		drm_live;
-	int				drm_opened;
+	us_fpsi_s		*drm_fpsi;
-	ldf				drm_blank_at_ts;
 #	endif
 
-	us_ring_s		*http_jpeg_ring;
+	atomic_bool		h264_online;
-	atomic_bool		http_has_clients;
+	us_fpsi_s		*h264_fpsi;
-	atomic_uint		http_snapshot_requested;
-	atomic_ullong	http_last_request_ts; // Seconds
-	atomic_ullong	http_capture_state; // Bits
 
-	us_blank_s		*blank;
+	us_ring_s		*jpeg_ring;
+	atomic_bool		has_clients;
+	atomic_uint		snapshot_requested;
+	atomic_ullong	last_request_ts; // Seconds
+	us_fpsi_s		*captured_fpsi;
+} us_stream_http_s;
 
-	atomic_bool		stop;
+typedef struct {
+	us_stream_http_s	*http;
+
+	us_m2m_encoder_s	*h264_enc;
+	us_frame_s			*h264_tmp_src;
+	us_frame_s			*h264_dest;
+	bool				h264_key_requested;
+
+	us_blank_s			*blank;
+
+	atomic_bool			stop;
 } us_stream_runtime_s;
 
 typedef struct {
 	us_capture_s	*cap;
 	us_encoder_s	*enc;
 
-	int				last_as_blank;
 	bool			slowdown;
 	uint			error_delay;
 	uint			exit_on_no_clients;
@@ -78,7 +88,7 @@ typedef struct {
 	char			*h264_m2m_path;
 
 #	ifdef WITH_V4P
-	bool			v4p;
+	us_drm_s		*drm;
 #	endif
 
 	us_stream_runtime_s	*run;
@@ -90,5 +100,3 @@ void us_stream_destroy(us_stream_s *stream);
 
 void us_stream_loop(us_stream_s *stream);
 void us_stream_loop_break(us_stream_s *stream);
-
-void us_stream_get_capture_state(us_stream_s *stream, uint *width, uint *height, bool *online, uint *captured_fps);

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) {
+	for (uint index = 0; index < pool->n_workers; ++index) {
-#		define WR(x_next) pool->workers[number].x_next
+		us_worker_s *wr;
+		US_CALLOC(wr, 1);
 
-		WR(number) = number;
+		wr->number = index;
-		US_ASPRINTF(WR(name), "%s-%u", wr_prefix, number);
+		US_ASPRINTF(wr->name, "%s-%u", wr_prefix, index);
 
-		US_MUTEX_INIT(WR(has_job_mutex));
+		US_MUTEX_INIT(wr->has_job_mutex);
-		atomic_init(&WR(has_job), false);
+		atomic_init(&wr->has_job, false);
-		US_COND_INIT(WR(has_job_cond));
+		US_COND_INIT(wr->has_job_cond);
 
-		WR(pool) = pool;
+		wr->pool = pool;
-		WR(job) = job_init(job_init_arg);
+		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) {
+	US_LIST_ITERATE(pool->workers, wr, { // cppcheck-suppress constStatement
-#		define WR(x_next) pool->workers[number].x_next
+		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));
+		US_THREAD_JOIN(wr->tid);
-		atomic_store(&WR(has_job), true); // Final job: die
+		US_MUTEX_DESTROY(wr->has_job_mutex);
-		US_MUTEX_UNLOCK(WR(has_job_mutex));
+		US_COND_DESTROY(wr->has_job_cond);
-		US_COND_SIGNAL(WR(has_job_cond));
 
-		US_THREAD_JOIN(WR(tid));
+		pool->job_destroy(wr->job);
-		US_MUTEX_DESTROY(WR(has_job_mutex));
-		US_COND_DESTROY(WR(has_job_cond));
 
-		free(WR(name));
+		free(wr->name);
-
+		free(wr);
-		pool->job_destroy(WR(job));
+	});
-
-#		undef 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)) {
+	us_worker_s *found = NULL;
-		ready_wr = pool->oldest_wr;
+	US_LIST_ITERATE(pool->workers, wr, { // cppcheck-suppress constStatement
-		ready_wr->job_timely = true;
+		if (!atomic_load(&wr->has_job) && (found == NULL || found->job_start_ts <= wr->job_start_ts)) {
-		pool->oldest_wr = pool->oldest_wr->next_wr;
+			found = 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;
-			}
 		}
-		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*/) {
+void us_workers_pool_assign(us_workers_pool_s *pool, us_worker_s *wr) {
-	if (pool->oldest_wr == NULL) {
+	US_MUTEX_LOCK(wr->has_job_mutex);
-		pool->oldest_wr = ready_wr;
+	atomic_store(&wr->has_job, true);
-		pool->latest_wr = pool->oldest_wr;
+	US_MUTEX_UNLOCK(wr->has_job_mutex);
-	} else {
+	US_COND_SIGNAL(wr->has_job_cond);
-		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);
 
 	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) {
+ldf us_workers_pool_get_fluency_delay(us_workers_pool_s *pool, const us_worker_s *wr) {
-	const long double approx_job_time = pool->approx_job_time * 0.9 + ready_wr->last_job_time * 0.1;
+	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/types.h"
-#include "../libs/threading.h"
+#include "../libs/list.h"
-#include "../libs/logging.h"
 
 
 typedef struct us_worker_sx {
-	pthread_t		tid;
+	pthread_t	tid;
-	unsigned		number;
+	uint		number;
-	char			*name;
+	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;
+	ldf				job_timely_ts;
-	us_worker_s		*latest_wr;
 
-	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);

src/v4p/main.c

@@ -23,6 +23,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <stdatomic.h>
+#include <limits.h>
 #include <string.h>
 #include <unistd.h>
 #include <getopt.h>
@@ -36,6 +37,7 @@
 #include <sys/stat.h>
 
 #include "../libs/types.h"
+#include "../libs/errors.h"
 #include "../libs/const.h"
 #include "../libs/tools.h"
 #include "../libs/logging.h"
@@ -172,18 +174,14 @@ static void _main_loop(void) {
 	cap->dma_required = true;
 
 	int once = 0;
-	ldf blank_at_ts = 0;
-	int drm_opened = -1;
 	while (!atomic_load(&_g_stop)) {
 #		define CHECK(x_arg) if ((x_arg) < 0) { goto close; }
 
-		if (drm_opened <= 0) {
+		if (drm->run->opened <= 0) {
-			blank_at_ts = 0;
+			CHECK(us_drm_open(drm, NULL));
-			CHECK(drm_opened = us_drm_open(drm, NULL));
 		}
 
 		if (atomic_load(&_g_ustreamer_online)) {
-			blank_at_ts = 0;
 			US_ONCE({ US_LOG_INFO("DRM: Online stream is active, pausing the service ..."); });
 			CHECK(us_drm_wait_for_vsync(drm));
 			CHECK(us_drm_expose_stub(drm, US_DRM_STUB_BUSY, NULL));
@@ -192,25 +190,14 @@ static void _main_loop(void) {
 		}
 
 		if (us_capture_open(cap) < 0) {
-			ldf now_ts = us_get_now_monotonic();
+			CHECK(us_drm_ensure_no_signal(drm));
-			if (blank_at_ts == 0) {
-				blank_at_ts = now_ts + 5;
-			}
-			if (now_ts <= blank_at_ts) {
-				CHECK(us_drm_wait_for_vsync(drm));
-				CHECK(us_drm_expose_stub(drm, US_DRM_STUB_NO_SIGNAL, NULL));
-			} else {
-				US_ONCE({ US_LOG_INFO("DRM: Turning off the display by timeout ..."); });
-				CHECK(us_drm_dpms_power_off(drm));
-			}
 			_slowdown();
 			continue;
 		}
 
 		once = 0;
-		blank_at_ts = 0;
 		us_drm_close(drm);
-		CHECK(drm_opened = us_drm_open(drm, cap));
+		CHECK(us_drm_open(drm, cap));
 
 		us_capture_hwbuf_s *prev_hw = NULL;
 		while (!atomic_load(&_g_stop)) {
@@ -227,28 +214,25 @@ static void _main_loop(void) {
 
 			us_capture_hwbuf_s *hw;
 			switch (us_capture_hwbuf_grab(cap, &hw)) {
-				case -2: continue; // Broken frame
+				case 0 ... INT_MAX: break; // Grabbed buffer number
-				case -1: goto close; // Any error
+				case US_ERROR_NO_DATA: continue; // Broken frame
-				default: break; // Grabbed on >= 0
+				default: goto close; // Any error
 			}
 
-			if (drm_opened == 0) {
+			if (drm->run->opened == 0) {
 				CHECK(us_drm_expose_dma(drm, hw));
 				prev_hw = hw;
 				continue;
 			}
 
-			CHECK(us_drm_expose_stub(drm, drm_opened, cap));
+			CHECK(us_drm_expose_stub(drm, drm->run->opened, cap));
 			CHECK(us_capture_hwbuf_release(cap, hw));
 			_slowdown();
 		}
 
 	close:
 		us_drm_close(drm);
-		drm_opened = -1;
-
 		us_capture_close(cap);
-
 		_slowdown();
 
 #		undef CHECK