Implement (optional) rate-limiting for sACN

3 files changed, 74 insertions, 25 deletions

diff --git a/backends/sacn.c b/backends/sacn.c
index 5096123..6c08b5a 100644
--- a/backends/sacn.c
+++ b/backends/sacn.c
@@ -28,12 +28,14 @@ static struct /*_sacn_global_config*/ {
 	size_t fds;
 	sacn_fd* fd;
 	uint64_t last_announce;
+	uint32_t next_frame;
 } global_cfg = {
 	.source_name = "MIDIMonster",
 	.cid = {'M', 'I', 'D', 'I', 'M', 'o', 'n', 's', 't', 'e', 'r'},
 	.fds = 0,
 	.fd = NULL,
-	.last_announce = 0
+	.last_announce = 0,
+	.next_frame = 0
 };
 
 MM_PLUGIN_API int init(){
@@ -46,6 +48,7 @@ MM_PLUGIN_API int init(){
 		.handle = sacn_set,
 		.process = sacn_handle,
 		.start = sacn_start,
+		.interval = sacn_interval,
 		.shutdown = sacn_shutdown
 	};
 
@@ -63,6 +66,13 @@ MM_PLUGIN_API int init(){
 	return 0;
 }
 
+static uint32_t sacn_interval(){
+	if(global_cfg.next_frame){
+		return global_cfg.next_frame;
+	}
+	return SACN_KEEPALIVE_INTERVAL;
+}
+
 static int sacn_listener(char* host, char* port, uint8_t flags){
 	int fd = -1, yes = 1;
 	if(global_cfg.fds >= MAX_FDS){
@@ -87,7 +97,6 @@ static int sacn_listener(char* host, char* port, uint8_t flags){
 	global_cfg.fd[global_cfg.fds].fd = fd;
 	global_cfg.fd[global_cfg.fds].universes = 0;
 	global_cfg.fd[global_cfg.fds].universe = NULL;
-	global_cfg.fd[global_cfg.fds].last_frame = NULL;
 
 	if(flags & mcast_loop){
 		//set IP_MCAST_LOOP to allow local applications to receive output
@@ -190,6 +199,10 @@ static int sacn_configure_instance(instance* inst, char* option, char* value){
 		data->unicast_input = strtoul(value, NULL, 10);
 		return 0;
 	}
+	else if(!strcmp(option, "realtime")){
+		data->realtime = strtoul(value, NULL, 10);
+		return 0;
+	}
 
 	LOGPF("Unknown instance configuration option %s for instance %s", option, inst->name);
 	return 1;
@@ -289,10 +302,11 @@ static int sacn_transmit(instance* inst){
 		LOGPF("Failed to output frame for instance %s: %s", inst->name, strerror(errno));
 	}
 
-	//update last transmit timestamp
+	//update last transmit timestamp, unmark instance
 	for(u = 0; u < global_cfg.fd[data->fd_index].universes; u++){
-		if(global_cfg.fd[data->fd_index].universe[u] == data->uni){
-			global_cfg.fd[data->fd_index].last_frame[u] = mm_timestamp();
+		if(global_cfg.fd[data->fd_index].universe[u].universe == data->uni){
+			global_cfg.fd[data->fd_index].universe[u].last_frame = mm_timestamp();
+			global_cfg.fd[data->fd_index].universe[u].mark = 0;
 		}
 	}
 	return 0;
@@ -300,6 +314,7 @@ static int sacn_transmit(instance* inst){
 
 static int sacn_set(instance* inst, size_t num, channel** c, channel_value* v){
 	size_t u, mark = 0;
+	uint32_t frame_delta = 0;
 	sacn_instance_data* data = (sacn_instance_data*) inst->impl;
 
 	if(!num){
@@ -333,6 +348,25 @@ static int sacn_set(instance* inst, size_t num, channel** c, channel_value* v){
 
 	//send packet if required
 	if(mark){
+		if(!data->realtime){
+			//find output instance data
+			for(u = 0; u < global_cfg.fd[data->fd_index].universes; u++){
+				if(global_cfg.fd[data->fd_index].universe[u].universe == data->uni){
+					break;
+				}
+			}
+
+			frame_delta = mm_timestamp() - global_cfg.fd[data->fd_index].universe[u].last_frame;
+
+			//check if ratelimiting engaged
+			if(frame_delta < SACN_FRAME_TIMEOUT){
+				global_cfg.fd[data->fd_index].universe[u].mark = 1;
+				if(!global_cfg.next_frame || global_cfg.next_frame > (SACN_KEEPALIVE_INTERVAL - frame_delta)){
+					global_cfg.next_frame = (SACN_KEEPALIVE_INTERVAL - frame_delta);
+				}
+				return 0;
+			}
+		}
 		sacn_transmit(inst);
 	}
 
@@ -468,6 +502,7 @@ static void sacn_discovery(size_t fd){
 static int sacn_handle(size_t num, managed_fd* fds){
 	size_t u, c;
 	uint64_t timestamp = mm_timestamp();
+	uint32_t synthesize_delta = 0;
 	ssize_t bytes_read;
 	char recv_buf[SACN_RECV_BUF];
 	instance* inst = NULL;
@@ -477,7 +512,7 @@ static int sacn_handle(size_t num, managed_fd* fds){
 	sacn_frame_root* frame = (sacn_frame_root*) recv_buf;
 	sacn_frame_data* data = (sacn_frame_data*) (recv_buf + sizeof(sacn_frame_root));
 
-	if(mm_timestamp() - global_cfg.last_announce > SACN_DISCOVERY_TIMEOUT){
+	if(timestamp - global_cfg.last_announce > SACN_DISCOVERY_TIMEOUT){
 		//send universe discovery pdu
 		for(u = 0; u < global_cfg.fds; u++){
 			if(global_cfg.fd[u].universes){
@@ -487,17 +522,28 @@ static int sacn_handle(size_t num, managed_fd* fds){
 		global_cfg.last_announce = timestamp;
 	}
 
-	//check for keepalive frames
+	//check for keepalive frames, synthesize frames if necessary
 	for(u = 0; u < global_cfg.fds; u++){
 		for(c = 0; c < global_cfg.fd[u].universes; c++){
-			if(timestamp - global_cfg.fd[u].last_frame[c] >= SACN_KEEPALIVE_INTERVAL){
+			synthesize_delta = timestamp - global_cfg.fd[u].universe[c].last_frame;
+
+			if((global_cfg.fd[u].universe[c].mark
+						&& synthesize_delta >= SACN_FRAME_TIMEOUT + SACN_SYNTHESIZE_MARGIN)
+					|| synthesize_delta >= SACN_KEEPALIVE_INTERVAL){
 				instance_id.fields.fd_index = u;
-				instance_id.fields.uni = global_cfg.fd[u].universe[c];
+				instance_id.fields.uni = global_cfg.fd[u].universe[c].universe;
 				inst = mm_instance_find(BACKEND_NAME, instance_id.label);
 				if(inst){
 					sacn_transmit(inst);
 				}
 			}
+
+			//update next frame request
+			if(global_cfg.fd[u].universe[c].mark 
+					&& (!global_cfg.next_frame || global_cfg.next_frame > SACN_FRAME_TIMEOUT + SACN_SYNTHESIZE_MARGIN - synthesize_delta)){
+				global_cfg.next_frame = SACN_FRAME_TIMEOUT + SACN_SYNTHESIZE_MARGIN - synthesize_delta;
+			}
+
 		}
 	}
 
@@ -557,7 +603,6 @@ static int sacn_start(size_t n, instance** inst){
 
 	if(!global_cfg.fds){
 		LOG("Failed to start, no descriptors bound");
-		free(inst);
 		return 1;
 	}
 
@@ -590,13 +635,15 @@ static int sacn_start(size_t n, instance** inst){
 
 		if(data->xmit_prio){
 			//add to list of advertised universes for this fd
-			global_cfg.fd[data->fd_index].universe = realloc(global_cfg.fd[data->fd_index].universe, (global_cfg.fd[data->fd_index].universes + 1) * sizeof(uint16_t));
+			global_cfg.fd[data->fd_index].universe = realloc(global_cfg.fd[data->fd_index].universe, (global_cfg.fd[data->fd_index].universes + 1) * sizeof(sacn_output_universe));
 			if(!global_cfg.fd[data->fd_index].universe){
 				LOG("Failed to allocate memory");
 				goto bail;
 			}
 
-			global_cfg.fd[data->fd_index].universe[global_cfg.fd[data->fd_index].universes] = data->uni;
+			global_cfg.fd[data->fd_index].universe[global_cfg.fd[data->fd_index].universes].universe = data->uni;
+			global_cfg.fd[data->fd_index].universe[global_cfg.fd[data->fd_index].universes].last_frame = 0;
+			global_cfg.fd[data->fd_index].universe[global_cfg.fd[data->fd_index].universes].mark = 0;
 			global_cfg.fd[data->fd_index].universes++;
 
 			//generate multicast destination address if none set
@@ -612,12 +659,6 @@ static int sacn_start(size_t n, instance** inst){
 
 	LOGPF("Registering %" PRIsize_t " descriptors to core", global_cfg.fds);
 	for(u = 0; u < global_cfg.fds; u++){
-		//allocate memory for storing last frame transmission timestamp
-		global_cfg.fd[u].last_frame = calloc(global_cfg.fd[u].universes, sizeof(uint64_t));
-		if(!global_cfg.fd[u].last_frame){
-			LOG("Failed to allocate memory");
-			goto bail;
-		}
 		if(mm_manage_fd(global_cfg.fd[u].fd, BACKEND_NAME, 1, (void*) u)){
 			goto bail;
 		}
@@ -638,7 +679,6 @@ static int sacn_shutdown(size_t n, instance** inst){
 	for(p = 0; p < global_cfg.fds; p++){
 		close(global_cfg.fd[p].fd);
 		free(global_cfg.fd[p].universe);
-		free(global_cfg.fd[p].last_frame);
 	}
 	free(global_cfg.fd);
 	LOG("Backend shut down");
diff --git a/backends/sacn.h b/backends/sacn.h
index ac59441..4642e59 100644
--- a/backends/sacn.h
+++ b/backends/sacn.h
@@ -1,6 +1,7 @@
 #include "midimonster.h"
 
 MM_PLUGIN_API int init();
+static uint32_t sacn_interval();
 static int sacn_configure(char* option, char* value);
 static int sacn_configure_instance(instance* instance, char* option, char* value);
 static int sacn_instance(instance* inst);
@@ -12,7 +13,11 @@ static int sacn_shutdown(size_t n, instance** inst);
 
 #define SACN_PORT "5568"
 #define SACN_RECV_BUF 8192
-#define SACN_KEEPALIVE_INTERVAL 2000
+//spec 6.6.2.1
+#define SACN_KEEPALIVE_INTERVAL 1000
+//spec 6.6.1
+#define SACN_FRAME_TIMEOUT 15
+#define SACN_SYNTHESIZE_MARGIN 10
 #define SACN_DISCOVERY_TIMEOUT 9000
 #define SACN_PDU_MAGIC "ASC-E1.17\0\0\0"
 
@@ -35,6 +40,7 @@ typedef struct /*_sacn_universe_model*/ {
 
 typedef struct /*_sacn_instance_model*/ {
 	uint16_t uni;
+	uint8_t realtime;
 	uint8_t xmit_prio;
 	uint8_t cid_filter[16];
 	uint8_t filter_enabled;
@@ -54,11 +60,16 @@ typedef union /*_sacn_instance_id*/ {
 	uint64_t label;
 } sacn_instance_id;
 
+typedef struct /*_sacn_output_universe*/ {
+	uint16_t universe;
+	uint64_t last_frame;
+	uint8_t mark;
+} sacn_output_universe;
+
 typedef struct /*_sacn_socket*/ {
 	int fd;
 	size_t universes;
-	uint16_t* universe;
-	uint64_t* last_frame;
+	sacn_output_universe* universe;
 } sacn_fd;
 
 #pragma pack(push, 1)
diff --git a/backends/sacn.md b/backends/sacn.md
index f5f1db4..598f430 100644
--- a/backends/sacn.md
+++ b/backends/sacn.md
@@ -26,6 +26,7 @@ This has the side effect of mirroring the output of instances on those descripto
 | `destination`	| `10.2.2.2`		| Universe multicast	| Destination address for unicast output. If unset, the multicast destination for the specified universe is used. |
 | `from`	| `0xAA 0xBB` ...	| none			| 16-byte input source CID filter. Setting this option filters the input stream for this universe. |
 | `unicast`	| `1`			| `0`			| Prevent this instance from joining its universe multicast group |
+| `realtime`	| `1`			| `0`			| Disable the recommended rate-limiting (approx. 44 packets per second) for this instance |
 
 Note that instances accepting multicast input also process unicast frames directed at them, while
 instances in `unicast` mode will not receive multicast frames.
@@ -50,9 +51,6 @@ A normal channel that is part of a wide channel can not be mapped individually.
 
 The DMX start code of transmitted and received universes is fixed as `0`.
 
-The (upper) limit on packet transmission rate mandated by section 6.6.1 of the sACN specification is disregarded.
-The rate of packet transmission is influenced by the rate of incoming mapped events on the instance.
-
 Universe synchronization is currently not supported, though this feature may be implemented in the future.
 
 To use multicast input, all networking hardware in the path must support the IGMPv2 protocol.