tests/check: add an rtpsession unit test to verify all RBs are included in all SRs, roundrobin

This test checks that when we have multiple internal sender sources
in rtpsession, SRs contain RBs for every other sender source, and that
they are included roundrobin when they exceed ST_RTCP_MAX_RB_COUNT,
which is the max number of RBs that can fit in a SR.

tests/check/elements/rtpsession.c

@@ -3,6 +3,7 @@
  * unit test for gstrtpsession
  *
  * Copyright (C) <2009> Wim Taymans <wim.taymans@gmail.com>
+ * Copyright (C) 2013 Collabora Ltd.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Library General Public
@@ -115,7 +116,7 @@ destroy_testharness (TestData * data)
 }
 
 static void
-setup_testharness (TestData * data)
+setup_testharness (TestData * data, gboolean session_as_sender)
 {
   GstPad *rtp_sink_pad, *rtcp_src_pad, *rtp_src_pad;
   GstSegment seg;
@@ -141,14 +142,16 @@ setup_testharness (TestData * data)
   /* link in the test source-pad */
   data->src = gst_pad_new ("src", GST_PAD_SRC);
   g_assert (data->src);
-  rtp_sink_pad = gst_element_get_request_pad (data->session, "recv_rtp_sink");
+  rtp_sink_pad = gst_element_get_request_pad (data->session,
+      session_as_sender ? "send_rtp_sink" : "recv_rtp_sink");
   g_assert (rtp_sink_pad);
   g_assert_cmpint (gst_pad_link (data->src, rtp_sink_pad), ==, GST_PAD_LINK_OK);
   gst_object_unref (rtp_sink_pad);
 
   data->rtpsrc = gst_pad_new ("sink", GST_PAD_SINK);
   g_assert (data->rtpsrc);
-  rtp_src_pad = gst_element_get_static_pad (data->session, "recv_rtp_src");
+  rtp_src_pad = gst_element_get_static_pad (data->session,
+      session_as_sender ? "send_rtp_src" : "recv_rtp_src");
   g_assert (rtp_src_pad);
   g_assert_cmpint (gst_pad_link (rtp_src_pad, data->rtpsrc), ==,
       GST_PAD_LINK_OK);
@@ -191,7 +194,7 @@ GST_START_TEST (test_multiple_ssrc_rr)
   gint32 packetslost;
   guint8 fractionlost;
 
-  setup_testharness (&data);
+  setup_testharness (&data, FALSE);
 
   gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), 10 * GST_MSECOND);
 
@@ -247,6 +250,144 @@ GST_START_TEST (test_multiple_ssrc_rr)
 
 GST_END_TEST;
 
+/* This verifies that rtpsession will correctly place RBs round-robin
+ * across multiple SRs when there are too many senders that their RBs
+ * do not fit in one SR */
+GST_START_TEST (test_multiple_senders_roundrobin_rbs)
+{
+  TestData data;
+  GstFlowReturn res;
+  GstClockID id;
+  GstBuffer *buf;
+  GstRTCPBuffer rtcp = GST_RTCP_BUFFER_INIT;
+  GstRTCPPacket rtcp_packet;
+  GstClockTime time;
+  gint queue_length;
+  gint i, j, k;
+  guint32 ssrc;
+  GHashTable *sr_ssrcs, *rb_ssrcs, *tmp_set;
+
+  setup_testharness (&data, TRUE);
+
+  /* only the RTCP thread waits on the clock */
+  gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock), &id);
+
+  for (i = 0; i < 2; i++) {     /* cycles between SR reports */
+    for (j = 0; j < 5; j++) {   /* packets per ssrc */
+      gint seq = (i * 5) + j;
+      GST_DEBUG ("Push %i", seq);
+
+      gst_test_clock_advance_time (GST_TEST_CLOCK (data.clock),
+          200 * GST_MSECOND);
+
+      for (k = 0; k < 35; k++) {        /* number of ssrcs */
+        buf =
+            generate_test_buffer (seq * 200 * GST_MSECOND, FALSE, seq,
+            seq * 200, 10000 + k);
+        res = gst_pad_push (data.src, buf);
+        fail_unless (res == GST_FLOW_OK || res == GST_FLOW_FLUSHING);
+      }
+
+      GST_DEBUG ("pushed %i", seq);
+    }
+
+    queue_length = g_async_queue_length (data.rtcp_queue);
+
+    do {
+      /* crank the RTCP pad thread */
+      time = gst_clock_id_get_time (id);
+      GST_DEBUG ("Advancing time to %" GST_TIME_FORMAT, GST_TIME_ARGS (time));
+      gst_test_clock_set_time (GST_TEST_CLOCK (data.clock), time);
+      fail_unless_equals_pointer (gst_test_clock_process_next_clock_id
+          (GST_TEST_CLOCK (data.clock)), id);
+
+      /* wait for the RTCP pad thread to output its data
+       * and start waiting on the next timeout */
+      gst_test_clock_wait_for_next_pending_id (GST_TEST_CLOCK (data.clock),
+          &id);
+
+      /* and retry as long as there are no new RTCP packets out,
+       * because the RTCP thread may randomly decide to reschedule
+       * the RTCP timeout for later */
+    } while (g_async_queue_length (data.rtcp_queue) == queue_length);
+
+    GST_DEBUG ("RTCP timeout processed");
+  }
+
+  sr_ssrcs = g_hash_table_new (g_direct_hash, g_direct_equal);
+  rb_ssrcs = g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL,
+      (GDestroyNotify) g_hash_table_unref);
+
+  /* verify the rtcp packets */
+  for (i = 0; i < 2 * 35; i++) {
+    guint expected_rb_count = (i < 35) ? GST_RTCP_MAX_RB_COUNT :
+        (35 - GST_RTCP_MAX_RB_COUNT - 1);
+
+    GST_DEBUG ("pop %d", i);
+
+    buf = g_async_queue_pop (data.rtcp_queue);
+    g_assert (buf != NULL);
+    g_assert (gst_rtcp_buffer_validate (buf));
+
+    gst_rtcp_buffer_map (buf, GST_MAP_READ, &rtcp);
+    g_assert (gst_rtcp_buffer_get_first_packet (&rtcp, &rtcp_packet));
+    g_assert_cmpint (gst_rtcp_packet_get_type (&rtcp_packet), ==,
+        GST_RTCP_TYPE_SR);
+
+    gst_rtcp_packet_sr_get_sender_info (&rtcp_packet, &ssrc, NULL, NULL, NULL,
+        NULL);
+    g_assert_cmpint (ssrc, >=, 10000);
+    g_assert_cmpint (ssrc, <=, 10035);
+    g_hash_table_add (sr_ssrcs, GUINT_TO_POINTER (ssrc));
+
+    /* inspect the RBs */
+    g_assert_cmpint (gst_rtcp_packet_get_rb_count (&rtcp_packet), ==,
+        expected_rb_count);
+
+    if (i < 35) {
+      tmp_set = g_hash_table_new (g_direct_hash, g_direct_equal);
+      g_hash_table_insert (rb_ssrcs, GUINT_TO_POINTER (ssrc), tmp_set);
+    } else {
+      tmp_set = g_hash_table_lookup (rb_ssrcs, GUINT_TO_POINTER (ssrc));
+      g_assert (tmp_set);
+    }
+
+    for (j = 0; j < expected_rb_count; j++) {
+      gst_rtcp_packet_get_rb (&rtcp_packet, j, &ssrc, NULL, NULL,
+          NULL, NULL, NULL, NULL);
+      g_assert_cmpint (ssrc, >=, 10000);
+      g_assert_cmpint (ssrc, <=, 10035);
+      g_hash_table_add (tmp_set, GUINT_TO_POINTER (ssrc));
+    }
+
+    gst_rtcp_buffer_unmap (&rtcp);
+    gst_buffer_unref (buf);
+
+    /* cycle done, verify all ssrcs have issued SR reports */
+    if ((i + 1) == 35 || (i + 1) == (2 * 35)) {
+      g_assert_cmpint (g_hash_table_size (sr_ssrcs), ==, 35);
+      g_hash_table_remove_all (sr_ssrcs);
+    }
+  }
+
+  /* now verify all other ssrcs have been reported on each ssrc's SR */
+  g_assert_cmpint (g_hash_table_size (rb_ssrcs), ==, 35);
+  for (i = 10000; i < 10035; i++) {
+    tmp_set = g_hash_table_lookup (rb_ssrcs, GUINT_TO_POINTER (i));
+    g_assert (tmp_set);
+    /* SR contains RBs for each other ssrc except the ssrc of the SR */
+    g_assert_cmpint (g_hash_table_size (tmp_set), ==, 34);
+    g_assert (!g_hash_table_contains (tmp_set, GUINT_TO_POINTER (i)));
+  }
+
+  g_hash_table_unref (sr_ssrcs);
+  g_hash_table_unref (rb_ssrcs);
+
+  destroy_testharness (&data);
+}
+
+GST_END_TEST;
+
 static Suite *
 gstrtpsession_suite (void)
 {
@@ -255,6 +396,7 @@ gstrtpsession_suite (void)
 
   suite_add_tcase (s, tc_chain);
   tcase_add_test (tc_chain, test_multiple_ssrc_rr);
+  tcase_add_test (tc_chain, test_multiple_senders_roundrobin_rbs);
 
   return s;
 }