[core, mod_opus] more elastic jitterbuffer with Opus codec (#2069)

2026-07-24 05:02:10 +00:00 · 2023-06-23 09:45:31 -04:00
parent f64314ff48
commit 67840823c1
9 changed files with 756 additions and 3 deletions
@@ -44,6 +44,8 @@

 struct switch_jb_s;

+static inline int check_jb_size(switch_jb_t *jb);
+
 typedef struct switch_jb_node_s {
 	struct switch_jb_s *parent;
 	switch_rtp_packet_t packet;
@@ -56,6 +58,29 @@ typedef struct switch_jb_node_s {
 	switch_bool_t complete_frame_mark;
 } switch_jb_node_t;

+typedef struct switch_jb_stats_s {
+	uint32_t reset_too_big;
+	uint32_t reset_missing_frames;
+	uint32_t reset_ts_jump;
+	uint32_t reset_error;
+	uint32_t reset;
+	uint32_t size_max;
+	uint32_t size_est;
+	uint32_t acceleration;
+	uint32_t expand;
+	uint32_t jitter_max_ms;
+	int estimate_ms;
+	int buffer_size_ms;
+} switch_jb_stats_t;
+
+typedef struct switch_jb_jitter_s {
+	double *estimate;
+	uint32_t samples_per_second;
+	uint32_t samples_per_frame;
+	uint32_t drop_gap;
+	switch_jb_stats_t stats;
+} switch_jb_jitter_t;
+
 struct switch_jb_s {
 	struct switch_jb_node_s *node_list;
 	uint32_t last_target_seq;
@@ -104,6 +129,7 @@ struct switch_jb_s {
 	switch_jb_flag_t flags;
 	switch_jb_type_t type;
 	switch_core_session_t *session;
+	switch_jb_jitter_t jitter;
 	switch_channel_t *channel;
 	uint32_t buffer_lag;
 	uint32_t flush;
@@ -112,6 +138,8 @@ struct switch_jb_s {
 	uint32_t period_len;
 	uint32_t nack_saved_the_day;
 	uint32_t nack_didnt_save_the_day;
+	switch_bool_t elastic;
+	switch_codec_t *codec;
 };


@@ -233,6 +261,7 @@ static inline switch_jb_node_t *new_node(switch_jb_t *jb)

 		if (jb->allocated_nodes > jb->max_frame_len * mult) {
 			jb_debug(jb, 2, "ALLOCATED FRAMES TOO HIGH! %d\n", jb->allocated_nodes);
+			jb->jitter.stats.reset_too_big++;
 			switch_jb_reset(jb);
 			switch_mutex_unlock(jb->list_mutex);
 			return NULL;
@@ -332,6 +361,29 @@ static inline void hide_nodes(switch_jb_t *jb)
 	switch_mutex_unlock(jb->list_mutex);
 }

+static inline switch_bool_t packet_vad(switch_jb_t *jb, switch_rtp_packet_t *packet, switch_size_t len) {
+	void* payload;
+	uint16_t payload_len = len;
+
+	if (packet->ebody) {
+		payload = packet->ebody;
+	} else {
+		payload = packet->body;
+	}
+	if (payload && payload_len > 0) {
+		switch_bool_t ret;
+		switch_bool_t *ret_p = &ret;
+		switch_codec_control_type_t ret_t;
+		switch_core_media_codec_control(jb->session, SWITCH_MEDIA_TYPE_AUDIO,
+						SWITCH_IO_WRITE, SCC_AUDIO_VAD,
+						SCCT_STRING, (void *)payload,
+						SCCT_INT, (void *)&payload_len,
+						&ret_t, (void *)&ret_p);
+		return ret;
+	}
+	return SWITCH_TRUE;
+}
+
 static inline void drop_ts(switch_jb_t *jb, uint32_t ts)
 {
 	switch_jb_node_t *np;
@@ -667,6 +719,7 @@ static inline void add_node(switch_jb_t *jb, switch_rtp_packet_t *packet, switch

 		if (((seq_diff >= 100) || (ts_diff > (900000 * 5)))) {
 			jb_debug(jb, 2, "CHANGE DETECTED, PUNT %u\n", abs(((int)ntohs(packet->header.seq) - ntohs(jb->highest_wrote_seq))));
+			jb->jitter.stats.reset_ts_jump++;
 			switch_jb_reset(jb);
 		}
 	}
@@ -732,6 +785,12 @@ static inline void increment_seq(switch_jb_t *jb)
 	jb->target_seq = htons((ntohs(jb->target_seq) + 1));
 }

+static inline void decrement_seq(switch_jb_t *jb)
+{
+	jb->last_target_seq = jb->target_seq;
+	jb->target_seq = htons((ntohs(jb->target_seq) - 1));
+}
+
 static inline void set_read_seq(switch_jb_t *jb, uint16_t seq)
 {
 	jb->last_target_seq = seq;
@@ -854,12 +913,128 @@ static inline switch_status_t jb_next_packet_by_ts(switch_jb_t *jb, switch_jb_no

 }

+static inline int check_jb_size(switch_jb_t *jb)
+{
+	switch_jb_node_t *np;
+	uint16_t seq;
+	uint16_t l_seq=0;
+	uint16_t h_seq=0;
+	uint16_t count=0;
+	uint16_t old=0;
+	switch_mutex_lock(jb->list_mutex);
+
+	for (np = jb->node_list; np; np = np->next) {
+		if (!np->visible) {
+			continue;
+		}
+
+	seq = ntohs(np->packet.header.seq);
+	if (ntohs(jb->target_seq) > seq) {
+		hide_node(np, SWITCH_FALSE);
+		old++;
+		continue;
+	}
+	if (count == 0) {
+		l_seq = h_seq = seq;
+	}
+	count++;
+	if (seq < l_seq)
+		l_seq = seq;
+	if (seq > h_seq)
+		h_seq = seq;
+	}
+	if (count > jb->jitter.stats.size_max) {
+		jb->jitter.stats.size_max = count;
+	}
+	if (jb->jitter.stats.size_est == 0) {
+		jb->jitter.stats.size_est = count;
+	} else {
+		jb->jitter.stats.size_est = ((99*jb->jitter.stats.size_est)+(1*count))/100;
+	}
+	if (ntohs(jb->target_seq) % 50 == 0) { /* update the stats every x packets */
+		int packet_ms = jb->jitter.samples_per_frame / (jb->jitter.samples_per_second / 1000);
+		jb->jitter.stats.estimate_ms = (*jb->jitter.estimate) / jb->jitter.samples_per_second * 1000;
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_size_max_ms", "%u", jb->jitter.stats.size_max*packet_ms);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_size_est_ms", "%u", jb->jitter.stats.size_est*packet_ms);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_acceleration_ms", "%u", jb->jitter.stats.acceleration*packet_ms);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_expand_ms", "%u", jb->jitter.stats.expand*packet_ms);
+		if (jb->jitter.stats.jitter_max_ms <  jb->jitter.stats.estimate_ms) {
+			jb->jitter.stats.jitter_max_ms = jb->jitter.stats.estimate_ms;
+		}
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_jitter_max_ms", "%u", jb->jitter.stats.jitter_max_ms);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_jitter_est_ms", "%u", jb->jitter.stats.estimate_ms);
+	}
+	if (old) {
+		sort_free_nodes(jb);
+	}
+	switch_mutex_unlock(jb->list_mutex);
+	jb_debug(jb, SWITCH_LOG_INFO, "JITTER buffersize %u == %u old[%u] target[%u] seq[%u|%u]\n", count, h_seq-l_seq+1, old, ntohs(jb->target_seq), l_seq, h_seq);
+	return count;
+}
+
+static inline switch_status_t jb_next_packet_by_seq_with_acceleration(switch_jb_t *jb, switch_jb_node_t **nodep)
+{
+		switch_status_t status = jb_next_packet_by_seq(jb, nodep);
+		switch_rtp_packet_t *packet;
+		uint32_t len;
+		uint16_t seq = ntohs(jb->target_seq);
+
+		/* When using a Codec that provides voice activity detection ex. Opus, use it to
+		   select packet to drop/accelerate. */
+
+		if (jb->elastic && jb->jitter.estimate && (jb->visible_nodes*jb->jitter.samples_per_frame)>0 && jb->jitter.samples_per_second) {
+			int visible_not_old = check_jb_size(jb);
+			jb->jitter.stats.estimate_ms = (int)((*jb->jitter.estimate)/((jb->jitter.samples_per_second))*1000);
+			jb->jitter.stats.buffer_size_ms = (int)((visible_not_old*jb->jitter.samples_per_frame)/(jb->jitter.samples_per_second/1000));
+
+			// We try to accelerate in order to remove delay when the jitter buffer is 3x larger than the estimation.
+			if (jb->jitter.stats.buffer_size_ms > (3*jb->jitter.stats.estimate_ms) && jb->jitter.stats.buffer_size_ms > 60) {
+				if (status == SWITCH_STATUS_SUCCESS) {
+					packet = &(*nodep)->packet;
+					seq = ntohs((*nodep)->packet.header.seq);
+					len = (*nodep)->len;
+				}
+				if (jb->jitter.drop_gap > 0) {
+					jb_debug(jb, SWITCH_LOG_INFO, "JITTER estimation %dms buffersize %d/%d %dms seq:%u [drop-gap][%d]\n",
+						jb->jitter.stats.estimate_ms, jb->complete_frames , jb->frame_len, jb->jitter.stats.buffer_size_ms, seq, jb->jitter.drop_gap);
+					jb->jitter.drop_gap--;
+				} else {
+				        if (status != SWITCH_STATUS_SUCCESS || packet_vad(jb, packet, len) == SWITCH_FALSE) {
+						jb->jitter.drop_gap = 3;
+						if (status != SWITCH_STATUS_SUCCESS) {
+							jb_debug(jb, SWITCH_LOG_INFO, "JITTER estimation n/a buffersize %d/%d %dms seq:%u [drop-missing/no-plc]\n",
+								jb->complete_frames , jb->frame_len, jb->jitter.stats.buffer_size_ms, seq);
+						} else {
+							jb_debug(jb, SWITCH_LOG_INFO, "JITTER estimation %dms buffersize %d/%d %dms seq:%u ACCELERATE [drop]\n",
+								jb->jitter.stats.estimate_ms, jb->complete_frames , jb->frame_len, jb->jitter.stats.buffer_size_ms, seq);
+						}
+						jb->jitter.stats.acceleration++;
+						return jb_next_packet_by_seq(jb, nodep);
+					} else {
+						jb_debug(jb, SWITCH_LOG_INFO, "JITTER estimation %dms buffersize %d/%d %dms seq:%u [drop-skip-vad]\n",
+							jb->jitter.stats.estimate_ms, jb->complete_frames , jb->frame_len, jb->jitter.stats.buffer_size_ms, seq);
+					}
+				}
+			} else {
+				jb_debug(jb, 2, "JITTER estimation %dms buffersize %d/%d %dms\n",
+						jb->jitter.stats.estimate_ms, jb->complete_frames , jb->frame_len, jb->jitter.stats.buffer_size_ms);
+			}
+		}
+		return status;
+}
+
 static inline switch_status_t jb_next_packet(switch_jb_t *jb, switch_jb_node_t **nodep)
 {
 	if (jb->samples_per_frame) {
 		return jb_next_packet_by_ts(jb, nodep);
 	} else {
-		return jb_next_packet_by_seq(jb, nodep);
+		switch_status_t status;
+		if (jb->elastic && jb->jitter.estimate) {
+			status = jb_next_packet_by_seq_with_acceleration(jb, nodep);
+		} else {
+			status = jb_next_packet_by_seq(jb, nodep);
+		}
+		return status;
 	}
 }

@@ -877,13 +1052,50 @@ SWITCH_DECLARE(void) switch_jb_ts_mode(switch_jb_t *jb, uint32_t samples_per_fra
 	switch_core_inthash_init(&jb->node_hash_ts);
 }

+SWITCH_DECLARE(void) switch_jb_set_jitter_estimator(switch_jb_t *jb, double *jitter, uint32_t samples_per_frame, uint32_t samples_per_second)
+{
+	if (jb && jitter) {
+		memset(&jb->jitter,0,sizeof(switch_jb_jitter_t));
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_max_ms", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_size_ms", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_acceleration_ms", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_expand_ms", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_jitter_max_ms", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_jitter_ms", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_count", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_too_big", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_missing_frames", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_ts_jump", "%u", 0);
+		switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_error", "%u", 0);
+		jb->jitter.estimate = jitter;
+		jb->jitter.samples_per_frame = samples_per_frame;
+	        jb->jitter.samples_per_second = samples_per_second;
+		jb->jitter.drop_gap = 5;
+	}
+}
+
 SWITCH_DECLARE(void) switch_jb_set_session(switch_jb_t *jb, switch_core_session_t *session)
 {
 	const char *var;

 	if (session) {
+		jb->codec = switch_core_session_get_read_codec(session);
 		jb->session = session;
 		jb->channel = switch_core_session_get_channel(session);
+		if (!strcmp(jb->codec->implementation->iananame, "opus")) {
+			if (switch_true(switch_channel_get_variable(jb->channel, "rtp_jitter_buffer_accelerate"))) {
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "codec is %s, accelerate on\n", jb->codec->implementation->iananame);
+				jb->elastic = SWITCH_TRUE;
+			} else {
+				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "codec is %s, accelerate off\n", jb->codec->implementation->iananame);
+				jb->elastic = SWITCH_FALSE;
+			}
+
+		} else {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "codec not opus: %s\n", jb->codec->implementation->iananame);
+			jb->elastic = SWITCH_FALSE;
+		}
+
 		if (jb->type == SJB_VIDEO && !switch_test_flag(jb, SJB_QUEUE_ONLY) && 
 			(var = switch_channel_get_variable_dup(jb->channel, "jb_video_low_bitrate", SWITCH_FALSE, -1))) {
 			int tmp = atoi(var);
@@ -932,6 +1144,12 @@ SWITCH_DECLARE(void) switch_jb_debug_level(switch_jb_t *jb, uint8_t level)

 SWITCH_DECLARE(void) switch_jb_reset(switch_jb_t *jb)
 {
+	jb->jitter.stats.reset++;
+	switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_count", "%u", jb->jitter.stats.reset);
+	switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_too_big", "%u", jb->jitter.stats.reset_too_big);
+	switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_missing_frames", "%u", jb->jitter.stats.reset_missing_frames);
+	switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_ts_jump", "%u", jb->jitter.stats.reset_ts_jump);
+	switch_channel_set_variable_printf(jb->channel, "rtp_jb_reset_error", "%u", jb->jitter.stats.reset_error);

 	if (jb->type == SJB_VIDEO) {
 		switch_mutex_lock(jb->mutex);
@@ -1257,6 +1475,7 @@ SWITCH_DECLARE(switch_status_t) switch_jb_put_packet(switch_jb_t *jb, switch_rtp
 		if (got > want) {
 			if (got - want > jb->max_frame_len && got - want > 17) {
 				jb_debug(jb, 2, "Missing %u frames, Resetting\n", got - want);
+				jb->jitter.stats.reset_missing_frames++;
 				switch_jb_reset(jb);
 			} else {

@@ -1434,6 +1653,7 @@ SWITCH_DECLARE(switch_status_t) switch_jb_get_packet(switch_jb_t *jb, switch_rtp
 			switch(status) {
 			case SWITCH_STATUS_RESTART:
 				jb_debug(jb, 2, "%s", "Error encountered\n");
+				jb->jitter.stats.reset_error++;
 				switch_jb_reset(jb);
 				switch_goto_status(SWITCH_STATUS_RESTART, end);
 			case SWITCH_STATUS_NOTFOUND:
@@ -1444,7 +1664,22 @@ SWITCH_DECLARE(switch_status_t) switch_jb_get_packet(switch_jb_t *jb, switch_rtp
 					jb_debug(jb, 2, "%s", "Too many frames not found, RESIZE\n");
 					switch_goto_status(SWITCH_STATUS_RESTART, end);
 				} else {
-					jb_debug(jb, 2, "%s", "Frame not found suggest PLC\n");
+					if (jb->elastic) {
+						int visible_not_old = check_jb_size(jb);
+						jb->jitter.stats.estimate_ms = (int)((*jb->jitter.estimate)/((jb->jitter.samples_per_second))*1000);
+						jb->jitter.stats.buffer_size_ms = (int)((visible_not_old*jb->jitter.samples_per_frame)/(jb->jitter.samples_per_second/1000));
+						/* When playing PLC, we take the oportunity to expand the buffer if the jitter buffer is smaller than the 3x the estimated jitter. */
+						if (jb->jitter.stats.buffer_size_ms < (3*jb->jitter.stats.estimate_ms)) {
+							jb_debug(jb, SWITCH_LOG_INFO, "JITTER estimation %dms buffersize %d/%d %dms EXPAND [plc]\n",
+									jb->jitter.stats.estimate_ms, jb->complete_frames , jb->frame_len, jb->jitter.stats.buffer_size_ms);
+							jb->jitter.stats.expand++;
+							decrement_seq(jb);
+						} else {
+							jb_debug(jb, 2, "%s", "Frame not found suggest PLC\n");
+						}
+					} else {
+						jb_debug(jb, 2, "%s", "Frame not found suggest PLC\n");
+					}
 					plc = 1;
 					switch_goto_status(SWITCH_STATUS_NOTFOUND, end);
 				}