enabled slowdown logic for sinks

src/libs/memsink.c

@@ -113,6 +113,7 @@ 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());
 		memcpy(sink->mem->data, frame->data, frame->used);
 #		undef COPY
 
@@ -159,6 +160,7 @@ int memsink_client_get(memsink_s *sink, frame_s *frame) { // cppcheck-suppress u
 		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
 	}

src/libs/memsink.h

@@ -56,6 +56,7 @@ typedef struct {
 	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;
 
@@ -69,6 +70,7 @@ typedef struct {
 	int					fd;
 	memsink_shared_s	*mem;
 	uint64_t			last_id;
+	bool				has_clients; // Only for server
 } memsink_s;
 
 

src/ustreamer/http/server.c

@@ -139,10 +139,6 @@ int server_listen(server_s *server) {
 		assert(!event_add(RUN(refresh), &refresh_interval));
 	}
 
-	if (server->slowdown) {
-		stream_switch_slowdown(RUN(stream), true);
-	}
-
 	evhttp_set_timeout(RUN(http), server->timeout);
 
 	if (server->user[0] != '\0') {
@@ -476,10 +472,7 @@ static void _http_callback_stream(struct evhttp_request *request, void *v_server
 		RUN(stream_clients_count) += 1;
 
 		if (RUN(stream_clients_count) == 1) {
-			if (server->slowdown) {
-				stream_switch_slowdown(RUN(stream), false);
-			}
-
+			atomic_store(&VID(has_clients), true);
 #			ifdef WITH_GPIO
 			gpio_set_has_http_clients(true);
 #			endif
@@ -650,10 +643,7 @@ static void _http_callback_stream_error(UNUSED struct bufferevent *buf_event, UN
 	RUN(stream_clients_count) -= 1;
 
 	if (RUN(stream_clients_count) == 0) {
-		if (client->server->slowdown) {
-			stream_switch_slowdown(RUN(stream), true);
-		}
-
+		atomic_store(&VID(has_clients), false);
 #		ifdef WITH_GPIO
 		gpio_set_has_http_clients(false);
 #		endif

src/ustreamer/http/server.h

@@ -135,7 +135,6 @@ typedef struct server_sx {
 	char		*allow_origin;
 
 	unsigned	drop_same_frames;
-	bool		slowdown;
 	unsigned	fake_width;
 	unsigned	fake_height;
 

src/ustreamer/options.c

@@ -41,16 +41,16 @@ enum _OPT_VALUES {
 #	ifdef WITH_OMX
 	_O_GLITCHED_RESOLUTIONS = 'g',
 #	endif
+	_O_BLANK = 'k',
+	_O_LAST_AS_BLANK = 'K',
+	_O_SLOWDOWN = 'l',
 
 	_O_HOST = 's',
 	_O_PORT = 'p',
 	_O_UNIX = 'U',
 	_O_UNIX_RM = 'D',
 	_O_UNIX_MODE = 'M',
-	_O_BLANK = 'k',
-	_O_LAST_AS_BLANK = 'K',
 	_O_DROP_SAME_FRAMES = 'e',
-	_O_SLOWDOWN = 'l',
 	_O_FAKE_RESOLUTION = 'R',
 
 	_O_HELP = 'h',
@@ -141,6 +141,7 @@ static const struct option _LONG_OPTS[] = {
 #	endif
 	{"blank",					required_argument,	NULL,	_O_BLANK},
 	{"last-as-blank",			required_argument,	NULL,	_O_LAST_AS_BLANK},
+	{"slowdown",				no_argument,		NULL,	_O_SLOWDOWN},
 	{"device-timeout",			required_argument,	NULL,	_O_DEVICE_TIMEOUT},
 	{"device-error-delay",		required_argument,	NULL,	_O_DEVICE_ERROR_DELAY},
 
@@ -167,7 +168,6 @@ static const struct option _LONG_OPTS[] = {
 	{"passwd",					required_argument,	NULL,	_O_PASSWD},
 	{"static",					required_argument,	NULL,	_O_STATIC},
 	{"drop-same-frames",		required_argument,	NULL,	_O_DROP_SAME_FRAMES},
-	{"slowdown",				no_argument,		NULL,	_O_SLOWDOWN},
 	{"allow-origin",			required_argument,	NULL,	_O_ALLOW_ORIGIN},
 	{"fake-resolution",			required_argument,	NULL,	_O_FAKE_RESOLUTION},
 	{"tcp-nodelay",				no_argument,		NULL,	_O_TCP_NODELAY},
@@ -382,6 +382,7 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 #			endif
 			case _O_BLANK:				OPT_SET(blank_path, optarg);
 			case _O_LAST_AS_BLANK:		OPT_NUMBER("--last-as-blank", stream->last_as_blank, 0, 86400, 0);
+			case _O_SLOWDOWN:			OPT_SET(stream->slowdown, true);
 			case _O_DEVICE_TIMEOUT:		OPT_NUMBER("--device-timeout", dev->timeout, 1, 60, 0);
 			case _O_DEVICE_ERROR_DELAY:	OPT_NUMBER("--device-error-delay", stream->error_delay, 1, 60, 0);
 
@@ -421,7 +422,6 @@ int options_parse(options_s *options, device_s *dev, encoder_s *enc, stream_s *s
 			case _O_PASSWD:				OPT_SET(server->passwd, optarg);
 			case _O_STATIC:				OPT_SET(server->static_path, optarg);
 			case _O_DROP_SAME_FRAMES:	OPT_NUMBER("--drop-same-frames", server->drop_same_frames, 0, VIDEO_MAX_FPS, 0);
-			case _O_SLOWDOWN:			OPT_SET(server->slowdown, true);
 			case _O_FAKE_RESOLUTION:	OPT_RESOLUTION("--fake-resolution", server->fake_width, server->fake_height, false);
 			case _O_ALLOW_ORIGIN:		OPT_SET(server->allow_origin, optarg);
 			case _O_TCP_NODELAY:		OPT_SET(server->tcp_nodelay, true);
@@ -617,6 +617,8 @@ static void _help(FILE *fp, device_s *dev, encoder_s *enc, stream_s *stream, ser
 	SAY("                                           but no more than specified time (or endlessly if 0 is specified).");
 	SAY("                                           If the device has not yet been online, display 'NO SIGNAL' or the image");
 	SAY("                                           specified by option --blank. Default: disabled.\n");
+	SAY("    -l|--slowdown  ─────────────────────── Slowdown capturing to 1 FPS or less when no stream or sink clients");
+	SAY("                                           are connected. Useful to reduce CPU consumption. Default: disabled.\n");
 	SAY("    --device-timeout <sec>  ────────────── Timeout for device querying. Default: %u.\n", dev->timeout);
 	SAY("    --device-error-delay <sec>  ────────── Delay before trying to connect to the device again");
 	SAY("                                           after an error (timeout for example). Default: %u.\n", stream->error_delay);
@@ -651,8 +653,6 @@ static void _help(FILE *fp, device_s *dev, encoder_s *enc, stream_s *stream, ser
 	SAY("                                  It can significantly reduce the outgoing traffic, but will increase");
 	SAY("                                  the CPU loading. Don't use this option with analog signal sources");
 	SAY("                                  or webcams, it's useless. Default: disabled.\n");
-	SAY("    -l|--slowdown  ────────────── Slowdown capturing to 1 FPS or less when no stream clients are connected.");
-	SAY("                                  Useful to reduce CPU consumption. Default: disabled.\n");
 	SAY("    -R|--fake-resolution <WxH>  ─ Override image resolution for the /state. Default: disabled.\n");
 	SAY("    --tcp-nodelay  ────────────── Set TCP_NODELAY flag to the client /stream socket. Ignored for --unix.");
 	SAY("                                  Default: disabled.\n");

src/ustreamer/stream.c

@@ -41,13 +41,13 @@ stream_s *stream_init(device_s *dev, encoder_s *enc) {
 	stream_runtime_s *run;
 	A_CALLOC(run, 1);
 	atomic_init(&run->stop, false);
-	atomic_init(&run->slowdown, false);
 
 	video_s *video;
 	A_CALLOC(video, 1);
 	video->frame = frame_init("stream_video");
 	atomic_init(&video->updated, false);
 	A_MUTEX_INIT(&video->mutex);
+	atomic_init(&video->has_clients, false);
 	run->video = video;
 
 	stream_s *stream;
@@ -122,7 +122,14 @@ void stream_loop(stream_s *stream) {
 				break;
 			}
 
-			if (atomic_load(&RUN(slowdown))) {
+			if (
+				stream->slowdown
+				&& !atomic_load(&RUN(video->has_clients))
+				&& (stream->sink == NULL || !stream->sink->has_clients)
+#				ifdef WITH_OMX
+				&& (stream->h264 == NULL || /*stream->h264->sink == NULL ||*/ !stream->h264->sink->has_clients)
+#				endif
+			) {
 				usleep(1000000);
 			}
 
@@ -226,10 +233,6 @@ void stream_loop_break(stream_s *stream) {
 	atomic_store(&RUN(stop), true);
 }
 
-void stream_switch_slowdown(stream_s *stream, bool slowdown) {
-	atomic_store(&RUN(slowdown), slowdown);
-}
-
 static workers_pool_s *_stream_init_loop(stream_s *stream) {
 
 	workers_pool_s *pool = NULL;

src/ustreamer/stream.h

@@ -56,6 +56,8 @@ typedef struct {
 	unsigned		captured_fps;
 	atomic_bool		updated;
 	pthread_mutex_t	mutex;
+
+	atomic_bool		has_clients; // For slowdown
 } video_s;
 
 #ifdef WITH_OMX
@@ -76,17 +78,16 @@ typedef struct {
 #	endif
 
 	atomic_bool stop;
-	atomic_bool slowdown;
 } stream_runtime_s;
 
 typedef struct {
 	device_s	*dev;
 	encoder_s	*enc;
 
-	int			last_as_blank;
-	unsigned	error_delay;
-
 	frame_s		*blank;
+	int			last_as_blank;
+	bool		slowdown;
+	unsigned	error_delay;
 
 	memsink_s	*sink;
 
@@ -105,4 +106,3 @@ void stream_destroy(stream_s *stream);
 
 void stream_loop(stream_s *stream);
 void stream_loop_break(stream_s *stream);
-void stream_switch_slowdown(stream_s *stream, bool slowdown);