Bump version: 3.8 → 3.9

.bumpversion.cfg

@@ -1,7 +1,7 @@
 [bumpversion]
 commit = True
 tag = True
-current_version = 3.4
+current_version = 3.9
 parse = (?P<major>\d+)\.(?P<minor>\d+)(\.(?P<patch>\d+)(\-(?P<release>[a-z]+))?)?
 serialize = 
 	{major}.{minor}
@@ -10,6 +10,10 @@ serialize =
 search = VERSION "{current_version}"
 replace = VERSION "{new_version}"
 
+[bumpversion:file:python/setup.py]
+search = version="{current_version}"
+replace = version="{new_version}"
+
 [bumpversion:file:pkg/arch/PKGBUILD]
 search = pkgver={current_version}
 replace = pkgver={new_version}

.gitignore

@@ -6,8 +6,10 @@
 /pkg/arch/ustreamer-*.pkg.tar.xz
 /pkg/arch/ustreamer-*.pkg.tar.zst
 /build/
+/python/build/
 /config.mk
 /vgcore.*
 /ustreamer
 /ustreamer-dump
+/*.so
 /*.sock

Makefile

@@ -7,7 +7,8 @@ PREFIX ?= /usr/local
 MANPREFIX ?= $(PREFIX)/share/man
 
 CC ?= gcc
-CFLAGS ?= -O3 -MD
+PY ?= python3
+CFLAGS ?= -O3
 LDFLAGS ?=
 
 RPI_VC_HEADERS ?= /opt/vc/include
@@ -19,7 +20,8 @@ LINTERS_IMAGE ?= $(USTR)-linters
 
 
 # =====
-override CFLAGS += -c -std=c11 -Wall -Wextra -D_GNU_SOURCE
+_CFLAGS = -MD -c -std=c11 -Wall -Wextra -D_GNU_SOURCE $(CFLAGS)
+_LDFLAGS = $(LDFLAGS)
 
 _COMMON_LIBS = -lm -ljpeg -pthread -lrt
 
@@ -47,7 +49,7 @@ endef
 
 ifneq ($(call optbool,$(WITH_OMX)),)
 _USTR_LIBS += -lbcm_host -lvcos -lvcsm -lopenmaxil -lmmal -lmmal_core -lmmal_util -lmmal_vc_client -lmmal_components -L$(RPI_VC_LIBS)
-override CFLAGS += -DWITH_OMX -DOMX_SKIP64BIT -I$(RPI_VC_HEADERS)
+override _CFLAGS += -DWITH_OMX -DOMX_SKIP64BIT -I$(RPI_VC_HEADERS)
 _USTR_SRCS += $(shell ls \
 	src/ustreamer/encoders/omx/*.c \
 	src/ustreamer/h264/*.c \
@@ -57,14 +59,14 @@ endif
 
 ifneq ($(call optbool,$(WITH_GPIO)),)
 _USTR_LIBS += -lgpiod
-override CFLAGS += -DWITH_GPIO
+override _CFLAGS += -DWITH_GPIO
 _USTR_SRCS += $(shell ls src/ustreamer/gpio/*.c)
 endif
 
 
 WITH_PTHREAD_NP ?= 1
 ifneq ($(call optbool,$(WITH_PTHREAD_NP)),)
-override CFLAGS += -DWITH_PTHREAD_NP
+override _CFLAGS += -DWITH_PTHREAD_NP
 endif
 
 
@@ -73,12 +75,12 @@ ifneq ($(call optbool,$(WITH_SETPROCTITLE)),)
 ifeq ($(shell uname -s | tr A-Z a-z),linux)
 _USTR_LIBS += -lbsd
 endif
-override CFLAGS += -DWITH_SETPROCTITLE
+override _CFLAGS += -DWITH_SETPROCTITLE
 endif
 
 
 # =====
-all: $(USTR) $(DUMP)
+all: $(USTR) $(DUMP) python
 
 
 install: $(USTR) $(DUMP)
@@ -87,8 +89,11 @@ install: $(USTR) $(DUMP)
 	install -m755 $(DUMP) $(DESTDIR)$(PREFIX)/bin/$(DUMP)
 	install -m644 man/$(USTR).1 $(DESTDIR)$(MANPREFIX)/man1/$(USTR).1
 	install -m644 man/$(DUMP).1 $(DESTDIR)$(MANPREFIX)/man1/$(DUMP).1
-	gzip $(DESTDIR)$(MANPREFIX)/man1/$(USTR).1
-	gzip $(DESTDIR)$(MANPREFIX)/man1/$(DUMP).1
+	gzip -f $(DESTDIR)$(MANPREFIX)/man1/$(USTR).1
+	gzip -f $(DESTDIR)$(MANPREFIX)/man1/$(DUMP).1
+ifneq ($(call optbool,$(WITH_PYTHON)),)
+	cd python && $(PY) setup.py install --prefix=$(PREFIX) --root=$(if $(DESTDIR),$(DESTDIR),/)
+endif
 
 
 install-strip: install
@@ -96,13 +101,6 @@ install-strip: install
 	strip $(DESTDIR)$(PREFIX)/bin/$(DUMP)
 
 
-uninstall:
-	rm -f $(DESTDIR)$(PREFIX)/bin/$(USTR) \
-		$(DESTDIR)$(PREFIX)/bin/$(DUMP) \
-		$(DESTDIR)$(MANPREFIX)/man1/$(USTR).1 \
-		$(DESTDIR)$(MANPREFIX)/man1/$(DUMP).1
-
-
 regen:
 	tools/make-jpeg-h.py src/ustreamer/data/blank.jpeg src/ustreamer/data/blank_jpeg.c BLANK
 	tools/make-html-h.py src/ustreamer/data/index.html src/ustreamer/data/index_html.c INDEX
@@ -111,29 +109,38 @@ regen:
 $(USTR): $(_USTR_SRCS:%.c=$(BUILD)/%.o)
 #	$(info ========================================)
 	$(info == LD $@)
-	@ $(CC) $^ -o $@ $(LDFLAGS) $(_USTR_LIBS)
+	@ $(CC) $^ -o $@ $(_LDFLAGS) $(_USTR_LIBS)
 #	$(info :: CC      = $(CC))
 #	$(info :: LIBS    = $(_USTR_LIBS))
-#	$(info :: CFLAGS  = $(CFLAGS))
-#	$(info :: LDFLAGS = $(LDFLAGS))
+#	$(info :: CFLAGS  = $(_CFLAGS))
+#	$(info :: LDFLAGS = $(_LDFLAGS))
 
 
 $(DUMP): $(_DUMP_SRCS:%.c=$(BUILD)/%.o)
 #	$(info ========================================)
 	$(info == LD $@)
-	@ $(CC) $^ -o $@ $(LDFLAGS) $(_DUMP_LIBS)
+	@ $(CC) $^ -o $@ $(_LDFLAGS) $(_DUMP_LIBS)
 #	$(info :: CC      = $(CC))
 #	$(info :: LIBS    = $(_DUMP_LIBS))
-#	$(info :: CFLAGS  = $(CFLAGS))
-#	$(info :: LDFLAGS = $(LDFLAGS))
+#	$(info :: CFLAGS  = $(_CFLAGS))
+#	$(info :: LDFLAGS = $(_LDFLAGS))
 
 
 $(BUILD)/%.o: %.c
 	$(info -- CC $<)
 	@ mkdir -p $(dir $@) || true
-	@ $(CC) $< -o $@ $(CFLAGS)
+	@ $(CC) $< -o $@ $(_CFLAGS)
 
 
+python:
+ifneq ($(call optbool,$(WITH_PYTHON)),)
+	$(info == PY_BUILD ustreamer-*.so)
+	@ cd python && $(PY) setup.py build
+	@ ln -sf python/build/lib.*/*.so .
+else
+	@ true
+endif
+
 release:
 	make clean
 	make tox
@@ -176,10 +183,10 @@ clean-all: linters clean
 		-it $(LINTERS_IMAGE) bash -c "cd src && rm -rf linters/{.tox,.mypy_cache}"
 clean:
 	rm -rf pkg/arch/pkg pkg/arch/src pkg/arch/v*.tar.gz pkg/arch/ustreamer-*.pkg.tar.{xz,zst}
-	rm -rf $(USTR) $(DUMP) $(BUILD) vgcore.* *.sock
+	rm -rf $(USTR) $(DUMP) $(BUILD) python/build vgcore.* *.sock *.so
 
 
-.PHONY: linters
+.PHONY: python linters
 
 
 _OBJS = $(_USTR_SRCS:%.c=$(BUILD)/%.o) $(_DUMP_SRCS:%.c=$(BUILD)/%.o)

linters/tox.ini

@@ -1,12 +1,12 @@
 [tox]
-envlist = cppcheck, flake8, pylint, mypy, vulture, htmlhint
+envlist = cppcheck-src, cppcheck-python, flake8, pylint, mypy, vulture, htmlhint
 skipsdist = true
 
 [testenv]
 basepython = python3.9
 changedir = /src
 
-[testenv:cppcheck]
+[testenv:cppcheck-src]
 whitelist_externals = cppcheck
 commands = cppcheck \
 			--force \
@@ -17,33 +17,49 @@ commands = cppcheck \
 			--suppress=assignmentInAssert \
 			--suppress=variableScope \
 			--inline-suppr \
+			--library=python \
 			-DCHAR_BIT=8 \
 			-DWITH_OMX \
 			-DWITH_GPIO \
-		src
+		src python
+
+[testenv:cppcheck-python]
+whitelist_externals = cppcheck
+commands = cppcheck \
+			--force \
+			--std=c11 \
+			--error-exitcode=1 \
+			--quiet \
+			--enable=warning,unusedFunction,portability,performance,style \
+			--suppress=assignmentInAssert \
+			--suppress=variableScope \
+			--inline-suppr \
+			--library=python \
+			-DCHAR_BIT=8 \
+		python
 
 [testenv:flake8]
 whitelist_externals = bash
-commands = bash -c 'flake8 --config=linters/flake8.ini tools/*.py'
+commands = bash -c 'flake8 --config=linters/flake8.ini tools/*.py' python/*.py
 deps =
 	flake8
 	flake8-quotes
 
 [testenv:pylint]
 whitelist_externals = bash
-commands = bash -c 'pylint --rcfile=linters/pylint.ini --output-format=colorized --reports=no tools/*.py'
+commands = bash -c 'pylint --rcfile=linters/pylint.ini --output-format=colorized --reports=no tools/*.py python/*.py'
 deps =
 	pylint
 
 [testenv:mypy]
 whitelist_externals = bash
-commands = bash -c 'mypy --config-file=linters/mypy.ini tools/*.py'
+commands = bash -c 'mypy --config-file=linters/mypy.ini tools/*.py python/*.py'
 deps =
 	mypy
 
 [testenv:vulture]
 whitelist_externals = bash
-commands = bash -c 'vulture tools/*.py'
+commands = bash -c 'vulture tools/*.py python/*.py'
 deps =
 	vulture
 

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 3.4" "January 2021"
+.TH USTREAMER-DUMP 1 "version 3.9" "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 3.4" "November 2020"
+.TH USTREAMER 1 "version 3.9" "November 2020"
 
 .SH NAME
 ustreamer \- stream MJPG video from any V4L2 device to the network
@@ -210,6 +210,9 @@ Set JPEG sink permissions (like 777). Default: 660.
 .BR \-\-sink\-rm
 Remove shared memory on stop. Default: disabled.
 .TP
+.BR \-\-sink\-client\-ttl\ \fIsec
+Client TTL. Default: 10.
+.TP
 .BR \-\-sink\-timeout\ \fIsec
 Timeout for lock. Default: 1.
 
@@ -226,6 +229,9 @@ Set H264 sink permissions (like 777). Default: 660.
 .BR \-\-h264\-sink\-rm
 Remove shared memory on stop. Default: disabled.
 .TP
+.BR \-\-h264\-sink\-client\-ttl\ \fIsec
+Client TTL. Default: 10.
+.TP
 .BR \-\-h264\-sink\-timeout\ \fIsec
 Timeout for lock. Default: 1.
 

pkg/arch/PKGBUILD

@@ -3,28 +3,37 @@
 
 
 pkgname=ustreamer
-pkgver=3.4
+pkgver=3.9
 pkgrel=1
 pkgdesc="Lightweight and fast MJPG-HTTP streamer"
 url="https://github.com/pikvm/ustreamer"
 license=(GPL)
 arch=(i686 x86_64 armv6h armv7h aarch64)
 depends=(libjpeg libevent libutil-linux libbsd libgpiod)
-# optional: raspberrypi-firmware for OMX encoder
 makedepends=(gcc make)
 source=(${pkgname}::"git+https://github.com/pikvm/ustreamer#commit=v${pkgver}")
 md5sums=(SKIP)
 
 
+_options="WITH_GPIO=1"
+if [ -e /usr/bin/python3 ]; then
+	_options="$_options WITH_PYTHON=1"
+	depends+=(python)
+	makedepends+=(python-setuptools)
+fi
+if [ -e /opt/vc/include/IL/OMX_Core.h ]; then
+	depends+=(raspberrypi-firmware)
+	makedepends+=(raspberrypi-firmware)
+	_options="$_options WITH_OMX=1"
+fi
+
+
 build() {
 	cd "$srcdir"
 	rm -rf $pkgname-build
 	cp -r $pkgname $pkgname-build
 	cd $pkgname-build
 
-	local _options="WITH_GPIO=1"
-	[ -e /opt/vc/include/IL/OMX_Core.h ] && _options="$_options WITH_OMX=1"
-
 	# LD does not link mmal with this option
 	LDFLAGS="${LDFLAGS//--as-needed/}"
 	LDFLAGS="${LDFLAGS//,,/,}"
@@ -34,5 +43,5 @@ build() {
 
 package() {
 	cd "$srcdir/$pkgname-build"
-	make DESTDIR="$pkgdir" PREFIX=/usr install
+	make $_options DESTDIR="$pkgdir" PREFIX=/usr install
 }

pkg/openwrt/Makefile

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

python/setup.py

@@ -0,0 +1,26 @@
+from distutils.core import Extension
+from distutils.core import setup
+
+
+# =====
+if __name__ == "__main__":
+    setup(
+        name="ustreamer",
+        version="3.9",
+        description="uStreamer tools",
+        author="Maxim Devaev",
+        author_email="mdevaev@gmail.com",
+        url="https://github.com/pikvm/ustreamer",
+        ext_modules=[
+            Extension(
+                "ustreamer",
+                libraries=["rt", "m", "pthread"],
+                undef_macros=["NDEBUG"],
+                sources=["ustreamer.c"],
+                depends=[
+                    "../src/libs/tools.h",
+                    "../src/libs/memsinksh.h",
+                ],
+            ),
+        ],
+    )

python/ustreamer.c

@@ -0,0 +1,325 @@
+#include <stdio.h>
+#include <stdint.h>
+#include <stdbool.h>
+#include <stdlib.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <assert.h>
+
+#include <sys/types.h>
+#include <sys/file.h>
+#include <sys/stat.h>
+#include <sys/mman.h>
+
+#include <Python.h>
+
+#include "../src/libs/tools.h" // Just a header without C-sources
+#include "../src/libs/memsinksh.h" // No sources again
+
+
+typedef struct {
+	PyObject_HEAD
+
+	char	*obj;
+	double	lock_timeout;
+	double	wait_timeout;
+
+	int					fd;
+	memsink_shared_s	*mem;
+	uint8_t				*tmp_data;
+	size_t				tmp_data_allocated;
+	uint64_t			last_id;
+
+	PyObject *frame; // PyDict
+} MemsinkObject;
+
+
+static void MemsinkObject_destroy_internals(MemsinkObject *self) {
+	if (self->frame != NULL) {
+		Py_DECREF(self->frame);
+		self->frame = NULL;
+	}
+	if (self->mem != NULL) {
+		munmap(self->mem, sizeof(memsink_shared_s));
+		self->mem = NULL;
+	}
+	if (self->fd > 0) {
+		close(self->fd);
+		self->fd = -1;
+	}
+	if (self->tmp_data) {
+		free(self->tmp_data);
+		self->tmp_data = NULL;
+		self->tmp_data_allocated = 0;
+	}
+}
+
+static int MemsinkObject_init(MemsinkObject *self, PyObject *args, PyObject *kwargs) {
+	self->lock_timeout = 1;
+	self->wait_timeout = 1;
+
+	static char *kws[] = {"obj", "lock_timeout", "wait_timeout", NULL};
+	if (!PyArg_ParseTupleAndKeywords(
+		args, kwargs, "s|dd", kws,
+		&self->obj, &self->lock_timeout, &self->wait_timeout)) {
+		return -1;
+	}
+
+#	define SET_TIMEOUT(_timeout) { \
+			if (self->_timeout <= 0) { \
+				PyErr_SetString(PyExc_ValueError, #_timeout " must be > 0"); \
+				return -1; \
+			} \
+		}
+
+	SET_TIMEOUT(lock_timeout);
+	SET_TIMEOUT(wait_timeout);
+
+#	undef CHECK_TIMEOUT
+
+	self->tmp_data_allocated = 512 * 1024;
+	A_REALLOC(self->tmp_data, self->tmp_data_allocated);
+
+	if ((self->fd = shm_open(self->obj, O_RDWR, 0)) == -1) {
+		PyErr_SetFromErrno(PyExc_OSError);
+		goto error;
+	}
+
+	if ((self->mem = mmap(
+		NULL,
+		sizeof(memsink_shared_s),
+		PROT_READ | PROT_WRITE,
+		MAP_SHARED,
+		self->fd,
+		0
+	)) == MAP_FAILED) {
+		PyErr_SetFromErrno(PyExc_OSError);
+		self->mem = NULL;
+		goto error;
+	}
+	if (self->mem == NULL) {
+		PyErr_SetString(PyExc_RuntimeError, "Memory mapping is NULL"); \
+		goto error;
+	}
+
+	if ((self->frame = PyDict_New()) == NULL) {
+		goto error;
+	}
+
+	return 0;
+
+	error:
+		MemsinkObject_destroy_internals(self);
+		return -1;
+}
+
+static PyObject *MemsinkObject_repr(MemsinkObject *self) {
+	char repr[1024];
+	snprintf(repr, 1023, "<Memsink(%s)>", self->obj);
+	return Py_BuildValue("s", repr);
+}
+
+static void MemsinkObject_dealloc(MemsinkObject *self) {
+	MemsinkObject_destroy_internals(self);
+	PyObject_Del(self);
+}
+
+static PyObject *MemsinkObject_close(MemsinkObject *self, PyObject *Py_UNUSED(ignored)) {
+	MemsinkObject_destroy_internals(self);
+	Py_RETURN_NONE;
+}
+
+static PyObject *MemsinkObject_enter(MemsinkObject *self, PyObject *Py_UNUSED(ignored)) {
+	Py_INCREF(self);
+	return (PyObject *)self;
+}
+
+static PyObject *MemsinkObject_exit(MemsinkObject *self, PyObject *Py_UNUSED(ignored)) {
+	return PyObject_CallMethod((PyObject *)self, "close", "");
+}
+
+#define MEM(_next) self->mem->_next
+
+static int wait_frame(MemsinkObject *self) {
+	long double deadline_ts = get_now_monotonic() + self->wait_timeout;
+
+#	define RETURN_OS_ERROR { \
+			Py_BLOCK_THREADS \
+			PyErr_SetFromErrno(PyExc_OSError); \
+			return -1; \
+		}
+
+	do {
+		Py_BEGIN_ALLOW_THREADS
+
+		int retval = flock_timedwait_monotonic(self->fd, self->lock_timeout);
+		if (retval < 0 && errno != EWOULDBLOCK) {
+			RETURN_OS_ERROR;
+		} else if (retval == 0) {
+			if (MEM(magic) == MEMSINK_MAGIC && MEM(version) == MEMSINK_VERSION && MEM(id) != self->last_id) {
+				Py_BLOCK_THREADS
+				return 0;
+			}
+			if (flock(self->fd, LOCK_UN) < 0) {
+				RETURN_OS_ERROR;
+			}
+		}
+		if (usleep(1000) < 0) {
+			RETURN_OS_ERROR;
+		}
+
+		Py_END_ALLOW_THREADS
+
+		if (PyErr_CheckSignals() < 0) {
+			return -1;
+		}
+	} while (get_now_monotonic() < deadline_ts);
+
+#	undef RETURN_OS_ERROR
+
+	return -2;
+}
+
+static PyObject *MemsinkObject_wait_frame(MemsinkObject *self, PyObject *Py_UNUSED(ignored)) {
+	if (self->mem == NULL || self->fd <= 0) {
+		PyErr_SetString(PyExc_RuntimeError, "Closed");
+		return NULL;
+	}
+
+	switch (wait_frame(self)) {
+		case 0: break;
+		case -2: Py_RETURN_NONE;
+		default: return NULL;
+	}
+
+#	define COPY(_type, _field) _type tmp_##_field = MEM(_field)
+	COPY(unsigned, width);
+	COPY(unsigned, height);
+	COPY(unsigned, format);
+	COPY(unsigned, stride);
+	COPY(bool, online);
+	COPY(double, grab_ts);
+	COPY(double, encode_begin_ts);
+	COPY(double, encode_end_ts);
+	COPY(unsigned, used);
+#	undef COPY
+
+	// Временный буффер используется для скорейшего разблокирования синка
+	if (self->tmp_data_allocated < MEM(used)) {
+		size_t size = MEM(used) + (512 * 1024);
+		A_REALLOC(self->tmp_data, size);
+		self->tmp_data_allocated = size;
+	}
+	memcpy(self->tmp_data, MEM(data), MEM(used));
+
+	MEM(last_client_ts) = get_now_monotonic();
+	self->last_id = MEM(id);
+
+	if (flock(self->fd, LOCK_UN) < 0) {
+		return PyErr_SetFromErrno(PyExc_OSError);
+	}
+
+	PyDict_Clear(self->frame);
+
+#	define SET_VALUE(_key, _maker) { \
+			PyObject *_tmp = _maker; \
+			if (_tmp == NULL) { \
+				return NULL; \
+			} \
+			if (PyDict_SetItemString(self->frame, _key, _tmp) < 0) { \
+				Py_DECREF(_tmp); \
+				return NULL; \
+			} \
+			Py_DECREF(_tmp); \
+		}
+
+	SET_VALUE("width", PyLong_FromLong(tmp_width));
+	SET_VALUE("height", PyLong_FromLong(tmp_height));
+	SET_VALUE("format", PyLong_FromLong(tmp_format));
+	SET_VALUE("stride", PyLong_FromLong(tmp_stride));
+	SET_VALUE("online", PyBool_FromLong(tmp_online));
+	SET_VALUE("grab_ts", PyFloat_FromDouble(tmp_grab_ts));
+	SET_VALUE("encode_begin_ts", PyFloat_FromDouble(tmp_encode_begin_ts));
+	SET_VALUE("encode_end_ts", PyFloat_FromDouble(tmp_encode_end_ts));
+	SET_VALUE("data", PyBytes_FromStringAndSize((const char *)self->tmp_data, tmp_used));
+
+#	undef SET_VALUE
+
+	Py_INCREF(self->frame);
+	return self->frame;
+}
+
+#undef MEM
+
+static PyObject *MemsinkObject_is_opened(MemsinkObject *self, PyObject *Py_UNUSED(ignored)) {
+	return PyBool_FromLong(self->mem != NULL && self->fd > 0);
+}
+
+#define FIELD_GETTER(_field, _from, _to) \
+	static PyObject *MemsinkObject_getter_##_field(MemsinkObject *self, void *Py_UNUSED(closure)) { \
+		return Py##_to##_From##_from(self->_field); \
+	}
+
+FIELD_GETTER(obj, String, Unicode)
+FIELD_GETTER(lock_timeout, Double, Float)
+FIELD_GETTER(wait_timeout, Double, Float)
+
+#undef FIELD_GETTER
+
+static PyMethodDef MemsinkObject_methods[] = {
+#	define ADD_METHOD(_meth, _flags) {.ml_name = #_meth, .ml_meth = (PyCFunction)MemsinkObject_##_meth, .ml_flags = (_flags)}
+	ADD_METHOD(close, METH_NOARGS),
+	ADD_METHOD(enter, METH_NOARGS),
+	ADD_METHOD(exit, METH_VARARGS),
+	ADD_METHOD(wait_frame, METH_NOARGS),
+	ADD_METHOD(is_opened, METH_NOARGS),
+	{},
+#	undef ADD_METHOD
+};
+
+static PyGetSetDef MemsinkObject_getsets[] = {
+#	define ADD_GETTER(_field) {.name = #_field, .get = (getter)MemsinkObject_getter_##_field}
+	ADD_GETTER(obj),
+	ADD_GETTER(lock_timeout),
+	ADD_GETTER(wait_timeout),
+	{},
+#	undef ADD_GETTER
+};
+
+static PyTypeObject MemsinkType = {
+	PyVarObject_HEAD_INIT(NULL, 0)
+	.tp_name		= "ustreamer.Memsink",
+	.tp_basicsize	= sizeof(MemsinkObject),
+	.tp_flags		= Py_TPFLAGS_DEFAULT,
+	.tp_new			= PyType_GenericNew,
+	.tp_init		= (initproc)MemsinkObject_init,
+	.tp_dealloc		= (destructor)MemsinkObject_dealloc,
+	.tp_repr		= (reprfunc)MemsinkObject_repr,
+	.tp_methods		= MemsinkObject_methods,
+	.tp_getset		= MemsinkObject_getsets,
+};
+
+static PyModuleDef ustreamer_Module = {
+	PyModuleDef_HEAD_INIT,
+	.m_name = "ustreamer",
+	.m_size = -1,
+};
+
+PyMODINIT_FUNC PyInit_ustreamer(void) { // cppcheck-suppress unusedFunction
+	PyObject *module = PyModule_Create(&ustreamer_Module);
+	if (module == NULL) {
+		return NULL;
+	}
+
+	if (PyType_Ready(&MemsinkType) < 0) {
+		return NULL;
+	}
+
+	Py_INCREF(&MemsinkType);
+
+	if (PyModule_AddObject(module, "Memsink", (PyObject *)&MemsinkType) < 0) {
+		return NULL;
+	}
+
+	return module;
+}

src/dump/file.c

@@ -0,0 +1,79 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPG-HTTP streamer.                    #
+#                                                                            #
+#    Copyright (C) 2018-2021  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 "file.h"
+
+
+output_file_s *output_file_init(const char *path, bool json) {
+	output_file_s *output;
+	A_CALLOC(output, 1);
+
+	if (!strcmp(path, "-")) {
+		LOG_INFO("Using output: <stdout>");
+		output->fp = stdout;
+	} else {
+		LOG_INFO("Using output: %s", path);
+		if ((output->fp = fopen(path, "wb")) == NULL) {
+			LOG_PERROR("Can't open output file");
+			goto error;
+		}
+	}
+
+	output->json = json;
+	return output;
+
+	error:
+		output_file_destroy(output);
+		return NULL;
+}
+
+void output_file_write(void *v_output, const frame_s *frame) {
+	output_file_s *output = (output_file_s *)v_output;
+	if (output->json) {
+		base64_encode(frame->data, frame->used, &output->base64_data, &output->base64_allocated);
+		fprintf(output->fp,
+			"{\"size\": %zu, \"width\": %u, \"height\": %u,"
+			" \"format\": %u, \"stride\": %u, \"online\": %u,"
+			" \"grab_ts\": %.3Lf, \"encode_begin_ts\": %.3Lf, \"encode_end_ts\": %.3Lf,"
+			" \"data\": \"%s\"}\n",
+			frame->used, frame->width, frame->height,
+			frame->format, frame->stride, frame->online,
+			frame->grab_ts, frame->encode_begin_ts, frame->encode_end_ts,
+			output->base64_data);
+	} else {
+		fwrite(frame->data, 1, frame->used, output->fp);
+	}
+	fflush(output->fp);
+}
+
+void output_file_destroy(void *v_output) {
+	output_file_s *output = (output_file_s *)v_output;
+	if (output->base64_data) {
+		free(output->base64_data);
+	}
+	if (output->fp && output->fp != stdout) {
+		if (fclose(output->fp) < 0) {
+			LOG_PERROR("Can't close output file");
+		}
+	}
+	free(output);
+}

src/dump/file.h

@@ -0,0 +1,49 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPG-HTTP streamer.                    #
+#                                                                            #
+#    Copyright (C) 2018-2021  Maxim Devaev <mdevaev@gmail.com>               #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#pragma once
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdbool.h>
+
+#include <sys/types.h>
+
+#include "../libs/tools.h"
+#include "../libs/logging.h"
+#include "../libs/frame.h"
+#include "../libs/base64.h"
+
+
+typedef struct {
+	const char *path;
+	bool		json;
+
+	FILE		*fp;
+	char		*base64_data;
+	size_t		base64_allocated;
+} output_file_s;
+
+
+output_file_s *output_file_init(const char *path, bool json);
+void output_file_write(void *v_output, const frame_s *frame);
+void output_file_destroy(void *v_output);

src/dump/main.c

@@ -23,6 +23,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <stdbool.h>
+#include <unistd.h>
 #include <signal.h>
 #include <getopt.h>
 #include <errno.h>
@@ -33,7 +34,9 @@
 #include "../libs/logging.h"
 #include "../libs/frame.h"
 #include "../libs/memsink.h"
-#include "../libs/base64.h"
+#include "../libs/options.h"
+
+#include "file.h"
 
 
 enum _OPT_VALUES {
@@ -73,13 +76,20 @@ static const struct option _LONG_OPTS[] = {
 };
 
 
-volatile bool stop = false;
+volatile bool global_stop = false;
+
+
+typedef struct {
+	void *v_output;
+	void (*write)(void *v_output, const frame_s *frame);
+	void (*destroy)(void *v_output);
+} _output_context_s;
 
 
 static void _signal_handler(int signum);
 static void _install_signal_handlers(void);
 
-static int _dump_sink(const char *sink_name, unsigned sink_timeout, const char *output_path, bool output_json);
+static int _dump_sink(const char *sink_name, unsigned sink_timeout, _output_context_s *ctx);
 
 static void _help(FILE *fp);
 
@@ -108,7 +118,10 @@ int main(int argc, char *argv[]) {
 			break; \
 		}
 
-	for (int ch; (ch = getopt_long(argc, argv, "s:t:o:jhv", _LONG_OPTS, NULL)) >= 0;) {
+	char short_opts[128];
+	build_short_options(_LONG_OPTS, short_opts, 128);
+
+	for (int ch; (ch = getopt_long(argc, argv, short_opts, _LONG_OPTS, NULL)) >= 0;) {
 		switch (ch) {
 			case _O_SINK:			OPT_SET(sink_name, optarg);
 			case _O_SINK_TIMEOUT:	OPT_NUMBER("--sink-timeout", sink_timeout, 1, 60, 0);
@@ -126,7 +139,7 @@ int main(int argc, char *argv[]) {
 			case _O_VERSION:	puts(VERSION); return 0;
 
 			case 0:		break;
-			default:	_help(stderr); return 1;
+			default:	return 1;
 		}
 	}
 
@@ -138,8 +151,23 @@ int main(int argc, char *argv[]) {
 		return 1;
 	}
 
+	_output_context_s ctx;
+	MEMSET_ZERO(ctx);
+
+	if (output_path && output_path[0] != '\0') {
+		if ((ctx.v_output = (void *)output_file_init(output_path, output_json)) == NULL) {
+			return 1;
+		}
+		ctx.write = output_file_write;
+		ctx.destroy = output_file_destroy;
+	}
+
 	_install_signal_handlers();
-	return abs(_dump_sink(sink_name, sink_timeout, output_path, output_json));
+	int retval = abs(_dump_sink(sink_name, sink_timeout, &ctx));
+	if (ctx.v_output && ctx.destroy) {
+		ctx.destroy(ctx.v_output);
+	}
+	return retval;
 }
 
 
@@ -150,7 +178,7 @@ static void _signal_handler(int signum) {
 		case SIGPIPE:	LOG_INFO_NOLOCK("===== Stopping by SIGPIPE ====="); break;
 		default:		LOG_INFO_NOLOCK("===== Stopping by %d =====", signum); break;
 	}
-	stop = true;
+	global_stop = true;
 }
 
 static void _install_signal_handlers(void) {
@@ -173,27 +201,11 @@ static void _install_signal_handlers(void) {
 	assert(!sigaction(SIGPIPE, &sig_act, NULL));
 }
 
-static int _dump_sink(const char *sink_name, unsigned sink_timeout, const char *output_path, bool output_json) {
+static int _dump_sink(const char *sink_name, unsigned sink_timeout, _output_context_s *ctx) {
 	frame_s *frame = frame_init("input");
 	memsink_s *sink = NULL;
-	FILE *output_fp = NULL;
-	char *base64_data = NULL;
-	size_t base64_allocated = 0;
 
-	if (output_path && output_path[0] != '\0') {
-		if (!strcmp(output_path, "-")) {
-			LOG_INFO("Using output: <stdout>");
-			output_fp = stdout;
-		} else {
-			LOG_INFO("Using output: %s", output_path);
-			if ((output_fp = fopen(output_path, "wb")) == NULL) {
-				LOG_PERROR("Can't open output file");
-				goto error;
-			}
-		}
-	}
-
-	if ((sink = memsink_init("input", sink_name, false, 0, false, sink_timeout)) == NULL) {
+	if ((sink = memsink_init("input", sink_name, false, 0, false, 0, sink_timeout)) == NULL) {
 		goto error;
 	}
 
@@ -201,7 +213,7 @@ static int _dump_sink(const char *sink_name, unsigned sink_timeout, const char *
 	unsigned fps_accum = 0;
 	long long fps_second = 0;
 
-	while (!stop) {
+	while (!global_stop) {
 		int error = memsink_client_get(sink, frame);
 		if (error == 0) {
 			const long double now = get_now_monotonic();
@@ -225,24 +237,12 @@ static int _dump_sink(const char *sink_name, unsigned sink_timeout, const char *
 			}
 			fps_accum += 1;
 
-			if (output_fp) {
-				if (output_json) {
-					base64_encode(frame->data, frame->used, &base64_data, &base64_allocated);
-					fprintf(output_fp,
-						"{\"size\": %zu, \"width\": %u, \"height\": %u,"
-						" \"format\": %u, \"stride\": %u, \"online\": %u,"
-						" \"grab_ts\": %.3Lf, \"encode_begin_ts\": %.3Lf, \"encode_end_ts\": %.3Lf,"
-						" \"data\": \"%s\"}\n",
-						frame->used, frame->width, frame->height,
-						frame->format, frame->stride, frame->online,
-						frame->grab_ts, frame->encode_begin_ts, frame->encode_end_ts,
-						base64_data);
-				} else {
-					fwrite(frame->data, 1, frame->used, output_fp);
-				}
-				fflush(output_fp);
+			if (ctx->v_output) {
+				ctx->write(ctx->v_output, frame);
 			}
-		} else if (error != -2) {
+		} else if (error == -2) {
+			usleep(1000);
+		} else {
 			goto error;
 		}
 	}
@@ -254,14 +254,6 @@ static int _dump_sink(const char *sink_name, unsigned sink_timeout, const char *
 		retval = -1;
 
 	ok:
-		if (base64_data) {
-			free(base64_data);
-		}
-		if (output_fp && output_fp != stdout) {
-			if (fclose(output_fp) < 0) {
-				LOG_PERROR("Can't close output file");
-			}
-		}
 		if (sink) {
 			memsink_destroy(sink);
 		}

src/libs/config.h

@@ -23,5 +23,5 @@
 #pragma once
 
 #ifndef VERSION
-#	define VERSION "3.4"
+#	define VERSION "3.9"
 #endif

src/libs/logging.h

@@ -150,7 +150,7 @@ extern pthread_mutex_t log_mutex;
 #define LOG_VERBOSE_PERROR(_msg, ...) { \
 		if (log_level >= LOG_LEVEL_VERBOSE) { \
 			char _perror_buf[1024] = {0}; \
-			char *_perror_ptr = errno_to_string(errno, _perror_buf, 1024); \
+			char *_perror_ptr = errno_to_string(errno, _perror_buf, 1023); \
 			LOG_PRINTF(COLOR_BLUE, "VERB ", COLOR_BLUE, _msg ": %s", ##__VA_ARGS__, _perror_ptr); \
 		} \
 	}

src/libs/memsink.c

@@ -23,23 +23,28 @@
 #include "memsink.h"
 
 
-static int _flock_timedwait_monotonic(int fd, long double timeout);
+memsink_s *memsink_init(
+	const char *name, const char *obj, bool server,
+	mode_t mode, bool rm, unsigned client_ttl, unsigned timeout) {
 
-
-memsink_s *memsink_init(const char *name, const char *obj, bool server, mode_t mode, bool rm, unsigned timeout) {
 	memsink_s *sink;
 	A_CALLOC(sink, 1);
 	sink->name = name;
 	sink->obj = obj;
 	sink->server = server;
 	sink->rm = rm;
+	sink->client_ttl = client_ttl;
 	sink->timeout = timeout;
 	sink->fd = -1;
 	sink->mem = MAP_FAILED;
 
 	LOG_INFO("Using %s-sink: %s", name, obj);
 
-	if ((sink->fd = shm_open(sink->obj, (server ? O_RDWR | O_CREAT : O_RDWR), mode)) == -1) {
+	mode_t mask = umask(0);
+	sink->fd = shm_open(sink->obj, (server ? O_RDWR | O_CREAT : O_RDWR), mode);
+	umask(mask);
+	if (sink->fd == -1) {
+		umask(mask);
 		LOG_PERROR("%s-sink: Can't open shared memory", name);
 		goto error;
 	}
@@ -87,6 +92,33 @@ void memsink_destroy(memsink_s *sink) {
 	free(sink);
 }
 
+bool memsink_server_check(memsink_s *sink, const frame_s *frame) {
+	// Возвращает true, если если клиенты ИЛИ изменились метаданные
+
+	assert(sink->server);
+
+	if (flock(sink->fd, LOCK_EX | LOCK_NB) < 0) {
+		if (errno == EWOULDBLOCK) {
+			sink->has_clients = true;
+			return true;
+		}
+		LOG_PERROR("%s-sink: Can't lock memory", sink->name);
+		return false;
+	}
+
+	sink->has_clients = (sink->mem->last_client_ts + sink->client_ttl > get_now_monotonic());
+
+#	define NEQ(_field) (sink->mem->_field != frame->_field)
+	bool retval = (sink->has_clients || NEQ(width) || NEQ(height) || NEQ(format) || NEQ(stride) || NEQ(online));
+#	undef NEQ
+
+	if (flock(sink->fd, LOCK_UN) < 0) {
+		LOG_PERROR("%s-sink: Can't unlock memory", sink->name);
+		return false;
+	}
+	return retval;
+}
+
 int memsink_server_put(memsink_s *sink, const frame_s *frame) {
 	assert(sink->server);
 
@@ -98,7 +130,7 @@ int memsink_server_put(memsink_s *sink, const frame_s *frame) {
 		return 0; // -2
 	}
 
-	if (_flock_timedwait_monotonic(sink->fd, 1) == 0) {
+	if (flock_timedwait_monotonic(sink->fd, 1) == 0) {
 		LOG_VERBOSE("%s-sink: >>>>> Exposing new frame ...", sink->name);
 
 #		define COPY(_field) sink->mem->_field = frame->_field
@@ -113,8 +145,10 @@ int memsink_server_put(memsink_s *sink, const frame_s *frame) {
 		COPY(grab_ts);
 		COPY(encode_begin_ts);
 		COPY(encode_end_ts);
-		sink->has_clients = (sink->mem->last_consumed_ts + 10 > get_now_monotonic());
+		sink->has_clients = (sink->mem->last_client_ts + sink->client_ttl > get_now_monotonic());
 		memcpy(sink->mem->data, frame->data, frame->used);
+		sink->mem->magic = MEMSINK_MAGIC;
+		sink->mem->version = MEMSINK_VERSION;
 #		undef COPY
 
 		if (flock(sink->fd, LOCK_UN) < 0) {
@@ -137,7 +171,7 @@ int memsink_server_put(memsink_s *sink, const frame_s *frame) {
 int memsink_client_get(memsink_s *sink, frame_s *frame) { // cppcheck-suppress unusedFunction
 	assert(!sink->server); // Client only
 
-	if (_flock_timedwait_monotonic(sink->fd, sink->timeout) < 0) {
+	if (flock_timedwait_monotonic(sink->fd, sink->timeout) < 0) {
 		if (errno == EWOULDBLOCK) {
 			return -2;
 		}
@@ -145,48 +179,36 @@ int memsink_client_get(memsink_s *sink, frame_s *frame) { // cppcheck-suppress u
 		return -1;
 	}
 
-	bool same = false;
-
-	if (sink->mem->id == sink->last_id) {
-		same = true;
-	} else {
-#		define COPY(_field) frame->_field = sink->mem->_field
-		sink->last_id = sink->mem->id;
-		COPY(width);
-		COPY(height);
-		COPY(format);
-		COPY(stride);
-		COPY(online);
-		COPY(grab_ts);
-		COPY(encode_begin_ts);
-		COPY(encode_end_ts);
-		sink->mem->last_consumed_ts = get_now_monotonic();
-		frame_set_data(frame, sink->mem->data, sink->mem->used);
-#		undef COPY
-	}
-
-	if (flock(sink->fd, LOCK_UN) < 0) {
-		LOG_PERROR("%s-sink: Can't unlock memory", sink->name);
-		return -1;
-	}
-
-	if (same) {
-		usleep(1000);
-		return -2;
-	}
-	return 0;
-}
-
-static int _flock_timedwait_monotonic(int fd, long double timeout) {
-	long double deadline_ts = get_now_monotonic() + timeout;
-	int retval = -1;
-
-	while (true) {
-		retval = flock(fd, LOCK_EX | LOCK_NB);
-		if (retval == 0 || errno != EWOULDBLOCK || get_now_monotonic() > deadline_ts) {
-			break;
+	int retval = -2; // Not updated
+	if (sink->mem->magic == MEMSINK_MAGIC) {
+		if (sink->mem->version != MEMSINK_VERSION) {
+			LOG_ERROR("%s-sink: Protocol version mismatch: sink=%u, required=%u",
+				sink->name, sink->mem->version, MEMSINK_VERSION);
+			retval = -1;
+			goto done;
 		}
-		usleep(1000);
+		if (sink->mem->id != sink->last_id) { // When updated
+#			define COPY(_field) frame->_field = sink->mem->_field
+			sink->last_id = sink->mem->id;
+			COPY(width);
+			COPY(height);
+			COPY(format);
+			COPY(stride);
+			COPY(online);
+			COPY(grab_ts);
+			COPY(encode_begin_ts);
+			COPY(encode_end_ts);
+			frame_set_data(frame, sink->mem->data, sink->mem->used);
+#			undef COPY
+			retval = 0;
+		}
+		sink->mem->last_client_ts = get_now_monotonic();
 	}
-	return retval;
+
+	done:
+		if (flock(sink->fd, LOCK_UN) < 0) {
+			LOG_PERROR("%s-sink: Can't unlock memory", sink->name);
+			return -1;
+		}
+		return retval;
 }

src/libs/memsink.h

@@ -37,34 +37,15 @@
 #include "tools.h"
 #include "logging.h"
 #include "frame.h"
+#include "memsinksh.h"
 
 
-#ifndef CFG_MEMSINK_MAX_DATA
-#	define CFG_MEMSINK_MAX_DATA 33554432
-#endif
-#define MEMSINK_MAX_DATA ((size_t)(CFG_MEMSINK_MAX_DATA))
-
-
-typedef struct {
-	uint64_t	id;
-	size_t		used;
-	unsigned	width;
-	unsigned	height;
-	unsigned	format;
-	unsigned	stride;
-	bool		online;
-	long double	grab_ts;
-	long double	encode_begin_ts;
-	long double	encode_end_ts;
-	long double	last_consumed_ts;
-	uint8_t		data[MEMSINK_MAX_DATA];
-} memsink_shared_s;
-
 typedef struct {
 	const char	*name;
 	const char	*obj;
 	bool		server;
 	bool		rm;
+	unsigned	client_ttl; // Only for server
 	unsigned	timeout;
 
 	int					fd;
@@ -74,8 +55,13 @@ typedef struct {
 } memsink_s;
 
 
-memsink_s *memsink_init(const char *name, const char *obj, bool server, mode_t mode, bool rm, unsigned timeout);
+memsink_s *memsink_init(
+	const char *name, const char *obj, bool server,
+	mode_t mode, bool rm, unsigned client_ttl, unsigned timeout);
+
 void memsink_destroy(memsink_s *sink);
 
+bool memsink_server_check(memsink_s *sink, const frame_s *frame);
 int memsink_server_put(memsink_s *sink, const frame_s *frame);
+
 int memsink_client_get(memsink_s *sink, frame_s *frame);

src/libs/memsinksh.h

@@ -0,0 +1,60 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPG-HTTP streamer.                    #
+#                                                                            #
+#    Copyright (C) 2018-2021  Maxim Devaev <mdevaev@gmail.com>               #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#pragma once
+
+#include <stdint.h>
+#include <stdbool.h>
+
+#include <sys/types.h>
+
+
+#define MEMSINK_MAGIC	((uint64_t)0xCAFEBABECAFEBABE)
+#define MEMSINK_VERSION	((uint32_t)1)
+
+#ifndef CFG_MEMSINK_MAX_DATA
+#	define CFG_MEMSINK_MAX_DATA 33554432
+#endif
+#define MEMSINK_MAX_DATA ((size_t)(CFG_MEMSINK_MAX_DATA))
+
+
+typedef struct {
+	uint64_t	magic;
+	uint32_t	version;
+
+	uint64_t	id;
+
+	size_t		used;
+	unsigned	width;
+	unsigned	height;
+	unsigned	format;
+	unsigned	stride;
+	bool		online;
+
+	long double	grab_ts;
+	long double	encode_begin_ts;
+	long double	encode_end_ts;
+
+	long double	last_client_ts;
+
+	uint8_t		data[MEMSINK_MAX_DATA];
+} memsink_shared_s;

src/libs/options.c

@@ -0,0 +1,39 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPG-HTTP streamer.                    #
+#                                                                            #
+#    Copyright (C) 2018-2021  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 "options.h"
+
+
+void build_short_options(const struct option opts[], char *short_opts, size_t size) {
+	memset(short_opts, 0, size);
+    for (unsigned short_index = 0, opt_index = 0; opts[opt_index].name != NULL; ++opt_index) {
+		assert(short_index < size - 3);
+        if (isalpha(opts[opt_index].val)) {
+            short_opts[short_index] = opts[opt_index].val;
+            ++short_index;
+            if (opts[opt_index].has_arg == required_argument) {
+                short_opts[short_index] = ':';
+                ++short_index;
+            }
+        }
+    }
+}

src/libs/options.h

@@ -0,0 +1,33 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPG-HTTP streamer.                    #
+#                                                                            #
+#    Copyright (C) 2018-2021  Maxim Devaev <mdevaev@gmail.com>               #
+#                                                                            #
+#    This program is free software: you can redistribute it and/or modify    #
+#    it under the terms of the GNU General Public License as published by    #
+#    the Free Software Foundation, either version 3 of the License, or       #
+#    (at your option) any later version.                                     #
+#                                                                            #
+#    This program is distributed in the hope that it will be useful,         #
+#    but WITHOUT ANY WARRANTY; without even the implied warranty of          #
+#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the           #
+#    GNU General Public License for more details.                            #
+#                                                                            #
+#    You should have received a copy of the GNU General Public License       #
+#    along with this program.  If not, see <https://www.gnu.org/licenses/>.  #
+#                                                                            #
+*****************************************************************************/
+
+
+#pragma once
+
+#include <string.h>
+#include <ctype.h>
+#include <getopt.h>
+#include <assert.h>
+
+#include <sys/types.h>
+
+
+void build_short_options(const struct option opts[], char *short_opts, size_t size);

src/libs/tools.h

@@ -32,6 +32,8 @@
 #include <time.h>
 #include <assert.h>
 
+#include <sys/file.h>
+
 
 #define A_CALLOC(_dest, _nmemb)		assert((_dest = calloc(_nmemb, sizeof(*(_dest)))))
 #define A_REALLOC(_dest, _nmemb)	assert((_dest = realloc(_dest, _nmemb * sizeof(*(_dest)))))
@@ -123,3 +125,19 @@ INLINE unsigned get_cores_available(void) {
 	cores_sysconf = (cores_sysconf < 0 ? 0 : cores_sysconf);
 	return max_u(min_u(cores_sysconf, 4), 1);
 }
+
+INLINE int flock_timedwait_monotonic(int fd, long double timeout) {
+	long double deadline_ts = get_now_monotonic() + timeout;
+	int retval = -1;
+
+	while (true) {
+		retval = flock(fd, LOCK_EX | LOCK_NB);
+		if (retval == 0 || errno != EWOULDBLOCK || get_now_monotonic() > deadline_ts) {
+			break;
+		}
+		if (usleep(1000) < 0) {
+			break;
+		}
+	}
+	return retval;
+}

src/ustreamer/encoders/omx/encoder.c

@@ -221,18 +221,20 @@ static int _vcos_semwait(VCOS_SEMAPHORE_T *sem) {
 		if (sem_status == VCOS_SUCCESS) {
 			return 0;
 		} else if (sem_status != VCOS_EAGAIN || get_now_monotonic() > deadline_ts) {
-			goto error;
+			break;
+		}
+		if (usleep(1000) < 0) {
+			break;
 		}
-		usleep(1000);
 	}
 
-	error:
-		switch (sem_status) {
-			case VCOS_EAGAIN: LOG_ERROR("Can't wait VCOS semaphore: EAGAIN (timeout)"); break;
-			case VCOS_EINVAL: LOG_ERROR("Can't wait VCOS semaphore: EINVAL"); break;
-			default: LOG_ERROR("Can't wait VCOS semaphore: %d", sem_status); break;
-		}
-		return -1;
+	switch (sem_status) {
+		case VCOS_EAGAIN: LOG_ERROR("Can't wait VCOS semaphore: EAGAIN (timeout)"); break;
+		case VCOS_EINVAL: LOG_ERROR("Can't wait VCOS semaphore: EINVAL"); break;
+		default: LOG_ERROR("Can't wait VCOS semaphore: %d", sem_status); break;
+	}
+	return -1;
+
 #	else
 	return (vcos_semaphore_wait(sem) == VCOS_SUCCESS ? 0 : -1);
 #	endif

src/ustreamer/h264/encoder.c

@@ -250,7 +250,9 @@ static void _h264_encoder_cleanup(h264_encoder_s *enc) {
 
 #	undef DISABLE_PORT
 
-	enc->wrapper->status = MMAL_SUCCESS; // Это реально надо?
+	if (enc->wrapper) {
+		enc->wrapper->status = MMAL_SUCCESS; // Это реально надо?
+	}
 
 	enc->last_online = -1;
 	enc->ready = false;

src/ustreamer/h264/stream.c

@@ -53,6 +53,10 @@ void h264_stream_destroy(h264_stream_s *h264) {
 }
 
 void h264_stream_process(h264_stream_s *h264, const frame_s *frame, int vcsm_handle, bool force_key) {
+	if (!memsink_server_check(h264->sink, frame)) {
+		return;
+	}
+
 	long double now = get_now_monotonic();
 	bool zero_copy = false;
 

src/ustreamer/options.c

@@ -86,8 +86,10 @@ enum _OPT_VALUES {
 		_O_##_prefix, \
 		_O_##_prefix##_MODE, \
 		_O_##_prefix##_RM, \
+		_O_##_prefix##_CLIENT_TTL, \
 		_O_##_prefix##_TIMEOUT,
 	ADD_SINK(SINK)
+	ADD_SINK(RAW_SINK)
 #	ifdef WITH_OMX
 	ADD_SINK(H264_SINK)
 	_O_H264_BITRATE,
@@ -174,11 +176,13 @@ static const struct option _LONG_OPTS[] = {
 	{"server-timeout",			required_argument,	NULL,	_O_SERVER_TIMEOUT},
 
 #	define ADD_SINK(_opt, _prefix) \
-		{_opt "sink",			required_argument,	NULL,	_O_##_prefix}, \
-		{_opt "sink-mode",		required_argument,	NULL,	_O_##_prefix##_MODE}, \
-		{_opt "sink-rm",		no_argument,		NULL,	_O_##_prefix##_RM}, \
-		{_opt "sink-timeout",	required_argument,	NULL,	_O_##_prefix##_TIMEOUT},
+		{_opt "sink",				required_argument,	NULL,	_O_##_prefix}, \
+		{_opt "sink-mode",			required_argument,	NULL,	_O_##_prefix##_MODE}, \
+		{_opt "sink-rm",			no_argument,		NULL,	_O_##_prefix##_RM}, \
+		{_opt "sink-client-ttl",	required_argument,	NULL,	_O_##_prefix##_CLIENT_TTL}, \
+		{_opt "sink-timeout",		required_argument,	NULL,	_O_##_prefix##_TIMEOUT},
 	ADD_SINK("", SINK)
+	ADD_SINK("raw-", RAW_SINK)
 #	ifdef WITH_OMX
 	ADD_SINK("h264-", H264_SINK)
 	{"h264-bitrate",			required_argument,	NULL,	_O_H264_BITRATE},
@@ -243,6 +247,7 @@ void options_destroy(options_s *options) {
 			} \
 		}
 	ADD_SINK(sink);
+	ADD_SINK(raw_sink);
 #	ifdef WITH_OMX
 	ADD_SINK(h264_sink);
 #	endif
@@ -334,8 +339,10 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 		char *_prefix##_name = NULL; \
 		mode_t _prefix##_mode = 0660; \
 		bool _prefix##_rm = false; \
+		unsigned _prefix##_client_ttl = 10; \
 		unsigned _prefix##_timeout = 1;
 	ADD_SINK(sink);
+	ADD_SINK(raw_sink);
 #	ifdef WITH_OMX
 	ADD_SINK(h264_sink);
 #	endif
@@ -345,18 +352,8 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 	char *process_name_prefix = NULL;
 #	endif
 
-	char short_opts[1024] = {0};
-
-	for (int short_index = 0, opt_index = 0; _LONG_OPTS[opt_index].name != NULL; ++opt_index) {
-		if (isalpha(_LONG_OPTS[opt_index].val)) {
-			short_opts[short_index] = _LONG_OPTS[opt_index].val;
-			++short_index;
-			if (_LONG_OPTS[opt_index].has_arg == required_argument) {
-				short_opts[short_index] = ':';
-				++short_index;
-			}
-		}
-	}
+	char short_opts[128];
+	build_short_options(_LONG_OPTS, short_opts, 128);
 
 	for (int ch; (ch = getopt_long(options->argc, options->argv_copy, short_opts, _LONG_OPTS, NULL)) >= 0;) {
 		switch (ch) {
@@ -431,8 +428,10 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 				case _O_##_up:				OPT_SET(_lp##_name, optarg); \
 				case _O_##_up##_MODE:		OPT_NUMBER("--" #_opt "sink-mode", _lp##_mode, INT_MIN, INT_MAX, 8); \
 				case _O_##_up##_RM:			OPT_SET(_lp##_rm, true); \
+				case _O_##_up##_CLIENT_TTL:	OPT_NUMBER("--" #_opt "sink-client-ttl", _lp##_client_ttl, 1, 60, 0); \
 				case _O_##_up##_TIMEOUT:	OPT_NUMBER("--" #_opt "sink-timeout", _lp##_timeout, 1, 60, 0);
 			ADD_SINK("", sink, SINK)
+			ADD_SINK("raw-", raw_sink, RAW_SINK)
 #			ifdef WITH_OMX
 			ADD_SINK("h264-", h264_sink, H264_SINK)
 			case _O_H264_BITRATE:	OPT_NUMBER("--h264-bitrate", stream->h264_bitrate, 100, 16000, 0);
@@ -472,7 +471,7 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 			case _O_FEATURES:	_features(); return 1;
 
 			case 0:		break;
-			default:	_help(stderr, dev, enc, stream, server); return -1;
+			default:	return -1;
 		}
 	}
 
@@ -487,12 +486,14 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 					true, \
 					_prefix##_mode, \
 					_prefix##_rm, \
+					_prefix##_client_ttl, \
 					_prefix##_timeout \
 				); \
 			} \
 			stream->_prefix = options->_prefix; \
 		}
 	ADD_SINK("JPEG", sink);
+	ADD_SINK("RAW", raw_sink);
 #	ifdef WITH_OMX
 	ADD_SINK("H264", h264_sink);
 #	endif
@@ -661,15 +662,17 @@ static void _help(FILE *fp, device_s *dev, encoder_s *enc, stream_s *stream, ser
 #	define ADD_SINK(_name, _opt) \
 		SAY(_name " sink options:"); \
 		SAY("══════════════════"); \
-		SAY("    --" _opt "sink <name>  ──────── Use the shared memory to sink " _name " frames. Default: disabled.\n"); \
-		SAY("    --" _opt "sink-mode <mode>  ─── Set " _name " sink permissions (like 777). Default: 660.\n"); \
-		SAY("    --" _opt "sink-rm  ──────────── Remove shared memory on stop. Default: disabled.\n"); \
-		SAY("    --" _opt "sink-timeout <sec>  ─ Timeout for lock. Default: 1.\n");
+		SAY("    --" _opt "sink <name>  ─────────── Use the shared memory to sink " _name " frames. Default: disabled.\n"); \
+		SAY("    --" _opt "sink-mode <mode>  ────── Set " _name " sink permissions (like 777). Default: 660.\n"); \
+		SAY("    --" _opt "sink-rm  ─────────────── Remove shared memory on stop. Default: disabled.\n"); \
+		SAY("    --" _opt "sink-client-ttl <sec>  ─ Client TTL. Default: 10.\n"); \
+		SAY("    --" _opt "sink-timeout <sec>  ──── Timeout for lock. Default: 1.\n");
 	ADD_SINK("JPEG", "")
+	ADD_SINK("RAW", "raw-")
 #	ifdef WITH_OMX
 	ADD_SINK("H264", "h264-")
-	SAY("    --h264-bitrate <kbps>  ───── H264 bitrate in Kbps. Default: %u.\n", stream->h264_bitrate);
-	SAY("    --h264-gop <N>  ──────────── Intarval between keyframes. Default: %u.\n", stream->h264_gop);
+	SAY("    --h264-bitrate <kbps>  ──────── H264 bitrate in Kbps. Default: %u.\n", stream->h264_bitrate);
+	SAY("    --h264-gop <N>  ─────────────── Intarval between keyframes. Default: %u.\n", stream->h264_gop);
 #	endif
 #	undef ADD_SINK
 #	ifdef WITH_GPIO

src/ustreamer/options.h

@@ -27,7 +27,6 @@
 #include <stdbool.h>
 #include <string.h>
 #include <strings.h>
-#include <ctype.h>
 #include <limits.h>
 #include <getopt.h>
 #include <errno.h>
@@ -38,6 +37,7 @@
 #include "../libs/process.h"
 #include "../libs/frame.h"
 #include "../libs/memsink.h"
+#include "../libs/options.h"
 
 #include "device.h"
 #include "encoder.h"
@@ -55,6 +55,7 @@ typedef struct {
 	char		**argv_copy;
 	frame_s		*blank;
 	memsink_s	*sink;
+	memsink_s	*raw_sink;
 #	ifdef WITH_OMX
 	memsink_s	*h264_sink;
 #	endif

src/ustreamer/stream.c

@@ -194,6 +194,12 @@ void stream_loop(stream_s *stream) {
 							workers_pool_assign(pool, ready_wr);
 							LOG_DEBUG("Assigned new frame in buffer %d to worker %s", buf_index, ready_wr->name);
 
+							if (stream->raw_sink) {
+								if (memsink_server_check(stream->raw_sink, &hw->raw)) {
+									memsink_server_put(stream->raw_sink, &hw->raw);
+								}
+							}
+
 #							ifdef WITH_OMX
 							if (RUN(h264)) {
 								h264_stream_process(RUN(h264), &hw->raw, hw->vcsm_handle, h264_force_key);
@@ -248,6 +254,11 @@ static workers_pool_s *_stream_init_loop(stream_s *stream) {
 
 	while (!atomic_load(&RUN(stop))) {
 		if (_stream_expose_frame(stream, NULL, 0)) {
+			if (stream->raw_sink) {
+				if (memsink_server_check(stream->raw_sink, stream->blank)) {
+					memsink_server_put(stream->raw_sink, stream->blank);
+				}
+			}
 #			ifdef WITH_OMX
 			if (RUN(h264)) {
 				h264_stream_process(RUN(h264), stream->blank, -1, false);
@@ -351,7 +362,9 @@ static bool _stream_expose_frame(stream_s *stream, frame_s *frame, unsigned capt
 	A_MUTEX_UNLOCK(&VID(mutex));
 
 	if (changed && stream->sink) {
-		memsink_server_put(stream->sink, VID(frame));
+		if (memsink_server_check(stream->sink, VID(frame))) {
+			memsink_server_put(stream->sink, VID(frame));
+		}
 	}
 
 	return changed;

src/ustreamer/stream.h

@@ -80,6 +80,7 @@ typedef struct {
 	unsigned	error_delay;
 
 	memsink_s	*sink;
+	memsink_s	*raw_sink;
 
 #	ifdef WITH_OMX
 	memsink_s	*h264_sink;