Bump version: 6.11 → 6.12

Include sys/types.h to avoid the following uclibc build failure since
version 5.52 and
2d6716aa47:

In file included from libs/base64.h:25,
                 from libs/base64.c:23:
libs/types.h:30:9: error: unknown type name 'ssize_t'
   30 | typedef ssize_t sz;
      |         ^~~~~~~

Fixes:
 - http://autobuild.buildroot.org/results/24498049d7beb4afaaf9f9a0c2fc0bcd26a3ee04

Signed-off-by: Fabrice Fontaine <fontaine.fabrice@gmail.com>

.bumpversion.cfg

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

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]
@@ -25,7 +25,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 +33,7 @@ allowlist_externals = bash
 commands = bash -c 'pylint --rcfile=linters/pylint.ini --output-format=colorized --reports=no tools/*.py python/*.py'
 deps =
 	pylint
+	setuptools
 
 [testenv:mypy]
 allowlist_externals = bash

man/ustreamer-dump.1

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

man/ustreamer.1

@@ -1,6 +1,6 @@
 .\" Manpage for ustreamer.
 .\" Open an issue or pull request to https://github.com/pikvm/ustreamer to correct errors or typos
-.TH USTREAMER 1 "version 6.10" "November 2020"
+.TH USTREAMER 1 "version 6.12" "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.10
+pkgver=6.12
 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.10
+PKG_VERSION:=6.12
 PKG_RELEASE:=1
 PKG_MAINTAINER:=Maxim Devaev <mdevaev@gmail.com>
 

python/setup.py

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

src/libs/const.h

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

src/libs/list.h

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

src/libs/types.h

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

src/ustreamer/encoder.c

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

src/ustreamer/encoder.h

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

src/ustreamer/encoders/cpu/encoder.c

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

src/ustreamer/options.c

@@ -651,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("                                             * NOOP  ─────── Don't compress MJPEG stream (do nothing).\n");
+	SAY("                                             * M2M-IMAGE  ── GPU-accelerated JPEG encoding using V4L2 M2M image interface.\n");
 	SAY("    -g|--glitched-resolutions <WxH,...>  ─ It doesn't do anything. Still here for compatibility.\n");
 	SAY("    -k|--blank <path>  ─────────────────── It doesn't do anything. Still here for compatibility.\n");
 	SAY("    -K|--last-as-blank <sec>  ──────────── It doesn't do anything. Still here for compatibility.\n");

src/ustreamer/stream.c

@@ -313,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_encoder_job_s *const ready_job = ready_wr->job;
+		us_worker_s *const wr = us_workers_pool_wait(stream->enc->run->pool);
+		us_encoder_job_s *const job = wr->job;
 
-		if (ready_job->hw != NULL) {
-			us_capture_hwbuf_decref(ready_job->hw);
-			ready_job->hw = NULL;
-			if (ready_wr->job_failed) {
+		if (job->hw != NULL) {
+			us_capture_hwbuf_decref(job->hw);
+			job->hw = NULL;
+			if (wr->job_failed) {
 				// pass
-			} else if (ready_wr->job_timely) {
-				_stream_expose_jpeg(stream, ready_job->dest);
+			} else if (wr->job_timely) {
+				_stream_expose_jpeg(stream, job->dest);
 				if (atomic_load(&stream->run->http->snapshot_requested) > 0) { // Process real snapshots
 					atomic_fetch_sub(&stream->run->http->snapshot_requested, 1);
 				}
 				US_LOG_PERF("JPEG: ##### Encoded JPEG exposed; worker=%s, latency=%.3Lf",
-					ready_wr->name, us_get_now_monotonic() - ready_job->dest->grab_ts);
+					wr->name, us_get_now_monotonic() - job->dest->grab_ts);
 			} else {
-				US_LOG_PERF("JPEG: ----- Encoded JPEG dropped; worker=%s", ready_wr->name);
+				US_LOG_PERF("JPEG: ----- Encoded JPEG dropped; worker=%s", wr->name);
 			}
 		}
 
@@ -355,13 +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;
-		us_workers_pool_assign(stream->enc->run->pool, ready_wr);
-		US_LOG_DEBUG("JPEG: Assigned new frame in buffer=%d to worker=%s", hw->buf.index, ready_wr->name);
+		job->hw = hw;
+		us_workers_pool_assign(stream->enc->run->pool, wr);
+		US_LOG_DEBUG("JPEG: Assigned new frame in buffer=%d to worker=%s", hw->buf.index, wr->name);
 	}
 	return NULL;
 }

src/ustreamer/workers.c

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

src/ustreamer/workers.h

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