Dynamically update NACK queue size depending on RTT

README.md

@@ -281,8 +281,8 @@ or on the command line:
 	-T, --ice-tcp                 Whether to enable ICE-TCP or not (warning: only
                                   works with ICE Lite)
                                   (default=off)
-	-q, --max-nack-queue=number   Maximum size of the NACK queue (in ms) per user
-                                  for retransmissions
+	-Q, --min-nack-queue=number   Minimum size of the NACK queue (in ms) per user
+                                  for retransmissions, no matter the RTT
 	-t, --no-media-timer=number   Time (in s) that should pass with no media
                                   (audio or video) being received before Janus
                                   notifies you about this

conf/janus.jcfg.sample.in

@@ -131,8 +131,8 @@ certificates: {
 }
 
 # Media-related stuff: you can configure whether if you want
-# to enable IPv6 support, the maximum size of the NACK queue (in
-# milliseconds, defaults to 500ms) for retransmissions, the range of
+# to enable IPv6 support, the minimum size of the NACK queue (in ms,
+# defaults to 200ms) for retransmissions no matter the RTT, the range of
 # ports to use for RTP and RTCP (by default, no range is envisaged), the
 # starting MTU for DTLS (1200 by default, it adapts automatically),
 # how much time, in seconds, should pass with no media (audio or
@@ -148,7 +148,7 @@ certificates: {
 # you should pick a reasonable trade-off (usually 2*max expected RTT).
 media: {
 	#ipv6 = true
-	#max_nack_queue = 500
+	#min_nack_queue = 500
 	#rtp_port_range = "20000-40000"
 	#dtls_mtu = 1200
 	#no_media_timer = 1

html/admin.js

@@ -339,20 +339,20 @@ function updateSettings() {
 							setLogLevel(result);
 						});
 					});
-				} else if(k === 'max_nack_queue') {
-					$('#'+k).append('<button id="' + k + '_button" type="button" class="btn btn-xs btn-primary">Edit max NACK queue</button>');
+				} else if(k === 'min_nack_queue') {
+					$('#'+k).append('<button id="' + k + '_button" type="button" class="btn btn-xs btn-primary">Edit min NACK queue</button>');
 					$('#'+k + "_button").click(function() {
-						bootbox.prompt("Set the new desired max NACK queue (a positive integer, currently " + settings["max_nack_queue"] + ")", function(result) {
+						bootbox.prompt("Set the new desired min NACK queue (a positive integer, currently " + settings["min_nack_queue"] + ")", function(result) {
 							if(isNaN(result)) {
-								bootbox.alert("Invalid max NACK queue (should be a positive integer)");
+								bootbox.alert("Invalid min NACK queue (should be a positive integer)");
 								return;
 							}
 							result = parseInt(result);
 							if(result < 0) {
-								bootbox.alert("Invalid max NACK queue (should be a positive integer)");
+								bootbox.alert("Invalid min NACK queue (should be a positive integer)");
 								return;
 							}
-							setMaxNackQueue(result);
+							setMinNackQueue(result);
 						});
 					});
 				} else if(k === 'no_media_timer') {
@@ -510,8 +510,8 @@ function setLibniceDebug(enable) {
 	sendSettingsRequest(request);
 }
 
-function setMaxNackQueue(queue) {
-	var request = { "janus": "set_max_nack_queue", "max_nack_queue": queue, "transaction": randomString(12), "admin_secret": secret };
+function setMinNackQueue(queue) {
+	var request = { "janus": "set_min_nack_queue", "min_nack_queue": queue, "transaction": randomString(12), "admin_secret": secret };
 	sendSettingsRequest(request);
 }
 

ice.c

@@ -459,29 +459,30 @@ static void janus_ice_free_queued_packet(janus_ice_queued_packet *pkt) {
 	g_free(pkt);
 }
 
-/* Maximum value, in milliseconds, for the NACK queue/retransmissions (default=500ms) */
-#define DEFAULT_MAX_NACK_QUEUE	500
+/* Minimum and maximum value, in milliseconds, for the NACK queue/retransmissions (default=200ms/1000ms) */
+#define DEFAULT_MIN_NACK_QUEUE	200
+#define DEFAULT_MAX_NACK_QUEUE	1000
 /* Maximum ignore count after retransmission (200ms) */
 #define MAX_NACK_IGNORE			200000
 
-static uint max_nack_queue = DEFAULT_MAX_NACK_QUEUE;
-void janus_set_max_nack_queue(uint mnq) {
-	max_nack_queue = mnq;
-	if(max_nack_queue == 0)
+static uint16_t min_nack_queue = DEFAULT_MIN_NACK_QUEUE;
+void janus_set_min_nack_queue(uint16_t mnq) {
+	min_nack_queue = mnq < DEFAULT_MAX_NACK_QUEUE ? mnq : DEFAULT_MAX_NACK_QUEUE;
+	if(min_nack_queue == 0)
 		JANUS_LOG(LOG_VERB, "Disabling NACK queue\n");
 	else
-		JANUS_LOG(LOG_VERB, "Setting max NACK queue to %dms\n", max_nack_queue);
+		JANUS_LOG(LOG_VERB, "Setting min NACK queue to %dms\n", min_nack_queue);
 }
-uint janus_get_max_nack_queue(void) {
-	return max_nack_queue;
+uint16_t janus_get_min_nack_queue(void) {
+	return min_nack_queue;
 }
 /* Helper to clean old NACK packets in the buffer when they exceed the queue time limit */
 static void janus_cleanup_nack_buffer(gint64 now, janus_ice_stream *stream, gboolean audio, gboolean video) {
 	if(stream && stream->component) {
 		janus_ice_component *component = stream->component;
 		if(audio && component->audio_retransmit_buffer) {
 			janus_rtp_packet *p = (janus_rtp_packet *)g_queue_peek_head(component->audio_retransmit_buffer);
-			while(p && (!now || (now - p->created >= (gint64)max_nack_queue*1000))) {
+			while(p && (!now || (now - p->created >= (gint64)stream->nack_queue_ms*1000))) {
 				/* Packet is too old, get rid of it */
 				g_queue_pop_head(component->audio_retransmit_buffer);
 				/* Remove from hashtable too */
@@ -495,7 +496,7 @@ static void janus_cleanup_nack_buffer(gint64 now, janus_ice_stream *stream, gboo
 		}
 		if(video && component->video_retransmit_buffer) {
 			janus_rtp_packet *p = (janus_rtp_packet *)g_queue_peek_head(component->video_retransmit_buffer);
-			while(p && (!now || (now - p->created >= (gint64)max_nack_queue*1000))) {
+			while(p && (!now || (now - p->created >= (gint64)stream->nack_queue_ms*1000))) {
 				/* Packet is too old, get rid of it */
 				g_queue_pop_head(component->video_retransmit_buffer);
 				/* Remove from hashtable too */
@@ -2746,10 +2747,28 @@ static void janus_ice_cb_nice_recv(NiceAgent *agent, guint stream_id, guint comp
 
 				/* Let's process this RTCP (compound?) packet, and update the RTCP context for this stream in case */
 				rtcp_context *rtcp_ctx = video ? stream->video_rtcp_ctx[vindex] : stream->audio_rtcp_ctx;
-				if (janus_rtcp_parse(rtcp_ctx, buf, buflen) < 0) {
+				uint32_t rtt = rtcp_ctx ? rtcp_ctx->rtt : 0;
+				if(janus_rtcp_parse(rtcp_ctx, buf, buflen) < 0) {
 					/* Drop the packet if the parsing function returns with an error */
 					return;
 				}
+				if(rtcp_ctx && rtcp_ctx->rtt != rtt) {
+					/* Check the current RTT, to see if we need to update the size of the queue: we take
+					 * the highest RTT (audio or video) and add 100ms just to be conservative */
+					uint32_t audio_rtt = janus_rtcp_context_get_rtt(stream->audio_rtcp_ctx),
+						video_rtt = janus_rtcp_context_get_rtt(stream->video_rtcp_ctx[0]);
+					uint16_t nack_queue_ms = (audio_rtt > video_rtt ? audio_rtt : video_rtt) + 100;
+					if(nack_queue_ms > DEFAULT_MAX_NACK_QUEUE)
+						nack_queue_ms = DEFAULT_MAX_NACK_QUEUE;
+					else if(nack_queue_ms < min_nack_queue)
+						nack_queue_ms = min_nack_queue;
+					uint16_t mavg = rtt ? ((7*stream->nack_queue_ms + nack_queue_ms)/8) : nack_queue_ms;
+					if(mavg > DEFAULT_MAX_NACK_QUEUE)
+						mavg = DEFAULT_MAX_NACK_QUEUE;
+					else if(mavg < min_nack_queue)
+						mavg = min_nack_queue;
+					stream->nack_queue_ms = mavg;
+				}
 				JANUS_LOG(LOG_HUGE, "[%"SCNu64"] Got %s RTCP (%d bytes)\n", handle->handle_id, video ? "video" : "audio", buflen);
 
 				/* Now let's see if there are any NACKs to handle */
@@ -3344,6 +3363,7 @@ int janus_ice_setup_local(janus_ice_handle *handle, int offer, int audio, int vi
 	stream->audio_payload_type = -1;
 	stream->video_payload_type = -1;
 	stream->video_rtx_payload_type = -1;
+	stream->nack_queue_ms = min_nack_queue;
 	/* FIXME By default, if we're being called we're DTLS clients, but this may be changed by ICE... */
 	stream->dtls_role = offer ? JANUS_DTLS_ROLE_CLIENT : JANUS_DTLS_ROLE_ACTPASS;
 	if(audio) {
@@ -4105,7 +4125,7 @@ static gboolean janus_ice_outgoing_traffic_handle(janus_ice_handle *handle, janu
 				}
 				/* Before encrypting, check if we need to copy the unencrypted payload (e.g., for rtx/90000) */
 				janus_rtp_packet *p = NULL;
-				if(max_nack_queue > 0 && !pkt->retransmission && pkt->type == JANUS_ICE_PACKET_VIDEO && component->do_video_nacks &&
+				if(stream->nack_queue_ms > 0 && !pkt->retransmission && pkt->type == JANUS_ICE_PACKET_VIDEO && component->do_video_nacks &&
 						janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_RFC4588_RTX)) {
 					/* Save the packet for retransmissions that may be needed later: start by
 					 * making room for two more bytes to store the original sequence number */
@@ -4212,7 +4232,7 @@ static gboolean janus_ice_outgoing_traffic_handle(janus_ice_handle *handle, janu
 						if(rtcp_ctx)
 							g_atomic_int_inc(&rtcp_ctx->sent_packets_since_last_rr);
 					}
-					if(max_nack_queue > 0 && !pkt->retransmission) {
+					if(stream->nack_queue_ms > 0 && !pkt->retransmission) {
 						/* Save the packet for retransmissions that may be needed later */
 						if((pkt->type == JANUS_ICE_PACKET_AUDIO && !component->do_audio_nacks) ||
 								(pkt->type == JANUS_ICE_PACKET_VIDEO && !component->do_video_nacks)) {

ice.h

@@ -122,12 +122,12 @@ gboolean janus_ice_is_full_trickle_enabled(void);
 /*! \brief Method to check whether IPv6 candidates are enabled/supported or not (still WIP)
  * @returns true if IPv6 candidates are enabled/supported, false otherwise */
 gboolean janus_ice_is_ipv6_enabled(void);
-/*! \brief Method to modify the max NACK value (i.e., the number of packets per handle to store for retransmissions)
- * @param[in] mnq The new max NACK value */
-void janus_set_max_nack_queue(uint mnq);
-/*! \brief Method to get the current max NACK value (i.e., the number of packets per handle to store for retransmissions)
- * @returns The current max NACK value */
-uint janus_get_max_nack_queue(void);
+/*! \brief Method to modify the min NACK value (i.e., the minimum time window of packets per handle to store for retransmissions)
+ * @param[in] mnq The new min NACK value */
+void janus_set_min_nack_queue(uint16_t mnq);
+/*! \brief Method to get the current min NACK value (i.e., the minimum time window of packets per handle to store for retransmissions)
+ * @returns The current min NACK value */
+uint16_t janus_get_min_nack_queue(void);
 /*! \brief Method to modify the no-media event timer (i.e., the number of seconds where no media arrives before Janus notifies this)
  * @param[in] timer The new timer value, in seconds */
 void janus_set_no_media_timer(uint timer);
@@ -386,6 +386,8 @@ struct janus_ice_stream {
 	janus_rtcp_context *audio_rtcp_ctx;
 	/*! \brief RTCP context(s) for the video stream (may be simulcasting) */
 	janus_rtcp_context *video_rtcp_ctx[3];
+	/*! \brief Size of the NACK queue (in ms), dynamically updated per the RTT */
+	uint16_t nack_queue_ms;
 	/*! \brief Map(s) of the NACKed packets (to track retransmissions and avoid duplicates) */
 	GHashTable *rtx_nacked[3];
 	/*! \brief First received audio NTP timestamp */

janus.1

@@ -80,8 +80,8 @@ Whether to enable the ICE Lite mode or not (default=off)
 .BR \-T ", " \-\-ice-tcp
 Whether to enable ICE-TCP or not (warning: only works with ICE Lite) (default=off)
 .TP
-.BR \-q ", " \-\-max-nack-queue=\fInumber\fR
-Maximum size of the NACK queue per user for retransmissions
+.BR \-Q ", " \-\-min-nack-queue=\fInumber\fR
+Minimum size of the NACK queue (in ms) per user for retransmissions, no matter the RTT
 .TP
 .BR \-t ", " \-\-no-media-timer=\fInumber\fR
 Time (in s) that should pass with no media (audio or video) being received before Janus notifies you about this

janus.c

@@ -132,7 +132,7 @@ static struct janus_json_parameter colors_parameters[] = {
 	{"colors", JANUS_JSON_BOOL, JANUS_JSON_PARAM_REQUIRED}
 };
 static struct janus_json_parameter mnq_parameters[] = {
-	{"max_nack_queue", JSON_INTEGER, JANUS_JSON_PARAM_REQUIRED | JANUS_JSON_PARAM_POSITIVE}
+	{"min_nack_queue", JSON_INTEGER, JANUS_JSON_PARAM_REQUIRED | JANUS_JSON_PARAM_POSITIVE}
 };
 static struct janus_json_parameter nmt_parameters[] = {
 	{"no_media_timer", JSON_INTEGER, JANUS_JSON_PARAM_REQUIRED | JANUS_JSON_PARAM_POSITIVE}
@@ -296,6 +296,7 @@ static json_t *janus_info(const char *transaction) {
 	json_object_set_new(info, "ice-lite", janus_ice_is_ice_lite_enabled() ? json_true() : json_false());
 	json_object_set_new(info, "ice-tcp", janus_ice_is_ice_tcp_enabled() ? json_true() : json_false());
 	json_object_set_new(info, "full-trickle", janus_ice_is_full_trickle_enabled() ? json_true() : json_false());
+	json_object_set_new(info, "min-nack-queue", json_integer(janus_get_min_nack_queue()));
 	json_object_set_new(info, "twcc-period", json_integer(janus_get_twcc_period()));
 	if(janus_ice_get_stun_server() != NULL) {
 		char server[255];
@@ -1817,7 +1818,7 @@ int janus_process_incoming_admin_request(janus_request *request) {
 			json_object_set_new(status, "locking_debug", lock_debug ? json_true() : json_false());
 			json_object_set_new(status, "refcount_debug", refcount_debug ? json_true() : json_false());
 			json_object_set_new(status, "libnice_debug", janus_ice_is_ice_debugging_enabled() ? json_true() : json_false());
-			json_object_set_new(status, "max_nack_queue", json_integer(janus_get_max_nack_queue()));
+			json_object_set_new(status, "min_nack_queue", json_integer(janus_get_min_nack_queue()));
 			json_object_set_new(status, "no_media_timer", json_integer(janus_get_no_media_timer()));
 			json_object_set_new(status, "slowlink_threshold", json_integer(janus_get_slowlink_threshold()));
 			json_object_set_new(reply, "status", status);
@@ -1963,25 +1964,21 @@ int janus_process_incoming_admin_request(janus_request *request) {
 			/* Send the success reply */
 			ret = janus_process_success(request, reply);
 			goto jsondone;
-		} else if(!strcasecmp(message_text, "set_max_nack_queue")) {
-			/* Change the current value for the max NACK queue */
+		} else if(!strcasecmp(message_text, "set_min_nack_queue")) {
+			/* Change the current value for the min NACK queue */
 			JANUS_VALIDATE_JSON_OBJECT(root, mnq_parameters,
 				error_code, error_cause, FALSE,
 				JANUS_ERROR_MISSING_MANDATORY_ELEMENT, JANUS_ERROR_INVALID_ELEMENT_TYPE);
 			if(error_code != 0) {
 				ret = janus_process_error_string(request, session_id, transaction_text, error_code, error_cause);
 				goto jsondone;
 			}
-			json_t *mnq = json_object_get(root, "max_nack_queue");
+			json_t *mnq = json_object_get(root, "min_nack_queue");
 			int mnq_num = json_integer_value(mnq);
-			if(mnq_num < 0 || (mnq_num > 0 && mnq_num < 200)) {
-				ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_ELEMENT_TYPE, "Invalid element type (max_nack_queue, if provided, should be greater than 200)");
-				goto jsondone;
-			}
-			janus_set_max_nack_queue(mnq_num);
+			janus_set_min_nack_queue(mnq_num);
 			/* Prepare JSON reply */
 			json_t *reply = janus_create_message("success", 0, transaction_text);
-			json_object_set_new(reply, "max_nack_queue", json_integer(janus_get_max_nack_queue()));
+			json_object_set_new(reply, "min_nack_queue", json_integer(janus_get_min_nack_queue()));
 			/* Send the success reply */
 			ret = janus_process_success(request, reply);
 			goto jsondone;
@@ -2776,6 +2773,7 @@ json_t *janus_admin_stream_summary(janus_ice_stream *stream) {
 	if(stream->transport_wide_cc_ext_id > 0)
 		json_object_set_new(bwe, "twcc-ext-id", json_integer(stream->transport_wide_cc_ext_id));
 	json_object_set_new(s, "bwe", bwe);
+	json_object_set_new(s, "nack-queue-ms", json_integer(stream->nack_queue_ms));
 	json_t *components = json_array();
 	if(stream->component) {
 		json_t *c = janus_admin_component_summary(stream->component);
@@ -4121,10 +4119,10 @@ gint main(int argc, char *argv[])
 	if(args_info.ipv6_candidates_given) {
 		janus_config_add(config, config_media, janus_config_item_create("ipv6", "true"));
 	}
-	if(args_info.max_nack_queue_given) {
+	if(args_info.min_nack_queue_given) {
 		char mnq[20];
-		g_snprintf(mnq, 20, "%d", args_info.max_nack_queue_arg);
-		janus_config_add(config, config_media, janus_config_item_create("max_nack_queue", mnq));
+		g_snprintf(mnq, 20, "%d", args_info.min_nack_queue_arg);
+		janus_config_add(config, config_media, janus_config_item_create("min_nack_queue", mnq));
 	}
 	if(args_info.no_media_timer_given) {
 		char nmt[20];
@@ -4486,15 +4484,13 @@ gint main(int argc, char *argv[])
 		}
 	}
 	/* NACK related stuff */
-	item = janus_config_get(config, config_media, janus_config_type_item, "max_nack_queue");
+	item = janus_config_get(config, config_media, janus_config_type_item, "min_nack_queue");
 	if(item && item->value) {
 		int mnq = atoi(item->value);
 		if(mnq < 0) {
-			JANUS_LOG(LOG_WARN, "Ignoring max_nack_queue value as it's not a positive integer\n");
-		} else if(mnq > 0 && mnq < 200) {
-			JANUS_LOG(LOG_WARN, "Ignoring max_nack_queue value as it's less than 200\n");
+			JANUS_LOG(LOG_WARN, "Ignoring min_nack_queue value as it's not a positive integer\n");
 		} else {
-			janus_set_max_nack_queue(mnq);
+			janus_set_min_nack_queue(mnq);
 		}
 	}
 	/* no-media timer */

janus.ggo

@@ -20,7 +20,7 @@ option "libnice-debug" l "Whether to enable libnice debugging or not" flag off
 option "full-trickle" f "Do full-trickle instead of half-trickle" flag off
 option "ice-lite" I "Whether to enable the ICE Lite mode or not" flag off
 option "ice-tcp" T "Whether to enable ICE-TCP or not (warning: only works with ICE Lite)" flag off
-option "max-nack-queue" q "Maximum size of the NACK queue (in ms) per user for retransmissions" int typestr="number" optional
+option "min-nack-queue" Q "Minimum size of the NACK queue (in ms) per user for retransmissions, no matter the RTT" int typestr="number" optional
 option "no-media-timer" t "Time (in s) that should pass with no media (audio or video) being received before Janus notifies you about this" int typestr="number" optional
 option "slowlink-threshold" W "Number of lost packets (per s) that should trigger a 'slowlink' Janus API event to users" int typestr="number" optional
 option "rtp-port-range" r "Port range to use for RTP/RTCP" string typestr="min-max" optional

mainpage.dox

@@ -2271,7 +2271,7 @@ const token = getJanusToken('janus', ['janus.plugin.videoroom']),
  * the reference counters in Janus on the fly (useful if you're experiencing
  * memory leaks in the Janus structures and want to investigate them);
  * - \c set_libnice_debug: selectively enable/disable libnice debugging;
- * - \c set_max_nack_queue: change the value of the max NACK queue window;
+ * - \c set_min_nack_queue: change the value of the min NACK queue window;
  * - \c set_no_media_timer: change the value of the no-media timer property;
  * - \c set_slowlink_threshold: change the value of the slowlink-threshold property.
  *