memsink: suffix-based memory limites

janus/src/memsinkfd.c

@@ -58,7 +58,7 @@ int us_memsink_fd_wait_frame(int fd, us_memsink_shared_s *mem, u64 last_id) {
 }
 
 int us_memsink_fd_get_frame(int fd, us_memsink_shared_s *mem, us_frame_s *frame, u64 *frame_id, bool key_required) {
-	us_frame_set_data(frame, mem->data, mem->used);
+	us_frame_set_data(frame, us_memsink_get_data(mem), mem->used);
 	US_FRAME_COPY_META(mem, frame);
 	*frame_id = mem->id;
 	mem->last_client_ts = us_get_now_monotonic();

janus/src/plugin.c

@@ -136,12 +136,18 @@ static void *_video_sink_thread(void *arg) {
 		int fd = -1;
 		us_memsink_shared_s *mem = NULL;
 
+		const uz data_size = us_memsink_calculate_size(_g_config->video_sink_name);
+		if (data_size == 0) {
+			US_ONCE({ US_JLOG_ERROR("video", "Invalid memsink object suffix"); });
+			goto close_memsink;
+		}
+
 		if ((fd = shm_open(_g_config->video_sink_name, O_RDWR, 0)) <= 0) {
 			US_ONCE({ US_JLOG_PERROR("video", "Can't open memsink"); });
 			goto close_memsink;
 		}
 
-		if ((mem = us_memsink_shared_map(fd)) == NULL) {
+		if ((mem = us_memsink_shared_map(fd, data_size)) == NULL) {
 			US_ONCE({ US_JLOG_PERROR("video", "Can't map memsink"); });
 			goto close_memsink;
 		}
@@ -178,7 +184,10 @@ static void *_video_sink_thread(void *arg) {
 		}
 
 	close_memsink:
-		US_DELETE(mem, us_memsink_shared_unmap);
+		if (mem != NULL) {
+			us_memsink_shared_unmap(mem, data_size);
+			mem = NULL;
+		}
 		US_CLOSE_FD(fd);
 		US_JLOG_INFO("video", "Memsink closed");
 		sleep(1); // error_delay

python/src/ustreamer.c

@@ -26,6 +26,7 @@ typedef struct {
 	double	lock_timeout;
 	double	wait_timeout;
 	double	drop_same_frames;
+	uz		data_size;
 
 	int					fd;
 	us_memsink_shared_s	*mem;
@@ -37,7 +38,10 @@ typedef struct {
 
 
 static void _MemsinkObject_destroy_internals(_MemsinkObject *self) {
-	US_DELETE(self->mem, us_memsink_shared_unmap);
+	if (self->mem != NULL) {
+		us_memsink_shared_unmap(self->mem, self->data_size);
+		self->mem = NULL;
+	}
 	US_CLOSE_FD(self->fd);
 	US_DELETE(self->frame, us_frame_destroy);
 }
@@ -64,13 +68,18 @@ static int _MemsinkObject_init(_MemsinkObject *self, PyObject *args, PyObject *k
 	SET_DOUBLE(drop_same_frames, >= 0);
 #	undef SET_DOUBLE
 
+	if ((self->data_size = us_memsink_calculate_size(self->obj)) == 0) {
+		PyErr_SetString(PyExc_ValueError, "Invalid memsink object suffix");
+		return -1;
+	}
+
 	self->frame = us_frame_init();
 
 	if ((self->fd = shm_open(self->obj, O_RDWR, 0)) == -1) {
 		PyErr_SetFromErrno(PyExc_OSError);
 		goto error;
 	}
-	if ((self->mem = us_memsink_shared_map(self->fd)) == NULL) {
+	if ((self->mem = us_memsink_shared_map(self->fd, self->data_size)) == NULL) {
 		PyErr_SetFromErrno(PyExc_OSError);
 		goto error;
 	}
@@ -139,7 +148,7 @@ static int _wait_frame(_MemsinkObject *self) {
 			if (
 				US_FRAME_COMPARE_GEOMETRY(self->mem, self->frame)
 				&& (self->frame_ts + self->drop_same_frames > now_ts)
-				&& !memcmp(self->frame->data, mem->data, mem->used)
+				&& !memcmp(self->frame->data, us_memsink_get_data(mem), mem->used)
 			) {
 				self->frame_id = mem->id;
 				goto retry;
@@ -190,7 +199,7 @@ static PyObject *_MemsinkObject_wait_frame(_MemsinkObject *self, PyObject *args,
 	}
 
 	us_memsink_shared_s *mem = self->mem;
-	us_frame_set_data(self->frame, mem->data, mem->used);
+	us_frame_set_data(self->frame, us_memsink_get_data(mem), mem->used);
 	US_FRAME_COPY_META(self->mem, self->frame);
 	self->frame_id = mem->id;
 	self->frame_ts = us_get_now_monotonic();

src/libs/memsink.c

@@ -56,6 +56,11 @@ us_memsink_s *us_memsink_init(
 
 	US_LOG_INFO("Using %s-sink: %s", name, obj);
 
+	if ((sink->data_size = us_memsink_calculate_size(obj)) == 0) {
+		US_LOG_ERROR("%s-sink: Invalid object suffix", name);
+		goto error;
+	}
+
 	const mode_t mask = umask(0);
 	sink->fd = shm_open(sink->obj, (server ? O_RDWR | O_CREAT : O_RDWR), mode);
 	umask(mask);
@@ -65,12 +70,12 @@ us_memsink_s *us_memsink_init(
 		goto error;
 	}
 
-	if (sink->server && ftruncate(sink->fd, sizeof(us_memsink_shared_s)) < 0) {
+	if (sink->server && ftruncate(sink->fd, sizeof(us_memsink_shared_s) + sink->data_size) < 0) {
 		US_LOG_PERROR("%s-sink: Can't truncate shared memory", name);
 		goto error;
 	}
 
-	if ((sink->mem = us_memsink_shared_map(sink->fd)) == NULL) {
+	if ((sink->mem = us_memsink_shared_map(sink->fd, sink->data_size)) == NULL) {
 		US_LOG_PERROR("%s-sink: Can't mmap shared memory", name);
 		goto error;
 	}
@@ -83,7 +88,7 @@ error:
 
 void us_memsink_destroy(us_memsink_s *sink) {
 	if (sink->mem != NULL) {
-		if (us_memsink_shared_unmap(sink->mem) < 0) {
+		if (us_memsink_shared_unmap(sink->mem, sink->data_size) < 0) {
 			US_LOG_PERROR("%s-sink: Can't unmap shared memory", sink->name);
 		}
 	}
@@ -160,9 +165,9 @@ int us_memsink_server_put(us_memsink_s *sink, const us_frame_s *frame, bool *key
 
 	const ldf now = us_get_now_monotonic();
 
-	if (frame->used > US_MEMSINK_MAX_DATA) {
+	if (frame->used > sink->data_size) {
 		US_LOG_ERROR("%s-sink: Can't put frame: is too big (%zu > %zu)",
-			sink->name, frame->used, US_MEMSINK_MAX_DATA);
+			sink->name, frame->used, sink->data_size);
 		return 0; // -2
 	}
 
@@ -177,7 +182,7 @@ int us_memsink_server_put(us_memsink_s *sink, const us_frame_s *frame, bool *key
 			*key_requested = sink->mem->key_requested;
 		}
 
-		memcpy(sink->mem->data, frame->data, frame->used);
+		memcpy(us_memsink_get_data(sink->mem), frame->data, frame->used);
 		sink->mem->used = frame->used;
 		US_FRAME_COPY_META(frame, sink->mem);
 
@@ -236,7 +241,7 @@ int us_memsink_client_get(us_memsink_s *sink, us_frame_s *frame, bool *key_reque
 	}
 
 	sink->last_readed_id = sink->mem->id;
-	us_frame_set_data(frame, sink->mem->data, sink->mem->used);
+	us_frame_set_data(frame, us_memsink_get_data(sink->mem), sink->mem->used);
 	US_FRAME_COPY_META(sink->mem, frame);
 	if (key_requested != NULL) { // We don't need it for non-H264 sinks
 		*key_requested = sink->mem->key_requested;

src/libs/memsink.h

@@ -34,6 +34,7 @@
 typedef struct {
 	const char	*name;
 	const char	*obj;
+	uz			data_size;
 	bool		server;
 	bool		rm;
 	uint		client_ttl; // Only for server

src/libs/memsinksh.c

@@ -22,6 +22,8 @@
 
 #include "memsinksh.h"
 
+#include <string.h>
+#include <strings.h>
 #include <assert.h>
 
 #include <sys/mman.h>
@@ -29,10 +31,10 @@
 #include "types.h"
 
 
-us_memsink_shared_s *us_memsink_shared_map(int fd) {
+us_memsink_shared_s *us_memsink_shared_map(int fd, uz data_size) {
 	us_memsink_shared_s *mem = mmap(
 		NULL,
-		sizeof(us_memsink_shared_s),
+		sizeof(us_memsink_shared_s) + data_size,
 		PROT_READ | PROT_WRITE, MAP_SHARED,
 		fd, 0);
 	if (mem == MAP_FAILED) {
@@ -42,7 +44,29 @@ us_memsink_shared_s *us_memsink_shared_map(int fd) {
 	return mem;
 }
 
-int us_memsink_shared_unmap(us_memsink_shared_s *mem) {
+int us_memsink_shared_unmap(us_memsink_shared_s *mem, uz data_size) {
 	assert(mem != NULL);
-	return munmap(mem, sizeof(us_memsink_shared_s));
+	return munmap(mem, sizeof(us_memsink_shared_s) + data_size);
+}
+
+uz us_memsink_calculate_size(const char *obj) {
+	const char *ptr = strrchr(obj, ':');
+	if (ptr == NULL) {
+		ptr = strrchr(obj, '.');
+	}
+	if (ptr != NULL) {
+		ptr += 1;
+		if (!strcasecmp(ptr, "jpeg")) {
+			return 4 * 1024 * 1024;
+		} else if (!strcasecmp(ptr, "h264")) {
+			return 2 * 1024 * 1024;
+		} else if (!strcasecmp(ptr, "raw")) {
+			return 1920 * 1200 * 3; // RGB
+		}
+	}
+	return 0;
+}
+
+u8 *us_memsink_get_data(us_memsink_shared_s *mem) {
+	return (u8*)(mem + sizeof(us_memsink_shared_s));
 }

src/libs/memsinksh.h

@@ -26,12 +26,7 @@
 
 
 #define US_MEMSINK_MAGIC	((u64)0xCAFEBABECAFEBABE)
-#define US_MEMSINK_VERSION	((u32)4)
+#define US_MEMSINK_VERSION	((u32)5)
-
-#ifndef US_CFG_MEMSINK_MAX_DATA
-#	define US_CFG_MEMSINK_MAX_DATA 33554432
-#endif
-#define US_MEMSINK_MAX_DATA ((uz)(US_CFG_MEMSINK_MAX_DATA))
 
 
 typedef struct {
@@ -55,10 +50,11 @@ typedef struct {
 
 	ldf		last_client_ts;
 	bool	key_requested;
-
-	u8		data[US_MEMSINK_MAX_DATA];
 } us_memsink_shared_s;
 
 
-us_memsink_shared_s *us_memsink_shared_map(int fd);
+us_memsink_shared_s *us_memsink_shared_map(int fd, uz data_size);
-int us_memsink_shared_unmap(us_memsink_shared_s *mem);
+int us_memsink_shared_unmap(us_memsink_shared_s *mem, uz data_size);
+
+uz us_memsink_calculate_size(const char *obj);
+u8 *us_memsink_get_data(us_memsink_shared_s *mem);