Report locally measured loss and jitter per ingress SSRC without RTCP

daemon/call.c

@@ -5376,7 +5376,7 @@ void call_destroy(call_t *c) {
 					se->highest_mos->mos % 10,
 					((se->highest_mos->reported - c->created) / 1000000L) / 60,
 					((se->highest_mos->reported - c->created) / 1000000L) % 60,
-					(unsigned int) se->packets_lost);
+					ssrc_packets_lost(se));
 				ilog(LOG_INFO, "------ respective (avg/min/max) jitter %" PRIu64 "/%" PRIu64 "/%" PRIu64 " ms, "
 						"RTT-e2e %" PRIu64 ".%" PRIu64 "/%" PRIu64 ".%" PRIu64
 						"/%" PRIu64 ".%" PRIu64 " ms, "

daemon/call_interfaces.c

@@ -54,7 +54,7 @@ static const char* const _ng_basic_errors[] = {
 };
 
 static void ng_stats_ssrc(const ng_parser_t *parser, parser_arg dict, parser_arg list,
-		const struct ssrc_hash *ht);
+		const struct ssrc_hash *ht, const struct call_media *rx_media);
 static void call_set_dtmf_block(call_t *call, struct call_monologue *monologue, sdp_ng_flags *flags);
 
 
@@ -1000,8 +1000,8 @@ static void ng_stats_media(ng_command_ctx_t *ctx, parser_arg list, const struct
 	BF_M("transcoding", TRANSCODING);
 	BF_M("block egress", BLOCK_EGRESS);
 
-	ng_stats_ssrc(parser, ssrc, parser->dict_add_list(dict, "ingress SSRCs"), &m->ssrc_hash_in);
-	ng_stats_ssrc(parser, NULL, parser->dict_add_list(dict,  "egress SSRCs"), &m->ssrc_hash_out);
+	ng_stats_ssrc(parser, ssrc, parser->dict_add_list(dict, "ingress SSRCs"), &m->ssrc_hash_in, m);
+	ng_stats_ssrc(parser, NULL, parser->dict_add_list(dict,  "egress SSRCs"), &m->ssrc_hash_out, NULL);
 
 stats:
 	IQUEUE_FOREACH(&m->streams, ps)
@@ -1133,17 +1133,36 @@ static void ng_stats_ssrc_mos_entry_dict_avg(const ng_parser_t *parser, parser_a
 	parser->dict_add_int(subent, "samples", div);
 }
 
-static void ng_stats_ssrc_1(const ng_parser_t *parser, parser_arg ent, struct ssrc_entry_call *se) {
+// locally measured jitter in ms, or -1 if the clock rate is unknown
+static int64_t ng_stats_ssrc_jitter(struct ssrc_entry_call *se, const struct call_media *rx_media) {
+	if (!se->tracker.most_len || se->tracker.most[0] == 255)
+		return -1;
+	const rtp_payload_type *rpt = get_rtp_payload_type(se->tracker.most[0], &rx_media->codecs);
+	if (!rpt || !rpt->clock_rate)
+		return -1;
+	return (int64_t) (ssrc_jitter(se) >> 4) * 1000 / rpt->clock_rate;
+}
+
+// `rx_media` is set for received (ingress) SSRCs, for which loss and jitter are measured
+// locally from the RTP stream itself and so are available without any RTCP
+static void ng_stats_ssrc_1(const ng_parser_t *parser, parser_arg ent, struct ssrc_entry_call *se,
+		const struct call_media *rx_media)
+{
 	parser->dict_add_int(ent, "bytes", atomic64_get_na(&se->stats->bytes));
 	parser->dict_add_int(ent, "packets", atomic64_get_na(&se->stats->packets));
 	parser->dict_add_int(ent, "last RTP timestamp", atomic_get_na(&se->stats->timestamp));
 	parser->dict_add_int(ent, "last RTP seq", atomic_get_na(&se->stats->ext_seq));
 
+	if (rx_media) {
+		parser->dict_add_int(ent, "cumulative loss", ssrc_packets_lost(se));
+		int64_t jitter = ng_stats_ssrc_jitter(se, rx_media);
+		if (jitter >= 0)
+			parser->dict_add_int(ent, "jitter", jitter);
+	}
+
 	if (!se->stats_blocks.length || !se->lowest_mos || !se->highest_mos)
 		return;
 
-	parser->dict_add_int(ent, "cumulative loss", se->packets_lost);
-
 	int mos_samples = se->mos_samples;
 	if (mos_samples < 1) mos_samples = 1;
 	ng_stats_ssrc_mos_entry_dict_avg(parser, ent, "average MOS", &se->average_mos, mos_samples);
@@ -1173,7 +1192,7 @@ static void ng_stats_ssrc_1(const ng_parser_t *parser, parser_arg ent, struct ss
 }
 
 static void ng_stats_ssrc(const ng_parser_t *parser, parser_arg dict, parser_arg list,
-		const struct ssrc_hash *ht)
+		const struct ssrc_hash *ht, const struct call_media *rx_media)
 {
 	if (rtpe_config.ssrc_reporting == SRP_NONE)
 		return;
@@ -1188,11 +1207,11 @@ static void ng_stats_ssrc(const ng_parser_t *parser, parser_arg dict, parser_arg
 		parser->dict_add_int(ent, "SSRC", se->h.ssrc);
 
 		if ((rtpe_config.ssrc_reporting & 0x2) == 0)
-			ng_stats_ssrc_1(parser, ent, se);
+			ng_stats_ssrc_1(parser, ent, se, rx_media);
 
 		if (dict.gen && !parser->dict_contains(dict, tmp)) {
 			ent = parser->dict_add_dict_dup(dict, tmp);
-			ng_stats_ssrc_1(parser, ent, se);
+			ng_stats_ssrc_1(parser, ent, se, rx_media);
 		}
 	}
 }

daemon/mqtt.c

@@ -213,7 +213,7 @@ static void mqtt_ssrc_stats(struct ssrc_entry_call *ssrc, JsonBuilder *json, str
 	int64_t packets, octets, packets_lost, packets_lost_rtcp, duplicates;
 	packets = atomic64_get_na(&ssrc->stats->packets);
 	octets = atomic64_get_na(&ssrc->stats->bytes);
-	packets_lost = ssrc->packets_lost;
+	packets_lost = ssrc_packets_lost(ssrc);
 	packets_lost_rtcp = ssrc->packets_lost_rtcp;
 	duplicates = ssrc->duplicates;
 

daemon/rtcp.c

@@ -1325,7 +1325,7 @@ static void transcode_rr(struct rtcp_process_ctx *ctx, struct report_block *rr)
 	if (!packets)
 		goto out;
 
-	unsigned int lost = input_ctx->packets_lost;
+	unsigned int lost = ssrc_packets_lost(input_ctx);
 	unsigned int dupes = input_ctx->duplicates;
 	unsigned int tot_lost = lost - dupes; // can be negative/rollover
 
@@ -1521,10 +1521,10 @@ static GString *rtcp_sender_report(struct ssrc_sender_report *ssr,
 				tv_diff = rtpe_now - si->time_item.received;
 				ntp_middle_bits = si->time_item.ntp_middle_bits;
 			}
-			uint32_t jitter = s->jitter;
+			uint32_t jitter = ssrc_jitter(s);
 			mutex_unlock(&s->h.lock);
 
-			uint64_t lost = s->packets_lost;
+			uint64_t lost = ssrc_packets_lost(s);
 			uint64_t tot = atomic64_get(&s->stats->packets);
 
 			*rr = (struct report_block) {

daemon/rtp.c

@@ -325,7 +325,7 @@ int srtp_payloads(str *to_auth, str *to_decrypt, str *auth_tag, str *mki,
 	return -1;
 }
 
-const rtp_payload_type *get_rtp_payload_type(unsigned int type, struct codec_store *cs) {
+const rtp_payload_type *get_rtp_payload_type(unsigned int type, const struct codec_store *cs) {
 	const rtp_payload_type *rtp_pt;
 
 	if (!cs)

daemon/ssrc.c

@@ -727,6 +727,19 @@ void payload_tracker_add(struct payload_tracker *t, int pt) {
 }
 
 
+// sequence-gap based loss, counted in userspace and in the kernel for packets each one handled
+uint32_t ssrc_packets_lost(struct ssrc_entry_call *se) {
+	return se->packets_lost + atomic_get_na(&se->stats->total_lost);
+}
+
+// RFC 3550 interarrival jitter (scaled by 16, in RTP timestamp units). The kernel keeps
+// its own estimate if it measures RTP stats, which is the current one once it has taken
+// over forwarding this SSRC.
+uint32_t ssrc_jitter(struct ssrc_entry_call *se) {
+	uint32_t kernel_jitter = atomic_get_na(&se->stats->jitter);
+	return kernel_jitter ? kernel_jitter : se->jitter;
+}
+
 // call master lock held in R
 void ssrc_collect_metrics(struct call_media *media) {
 	for (GList *l = media->ssrc_hash_in.nq.head; l; l = l->next) {
@@ -735,17 +748,20 @@ void ssrc_collect_metrics(struct call_media *media) {
 			break; // end of list
 
 		// work on a copy, as the estimator itself must stay in RTP timestamp units
-		uint32_t jitter = s->jitter;
+		uint32_t jitter = ssrc_jitter(s);
 		// exclude zero values - technically possible but unlikely and probably just unset
 		if (!jitter)
 			continue;
 
-		if (s->tracker.most_len > 0 && s->tracker.most[0] != 255) {
-			const rtp_payload_type *rpt = get_rtp_payload_type(s->tracker.most[0],
-					&media->codecs);
-			if (rpt && rpt->clock_rate)
-				jitter = jitter * 1000 / rpt->clock_rate;
-		}
+		// The sampled metric is in ms. Without a known RTP clock rate,
+		// publishing raw clock ticks here would mislabel its units.
+		if (!s->tracker.most_len || s->tracker.most[0] == 255)
+			continue;
+		const rtp_payload_type *rpt = get_rtp_payload_type(s->tracker.most[0],
+				&media->codecs);
+		if (!rpt || !rpt->clock_rate)
+			continue;
+		jitter = jitter * 1000 / rpt->clock_rate;
 
 		if (media->streams.head) {
 			LOCK(&media->streams.head->lock);

docs/ng_control_protocol.md

@@ -2063,6 +2063,33 @@ The response dictionary contains the following keys:
 			A list of strings containing various status flags. Contains zero of more
 			of: `initialized`, `rtcp-mux`, `DTLS-SRTP`, `SDES`, `passthrough`, `ICE`.
 
+		+ `ingress SSRCs` and `egress SSRCs`
+
+			Lists of dictionaries, one for each SSRC received from (ingress) or
+			sent to (egress) this media. Each contains the keys `SSRC`,
+			`bytes`, `packets`, `last RTP timestamp` and `last RTP seq`.
+			Ingress entries of SSRCs which sent RTCP receiver reports
+			additionally contain MOS statistics (`average MOS`, `lowest MOS`,
+			`highest MOS`, `MOS progression`).
+
+			Ingress entries also contain values measured locally from the
+			received RTP itself, which are therefore available per call leg
+			even if no RTCP is present:
+
+			- `cumulative loss`: number of packets lost, based on gaps in the
+			  RTP sequence numbers. A packet is counted as lost once it falls
+			  out of a 32-packet reordering window. Bursts of loss of about
+			  32 packets or more are only approximated, and a jump of more
+			  than 256 is treated as a sequence reset and not counted.
+
+			- `jitter`: RFC 3550 interarrival jitter in milliseconds. Missing
+			  if the clock rate of the payload type is not known.
+
+			Packets forwarded by the kernel module are included only if it
+			measures RTP statistics, which is the case with `measure-rtp`
+			enabled (or with RTCP generation or MQTT publishing active).
+			Otherwise only packets handled in userspace are accounted for.
+
 		+ `streams`
 
 			Contains a list of dictionary representing the packet streams associated

include/rtp.h

@@ -15,7 +15,7 @@ struct packet_stream;
 
 typedef GString crypto_debug_string;
 
-const rtp_payload_type *get_rtp_payload_type(unsigned int, struct codec_store *);
+const rtp_payload_type *get_rtp_payload_type(unsigned int, const struct codec_store *);
 
 int rtp_avp2savp(const struct rtp_header *, str *packet, str *payload, struct crypto_context *,
 		struct ssrc_entry_call *);

include/ssrc.h

@@ -196,6 +196,9 @@ INLINE struct ssrc_entry_call *call_get_first_ssrc(struct ssrc_hash *ht) {
 	return ht->nq.head ? ht->nq.head->data : NULL;
 }
 
+uint32_t ssrc_packets_lost(struct ssrc_entry_call *se);
+uint32_t ssrc_jitter(struct ssrc_entry_call *se);
+
 void ssrc_sender_report(struct call_media *, const struct ssrc_sender_report *, int64_t);
 void ssrc_receiver_report(struct call_media *, stream_fd *, const struct ssrc_receiver_report *, int64_t);
 void ssrc_receiver_rr_time(struct call_media *m, const struct ssrc_xr_rr_time *rr, int64_t);

t/auto-daemon-tests-ssrc.pl

@@ -103,5 +103,76 @@
 rcv($sock_b, $port_a, rtpm(8, 1002, 90000, $egress, "\x00" x 160));
 
 
+# loss and jitter of received RTP are measured locally and reported without any RTCP
+
+($sock_a, $sock_b) = new_call([qw(198.51.100.1 7010)], [qw(198.51.100.3 7012)]);
+
+($port_a) = offer('local loss and jitter without RTCP', { }, <<SDP);
+v=0
+o=- 1545997027 1 IN IP4 198.51.100.1
+s=tester
+t=0 0
+m=audio 7010 RTP/AVP 8
+c=IN IP4 198.51.100.1
+a=sendrecv
+----------------------------------
+v=0
+o=- 1545997027 1 IN IP4 198.51.100.1
+s=tester
+t=0 0
+m=audio PORT RTP/AVP 8
+c=IN IP4 203.0.113.1
+a=rtpmap:8 PCMA/8000
+a=sendrecv
+a=rtcp:PORT
+SDP
+
+($port_b) = answer('local loss and jitter without RTCP', { }, <<SDP);
+v=0
+o=- 1545997027 1 IN IP4 198.51.100.3
+s=tester
+t=0 0
+m=audio 7012 RTP/AVP 8
+c=IN IP4 198.51.100.3
+a=rtpmap:8 PCMA/8000
+a=sendrecv
+--------------------------------------
+v=0
+o=- 1545997027 1 IN IP4 198.51.100.3
+s=tester
+t=0 0
+m=audio PORT RTP/AVP 8
+c=IN IP4 203.0.113.1
+a=rtpmap:8 PCMA/8000
+a=sendrecv
+a=rtcp:PORT
+SDP
+
+# seq 2003 goes missing. Loss is counted once it falls out of the 32 packet
+# reordering window, so send enough packets after it.
+for my $i (0 .. 40) {
+	next if $i == 3;
+	snd($sock_a, $port_b, rtp(8, 2000 + $i, 4000 + $i * 160, 0x5678, "\x00" x 160));
+	rcv($sock_b, $port_a, rtpm(8, 2000 + $i, 4000 + $i * 160, 0x5678, "\x00" x 160));
+}
+
+$resp = rtpe_req('query', 'local loss and jitter without RTCP', { 'call-id' => cid() });
+
+my $ingress = $resp->{tags}{ft()}{medias}[0]{'ingress SSRCs'};
+is scalar(@{$ingress // []}), 1, 'one ingress SSRC';
+is $ingress->[0]{SSRC}, 0x5678, 'ingress SSRC';
+is $ingress->[0]{packets}, 40, 'ingress packets';
+is $ingress->[0]{'cumulative loss'}, 1, 'loss reported without RTCP';
+# packets are sent back to back, much faster than their 20 ms of timestamp spacing
+ok(($ingress->[0]{jitter} // 0) > 0, 'jitter reported without RTCP');
+ok(!exists($ingress->[0]{'highest MOS'}), 'no MOS without RTCP');
+is $resp->{SSRC}{0x5678}{'cumulative loss'}, 1, 'loss in top level SSRC dictionary';
+
+my $egress_ssrcs = $resp->{tags}{tt()}{medias}[0]{'egress SSRCs'};
+is scalar(@{$egress_ssrcs // []}), 1, 'one egress SSRC';
+ok(!exists($egress_ssrcs->[0]{'cumulative loss'}), 'no local loss for egress SSRC');
+ok(!exists($egress_ssrcs->[0]{jitter}), 'no local jitter for egress SSRC');
+
+
 #done_testing;NGCP::Rtpengine::AutoTest::terminate('f00');exit;
 done_testing();