rtpmanager: save the report block statistics in each RTPSource

Move RB info from receiver reports into the internal source that the RR
are about, and deprecate (but retain) the old mapping where each
external source has only a single RB entry in the rtp statistics.

The old method is broken if a remote peer uses a single ssrc to send
receiver reports for more than one of our internal sources, other
as multiple RB in a single packet, or alternate RB in different reports.
In each case only the most recent entry was kept, overwriting data for
other internal sources.

In multicast scenarios each internal source may receive multiple
receiver reports from different peers. To support that, all received
RR's are now stored into a hash table indexed by the sender's SSRC,
and all RRs are placed into an array when generating statistics, so
that the information from all peers is retrievable.

The current deficient behaviour (adding RB info into non-internal RTPSources) is
deprecated but kept in order to be backward compatible, and retained
that way in the generated statistics structure.

Refs
[1] https://tools.ietf.org/html/rfc3550#section-6.4.1

Based on a patch by Fede Claramonte <fclaramonte@twilio.com>

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/7424>

subprojects/gst-plugins-good/gst/rtpmanager/rtpsession.c

@@ -2441,6 +2441,7 @@ rtp_session_process_rb (RTPSession * sess, RTPSource * source,
     GstRTCPPacket * packet, RTPPacketInfo * pinfo)
 {
   guint count, i;
+  guint32 sender_ssrc = source->ssrc;
 
   count = gst_rtcp_packet_get_rb_count (packet);
   for (i = 0; i < count; i++) {
@@ -2452,7 +2453,8 @@ rtp_session_process_rb (RTPSession * sess, RTPSource * source,
     gst_rtcp_packet_get_rb (packet, i, &ssrc, &fractionlost,
         &packetslost, &exthighestseq, &jitter, &lsr, &dlsr);
 
-    GST_DEBUG ("RB %d: SSRC %08x, jitter %" G_GUINT32_FORMAT, i, ssrc, jitter);
+    GST_DEBUG ("RB %d from SSRC %08x: SSRC %08x, jitter %" G_GUINT32_FORMAT, i,
+        sender_ssrc, ssrc, jitter);
 
     /* find our own source */
     src = find_source (sess, ssrc);
@@ -2461,11 +2463,13 @@ rtp_session_process_rb (RTPSession * sess, RTPSource * source,
 
     if (src->internal && RTP_SOURCE_IS_ACTIVE (src)) {
       /* only deal with report blocks for our session, we update the stats of
-       * the sender of the RTCP message. We could also compare our stats against
+       * the ssrc the RTCP message is about. We could also compare our stats against
        * the other sender to see if we are better or worse. */
-      /* FIXME, need to keep track who the RB block is from */
+      rtp_source_process_rb (src, ssrc, sender_ssrc, pinfo->ntpnstime,
-      rtp_source_process_rb (source, ssrc, pinfo->ntpnstime, fractionlost,
+          fractionlost, packetslost, exthighestseq, jitter, lsr, dlsr);
-          packetslost, exthighestseq, jitter, lsr, dlsr);
+      /* deprecated: it is also stored in the non-internal source */
+      rtp_source_process_rb (source, ssrc, sender_ssrc, pinfo->ntpnstime,
+          fractionlost, packetslost, exthighestseq, jitter, lsr, dlsr);
     }
   }
   on_ssrc_active (sess, source);

subprojects/gst-plugins-good/gst/rtpmanager/rtpsource.c

@@ -198,24 +198,48 @@ rtp_source_class_init (RTPSourceClass * klass)
    * * "sent-rb-lsr"           G_TYPE_UINT     last SR time (seconds in NTP Short Format, 16.16 fixed point)
    * * "sent-rb-dlsr"          G_TYPE_UINT     delay since last SR (seconds in NTP Short Format, 16.16 fixed point)
    *
-   * The following fields are only present for non-internal sources and
+   * The following fields are present for non-internal and internal sources, but
-   * represents the last RB that this source sent. This is only updated
+   * the meaning is different.
-   * when the source is receiving data and sending RB blocks.
+   *   non-internal: it represents the last RB (RR in this case) that this source sent. This is
+   * only updated when the source is receiving data and sending RB blocks. It is
+   * deprecated and is present for backwards compatibility. Statistics about internal local sources
+   * should be retrieved from the source.
+   *   internal: it represents the last RB (RR in this case) received with remote statistics about this source.
    *
    * * "have-rb"          G_TYPE_BOOLEAN  the source has sent RB
+   * * "rb-ssrc"          G_TYPE_UINT     The SSRC of the source the RB is about
    * * "rb-fractionlost"  G_TYPE_UINT     lost 8-bit fraction
    * * "rb-packetslost"   G_TYPE_INT      lost packets
    * * "rb-exthighestseq" G_TYPE_UINT     highest received seqnum
    * * "rb-jitter"        G_TYPE_UINT     reception jitter (in clock rate units)
    * * "rb-lsr"           G_TYPE_UINT     last SR time (seconds in NTP Short Format, 16.16 fixed point)
    * * "rb-dlsr"          G_TYPE_UINT     delay since last SR (seconds in NTP Short Format, 16.16 fixed point)
+   * * "rb-round-trip"    G_TYPE_UINT     the round-trip time (seconds in NTP Short Format, 16.16 fixed point)
    *
    * The round trip of this source is calculated from the last RB
-   * values and the reception time of the last RB packet. It is only present for
+   * values and the reception time of the last RB packet.
-   * non-internal sources.
    *
+   * The following field is present only for internal sources and
+   * contains an array of the the most recent receiver reports from each peer. In unicast
+   * scenarios this will be a single entry that is identical to the data provided by the have-rb and rb-*
+   * fields, but in multicast there will be one entry in the array for each peer that is sending
+   * receiver reports.
+   *
+   * * "received-rr"  GST_TYPE_LIST   Array of GstStructure entries, one for each peer that sent an RR.
+   *
+   * Each entry in the array contains the following fields:
+   *
+   *   * "rb-ssrc"          G_TYPE_UINT     The SSRC of this source
+   *   * "rb-sender-ssrc"   G_TYPE_UINT     The SSRC of the sender of this RR
+   *   * "rb-fractionlost"  G_TYPE_UINT     lost 8-bit fraction
+   *   * "rb-packetslost"   G_TYPE_INT      lost packets
+   *   * "rb-exthighestseq" G_TYPE_UINT     highest received seqnum
+   *   * "rb-jitter"        G_TYPE_UINT     reception jitter (in clock rate units)
+   *   * "rb-lsr"           G_TYPE_UINT     last SR time (seconds in NTP Short Format, 16.16 fixed point)
+   *   * "rb-dlsr"          G_TYPE_UINT     delay since last SR (seconds in NTP Short Format, 16.16 fixed point)
    *   * "rb-round-trip"    G_TYPE_UINT     the round-trip time (seconds in NTP Short Format, 16.16 fixed point)
    *
+   *
    */
   g_object_class_install_property (gobject_class, PROP_STATS,
       g_param_spec_boxed ("stats", "Stats",
@@ -268,6 +292,7 @@ rtp_source_reset (RTPSource * src)
   src->bye_reason = NULL;
   src->sent_bye = FALSE;
   g_hash_table_remove_all (src->reported_in_sr_of);
+  g_hash_table_remove_all (src->received_rr);
   g_queue_foreach (src->retained_feedback, (GFunc) gst_buffer_unref, NULL);
   g_queue_clear (src->retained_feedback);
   src->last_rtptime = -1;
@@ -316,6 +341,8 @@ rtp_source_init (RTPSource * src)
   src->nack_deadlines = g_array_new (FALSE, FALSE, sizeof (GstClockTime));
 
   src->reported_in_sr_of = g_hash_table_new (g_direct_hash, g_direct_equal);
+  src->received_rr =
+      g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL, g_free);
 
   src->last_keyframe_request = GST_CLOCK_TIME_NONE;
 
@@ -361,17 +388,14 @@ rtp_source_finalize (GObject * object)
     g_object_unref (src->rtcp_from);
 
   g_hash_table_unref (src->reported_in_sr_of);
+  g_hash_table_unref (src->received_rr);
 
   G_OBJECT_CLASS (rtp_source_parent_class)->finalize (object);
 }
 
-static GstStructure *
+static void
-rtp_source_create_stats (RTPSource * src)
+rtp_source_get_rb_stats (RTPSource * src, GstStructure * s)
 {
-  GstStructure *s;
-  gboolean is_sender = src->is_sender;
-  gboolean internal = src->internal;
-  gchar *address_str;
   gboolean have_rb;
   guint32 ssrc = 0;
   guint8 fractionlost = 0;
@@ -381,6 +405,63 @@ rtp_source_create_stats (RTPSource * src)
   guint32 lsr = 0;
   guint32 dlsr = 0;
   guint32 round_trip = 0;
+
+  have_rb = rtp_source_get_last_rb (src, &ssrc, &fractionlost,
+      &packetslost, &exthighestseq, &jitter, &lsr, &dlsr, &round_trip);
+
+  gst_structure_set (s,
+      "have-rb", G_TYPE_BOOLEAN, have_rb,
+      "rb-ssrc", G_TYPE_UINT, ssrc,
+      "rb-fractionlost", G_TYPE_UINT, (guint) fractionlost,
+      "rb-packetslost", G_TYPE_INT, (gint) packetslost,
+      "rb-exthighestseq", G_TYPE_UINT, (guint) exthighestseq,
+      "rb-jitter", G_TYPE_UINT, (guint) jitter,
+      "rb-lsr", G_TYPE_UINT, (guint) lsr,
+      "rb-dlsr", G_TYPE_UINT, (guint) dlsr,
+      "rb-round-trip", G_TYPE_UINT, (guint) round_trip, NULL);
+}
+
+static void
+_create_rr_entry (gpointer * key, gpointer val, gpointer userdata)
+{
+  GValue *res = userdata;
+  guint32 sender_ssrc = GPOINTER_TO_UINT (key);
+  RTPReceiverReport *rr = val;
+  GstStructure *s = gst_structure_new ("application/x-rtp-receiver-report",
+      "rb-ssrc", G_TYPE_UINT, (guint) rr->ssrc,
+      "rb-sender-ssrc", G_TYPE_UINT, (guint) sender_ssrc,
+      "rb-fractionlost", G_TYPE_UINT, (guint) rr->fractionlost,
+      "rb-packetslost", G_TYPE_INT, (gint) rr->packetslost,
+      "rb-exthighestseq", G_TYPE_UINT, (guint) rr->exthighestseq,
+      "rb-jitter", G_TYPE_UINT, (guint) rr->jitter,
+      "rb-lsr", G_TYPE_UINT, (guint) rr->lsr,
+      "rb-dlsr", G_TYPE_UINT, (guint) rr->dlsr,
+      "rb-round-trip", G_TYPE_UINT, (guint) rr->round_trip, NULL);
+
+  GValue v = G_VALUE_INIT;
+  g_value_init (&v, GST_TYPE_STRUCTURE);
+  gst_value_set_structure (&v, s);
+  gst_value_list_append_and_take_value (res, &v);
+  gst_structure_free (s);
+}
+
+static void
+rtp_source_get_rr_stats (RTPSource * src, GstStructure * s)
+{
+  GValue rr_list = G_VALUE_INIT;
+  g_value_init (&rr_list, GST_TYPE_LIST);
+  g_hash_table_foreach (src->received_rr, (GHFunc) _create_rr_entry, &rr_list);
+
+  gst_structure_take_value (s, "received-rr", &rr_list);
+}
+
+static GstStructure *
+rtp_source_create_stats (RTPSource * src)
+{
+  GstStructure *s;
+  gboolean is_sender = src->is_sender;
+  gboolean internal = src->internal;
+  gchar *address_str;
   gboolean have_sr;
   GstClockTime time = 0;
   guint64 ntptime = 0;
@@ -453,20 +534,12 @@ rtp_source_create_stats (RTPSource * src)
         (guint) src->last_rr.lsr, "sent-rb-dlsr", G_TYPE_UINT,
         (guint) src->last_rr.dlsr, NULL);
 
+    /* Deprecated, report block information is reported in each RTPSource */
     /* get the last RB */
-    have_rb = rtp_source_get_last_rb (src, &ssrc, &fractionlost,
+    rtp_source_get_rb_stats (src, s);
-        &packetslost, &exthighestseq, &jitter, &lsr, &dlsr, &round_trip);
+  } else {
-
+    rtp_source_get_rb_stats (src, s);
-    gst_structure_set (s,
+    rtp_source_get_rr_stats (src, s);
-        "have-rb", G_TYPE_BOOLEAN, have_rb,
-        "rb-ssrc", G_TYPE_UINT, ssrc,
-        "rb-fractionlost", G_TYPE_UINT, (guint) fractionlost,
-        "rb-packetslost", G_TYPE_INT, (gint) packetslost,
-        "rb-exthighestseq", G_TYPE_UINT, (guint) exthighestseq,
-        "rb-jitter", G_TYPE_UINT, (guint) jitter,
-        "rb-lsr", G_TYPE_UINT, (guint) lsr,
-        "rb-dlsr", G_TYPE_UINT, (guint) dlsr,
-        "rb-round-trip", G_TYPE_UINT, (guint) round_trip, NULL);
   }
 
   return s;
@@ -1514,7 +1587,8 @@ rtp_source_process_sr (RTPSource * src, GstClockTime time, guint64 ntptime,
 /**
  * rtp_source_process_rb:
  * @src: an #RTPSource
- * @ssrc: SSRC of the local source for this this RB was sent
+ * @ssrc: SSRC of the local source for which this RB was sent
+ * @sender_ssrc: SSRC from which this RB was received
  * @ntpnstime: the current time in nanoseconds since 1970
  * @fractionlost: fraction lost since last SR/RR
  * @packetslost: the cumulative number of packets lost
@@ -1528,9 +1602,9 @@ rtp_source_process_sr (RTPSource * src, GstClockTime time, guint64 ntptime,
  * Update the report block in @src.
  */
 void
-rtp_source_process_rb (RTPSource * src, guint32 ssrc, guint64 ntpnstime,
+rtp_source_process_rb (RTPSource * src, guint32 ssrc, guint32 sender_ssrc,
-    guint8 fractionlost, gint32 packetslost, guint32 exthighestseq,
+    guint64 ntpnstime, guint8 fractionlost, gint32 packetslost,
-    guint32 jitter, guint32 lsr, guint32 dlsr)
+    guint32 exthighestseq, guint32 jitter, guint32 lsr, guint32 dlsr)
 {
   RTPReceiverReport *curr;
   gint curridx;
@@ -1574,6 +1648,14 @@ rtp_source_process_rb (RTPSource * src, guint32 ssrc, guint64 ntpnstime,
 
   /* make current */
   src->stats.curr_rr = curridx;
+
+  if (src->internal) {
+    /* Make a copy to store in the received rr's hash table, but only for
+     * internal sources to track which remote sources sent reports for this source */
+    RTPReceiverReport *copy = g_memdup2 (curr, sizeof (RTPReceiverReport));
+    g_hash_table_replace (src->received_rr, GUINT_TO_POINTER (sender_ssrc),
+        copy);
+  }
 }
 
 /**

subprojects/gst-plugins-good/gst/rtpmanager/rtpsource.h

@@ -187,7 +187,9 @@ struct _RTPSource {
   gpointer           user_data;
 
   RTPSourceStats stats;
-  RTPReceiverReport last_rr;
+  RTPReceiverReport last_rr; /* last_rr sent for this source */
+
+  GHashTable    *received_rr; /* set of sender SSRC -> (RTPReceiveReport *) */
 
   GList         *conflicting_addresses;
 
@@ -252,7 +254,8 @@ GstFlowReturn   rtp_source_send_rtp            (RTPSource *src, RTPPacketInfo *p
 /* RTCP messages */
 void            rtp_source_process_sr          (RTPSource *src, GstClockTime time, guint64 ntptime,
                                                 guint32 rtptime, guint32 packet_count, guint32 octet_count);
-void            rtp_source_process_rb          (RTPSource *src, guint32 ssrc, guint64 ntpnstime, guint8 fractionlost,
+void            rtp_source_process_rb          (RTPSource *src, guint32 ssrc, guint32 sender_ssrc,
+                                                guint64 ntpnstime, guint8 fractionlost,
                                                 gint32 packetslost, guint32 exthighestseq, guint32 jitter,
                                                 guint32 lsr, guint32 dlsr);
 

subprojects/gst-plugins-good/tests/check/elements/rtpsession.c

@@ -977,6 +977,72 @@ GST_START_TEST (test_ignore_suspicious_bye)
 
 GST_END_TEST;
 
+GST_START_TEST (test_rr_stats_assignment)
+{
+  SessionHarness *h = session_harness_new ();
+  GstFlowReturn res;
+  GstBuffer *in_buf, *rtcp_buf;
+  gint i, j;
+
+  guint ssrcs[] = {
+    0x01BADBAD,
+    0xDEADBEEF,
+  };
+
+  /* receive buffers with multiple ssrcs */
+  for (i = 0; i < 2; i++) {
+    for (j = 0; j < G_N_ELEMENTS (ssrcs); j++) {
+      in_buf = generate_test_buffer (i, ssrcs[j]);
+      res = session_harness_recv_rtp (h, in_buf);
+      fail_unless_equals_int (GST_FLOW_OK, res);
+    }
+  }
+
+  /* crank the rtcp-thread and pull out the rtcp-packet we have generated */
+  session_harness_crank_clock (h);
+  rtcp_buf = session_harness_pull_rtcp (h);
+
+  g_assert (rtcp_buf != NULL);
+  fail_unless (gst_rtcp_buffer_validate (rtcp_buf));
+
+  /* Now take this RTCP buffer to a second 'sender' session and check
+   * that the RR info gets assigned to the correct internal senders */
+  session_harness_free (h);
+  h = session_harness_new ();
+
+  /* Send some packets to create the sources */
+  fail_unless_equals_int (GST_FLOW_OK,
+      session_harness_send_rtp (h, generate_test_buffer (0, 0x01BADBAD)));
+  fail_unless_equals_int (GST_FLOW_OK,
+      session_harness_send_rtp (h, generate_test_buffer (0, 0xDEADBEEF)));
+
+  session_harness_recv_rtcp (h, rtcp_buf);
+
+  for (i = 0; i < G_N_ELEMENTS (ssrcs); i++) {
+    guint32 ssrc = ssrcs[i], rb_ssrc;
+    GObject *source;
+    GstStructure *stats;
+    gboolean have_rb = FALSE;
+
+    g_signal_emit_by_name (h->internal_session, "get-source-by-ssrc", ssrc,
+        &source);
+
+    g_object_get (source, "stats", &stats, NULL);
+
+    GST_DEBUG ("Got stats from source %" GST_PTR_FORMAT " %" GST_PTR_FORMAT,
+        source, stats);
+
+    fail_unless (gst_structure_get_boolean (stats, "have-rb", &have_rb));
+    fail_unless (gst_structure_get_uint (stats, "rb-ssrc", &rb_ssrc));
+    fail_unless (rb_ssrc == ssrc);
+
+    gst_structure_free (stats);
+    g_object_unref (source);
+  }
+  session_harness_free (h);
+}
+
+GST_END_TEST;
 static GstBuffer *
 create_buffer (guint8 * data, gsize size)
 {
@@ -4318,6 +4384,7 @@ rtpsession_suite (void)
   tcase_add_test (tc_chain, test_receive_rtcp_app_packet);
   tcase_add_test (tc_chain, test_dont_lock_on_stats);
   tcase_add_test (tc_chain, test_ignore_suspicious_bye);
+  tcase_add_test (tc_chain, test_rr_stats_assignment);
 
   tcase_add_test (tc_chain, test_ssrc_collision_when_sending);
   tcase_add_test (tc_chain, test_ssrc_collision_when_sending_loopback);