[Core] EJB: revised elastic jitter buffer — reduces per-call max delay ~2× in canary A/B

src/include/private/switch_core_pvt.h

@@ -170,6 +170,8 @@ struct switch_core_session {
 	switch_buffer_t *text_line_buffer;
 	switch_mutex_t *text_mutex;
 	const char *external_id;
+
+	packet_stats_t stats;
 };
 
 struct switch_media_bug {

src/include/switch_core.h

@@ -90,6 +90,31 @@ typedef struct device_uuid_node_s {
 	struct device_uuid_node_s *next;
 } switch_device_node_t;
 
+typedef struct packet_stats_io_info {
+	const char* in_callid;
+	char* in_codec;
+	uint32_t in_ssrc;
+	switch_sockaddr_t *in_remote_addr;
+	switch_sockaddr_t *in_local_addr;
+	const char* out_callid;
+	char* out_codec;
+	uint32_t out_ssrc;
+	switch_sockaddr_t *out_remote_addr;
+	switch_sockaddr_t *out_local_addr;
+	uint32_t count; // count of packets going out
+} packet_stats_io_info_t;
+
+typedef struct packet_stats {
+	int max;
+	float average;
+	uint32_t in_count;
+	uint32_t in_plc;
+	uint32_t in_rx;     /* Packets received via RTP and about to enter the JB. */
+	uint32_t count;
+	packet_stats_io_info_t io_info;
+	switch_bool_t reported;
+} packet_stats_t;
+
 typedef struct switch_device_stats_s {
 	uint32_t total;
 	uint32_t total_in;
@@ -255,6 +280,11 @@ static inline void *switch_must_realloc(void *_b, size_t _z)
 ///\{
 
 
+SWITCH_DECLARE(void) switch_core_session_increment_read(switch_core_session_t *session);
+SWITCH_DECLARE(void) switch_core_session_increment_plc(switch_core_session_t *session);
+SWITCH_DECLARE(void) switch_core_session_increment_rx(switch_core_session_t *session);
+SWITCH_DECLARE(void) packet_stats_print(switch_core_session_t *session);
+SWITCH_DECLARE(void) switch_core_session_set_io_stats(switch_core_session_t *session, packet_stats_io_info_t *packet_stats_io_info);
 SWITCH_DECLARE(void) switch_core_screen_size(int *x, int *y);
 SWITCH_DECLARE(void) switch_core_session_sched_heartbeat(switch_core_session_t *session, uint32_t seconds);
 SWITCH_DECLARE(void) switch_core_session_unsched_heartbeat(switch_core_session_t *session);

src/include/switch_core_media.h

@@ -292,6 +292,7 @@ SWITCH_DECLARE(void) switch_core_media_resume(switch_core_session_t *session);
 SWITCH_DECLARE(void) switch_core_media_init(void);
 SWITCH_DECLARE(void) switch_core_media_deinit(void);
 SWITCH_DECLARE(void) switch_core_media_set_stats(switch_core_session_t *session);
+SWITCH_DECLARE(void) switch_core_media_export_jb_stats(switch_core_session_t *session);
 SWITCH_DECLARE(void) switch_core_media_sync_stats(switch_core_session_t *session);
 SWITCH_DECLARE(void) switch_core_session_wake_video_thread(switch_core_session_t *session);
 SWITCH_DECLARE(void) switch_core_session_clear_crypto(switch_core_session_t *session);

src/include/switch_frame.h

@@ -87,6 +87,7 @@ typedef struct switch_frame_geometry {
 	payload_map_t *pmap;
 	switch_image_t *img;
 	struct switch_frame_geometry geometry;
+	switch_time_t received_ts;
 };
 
 SWITCH_END_EXTERN_C

src/include/switch_jitterbuffer.h

@@ -67,6 +67,8 @@ SWITCH_DECLARE(void) switch_jb_set_flag(switch_jb_t *jb, switch_jb_flag_t flag);
 SWITCH_DECLARE(void) switch_jb_clear_flag(switch_jb_t *jb, switch_jb_flag_t flag);
 SWITCH_DECLARE(uint32_t) switch_jb_get_nack_success(switch_jb_t *jb);
 SWITCH_DECLARE(uint32_t) switch_jb_get_packets_per_frame(switch_jb_t *jb);
+SWITCH_DECLARE(switch_bool_t) switch_jb_is_elastic(switch_jb_t *jb);
+SWITCH_DECLARE(void) switch_jb_export_stats(switch_jb_t *jb);
 
 SWITCH_END_EXTERN_C
 #endif

src/include/switch_rtp.h

@@ -57,6 +57,7 @@ typedef struct {
 	char body[SWITCH_RTP_MAX_BUF_LEN+4+sizeof(char *)];
 	switch_rtp_hdr_ext_t *ext;
 	char *ebody;
+	switch_time_t received_ts;
 } switch_rtp_packet_t;
 
 typedef enum {
@@ -360,6 +361,7 @@ SWITCH_DECLARE(switch_status_t) switch_rtp_debug_jitter_buffer(switch_rtp_t *rtp
 SWITCH_DECLARE(switch_status_t) switch_rtp_deactivate_jitter_buffer(switch_rtp_t *rtp_session);
 SWITCH_DECLARE(switch_status_t) switch_rtp_pause_jitter_buffer(switch_rtp_t *rtp_session, switch_bool_t pause);
 SWITCH_DECLARE(switch_jb_t *) switch_rtp_get_jitter_buffer(switch_rtp_t *rtp_session);
+SWITCH_DECLARE(switch_jb_t *) switch_rtp_get_jitter_buffer_for_stats(switch_rtp_t *rtp_session);
 
 
 

src/mod/codecs/mod_opus/mod_opus.c

@@ -109,6 +109,20 @@ struct dec_stats {
 	uint32_t fec_counter;
 	uint32_t plc_counter;
 	uint32_t frame_counter;
+	/* PLC run-length classification. ≤3 PLC frames in a run (≤60ms) are
+	 * essentially inaudible — opus spectrally interpolates from the prior
+	 * packet. Beyond 3 the output degenerates into comfort noise.
+	 *
+	 * A "run" is gap-tolerant: as long as a new PLC arrives within
+	 * PLC_GAP_TOLERANCE frames (200ms) of the prior PLC, it continues the
+	 * same run. This catches alternating glitchy patterns like
+	 * PLC-real-PLC-real-PLC where each PLC is "isolated" but the listener
+	 * hears a sustained audible artifact. */
+	uint32_t plc_run_len;       /* current run length (gap-tolerant) */
+	uint32_t plc_gap_since;     /* frames since last PLC (caps at gap tolerance) */
+	uint32_t plc_short;         /* PLC frames in runs ≤3 — soft */
+	uint32_t plc_long;          /* PLC frames in runs >3 — audible */
+	uint32_t plc_long_starts;   /* count of runs that crossed the >3 threshold */
 };
 typedef struct dec_stats dec_stats_t;
 
@@ -732,8 +746,19 @@ static switch_status_t switch_opus_destroy(switch_codec_t *codec)
 		if (context->decoder_object) {
 			switch_core_session_t *session = codec->session;
 			if (session) {
+				switch_channel_t *channel = switch_core_session_get_channel(session);
 				switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG,"Opus decoder stats: Frames[%d] PLC[%d] FEC[%d]\n",
 										context->decoder_stats.frame_counter, context->decoder_stats.plc_counter-context->decoder_stats.fec_counter, context->decoder_stats.fec_counter);
+				if (channel) {
+					switch_channel_set_variable_printf(channel, "opus_decoder_frames", "%u", context->decoder_stats.frame_counter);
+					switch_channel_set_variable_printf(channel, "opus_decoder_plc", "%u",
+						context->decoder_stats.plc_counter > context->decoder_stats.fec_counter
+							? context->decoder_stats.plc_counter - context->decoder_stats.fec_counter : 0);
+					switch_channel_set_variable_printf(channel, "opus_decoder_fec", "%u", context->decoder_stats.fec_counter);
+					switch_channel_set_variable_printf(channel, "opus_decoder_plc_short", "%u", context->decoder_stats.plc_short);
+					switch_channel_set_variable_printf(channel, "opus_decoder_plc_long",  "%u", context->decoder_stats.plc_long);
+					switch_channel_set_variable_printf(channel, "opus_decoder_plc_long_starts", "%u", context->decoder_stats.plc_long_starts);
+				}
 			}
 			opus_decoder_destroy(context->decoder_object);
 			context->decoder_object = NULL;
@@ -916,15 +941,53 @@ static switch_status_t switch_opus_decode(switch_codec_t *codec,
 						 !encoded_data ? "PLC correction" : fec ?  "FEC correction" : "decode");
 	}
 
+	/* Gap-tolerant PLC run accounting. PLC_GAP_TOLERANCE is in frames at
+	 * 20ms ptime. If a new PLC fires within this gap of the previous one,
+	 * treat it as continuing the same audible event. Otherwise start fresh. */
+	#define PLC_GAP_TOLERANCE 10  /* 10 frames * 20ms = 200ms */
 	if (plc) {
 		context->decoder_stats.plc_counter++;
+		if (context->decoder_stats.plc_gap_since <= PLC_GAP_TOLERANCE) {
+			context->decoder_stats.plc_run_len++;
+		} else {
+			context->decoder_stats.plc_run_len = 1;
+		}
+		context->decoder_stats.plc_gap_since = 0;
+		if (context->decoder_stats.plc_run_len <= 3) {
+			context->decoder_stats.plc_short++;
+		} else {
+			if (context->decoder_stats.plc_run_len == 4) {
+				context->decoder_stats.plc_long_starts++;
+			}
+			context->decoder_stats.plc_long++;
+		}
+	} else {
+		if (context->decoder_stats.plc_gap_since <= PLC_GAP_TOLERANCE) {
+			context->decoder_stats.plc_gap_since++;
+		}
 	}
 	if (fec) {
 		context->decoder_stats.fec_counter++;
 	}
 	/* a frame for which we decode FEC will be counted twice */
 	context->decoder_stats.frame_counter++;
 
+	/* Keep channel variables fresh on each PLC/FEC event so the hangup hook
+	 * (api_hangup_hook fires in CS_HANGUP, before CS_DESTROY runs codec destroy)
+	 * sees current values. Normal decode path does not hit this branch. */
+	if ((plc || fec) && session) {
+		switch_channel_t *channel = switch_core_session_get_channel(session);
+		if (channel) {
+			switch_channel_set_variable_printf(channel, "opus_decoder_plc", "%u",
+				context->decoder_stats.plc_counter > context->decoder_stats.fec_counter
+					? context->decoder_stats.plc_counter - context->decoder_stats.fec_counter : 0);
+			switch_channel_set_variable_printf(channel, "opus_decoder_fec", "%u", context->decoder_stats.fec_counter);
+			switch_channel_set_variable_printf(channel, "opus_decoder_plc_short", "%u", context->decoder_stats.plc_short);
+			switch_channel_set_variable_printf(channel, "opus_decoder_plc_long",  "%u", context->decoder_stats.plc_long);
+			switch_channel_set_variable_printf(channel, "opus_decoder_plc_long_starts", "%u", context->decoder_stats.plc_long_starts);
+		}
+	}
+
 	samples = opus_decode(context->decoder_object, encoded_data, encoded_data_len, decoded_data, frame_size, fec);
 
 	if (samples < 0) {

src/switch_core_media.c

@@ -35,6 +35,7 @@
 #include <switch_stun.h>
 #include <switch_nat.h>
 #include "private/switch_core_pvt.h"
+#include <fspr_network_io.h>
 #include <switch_curl.h>
 #include <errno.h>
 #include <sofia-sip/sdp.h>
@@ -274,6 +275,72 @@ struct switch_media_handle_s {
 
 };
 
+void packet_stats_init(packet_stats_t *stats, int latency, int count) {
+	stats->max = latency;
+	stats->average = latency;
+	stats->count = count;
+}
+#define _VOR1(v) ((v)?(v):1)
+void packet_stats_update(packet_stats_t *stats, int latency)
+{
+	if (stats->count >= UINT32_MAX)
+		return;
+	stats->count++;
+
+	if (stats->count == 1)
+		packet_stats_init(stats, latency, 1);
+	if (stats->max < latency)
+		stats->max = latency;
+
+	if (stats->count > 1) {
+		float delta;
+		delta = latency - stats->average;
+		stats->average += delta/_VOR1(stats->count);
+	}
+}
+
+void packet_stats_print(switch_core_session_t *session) {
+	if (session->stats.io_info.in_remote_addr && session->stats.io_info.out_local_addr
+		       && !session->stats.reported) {
+		char in_ipbuf[48];
+		char in_l_ipbuf[48];
+		char out_ipbuf[48];
+		char out_l_ipbuf[48];
+		const char *v=NULL;
+
+		switch_channel_set_variable_printf(session->channel, "packet_stats_report",
+			"{"
+			"\"in\" : { \"ssrc\": \"0x%08X\", \"remote_socket\": \"%s:%u\", \"local_socket\": \"%s:%u\""
+			", \"codec\": \"%s\", \"count\": %u, \"plc\": %u, \"rx\": %u}"
+			","
+			"\"out\" : { \"ssrc\": \"0x%08X\", \"remote_socket\": \"%s:%u\", \"local_socket\": \"%s:%u\""
+			", \"codec\": \"%s\", \"count\": %u, \"max\": %d, \"avg\": %.2f }}",
+			session->stats.io_info.in_ssrc,
+               switch_get_addr(in_ipbuf, sizeof(in_ipbuf), session->stats.io_info.in_remote_addr),
+               session->stats.io_info.in_remote_addr->port,
+               switch_get_addr(in_l_ipbuf, sizeof(in_ipbuf), session->stats.io_info.in_local_addr),
+               session->stats.io_info.in_local_addr->port,
+			session->stats.io_info.in_codec,
+			session->stats.in_count,
+			session->stats.in_plc,
+			session->stats.in_rx,
+			session->stats.io_info.out_ssrc,
+               switch_get_addr(out_ipbuf, sizeof(out_ipbuf), session->stats.io_info.out_remote_addr),
+               session->stats.io_info.out_local_addr->port,
+               switch_get_addr(out_l_ipbuf, sizeof(out_ipbuf), session->stats.io_info.out_local_addr),
+               session->stats.io_info.out_remote_addr->port,
+			session->stats.io_info.out_codec,
+			session->stats.count,
+			session->stats.max,
+			session->stats.average
+		);
+		v = switch_channel_get_variable_dup(session->channel,"packet_stats_report", SWITCH_FALSE, -1);
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "[packet_stats_report] %s\n", v);
+		session->stats.reported = true;
+
+	}
+}
+
 switch_srtp_crypto_suite_t SUITES[CRYPTO_INVALID] = {
 	{ "AEAD_AES_256_GCM_8", "", AEAD_AES_256_GCM_8, 44, 12},
 	{ "AEAD_AES_256_GCM", "", AEAD_AES_256_GCM, 44, 12},
@@ -1894,6 +1961,14 @@ static void set_stats(switch_core_session_t *session, switch_media_type_t type,
 		add_stat(stats->rtcp.packet_count, "rtcp_packet_count");
 		add_stat(stats->rtcp.octet_count, "rtcp_octet_count");
 
+		/* Inbound sequence-loss counters from the RTCP receiver-report path.
+		 * cum_lost is the running total of packets FS never received (detected
+		 * via sequence-number gaps). Expected = high_ext_seq - base_seq + 1,
+		 * so cum_lost / expected is the true inbound loss ratio. Only populated
+		 * when rtcp-audio-interval-msec is set on the profile. */
+		add_stat(stats->rtcp.cum_lost, "in_cum_lost");
+		add_stat((stats->rtcp.high_ext_seq_recv - stats->rtcp.base_seq + 1), "in_expected_pkt");
+
 	}
 }
 
@@ -1926,13 +2001,66 @@ SWITCH_DECLARE(void) switch_core_media_sync_stats(switch_core_session_t *session
 
 }
 
+static void set_jb_stats(switch_core_session_t *session, switch_media_type_t type)
+{
+	switch_media_handle_t *smh;
+	switch_rtp_engine_t *engine;
+	switch_jb_t *jb;
+	const char *type_str = (type == SWITCH_MEDIA_TYPE_AUDIO) ? "audio" :
+	                       (type == SWITCH_MEDIA_TYPE_VIDEO) ? "video" : "text";
+
+	if (!(smh = session->media_handle)) {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG,
+			"set_jb_stats(%s): no media handle\n", type_str);
+		return;
+	}
+
+	engine = &smh->engines[type];
+	if (!engine->rtp_session) {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG,
+			"set_jb_stats(%s): no rtp_session\n", type_str);
+		return;
+	}
+
+	/* Use _for_stats variant which bypasses switch_rtp_ready() check
+	 * since RTP session may not be "ready" during hangup but JB still exists */
+	jb = switch_rtp_get_jitter_buffer_for_stats(engine->rtp_session);
+	if (jb) {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO,
+			"set_jb_stats(%s): exporting JB stats\n", type_str);
+		switch_jb_export_stats(jb);
+	} else {
+		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO,
+			"set_jb_stats(%s): no jitter buffer\n", type_str);
+	}
+}
+
+SWITCH_DECLARE(void) switch_core_media_export_jb_stats(switch_core_session_t *session)
+{
+	if (!session->media_handle) {
+		return;
+	}
+
+	switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO,
+		"switch_core_media_export_jb_stats() exporting JB stats before hangup handlers\n");
+
+	/* Export jitter buffer stats so they're available in hangup hooks */
+	set_jb_stats(session, SWITCH_MEDIA_TYPE_AUDIO);
+	set_jb_stats(session, SWITCH_MEDIA_TYPE_VIDEO);
+	set_jb_stats(session, SWITCH_MEDIA_TYPE_TEXT);
+}
+
 SWITCH_DECLARE(void) switch_core_media_set_stats(switch_core_session_t *session)
 {
+	switch_channel_t *channel = switch_core_session_get_channel(session);
 
 	if (!session->media_handle) {
 		return;
 	}
 
+	switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO,
+		"[%s] switch_core_media_set_stats() called\n", switch_channel_get_name(channel));
+
 	switch_core_media_sync_stats(session);
 
 	set_stats(session, SWITCH_MEDIA_TYPE_AUDIO, "audio");
@@ -15809,7 +15937,6 @@ SWITCH_DECLARE(switch_msrp_session_t *) switch_core_media_get_msrp_session(switc
 	return session->media_handle->msrp_session;
 }
 
-
 SWITCH_DECLARE(switch_status_t) switch_core_session_write_frame(switch_core_session_t *session, switch_frame_t *frame, switch_io_flag_t flags,
 																int stream_id)
 {
@@ -16168,6 +16295,7 @@ SWITCH_DECLARE(switch_status_t) switch_core_session_write_frame(switch_core_sess
 		goto done;
 	}
 
+	write_frame->received_ts = frame->received_ts;
 	if (session->write_codec) {
 		if (!ptime_mismatch && write_frame->codec && write_frame->codec->implementation &&
 			write_frame->codec->implementation->decoded_bytes_per_packet == session->write_impl.decoded_bytes_per_packet) {
@@ -16443,6 +16571,16 @@ SWITCH_DECLARE(switch_status_t) switch_core_session_write_frame(switch_core_sess
 	}
 
   error:
+	if (frame->received_ts > 0) {
+		int64_t d = (switch_micro_time_now() - frame->received_ts)/1000;
+		if (d > 2000) {
+			switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "excessive delay[%ld]-[%ld]=[%ldms] seq[%u]ssrc[0x%08X]\n",
+					(int64_t)(switch_micro_time_now()/1000), (int64_t)(frame->received_ts/1000), d, ntohs(frame->seq), frame->ssrc);
+		}
+		packet_stats_update(&session->stats, d);
+	}
+	session->stats.io_info.out_codec = write_frame->codec->implementation->iananame;
+	session->stats.io_info.in_codec = frame->codec->implementation->iananame;
 
 	switch_mutex_unlock(session->write_codec->mutex);
 	switch_mutex_unlock(frame->codec->mutex);