Bump version: 5.14 → 5.15

This reverts commit 021823bcba.

.bumpversion.cfg

@@ -1,18 +1,18 @@
 [bumpversion]
 commit = True
 tag = True
-current_version = 5.12
+current_version = 5.15
 parse = (?P<major>\d+)\.(?P<minor>\d+)
 serialize = 
 	{major}.{minor}
 
-[bumpversion:file:src/libs/config.h]
+[bumpversion:file:src/libs/const.h]
 parse = (?P<major>\d+)
 serialize = {major}
 search = VERSION_MAJOR {current_version}
 replace = VERSION_MAJOR {new_version}
 
-[bumpversion:file:./src/libs/config.h]
+[bumpversion:file:./src/libs/const.h]
 parse = <major>\d+\.(?P<minor>\d+)
 serialize = {minor}
 search = VERSION_MINOR {current_version}

janus/src/audio.c

@@ -146,17 +146,8 @@ void audio_destroy(audio_s *audio) {
 	if (audio->pcm_params) {
 		snd_pcm_hw_params_free(audio->pcm_params);
 	}
-#	define FREE_QUEUE(_suffix) { \
-			while (!queue_get_free(audio->_suffix##_queue)) { \
-				_##_suffix##_buffer_s *ptr; \
-				assert(!queue_get(audio->_suffix##_queue, (void **)&ptr, 1)); \
-				free(ptr); \
-			} \
-			queue_destroy(audio->_suffix##_queue); \
-		}
-	FREE_QUEUE(enc);
-	FREE_QUEUE(pcm);
-#	undef FREE_QUEUE
+	QUEUE_FREE_ITEMS_AND_DESTROY(audio->enc_queue, free);
+	QUEUE_FREE_ITEMS_AND_DESTROY(audio->pcm_queue, free);
 	if (audio->tids_created) {
 		JLOG_INFO("audio", "Pipeline closed");
 	}
@@ -168,7 +159,7 @@ int audio_get_encoded(audio_s *audio, uint8_t *data, size_t *size, uint64_t *pts
 		return -1;
 	}
 	_enc_buffer_s *buf;
-	if (!queue_get(audio->enc_queue, (void **)&buf, 1)) {
+	if (!queue_get(audio->enc_queue, (void **)&buf, 0.1)) {
 		if (*size < buf->used) {
 			free(buf);
 			return -3;
@@ -202,7 +193,7 @@ static void *_pcm_thread(void *v_audio) {
 			_pcm_buffer_s *out;
 			A_CALLOC(out, 1);
 			memcpy(out->data, in, audio->pcm_size);
-			assert(!queue_put(audio->pcm_queue, out, 1));
+			assert(!queue_put(audio->pcm_queue, out, 0));
 		} else {
 			JLOG_ERROR("audio", "PCM queue is full");
 		}
@@ -220,7 +211,7 @@ static void *_encoder_thread(void *v_audio) {
 
 	while (!atomic_load(&audio->stop)) {
 		_pcm_buffer_s *in;
-		if (!queue_get(audio->pcm_queue, (void **)&in, 1)) {
+		if (!queue_get(audio->pcm_queue, (void **)&in, 0.1)) {
 			int16_t *in_ptr;
 			if (audio->res) {
 				assert(audio->pcm_hz != ENCODER_INPUT_HZ);
@@ -247,9 +238,7 @@ static void *_encoder_thread(void *v_audio) {
 			// https://datatracker.ietf.org/doc/html/rfc7587#section-4.2
 			audio->pts += HZ_TO_FRAMES(ENCODER_INPUT_HZ);
 
-			if (queue_get_free(audio->enc_queue)) {
-				assert(!queue_put(audio->enc_queue, out, 1));
-			} else {
+			if (queue_put(audio->enc_queue, out, 0) != 0) {
 				JLOG_ERROR("audio", "OPUS encoder queue is full");
 				free(out);
 			}

janus/src/audio.h

@@ -38,7 +38,7 @@
 #include "uslibs/tools.h"
 #include "uslibs/threading.h"
 
-#include "jlogging.h"
+#include "logging.h"
 #include "queue.h"
 
 

janus/src/client.c

@@ -0,0 +1,120 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C) 2018-2022  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 "client.h"
+
+
+static void *_video_thread(void *v_client);
+static void *_audio_thread(void *v_client);
+static void *_common_thread(void *v_client, bool video);
+
+
+client_s *client_init(janus_callbacks *gw, janus_plugin_session *session, bool has_audio) {
+	client_s *client;
+	A_CALLOC(client, 1);
+	client->gw = gw;
+	client->session = session;
+	atomic_init(&client->transmit, true);
+
+	atomic_init(&client->stop, false);
+
+	client->video_queue = queue_init(1024);
+	A_THREAD_CREATE(&client->video_tid, _video_thread, client);
+
+	if (has_audio) {
+		client->audio_queue = queue_init(64);
+		A_THREAD_CREATE(&client->audio_tid, _audio_thread, client);
+	}
+	return client;
+}
+
+void client_destroy(client_s *client) {
+	atomic_store(&client->stop, true);
+	queue_put(client->video_queue, NULL, 0);
+	if (client->audio_queue != NULL) {
+		queue_put(client->audio_queue, NULL, 0);
+	}
+
+	A_THREAD_JOIN(client->video_tid);
+	QUEUE_FREE_ITEMS_AND_DESTROY(client->video_queue, rtp_destroy);
+	if (client->audio_queue != NULL) {
+		A_THREAD_JOIN(client->audio_tid);
+		QUEUE_FREE_ITEMS_AND_DESTROY(client->audio_queue, rtp_destroy);
+	}
+	free(client);
+}
+
+void client_send(client_s *client, const rtp_s *rtp) {
+	if (
+		!atomic_load(&client->transmit)
+		|| (!rtp->video && client->audio_queue == NULL)
+	) {
+		return;
+	}
+	rtp_s *new = rtp_dup(rtp);
+	if (queue_put((new->video ? client->video_queue : client->audio_queue), new, 0) != 0) {
+		JLOG_ERROR("client", "Session %p %s queue is full",
+			client->session, (new->video ? "video" : "audio"));
+		rtp_destroy(new);
+	}
+}
+
+static void *_video_thread(void *v_client) {
+	return _common_thread(v_client, true);
+}
+
+static void *_audio_thread(void *v_client) {
+	return _common_thread(v_client, false);
+}
+
+static void *_common_thread(void *v_client, bool video) {
+	client_s *client = (client_s *)v_client;
+	queue_s *queue = (video ? client->video_queue : client->audio_queue);
+	assert(queue != NULL); // Audio may be NULL
+
+	while (!atomic_load(&client->stop)) {
+		rtp_s *rtp;
+		if (!queue_get(queue, (void **)&rtp, 0.1)) {
+			if (rtp == NULL) {
+				break;
+			}
+			if (atomic_load(&client->transmit)) {
+				janus_plugin_rtp packet = {0};
+				packet.video = rtp->video;
+				packet.buffer = (char *)rtp->datagram;
+				packet.length = rtp->used;
+				janus_plugin_rtp_extensions_reset(&packet.extensions);
+				// FIXME: See rtpv.c
+				// Это очень эффективный способ уменьшить задержку, но WebRTC стек в хроме и фоксе
+				// слишком корявый, чтобы обработать это, тз-за чего на кейфреймах начинаются заикания.
+				//   - https://github.com/Glimesh/janus-ftl-plugin/issues/101
+				/*if (video) {
+					packet.extensions.min_delay = 0;
+					packet.extensions.max_delay = 0;
+				}*/
+				client->gw->relay_rtp(client->session, &packet);
+			}
+			rtp_destroy(rtp);
+		}
+	}
+	return NULL;
+}

janus/src/client.h

@@ -0,0 +1,61 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C) 2018-2022  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 <stdlib.h>
+#include <stdbool.h>
+#include <stdatomic.h>
+#include <string.h>
+
+#include <pthread.h>
+#include <janus/plugins/plugin.h>
+
+#include "uslibs/tools.h"
+#include "uslibs/threading.h"
+#include "uslibs/list.h"
+
+#include "logging.h"
+#include "queue.h"
+#include "rtp.h"
+
+
+typedef struct client_sx {
+	janus_callbacks			*gw;
+    janus_plugin_session	*session;
+    atomic_bool				transmit;
+
+	pthread_t	video_tid;
+	pthread_t	audio_tid;
+	atomic_bool	stop;
+
+	queue_s		*video_queue;
+	queue_s		*audio_queue;
+
+    LIST_STRUCT(struct client_sx);
+} client_s;
+
+
+client_s *client_init(janus_callbacks *gw, janus_plugin_session *session, bool has_audio);
+void client_destroy(client_s *client);
+
+void client_send(client_s *client, const rtp_s *rtp);

janus/src/config.c

@@ -0,0 +1,78 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C) 2018-2022  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 "config.h"
+
+
+static char *_get_value(janus_config *config, const char *section, const char *option);
+
+
+int read_config(const char *config_dir_path, char **video_sink_name, char **audio_dev_name, char **tc358743_dev_path) {
+	int retval = 0;
+
+	char *config_file_path;
+	janus_config *config = NULL;
+
+	A_ASPRINTF(config_file_path, "%s/%s.jcfg", config_dir_path, PLUGIN_PACKAGE);
+	JLOG_INFO("config", "Reading config file '%s' ...", config_file_path);
+
+	config = janus_config_parse(config_file_path);
+	if (config == NULL) {
+		JLOG_ERROR("config", "Can't read config");
+		goto error;
+	}
+	janus_config_print(config);
+
+	if (
+		(*video_sink_name = _get_value(config, "memsink", "object")) == NULL
+		&& (*video_sink_name = _get_value(config, "video", "sink")) == NULL
+	) {
+		JLOG_ERROR("config", "Missing config value: video.sink (ex. memsink.object)");
+		goto error;
+	}
+	if ((*audio_dev_name = _get_value(config, "audio", "device")) != NULL) {
+		JLOG_INFO("config", "Enabled the experimental AUDIO feature");
+		if ((*tc358743_dev_path = _get_value(config, "audio", "tc358743")) == NULL) {
+			JLOG_INFO("config", "Missing config value: audio.tc358743");
+			goto error;
+		}
+	}
+
+	goto ok;
+	error:
+		retval = -1;
+	ok:
+		if (config) {
+			janus_config_destroy(config);
+		}
+		free(config_file_path);
+		return retval;
+}
+
+static char *_get_value(janus_config *config, const char *section, const char *option) {
+	janus_config_category *section_obj = janus_config_get_create(config, NULL, janus_config_type_category, section);
+	janus_config_item *option_obj = janus_config_get(config, section_obj, janus_config_type_item, option);
+	if (option_obj == NULL || option_obj->value == NULL || option_obj->value[0] == '\0') {
+		return NULL;
+	}
+	return strdup(option_obj->value);
+}

janus/src/config.h

@@ -0,0 +1,37 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C) 2018-2022  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 <stdlib.h>
+#include <string.h>
+
+#include <janus/config.h>
+#include <janus/plugins/plugin.h>
+
+#include "uslibs/tools.h"
+
+#include "const.h"
+#include "logging.h"
+
+
+int read_config(const char *config_dir_path, char **video_sink_name, char **audio_dev_name, char **tc358743_dev_path);

janus/src/const.h

@@ -0,0 +1,26 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C) 2018-2022  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
+
+#define PLUGIN_NAME		"ustreamer"
+#define PLUGIN_PACKAGE	"janus.plugin.ustreamer"

janus/src/logging.h

@@ -24,13 +24,15 @@
 
 #include "uslibs/tools.h"
 
+#include "const.h"
 
-#define JLOG_INFO(_prefix, _msg, ...)	JANUS_LOG(LOG_INFO, "== ustreamer/%-9s -- " _msg "\n", _prefix, ##__VA_ARGS__)
-#define JLOG_WARN(_prefix, _msg, ...)	JANUS_LOG(LOG_WARN, "== ustreamer/%-9s -- " _msg "\n", _prefix, ##__VA_ARGS__)
-#define JLOG_ERROR(_prefix, _msg, ...)	JANUS_LOG(LOG_ERR, "== ustreamer/%-9s -- " _msg "\n", _prefix, ##__VA_ARGS__)
+
+#define JLOG_INFO(_prefix, _msg, ...)	JANUS_LOG(LOG_INFO, "== %s/%-9s -- " _msg "\n", PLUGIN_NAME, _prefix, ##__VA_ARGS__)
+#define JLOG_WARN(_prefix, _msg, ...)	JANUS_LOG(LOG_WARN, "== %s/%-9s -- " _msg "\n", PLUGIN_NAME, _prefix, ##__VA_ARGS__)
+#define JLOG_ERROR(_prefix, _msg, ...)	JANUS_LOG(LOG_ERR, "== %s/%-9s -- " _msg "\n", PLUGIN_NAME, _prefix, ##__VA_ARGS__)
 
 #define JLOG_PERROR(_prefix, _msg, ...) { \
 		char _perror_buf[1024] = {0}; \
 		char *_perror_ptr = errno_to_string(errno, _perror_buf, 1023); \
-		JANUS_LOG(LOG_ERR, "[ustreamer/%-9s] " _msg ": %s\n", _prefix, ##__VA_ARGS__, _perror_ptr); \
+		JANUS_LOG(LOG_ERR, "[%s/%-9s] " _msg ": %s\n", PLUGIN_NAME, _prefix, ##__VA_ARGS__, _perror_ptr); \
 	}

janus/src/memsinkfd.c

@@ -0,0 +1,70 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C) 2018-2022  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 "memsinkfd.h"
+
+
+int memsink_fd_wait_frame(int fd, memsink_shared_s* mem, uint64_t last_id) {
+	long double deadline_ts = get_now_monotonic() + 1; // wait_timeout
+	long double now;
+	do {
+		int result = flock_timedwait_monotonic(fd, 1); // lock_timeout
+		now = get_now_monotonic();
+		if (result < 0 && errno != EWOULDBLOCK) {
+			JLOG_PERROR("video", "Can't lock memsink");
+			return -1;
+		} else if (result == 0) {
+			if (mem->magic == MEMSINK_MAGIC && mem->version == MEMSINK_VERSION && mem->id != last_id) {
+				return 0;
+			}
+			if (flock(fd, LOCK_UN) < 0) {
+				JLOG_PERROR("video", "Can't unlock memsink");
+				return -1;
+			}
+		}
+		usleep(1000); // lock_polling
+	} while (now < deadline_ts);
+	return -2;
+}
+
+frame_s *memsink_fd_get_frame(int fd, memsink_shared_s *mem, uint64_t *frame_id) {
+	frame_s *frame = frame_init();
+	frame_set_data(frame, mem->data, mem->used);
+	FRAME_COPY_META(mem, frame);
+	*frame_id = mem->id;
+	mem->last_client_ts = get_now_monotonic();
+
+	bool ok = true;
+	if (frame->format != V4L2_PIX_FMT_H264) {
+		JLOG_ERROR("video", "Got non-H264 frame from memsink");
+		ok = false;
+	}
+	if (flock(fd, LOCK_UN) < 0) {
+		JLOG_PERROR("video", "Can't unlock memsink");
+		ok = false;
+	}
+	if (!ok) {
+		frame_destroy(frame);
+		frame = NULL;
+	}
+	return frame;
+}

janus/src/memsinkfd.h

@@ -0,0 +1,38 @@
+/*****************************************************************************
+#                                                                            #
+#    uStreamer - Lightweight and fast MJPEG-HTTP streamer.                   #
+#                                                                            #
+#    Copyright (C) 2018-2022  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 <unistd.h>
+
+#include <linux/videodev2.h>
+
+#include "uslibs/tools.h"
+#include "uslibs/frame.h"
+#include "uslibs/memsinksh.h"
+
+#include "logging.h"
+
+
+int memsink_fd_wait_frame(int fd, memsink_shared_s* mem, uint64_t last_id);
+frame_s *memsink_fd_get_frame(int fd, memsink_shared_s *mem, uint64_t *frame_id);

janus/src/plugin.c

@@ -31,190 +31,99 @@
 
 #include <sys/mman.h>
 #include <sys/stat.h>
-#include <linux/videodev2.h>
 
 #include <pthread.h>
 #include <jansson.h>
-#include <janus/config.h>
 #include <janus/plugins/plugin.h>
 
-#include "uslibs/config.h"
+#include "uslibs/const.h"
 #include "uslibs/tools.h"
 #include "uslibs/threading.h"
 #include "uslibs/list.h"
 #include "uslibs/memsinksh.h"
 
-#include "jlogging.h"
+#include "const.h"
+#include "logging.h"
+#include "queue.h"
+#include "client.h"
 #include "audio.h"
 #include "tc358743.h"
+#include "rtp.h"
 #include "rtpv.h"
 #include "rtpa.h"
-
-
-static int _plugin_init(janus_callbacks *gw, const char *config_file_path);
-static void _plugin_destroy(void);
-
-static void _plugin_create_session(janus_plugin_session *session, int *err);
-static void _plugin_destroy_session(janus_plugin_session *session, int *err);
-static json_t *_plugin_query_session(janus_plugin_session *session);
-
-static void _plugin_setup_media(janus_plugin_session *session);
-static void _plugin_hangup_media(janus_plugin_session *session);
-
-static struct janus_plugin_result *_plugin_handle_message(
-	janus_plugin_session *session, char *transaction, json_t *msg, json_t *jsep);
-
-
-static int _plugin_get_api_compatibility(void)		{ return JANUS_PLUGIN_API_VERSION; }
-static int _plugin_get_version(void)				{ return VERSION_U; }
-static const char *_plugin_get_version_string(void)	{ return VERSION; }
-static const char *_plugin_get_description(void)	{ return "PiKVM uStreamer Janus plugin for H.264 video"; }
-static const char *_plugin_get_name(void)			{ return "ustreamer"; }
-static const char *_plugin_get_author(void)			{ return "Maxim Devaev <mdevaev@gmail.com>"; }
-static const char *_plugin_get_package(void)		{ return "janus.plugin.ustreamer"; }
-
-// Just a stub to avoid logging spam about the plugin's purpose.
-static void _plugin_incoming_rtp(UNUSED janus_plugin_session *handle, UNUSED janus_plugin_rtp *packet) {}
-
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Woverride-init"
-static janus_plugin _plugin = JANUS_PLUGIN_INIT(
-	.init = _plugin_init,
-	.destroy = _plugin_destroy,
-
-	.create_session = _plugin_create_session,
-	.destroy_session = _plugin_destroy_session,
-	.query_session = _plugin_query_session,
-
-	.setup_media = _plugin_setup_media,
-	.hangup_media = _plugin_hangup_media,
-
-	.handle_message = _plugin_handle_message,
-
-	.get_api_compatibility = _plugin_get_api_compatibility,
-	.get_version = _plugin_get_version,
-	.get_version_string = _plugin_get_version_string,
-	.get_description = _plugin_get_description,
-	.get_name = _plugin_get_name,
-	.get_author = _plugin_get_author,
-	.get_package = _plugin_get_package,
-
-	.incoming_rtp = _plugin_incoming_rtp,
-);
-#pragma GCC diagnostic pop
-
-janus_plugin *create(void) { // cppcheck-suppress unusedFunction
-	return &_plugin;
-}
-
-
-typedef struct _client_sx {
-	janus_plugin_session *session;
-	bool transmit;
-
-	LIST_STRUCT(struct _client_sx);
-} _client_s;
+#include "memsinkfd.h"
+#include "config.h"
 
 
 static char				*_g_video_sink_name = NULL;
-const long double		_g_sink_wait_timeout = 1;
-const long double		_g_sink_lock_timeout = 1;
-const useconds_t		_g_sink_lock_polling = 1000;
-
 static char				*_g_audio_dev_name = NULL;
 static char				*_g_tc358743_dev_path = NULL;
 
 const useconds_t		_g_watchers_polling = 100000;
 
-static _client_s		*_g_clients = NULL;
+static client_s			*_g_clients = NULL;
 static janus_callbacks	*_g_gw = NULL;
+static queue_s			*_g_video_queue = NULL;
 static rtpv_s			*_g_rtpv = NULL;
 static rtpa_s			*_g_rtpa = NULL;
 
-static pthread_t		_g_video_tid;
-static atomic_bool		_g_video_tid_created = false;
+static pthread_t		_g_video_rtp_tid;
+static atomic_bool		_g_video_rtp_tid_created = false;
+static pthread_t		_g_video_sink_tid;
+static atomic_bool		_g_video_sink_tid_created = false;
 static pthread_t		_g_audio_tid;
 static atomic_bool		_g_audio_tid_created = false;
 
-static pthread_mutex_t	_g_lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_mutex_t	_g_video_lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_mutex_t	_g_audio_lock = PTHREAD_MUTEX_INITIALIZER;
 static atomic_bool		_g_ready = false;
 static atomic_bool		_g_stop = false;
 static atomic_bool		_g_has_watchers = false;
 
 
-#define LOCK			A_MUTEX_LOCK(&_g_lock)
-#define UNLOCK			A_MUTEX_UNLOCK(&_g_lock)
+#define LOCK_VIDEO		A_MUTEX_LOCK(&_g_video_lock)
+#define UNLOCK_VIDEO	A_MUTEX_UNLOCK(&_g_video_lock)
+
+#define LOCK_AUDIO		A_MUTEX_LOCK(&_g_audio_lock)
+#define UNLOCK_AUDIO	A_MUTEX_UNLOCK(&_g_audio_lock)
+
+#define LOCK_ALL		{ LOCK_VIDEO; LOCK_AUDIO; }
+#define UNLOCK_ALL		{ UNLOCK_AUDIO; UNLOCK_VIDEO; }
+
 #define READY			atomic_load(&_g_ready)
 #define STOP			atomic_load(&_g_stop)
 #define HAS_WATCHERS	atomic_load(&_g_has_watchers)
 
 
-static int _wait_frame(int fd, memsink_shared_s* mem, uint64_t last_id) {
-	long double deadline_ts = get_now_monotonic() + _g_sink_wait_timeout;
-	long double now;
-	do {
-		int result = flock_timedwait_monotonic(fd, _g_sink_lock_timeout);
-		now = get_now_monotonic();
-		if (result < 0 && errno != EWOULDBLOCK) {
-			JLOG_PERROR("video", "Can't lock memsink");
-			return -1;
-		} else if (result == 0) {
-			if (mem->magic == MEMSINK_MAGIC && mem->version == MEMSINK_VERSION && mem->id != last_id) {
-				return 0;
-			}
-			if (flock(fd, LOCK_UN) < 0) {
-				JLOG_PERROR("video", "Can't unlock memsink");
-				return -1;
-			}
-		}
-		usleep(_g_sink_lock_polling);
-	} while (now < deadline_ts);
-	return -2;
-}
+janus_plugin *create(void);
 
-static int _get_frame(int fd, memsink_shared_s *mem, frame_s *frame, uint64_t *frame_id) {
-	frame_set_data(frame, mem->data, mem->used);
-	FRAME_COPY_META(mem, frame);
-	*frame_id = mem->id;
-	mem->last_client_ts = get_now_monotonic();
-	int retval = 0;
-	if (frame->format != V4L2_PIX_FMT_H264) {
-		JLOG_ERROR("video", "Got non-H264 frame from memsink");
-		retval = -1;
-	}
-	if (flock(fd, LOCK_UN) < 0) {
-		JLOG_PERROR("video", "Can't unlock memsink");
-		retval = -1;
-	}
-	return retval;
-}
-
-static void _relay_rtp_clients(const rtp_s *rtp) {
-	janus_plugin_rtp packet = {0};
-	packet.video = rtp->video;
-	packet.buffer = (char *)rtp->datagram;
-	packet.length = rtp->used;
-	janus_plugin_rtp_extensions_reset(&packet.extensions);
-	LIST_ITERATE(_g_clients, client, {
-		if (client->transmit) {
-			_g_gw->relay_rtp(client->session, &packet);
-		}
-	});
-}
 
 #define IF_NOT_REPORTED(...) { \
 		unsigned _error_code = __LINE__; \
 		if (error_reported != _error_code) { __VA_ARGS__; error_reported = _error_code; } \
 	}
 
-static void *_clients_video_thread(UNUSED void *arg) {
-	A_THREAD_RENAME("us_v_clients");
-	atomic_store(&_g_video_tid_created, true);
-	atomic_store(&_g_ready, true);
+static void *_video_rtp_thread(UNUSED void *arg) {
+	A_THREAD_RENAME("us_video_rtp");
+	atomic_store(&_g_video_rtp_tid_created, true);
+
+	while (!STOP) {
+		frame_s *frame;
+		if (queue_get(_g_video_queue, (void **)&frame, 0.1) == 0) {
+			LOCK_VIDEO;
+			rtpv_wrap(_g_rtpv, frame);
+			UNLOCK_VIDEO;
+			frame_destroy(frame);
+		}
+	}
+	return NULL;
+}
+
+static void *_video_sink_thread(UNUSED void *arg) {
+	A_THREAD_RENAME("us_video_sink");
+	atomic_store(&_g_video_sink_tid_created, true);
 
-	frame_s *frame = frame_init();
 	uint64_t frame_id = 0;
-
 	unsigned error_reported = 0;
 
 	while (!STOP) {
@@ -241,14 +150,16 @@ static void *_clients_video_thread(UNUSED void *arg) {
 
 		JLOG_INFO("video", "Memsink opened; reading frames ...");
 		while (!STOP && HAS_WATCHERS) {
-			int result = _wait_frame(fd, mem, frame_id);
+			int result = memsink_fd_wait_frame(fd, mem, frame_id);
 			if (result == 0) {
-				if (_get_frame(fd, mem, frame, &frame_id) != 0) {
+				frame_s *frame = memsink_fd_get_frame(fd, mem, &frame_id);
+				if (frame == NULL) {
 					goto close_memsink;
 				}
-				LOCK;
-				rtpv_wrap(_g_rtpv, frame);
-				UNLOCK;
+				if (queue_put(_g_video_queue, frame, 0) != 0) {
+					IF_NOT_REPORTED({ JLOG_PERROR("video", "Video queue is full"); });
+					frame_destroy(frame);
+				}
 			} else if (result == -1) {
 				goto close_memsink;
 			}
@@ -266,13 +177,11 @@ static void *_clients_video_thread(UNUSED void *arg) {
 			}
 			sleep(1); // error_delay
 	}
-
-	frame_destroy(frame);
 	return NULL;
 }
 
-static void *_clients_audio_thread(UNUSED void *arg) {
-	A_THREAD_RENAME("us_a_clients");
+static void *_audio_thread(UNUSED void *arg) {
+	A_THREAD_RENAME("us_audio");
 	atomic_store(&_g_audio_tid_created, true);
 	assert(_g_audio_dev_name);
 	assert(_g_tc358743_dev_path);
@@ -316,9 +225,9 @@ static void *_clients_audio_thread(UNUSED void *arg) {
 			uint64_t pts;
 			int result = audio_get_encoded(audio, data, &size, &pts);
 			if (result == 0) {
-				LOCK;
+				LOCK_AUDIO;
 				rtpa_wrap(_g_rtpa, data, size, pts);
-				UNLOCK;
+				UNLOCK_AUDIO;
 			} else if (result == -1) {
 				goto close_audio;
 			}
@@ -335,55 +244,10 @@ static void *_clients_audio_thread(UNUSED void *arg) {
 
 #undef IF_NOT_REPORTED
 
-static char *_get_config_value(janus_config *config, const char *section, const char *option) {
-	janus_config_category *section_obj = janus_config_get_create(config, NULL, janus_config_type_category, section);
-	janus_config_item *option_obj = janus_config_get(config, section_obj, janus_config_type_item, option);
-	if (option_obj == NULL || option_obj->value == NULL || option_obj->value[0] == '\0') {
-		return NULL;
-	}
-	return strdup(option_obj->value);
-}
-
-static int _read_config(const char *config_dir_path) {
-	int retval = 0;
-
-	char *config_file_path;
-	janus_config *config = NULL;
-
-	A_ASPRINTF(config_file_path, "%s/%s.jcfg", config_dir_path, _plugin_get_package());
-	JLOG_INFO("main", "Reading config file '%s' ...", config_file_path);
-
-	config = janus_config_parse(config_file_path);
-	if (config == NULL) {
-		JLOG_ERROR("main", "Can't read config");
-		goto error;
-	}
-	janus_config_print(config);
-
-	if (
-		(_g_video_sink_name = _get_config_value(config, "memsink", "object")) == NULL
-		&& (_g_video_sink_name = _get_config_value(config, "video", "sink")) == NULL
-	) {
-		JLOG_ERROR("main", "Missing config value: video.sink (ex. memsink.object)");
-		goto error;
-	}
-	if ((_g_audio_dev_name = _get_config_value(config, "audio", "device")) != NULL) {
-		JLOG_INFO("main", "Enabled the experimental AUDIO feature");
-		if ((_g_tc358743_dev_path = _get_config_value(config, "audio", "tc358743")) == NULL) {
-			JLOG_INFO("main", "Missing config value: audio.tc358743");
-			goto error;
-		}
-	}
-
-	goto ok;
-	error:
-		retval = -1;
-	ok:
-		if (config) {
-			janus_config_destroy(config);
-		}
-		free(config_file_path);
-		return retval;
+static void _relay_rtp_clients(const rtp_s *rtp) {
+	LIST_ITERATE(_g_clients, client, {
+		client_send(client, rtp);
+	});
 }
 
 static int _plugin_init(janus_callbacks *gw, const char *config_dir_path) {
@@ -394,43 +258,48 @@ static int _plugin_init(janus_callbacks *gw, const char *config_dir_path) {
 	// sysctl -w net.core.wmem_max=1000000
 
 	JLOG_INFO("main", "Initializing plugin ...");
-	assert(!atomic_load(&_g_video_tid_created));
-	assert(!atomic_load(&_g_audio_tid_created));
-	assert(!READY);
-	assert(!STOP);
-	if (gw == NULL || config_dir_path == NULL || _read_config(config_dir_path) < 0) {
+	if (gw == NULL || config_dir_path == NULL || read_config(config_dir_path,
+		&_g_video_sink_name,
+		&_g_audio_dev_name,
+		&_g_tc358743_dev_path
+	) < 0) {
 		return -1;
 	}
 	_g_gw = gw;
+
+	_g_video_queue = queue_init(1024);
 	_g_rtpv = rtpv_init(_relay_rtp_clients);
 	if (_g_audio_dev_name) {
 		_g_rtpa = rtpa_init(_relay_rtp_clients);
-		A_THREAD_CREATE(&_g_audio_tid, _clients_audio_thread, NULL);
+		A_THREAD_CREATE(&_g_audio_tid, _audio_thread, NULL);
 	}
-	A_THREAD_CREATE(&_g_video_tid, _clients_video_thread, NULL);
+	A_THREAD_CREATE(&_g_video_rtp_tid, _video_rtp_thread, NULL);
+	A_THREAD_CREATE(&_g_video_sink_tid, _video_sink_thread, NULL);
+
+	atomic_store(&_g_ready, true);
 	return 0;
 }
 
 static void _plugin_destroy(void) {
 	JLOG_INFO("main", "Destroying plugin ...");
+
 	atomic_store(&_g_stop, true);
-	if (atomic_load(&_g_video_tid_created)) {
-		A_THREAD_JOIN(_g_video_tid);
-	}
-	if (atomic_load(&_g_audio_tid_created)) {
-		A_THREAD_JOIN(_g_audio_tid);
-	}
+#	define JOIN(_tid) { if (atomic_load(&_tid##_created)) { A_THREAD_JOIN(_tid); } }
+	JOIN(_g_video_sink_tid);
+	JOIN(_g_video_rtp_tid);
+	JOIN(_g_audio_tid);
+#	undef JOIN
 
 	LIST_ITERATE(_g_clients, client, {
 		LIST_REMOVE(_g_clients, client);
-		free(client);
+		client_destroy(client);
 	});
-	_g_clients = NULL;
 
-#	define DEL(_func, _var) { if (_var) { _func(_var); _var = NULL; } }
+	QUEUE_FREE_ITEMS_AND_DESTROY(_g_video_queue, frame_destroy);
+
+#	define DEL(_func, _var) { if (_var) { _func(_var); } }
 	DEL(rtpa_destroy, _g_rtpa);
 	DEL(rtpv_destroy, _g_rtpv);
-	_g_gw = NULL;
 	DEL(free, _g_tc358743_dev_path);
 	DEL(free, _g_audio_dev_name);
 	DEL(free, _g_video_sink_name);
@@ -441,30 +310,27 @@ static void _plugin_destroy(void) {
 
 static void _plugin_create_session(janus_plugin_session *session, int *err) {
 	IF_DISABLED({ *err = -1; return; });
-	LOCK;
+	LOCK_ALL;
 	JLOG_INFO("main", "Creating session %p ...", session);
-	_client_s *client;
-	A_CALLOC(client, 1);
-	client->session = session;
-	client->transmit = true;
+	client_s *client = client_init(_g_gw, session, (_g_audio_dev_name != NULL));
 	LIST_APPEND(_g_clients, client);
 	atomic_store(&_g_has_watchers, true);
-	UNLOCK;
+	UNLOCK_ALL;
 }
 
 static void _plugin_destroy_session(janus_plugin_session* session, int *err) {
 	IF_DISABLED({ *err = -1; return; });
-	LOCK;
+	LOCK_ALL;
 	bool found = false;
 	bool has_watchers = false;
 	LIST_ITERATE(_g_clients, client, {
 		if (client->session == session) {
 			JLOG_INFO("main", "Removing session %p ...", session);
 			LIST_REMOVE(_g_clients, client);
-			free(client);
+			client_destroy(client);
 			found = true;
 		} else {
-			has_watchers = (has_watchers || client->transmit);
+			has_watchers = (has_watchers || atomic_load(&client->transmit));
 		}
 	});
 	if (!found) {
@@ -472,41 +338,41 @@ static void _plugin_destroy_session(janus_plugin_session* session, int *err) {
 		*err = -2;
 	}
 	atomic_store(&_g_has_watchers, has_watchers);
-	UNLOCK;
+	UNLOCK_ALL;
 }
 
 static json_t *_plugin_query_session(janus_plugin_session *session) {
 	IF_DISABLED({ return NULL; });
 	json_t *info = NULL;
-	LOCK;
+	LOCK_ALL;
 	LIST_ITERATE(_g_clients, client, {
 		if (client->session == session) {
 			info = json_string("session_found");
 			break;
 		}
 	});
-	UNLOCK;
+	UNLOCK_ALL;
 	return info;
 }
 
 static void _set_transmit(janus_plugin_session *session, UNUSED const char *msg, bool transmit) {
 	IF_DISABLED({ return; });
-	LOCK;
+	LOCK_ALL;
 	bool found = false;
 	bool has_watchers = false;
 	LIST_ITERATE(_g_clients, client, {
 		if (client->session == session) {
-			client->transmit = transmit;
+			atomic_store(&client->transmit, transmit);
 			// JLOG_INFO("main", "%s session %p", msg, session);
 			found = true;
 		}
-		has_watchers = (has_watchers || client->transmit);
+		has_watchers = (has_watchers || atomic_load(&client->transmit));
 	});
 	if (!found) {
 		JLOG_WARN("main", "No session %p", session);
 	}
 	atomic_store(&_g_has_watchers, has_watchers);
-	UNLOCK;
+	UNLOCK_ALL;
 }
 
 #undef IF_DISABLED
@@ -536,7 +402,7 @@ static struct janus_plugin_result *_plugin_handle_message(
 			json_object_set_new(_event, "ustreamer", json_string("event")); \
 			json_object_set_new(_event, "error_code", json_integer(_error)); \
 			json_object_set_new(_event, "error", json_string(_reason)); \
-			_g_gw->push_event(session, &_plugin, transaction, _event, NULL); \
+			_g_gw->push_event(session, create(), transaction, _event, NULL); \
 			json_decref(_event); \
 		}
 
@@ -559,7 +425,7 @@ static struct janus_plugin_result *_plugin_handle_message(
 			json_t *_result = json_object(); \
 			json_object_set_new(_result, "status", json_string(_status)); \
 			json_object_set_new(_event, "result", _result); \
-			_g_gw->push_event(session, &_plugin, transaction, _event, _jsep); \
+			_g_gw->push_event(session, create(), transaction, _event, _jsep); \
 			json_decref(_event); \
 		}
 
@@ -602,6 +468,52 @@ static struct janus_plugin_result *_plugin_handle_message(
 		FREE_MSG_JSEP;
 		return janus_plugin_result_new(JANUS_PLUGIN_OK_WAIT, NULL, NULL);
 
+#	undef PUSH_STATUS
 #	undef PUSH_ERROR
 #	undef FREE_MSG_JSEP
 }
+
+
+// ***** Plugin *****
+
+static int _plugin_get_api_compatibility(void)		{ return JANUS_PLUGIN_API_VERSION; }
+static int _plugin_get_version(void)				{ return VERSION_U; }
+static const char *_plugin_get_version_string(void)	{ return VERSION; }
+static const char *_plugin_get_description(void)	{ return "PiKVM uStreamer Janus plugin for H.264 video"; }
+static const char *_plugin_get_name(void)			{ return PLUGIN_NAME; }
+static const char *_plugin_get_author(void)			{ return "Maxim Devaev <mdevaev@gmail.com>"; }
+static const char *_plugin_get_package(void)		{ return PLUGIN_PACKAGE; }
+
+static void _plugin_incoming_rtp(UNUSED janus_plugin_session *handle, UNUSED janus_plugin_rtp *packet) {
+	// Just a stub to avoid logging spam about the plugin's purpose
+}
+
+janus_plugin *create(void) {
+#	pragma GCC diagnostic push
+#	pragma GCC diagnostic ignored "-Woverride-init"
+	static janus_plugin plugin = JANUS_PLUGIN_INIT(
+		.init = _plugin_init,
+		.destroy = _plugin_destroy,
+
+		.create_session = _plugin_create_session,
+		.destroy_session = _plugin_destroy_session,
+		.query_session = _plugin_query_session,
+
+		.setup_media = _plugin_setup_media,
+		.hangup_media = _plugin_hangup_media,
+
+		.handle_message = _plugin_handle_message,
+
+		.get_api_compatibility = _plugin_get_api_compatibility,
+		.get_version = _plugin_get_version,
+		.get_version_string = _plugin_get_version_string,
+		.get_description = _plugin_get_description,
+		.get_name = _plugin_get_name,
+		.get_author = _plugin_get_author,
+		.get_package = _plugin_get_package,
+
+		.incoming_rtp = _plugin_incoming_rtp,
+	);
+#	pragma GCC diagnostic pop
+	return &plugin;
+}

janus/src/queue.c

@@ -45,11 +45,11 @@ void queue_destroy(queue_s *queue) {
 }
 
 #define WAIT_OR_UNLOCK(_var, _cond) { \
-		struct timespec ts; \
-		assert(!clock_gettime(CLOCK_MONOTONIC, &ts)); \
-		ts.tv_sec += timeout; \
+		struct timespec _ts; \
+		assert(!clock_gettime(CLOCK_MONOTONIC, &_ts)); \
+		ld_to_timespec(timespec_to_ld(&_ts) + timeout, &_ts); \
 		while (_var) { \
-			int err = pthread_cond_timedwait(_cond, &queue->mutex, &ts); \
+			int err = pthread_cond_timedwait(_cond, &queue->mutex, &_ts); \
 			if (err == ETIMEDOUT) { \
 				A_MUTEX_UNLOCK(&queue->mutex); \
 				return -1; \
@@ -58,9 +58,16 @@ void queue_destroy(queue_s *queue) {
 		} \
 	}
 
-int queue_put(queue_s *queue, void *item, unsigned timeout) {
+int queue_put(queue_s *queue, void *item, long double timeout) {
 	A_MUTEX_LOCK(&queue->mutex);
-	WAIT_OR_UNLOCK(queue->size == queue->capacity, &queue->full_cond);
+	if (timeout == 0) {
+		if (queue->size == queue->capacity) {
+			A_MUTEX_UNLOCK(&queue->mutex);
+			return -1;
+		}
+	} else {
+		WAIT_OR_UNLOCK(queue->size == queue->capacity, &queue->full_cond);
+	}
 	queue->items[queue->in] = item;
 	++queue->size;
 	++queue->in;
@@ -70,7 +77,7 @@ int queue_put(queue_s *queue, void *item, unsigned timeout) {
 	return 0;
 }
 
-int queue_get(queue_s *queue, void **item, unsigned timeout) {
+int queue_get(queue_s *queue, void **item, long double timeout) {
 	A_MUTEX_LOCK(&queue->mutex);
 	WAIT_OR_UNLOCK(queue->size == 0, &queue->empty_cond);
 	*item = queue->items[queue->out];

janus/src/queue.h

@@ -24,6 +24,7 @@
 
 #include <errno.h>
 #include <time.h>
+#include <assert.h>
 
 #include <pthread.h>
 
@@ -46,9 +47,21 @@ typedef struct {
 } queue_s;
 
 
+#define QUEUE_FREE_ITEMS_AND_DESTROY(_queue, _free_item) { \
+		while (!queue_get_free(_queue)) { \
+			void *_ptr; \
+			assert(!queue_get(_queue, &_ptr, 0.1)); \
+			if (_ptr != NULL) { \
+				_free_item(_ptr); \
+			} \
+		} \
+		queue_destroy(_queue); \
+	}
+
+
 queue_s *queue_init(unsigned capacity);
 void queue_destroy(queue_s *queue);
 
-int queue_put(queue_s *queue, void *item, unsigned timeout);
-int queue_get(queue_s *queue, void **item, unsigned timeout);
+int queue_put(queue_s *queue, void *item, long double timeout);
+int queue_get(queue_s *queue, void **item, long double timeout);
 int queue_get_free(queue_s *queue);

janus/src/rtp.c

@@ -35,6 +35,13 @@ rtp_s *rtp_init(unsigned payload, bool video) {
 	return rtp;
 }
 
+rtp_s *rtp_dup(const rtp_s *rtp) {
+	rtp_s *new;
+	A_CALLOC(new, 1);
+	memcpy(new, rtp, sizeof(rtp_s));
+	return new;
+}
+
 void rtp_destroy(rtp_s *rtp) {
 	free(rtp);
 }

janus/src/rtp.h

@@ -50,6 +50,7 @@ typedef void (*rtp_callback_f)(const rtp_s *rtp);
 
 
 rtp_s *rtp_init(unsigned payload, bool video);
+rtp_s *rtp_dup(const rtp_s *rtp);
 void rtp_destroy(rtp_s *rtp);
 
 void rtp_write_header(rtp_s *rtp, uint32_t pts, bool marked);

janus/src/rtpv.c

@@ -81,6 +81,8 @@ char *rtpv_make_sdp(rtpv_s *rtpv) {
 		"a=rtcp-fb:%u nack pli" RN
 		"a=rtcp-fb:%u goog-remb" RN
 		"a=ssrc:%" PRIu32 " cname:ustreamer" RN
+		// XXX: See client.c
+		// "a=extmap:1 http://www.webrtc.org/experiments/rtp-hdrext/playout-delay" RN
 		"a=sendonly" RN,
 		PAYLOAD, PAYLOAD, PAYLOAD, PAYLOAD,
 		PAYLOAD, sps,

janus/src/tc358743.h

@@ -34,7 +34,7 @@
 #include "uslibs/tools.h"
 #include "uslibs/xioctl.h"
 
-#include "jlogging.h"
+#include "logging.h"
 
 
 typedef struct {

janus/src/uslibs/config.h

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

janus/src/uslibs/const.h

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

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 5.12" "January 2021"
+.TH USTREAMER-DUMP 1 "version 5.15" "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 5.12" "November 2020"
+.TH USTREAMER 1 "version 5.15" "November 2020"
 
 .SH NAME
 ustreamer \- stream MJPEG video from any V4L2 device to the network

pkg/arch/PKGBUILD

@@ -3,7 +3,7 @@
 
 
 pkgname=ustreamer
-pkgver=5.12
+pkgver=5.15
 pkgrel=1
 pkgdesc="Lightweight and fast MJPEG-HTTP streamer"
 url="https://github.com/pikvm/ustreamer"

pkg/openwrt/Makefile

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

python/setup.py

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

src/dump/main.c

@@ -31,7 +31,7 @@
 #include <errno.h>
 #include <assert.h>
 
-#include "../libs/config.h"
+#include "../libs/const.h"
 #include "../libs/tools.h"
 #include "../libs/logging.h"
 #include "../libs/frame.h"

src/libs/const.h

@@ -23,7 +23,7 @@
 #pragma once
 
 #define VERSION_MAJOR 5
-#define VERSION_MINOR 12
+#define VERSION_MINOR 15
 
 #define MAKE_VERSION2(_major, _minor) #_major "." #_minor
 #define MAKE_VERSION1(_major, _minor) MAKE_VERSION2(_major, _minor)

src/libs/tools.h

@@ -147,6 +147,19 @@ INLINE unsigned get_cores_available(void) {
 	return max_u(min_u(cores_sysconf, 4), 1);
 }
 
+INLINE void ld_to_timespec(long double ld, struct timespec *ts) {
+	ts->tv_sec = (long)ld;
+	ts->tv_nsec = (ld - ts->tv_sec) * 1000000000L;
+	if (ts->tv_nsec > 999999999L) {
+		ts->tv_sec += 1;
+		ts->tv_nsec = 0;
+	}
+}
+
+INLINE long double timespec_to_ld(const struct timespec *ts) {
+	return ts->tv_sec + ((long double)ts->tv_nsec) / 1000000000;
+}
+
 INLINE int flock_timedwait_monotonic(int fd, long double timeout) {
 	long double deadline_ts = get_now_monotonic() + timeout;
 	int retval = -1;

src/ustreamer/data/index_html.h

@@ -24,7 +24,7 @@
 
 #include <sys/types.h>
 
-#include "../../libs/config.h"
+#include "../../libs/const.h"
 
 
 extern const char *const HTML_INDEX_PAGE;

src/ustreamer/m2m.c

@@ -25,7 +25,7 @@
 
 static m2m_encoder_s *_m2m_encoder_init(
 	const char *name, const char *path, unsigned output_format,
-	unsigned fps, bool allow_dma, m2m_option_s *options);
+	unsigned fps, unsigned bitrate, unsigned gop, unsigned quality, bool allow_dma);
 
 static void _m2m_encoder_prepare(m2m_encoder_s *enc, const frame_s *frame);
 
@@ -46,26 +46,11 @@ static int _m2m_encoder_compress_raw(m2m_encoder_s *enc, const frame_s *src, fra
 
 
 m2m_encoder_s *m2m_h264_encoder_init(const char *name, const char *path, unsigned bitrate, unsigned gop) {
-#	define OPTION(_key, _value) {#_key, V4L2_CID_MPEG_VIDEO_##_key, _value}
-
-	m2m_option_s options[] = {
-		OPTION(BITRATE, bitrate * 1000),
-		// OPTION(BITRATE_PEAK, bitrate * 1000),
-		OPTION(H264_I_PERIOD, gop),
-		OPTION(H264_PROFILE, V4L2_MPEG_VIDEO_H264_PROFILE_CONSTRAINED_BASELINE),
-		OPTION(H264_LEVEL, V4L2_MPEG_VIDEO_H264_LEVEL_4_0),
-		OPTION(REPEAT_SEQ_HEADER, 1),
-		OPTION(H264_MIN_QP, 16),
-		OPTION(H264_MAX_QP, 32),
-		{NULL, 0, 0},
-	};
-
-#	undef OPTION
-
 	// FIXME: 30 or 0? https://github.com/6by9/yavta/blob/master/yavta.c#L2100
 	// По логике вещей правильно 0, но почему-то на низких разрешениях типа 640x480
 	// енкодер через несколько секунд перестает производить корректные фреймы.
-	return _m2m_encoder_init(name, path, V4L2_PIX_FMT_H264, 30, true, options);
+	bitrate *= 1000; // From Kbps
+	return _m2m_encoder_init(name, path, V4L2_PIX_FMT_H264, 30, bitrate, gop, 0, true);
 }
 
 m2m_encoder_s *m2m_mjpeg_encoder_init(const char *name, const char *path, unsigned quality) {
@@ -76,30 +61,18 @@ m2m_encoder_s *m2m_mjpeg_encoder_init(const char *name, const char *path, unsign
 	bitrate = step * round(bitrate / step);
 	bitrate *= 1000; // From Kbps
 	assert(bitrate > 0);
-
-	m2m_option_s options[] = {
-		{"BITRATE", V4L2_CID_MPEG_VIDEO_BITRATE, bitrate},
-		{NULL, 0, 0},
-	};
-
 	// FIXME: То же самое про 30 or 0, но еще даже не проверено на низких разрешениях
-	return _m2m_encoder_init(name, path, V4L2_PIX_FMT_MJPEG, 30, true, options);
+	return _m2m_encoder_init(name, path, V4L2_PIX_FMT_MJPEG, 30, bitrate, 0, 0, true);
 }
 
 m2m_encoder_s *m2m_jpeg_encoder_init(const char *name, const char *path, unsigned quality) {
-	m2m_option_s options[] = {
-		{"QUALITY", V4L2_CID_JPEG_COMPRESSION_QUALITY, quality},
-		{NULL, 0, 0},
-	};
-
 	// FIXME: DMA не работает
-	return _m2m_encoder_init(name, path, V4L2_PIX_FMT_JPEG, 30, false, options);
+	return _m2m_encoder_init(name, path, V4L2_PIX_FMT_JPEG, 30, 0, 0, quality, false);
 }
 
 void m2m_encoder_destroy(m2m_encoder_s *enc) {
 	E_LOG_INFO("Destroying encoder ...");
 	_m2m_encoder_cleanup(enc);
-	free(enc->options);
 	free(enc->path);
 	free(enc->name);
 	free(enc);
@@ -142,7 +115,7 @@ int m2m_encoder_compress(m2m_encoder_s *enc, const frame_s *src, frame_s *dest,
 
 static m2m_encoder_s *_m2m_encoder_init(
 	const char *name, const char *path, unsigned output_format,
-	unsigned fps, bool allow_dma, m2m_option_s *options) {
+	unsigned fps, unsigned bitrate, unsigned gop, unsigned quality, bool allow_dma) {
 
 	LOG_INFO("%s: Initializing encoder ...", name);
 
@@ -161,15 +134,11 @@ static m2m_encoder_s *_m2m_encoder_init(
 	}
 	enc->output_format = output_format;
 	enc->fps = fps;
+	enc->bitrate = bitrate;
+	enc->gop = gop;
+	enc->quality = quality;
 	enc->allow_dma = allow_dma;
 	enc->run = run;
-
-	unsigned count = 0;
-	for (; options[count].name != NULL; ++count);
-	++count;
-	A_CALLOC(enc->options, count);
-	memcpy(enc->options, options, sizeof(m2m_option_s) * count);
-
 	return enc;
 }
 
@@ -199,15 +168,34 @@ static void _m2m_encoder_prepare(m2m_encoder_s *enc, const frame_s *frame) {
 	}
 	E_LOG_DEBUG("Encoder device fd=%d opened", RUN(fd));
 
-	for (m2m_option_s *option = enc->options; option->name != NULL; ++option) {
-		struct v4l2_control ctl = {0};
-		ctl.id = option->id;
-		ctl.value = option->value;
+#	define SET_OPTION(_cid, _value) { \
+			struct v4l2_control _ctl = {0}; \
+			_ctl.id = _cid; \
+			_ctl.value = _value; \
+			E_LOG_DEBUG("Configuring option " #_cid " ..."); \
+			E_XIOCTL(VIDIOC_S_CTRL, &_ctl, "Can't set option " #_cid); \
+		}
 
-		E_LOG_DEBUG("Configuring option %s ...", option->name);
-		E_XIOCTL(VIDIOC_S_CTRL, &ctl, "Can't set option %s", option->name);
+	if (enc->output_format == V4L2_PIX_FMT_H264) {
+		SET_OPTION(V4L2_CID_MPEG_VIDEO_BITRATE,				enc->bitrate);
+		SET_OPTION(V4L2_CID_MPEG_VIDEO_H264_I_PERIOD,		enc->gop);
+		SET_OPTION(V4L2_CID_MPEG_VIDEO_H264_PROFILE,		V4L2_MPEG_VIDEO_H264_PROFILE_CONSTRAINED_BASELINE);
+		if (RUN(width) * RUN(height) <= 1920 * 1080) { // https://forums.raspberrypi.com/viewtopic.php?t=291447#p1762296
+			SET_OPTION(V4L2_CID_MPEG_VIDEO_H264_LEVEL,		V4L2_MPEG_VIDEO_H264_LEVEL_4_0);
+		} else {
+			SET_OPTION(V4L2_CID_MPEG_VIDEO_H264_LEVEL,		V4L2_MPEG_VIDEO_H264_LEVEL_5_1);
+		}
+		SET_OPTION(V4L2_CID_MPEG_VIDEO_REPEAT_SEQ_HEADER,	1);
+		SET_OPTION(V4L2_CID_MPEG_VIDEO_H264_MIN_QP,			16);
+		SET_OPTION(V4L2_CID_MPEG_VIDEO_H264_MAX_QP,			32);
+	} else if (enc->output_format == V4L2_PIX_FMT_MJPEG) {
+		SET_OPTION(V4L2_CID_MPEG_VIDEO_BITRATE,				enc->bitrate);
+	} else if (enc->output_format == V4L2_PIX_FMT_JPEG) {
+		SET_OPTION(V4L2_CID_JPEG_COMPRESSION_QUALITY,		enc->quality);
 	}
 
+#	undef SET_OPTION
+
 	{
 		struct v4l2_format fmt = {0};
 		fmt.type = V4L2_BUF_TYPE_VIDEO_OUTPUT_MPLANE;

src/ustreamer/m2m.h

@@ -47,12 +47,6 @@ typedef struct {
 	size_t	allocated;
 } m2m_buffer_s;
 
-typedef struct {
-	char		*name;
-	uint32_t	id;
-	int32_t		value;
-} m2m_option_s;
-
 typedef struct {
 	int				fd;
 	m2m_buffer_s	*input_bufs;
@@ -75,8 +69,10 @@ typedef struct {
 	char			*path;
 	unsigned		output_format;
 	unsigned		fps;
+	unsigned		bitrate;
+	unsigned		gop;
+	unsigned		quality;
 	bool			allow_dma;
-	m2m_option_s	*options;
 
 	m2m_encoder_runtime_s *run;
 } m2m_encoder_s;

src/ustreamer/options.h

@@ -32,7 +32,7 @@
 #include <errno.h>
 #include <assert.h>
 
-#include "../libs/config.h"
+#include "../libs/const.h"
 #include "../libs/logging.h"
 #include "../libs/process.h"
 #include "../libs/frame.h"