From 9ba9837058c5877959008a9d3813a7d956b622e5 Mon Sep 17 00:00:00 2001 From: =?utf8?q?H=C3=A5vard=20Graff?= Date: Fri, 14 Feb 2020 10:08:05 +0000 Subject: [PATCH] rtpfunnel: various cleanups * Organize GstRtpFunnelPad and GstRtpFunnel separately * Use G_GNUC_UNUSED instead of (void) casts * Don't call an event "caps" * Use semicolons after GST_END_TEST (helps gst-indent) --- gst/rtpmanager/gstrtpfunnel.c | 201 +++++++++++++++++++++++++++++---- tests/check/elements/rtpfunnel.c | 235 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 410 insertions(+), 26 deletions(-) diff --git a/gst/rtpmanager/gstrtpfunnel.c b/gst/rtpmanager/gstrtpfunnel.c index 155875b..ec70cf6 100644 --- a/gst/rtpmanager/gstrtpfunnel.c +++ b/gst/rtpmanager/gstrtpfunnel.c @@ -20,15 +20,63 @@ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, * Boston, MA 02110-1301, USA. */ + + /** + * SECTION:element-rtpfunnel + * @title: rtpfunnel + * @see_also: rtpbasepaylaoder, rtpsession + * + * RTP funnel is basically like a normal funnel with a few added + * functionalities to support bundling. + * + * Bundle is the concept of sending multiple streams in a single RTP session. + * These can be both audio and video streams, and several of both. + * One of the advantages with bundling is that you can get away with fewer + * ports for sending and receiving media. Also the RTCP traffic gets more + * compact if you can report on multiple streams in a single sender/receiver + * report. + * + * One of the reasons for a specialized RTP funnel is that some messages + * coming upstream want to find their way back to the right stream, + * and a normal funnel can't know which of its sinkpads it should send + * these messages to. The RTP funnel achieves this by keeping track of the + * SSRC of each stream on its sinkpad, and then uses the fact that upstream + * events are tagged inside rtpbin with the appropriate SSRC, so that upon + * receiving such an event, the RTP funnel can do a simple lookup for the + * right pad to forward the event to. + * + * A good example here is the KeyUnit event. If several video encoders are + * being bundled together using the RTP funnel, and one of the decoders on + * the receiving side asks for a KeyUnit, typically a RTCP PLI message will + * be sent from the receiver to the sender, and this will be transformed into + * a GstForceKeyUnit event inside GstRTPSession, and sent upstream. The + * RTP funnel can than make sure that this event hits the right encoder based + * on the SSRC embedded in the event. + * + * Another feature of the RTP funnel is that it will mux together TWCC + * (Transport-Wide Congestion Control) sequence-numbers. The point being that + * it should increment "transport-wide", meaning potentially several + * bundled streams. Note that not *all* streams being bundled needs to be + * affected by this. As an example Google WebRTC will use bundle with audio + * and video, but will only use TWCC sequence-numbers for the video-stream(s). + * + */ + #ifdef HAVE_CONFIG_H #include "config.h" #endif +#include + #include "gstrtpfunnel.h" GST_DEBUG_CATEGORY_STATIC (gst_rtp_funnel_debug); #define GST_CAT_DEFAULT gst_rtp_funnel_debug +#define TWCC_EXTMAP_STR "http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01" + +/**************** GstRTPFunnelPad ****************/ + struct _GstRtpFunnelPadClass { GstPadClass class; @@ -38,30 +86,31 @@ struct _GstRtpFunnelPad { GstPad pad; guint32 ssrc; + gboolean has_twcc; }; -enum -{ - PROP_0, - PROP_COMMON_TS_OFFSET, -}; - -#define DEFAULT_COMMON_TS_OFFSET -1 - G_DEFINE_TYPE (GstRtpFunnelPad, gst_rtp_funnel_pad, GST_TYPE_PAD); static void -gst_rtp_funnel_pad_class_init (GstRtpFunnelPadClass * klass) +gst_rtp_funnel_pad_class_init (G_GNUC_UNUSED GstRtpFunnelPadClass * klass) { - (void) klass; } static void -gst_rtp_funnel_pad_init (GstRtpFunnelPad * pad) +gst_rtp_funnel_pad_init (G_GNUC_UNUSED GstRtpFunnelPad * pad) { - (void) pad; } +/**************** GstRTPFunnel ****************/ + +enum +{ + PROP_0, + PROP_COMMON_TS_OFFSET, +}; + +#define DEFAULT_COMMON_TS_OFFSET -1 + struct _GstRtpFunnelClass { GstElementClass class; @@ -78,6 +127,10 @@ struct _GstRtpFunnel /* The last pad data was chained on */ GstPad *current_pad; + guint8 twcc_ext_id; /* the negotiated twcc extmap id */ + guint16 twcc_seqnum; /* our internal twcc seqnum */ + guint twcc_pads; /* numer of sinkpads with negotiated twcc */ + /* properties */ gint common_ts_offset; }; @@ -102,7 +155,7 @@ static void gst_rtp_funnel_send_sticky (GstRtpFunnel * funnel, GstPad * pad) { GstEvent *stream_start; - GstEvent *caps; + GstEvent *caps_ev; if (!funnel->send_sticky_events) goto done; @@ -113,8 +166,8 @@ gst_rtp_funnel_send_sticky (GstRtpFunnel * funnel, GstPad * pad) goto done; } - caps = gst_event_new_caps (funnel->srccaps); - if (caps && !gst_pad_push_event (funnel->srcpad, caps)) { + caps_ev = gst_event_new_caps (funnel->srccaps); + if (caps_ev && !gst_pad_push_event (funnel->srcpad, caps_ev)) { GST_ERROR_OBJECT (funnel, "Could not push caps"); goto done; } @@ -146,6 +199,43 @@ done: return; } +static void +gst_rtp_funnel_set_twcc_seqnum (GstRtpFunnel * funnel, + GstPad * pad, GstBuffer ** buf) +{ + GstRtpFunnelPad *fpad = GST_RTP_FUNNEL_PAD_CAST (pad); + GstRTPBuffer rtp = GST_RTP_BUFFER_INIT; + + if (!funnel->twcc_ext_id || !fpad->has_twcc) + return; + + *buf = gst_buffer_make_writable (*buf); + + if (gst_rtp_buffer_map (*buf, GST_MAP_READWRITE, &rtp)) { + gpointer data; + + /* if there already is a twcc-seqnum inside the packet */ + if (gst_rtp_buffer_get_extension_onebyte_header (&rtp, funnel->twcc_ext_id, + 0, &data, NULL)) { + + /* with only one pad, we read the twcc-seqnum instead of writing it */ + if (funnel->twcc_pads == 1) { + funnel->twcc_seqnum = GST_READ_UINT16_BE (data); + } else { + GST_WRITE_UINT16_BE (data, funnel->twcc_seqnum); + } + } else { + guint16 seq_be; + GST_WRITE_UINT16_BE (&seq_be, funnel->twcc_seqnum); + gst_rtp_buffer_add_extension_onebyte_header (&rtp, funnel->twcc_ext_id, + &seq_be, 2); + } + } + gst_rtp_buffer_unmap (&rtp); + + funnel->twcc_seqnum++; +} + static GstFlowReturn gst_rtp_funnel_sink_chain_object (GstPad * pad, GstRtpFunnel * funnel, gboolean is_list, GstMiniObject * obj) @@ -159,11 +249,13 @@ gst_rtp_funnel_sink_chain_object (GstPad * pad, GstRtpFunnel * funnel, gst_rtp_funnel_send_sticky (funnel, pad); gst_rtp_funnel_forward_segment (funnel, pad); - if (is_list) + if (is_list) { res = gst_pad_push_list (funnel->srcpad, GST_BUFFER_LIST_CAST (obj)); - else - res = gst_pad_push (funnel->srcpad, GST_BUFFER_CAST (obj)); - + } else { + GstBuffer *buf = GST_BUFFER_CAST (obj); + gst_rtp_funnel_set_twcc_seqnum (funnel, pad, &buf); + res = gst_pad_push (funnel->srcpad, buf); + } GST_PAD_STREAM_UNLOCK (funnel->srcpad); return res; @@ -188,10 +280,55 @@ gst_rtp_funnel_sink_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer) GST_MINI_OBJECT_CAST (buffer)); } +static void +gst_rtp_funnel_set_twcc_ext_id (GstRtpFunnel * funnel, guint8 twcc_ext_id) +{ + gchar *name; + + if (funnel->twcc_ext_id == twcc_ext_id) + return; + + name = g_strdup_printf ("extmap-%u", twcc_ext_id); + gst_caps_set_simple (funnel->srccaps, name, G_TYPE_STRING, TWCC_EXTMAP_STR, + NULL); + g_free (name); + + /* make sure we update the sticky with the new caps */ + funnel->send_sticky_events = TRUE; + + GST_INFO_OBJECT (funnel, "Setting twcc-ext-id to %u", twcc_ext_id); + funnel->twcc_ext_id = twcc_ext_id; +} + +static guint8 +_get_extmap_id_for_attribute (const GstStructure * s, const gchar * ext_name) +{ + guint i; + guint8 extmap_id = 0; + guint n_fields = gst_structure_n_fields (s); + + for (i = 0; i < n_fields; i++) { + const gchar *field_name = gst_structure_nth_field_name (s, i); + if (g_str_has_prefix (field_name, "extmap-")) { + const gchar *str = gst_structure_get_string (s, field_name); + if (str && g_strcmp0 (str, ext_name) == 0) { + gint64 id = g_ascii_strtoll (field_name + 7, NULL, 10); + if (id > 0 && id < 15) { + extmap_id = id; + break; + } + } + } + } + return extmap_id; +} + static gboolean gst_rtp_funnel_sink_event (GstPad * pad, GstObject * parent, GstEvent * event) { GstRtpFunnel *funnel = GST_RTP_FUNNEL_CAST (parent); + GstRtpFunnelPad *fpad = GST_RTP_FUNNEL_PAD_CAST (pad); + gboolean forward = TRUE; gboolean ret = TRUE; @@ -207,6 +344,7 @@ gst_rtp_funnel_sink_event (GstPad * pad, GstObject * parent, GstEvent * event) GstCaps *caps; GstStructure *s; guint ssrc; + guint8 ext_id; gst_event_parse_caps (event, &caps); if (!gst_caps_can_intersect (funnel->srccaps, caps)) { @@ -217,13 +355,18 @@ gst_rtp_funnel_sink_event (GstPad * pad, GstObject * parent, GstEvent * event) s = gst_caps_get_structure (caps, 0); if (gst_structure_get_uint (s, "ssrc", &ssrc)) { - GstRtpFunnelPad *fpad = GST_RTP_FUNNEL_PAD_CAST (pad); fpad->ssrc = ssrc; GST_DEBUG_OBJECT (pad, "Got ssrc: %u", ssrc); GST_OBJECT_LOCK (funnel); g_hash_table_insert (funnel->ssrc_to_pad, GUINT_TO_POINTER (ssrc), pad); GST_OBJECT_UNLOCK (funnel); } + ext_id = _get_extmap_id_for_attribute (s, TWCC_EXTMAP_STR); + if (ext_id > 0) { + fpad->has_twcc = TRUE; + funnel->twcc_pads++; + gst_rtp_funnel_set_twcc_ext_id (funnel, ext_id); + } forward = FALSE; break; @@ -245,8 +388,7 @@ static gboolean gst_rtp_funnel_sink_query (GstPad * pad, GstObject * parent, GstQuery * query) { GstRtpFunnel *funnel = GST_RTP_FUNNEL_CAST (parent); - gboolean res = FALSE; - (void) funnel; + gboolean res = TRUE; switch (GST_QUERY_TYPE (query)) { case GST_QUERY_CAPS: @@ -271,7 +413,20 @@ gst_rtp_funnel_sink_query (GstPad * pad, GstObject * parent, GstQuery * query) GST_DEBUG_OBJECT (pad, "Answering caps-query with caps: %" GST_PTR_FORMAT, new_caps); gst_caps_unref (new_caps); - res = TRUE; + break; + } + case GST_QUERY_ACCEPT_CAPS: + { + GstCaps *caps; + gboolean result; + gst_query_parse_accept_caps (query, &caps); + result = gst_caps_is_subset (caps, funnel->srccaps); + if (!result) { + GST_ERROR_OBJECT (pad, + "caps: %" GST_PTR_FORMAT " were not compatible with: %" + GST_PTR_FORMAT, caps, funnel->srccaps); + } + gst_query_set_accept_caps_result (query, result); break; } default: diff --git a/tests/check/elements/rtpfunnel.c b/tests/check/elements/rtpfunnel.c index 5c690e2..5224dce 100644 --- a/tests/check/elements/rtpfunnel.c +++ b/tests/check/elements/rtpfunnel.c @@ -22,6 +22,7 @@ */ #include #include +#include GST_START_TEST (rtpfunnel_ssrc_demuxing) { @@ -81,7 +82,8 @@ GST_START_TEST (rtpfunnel_ssrc_demuxing) gst_harness_teardown (h1); } -GST_END_TEST +GST_END_TEST; + GST_START_TEST (rtpfunnel_ssrc_downstream_not_leaking_through) { GstHarness *h = gst_harness_new_with_padnames ("rtpfunnel", @@ -100,7 +102,8 @@ GST_START_TEST (rtpfunnel_ssrc_downstream_not_leaking_through) gst_harness_teardown (h); } -GST_END_TEST +GST_END_TEST; + GST_START_TEST (rtpfunnel_common_ts_offset) { GstHarness *h = gst_harness_new_with_padnames ("rtpfunnel", @@ -122,7 +125,8 @@ GST_START_TEST (rtpfunnel_common_ts_offset) gst_harness_teardown (h); } -GST_END_TEST +GST_END_TEST; + GST_START_TEST (rtpfunnel_stress) { GstHarness *h = gst_harness_new_with_padnames ("rtpfunnel", @@ -173,6 +177,226 @@ GST_START_TEST (rtpfunnel_stress) GST_END_TEST; +#define TWCC_EXTMAP_STR "http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01" + +GST_START_TEST (rtpfunnel_twcc_caps) +{ + GstHarness *h, *h0, *h1; + GstCaps *caps, *expected_caps; + + h = gst_harness_new_with_padnames ("rtpfunnel", NULL, "src"); + + /* request a sinkpad, set caps with twcc extmap */ + h0 = gst_harness_new_with_element (h->element, "sink_0", NULL); + gst_harness_set_src_caps_str (h0, "application/x-rtp, " + "ssrc=(uint)123, extmap-5=" TWCC_EXTMAP_STR ""); + + /* request a second sinkpad, and verify the extmap is + present in the caps when doing a caps-query downstream */ + h1 = gst_harness_new_with_element (h->element, "sink_1", NULL); + caps = gst_pad_query_caps (GST_PAD_PEER (h1->srcpad), NULL); + expected_caps = gst_caps_from_string ("application/x-rtp, " + "extmap-5=" TWCC_EXTMAP_STR ""); + fail_unless (gst_caps_is_equal (expected_caps, caps)); + gst_caps_unref (caps); + gst_caps_unref (expected_caps); + + /* now try and set a different extmap (4) on the other sinkpad, + and verify this does not work */ + gst_harness_set_src_caps_str (h1, "application/x-rtp, " + "ssrc=(uint)456, extmap-4=" TWCC_EXTMAP_STR ""); + caps = gst_pad_get_current_caps (GST_PAD_PEER (h1->srcpad)); + fail_unless (caps == NULL); + + /* ...but setting the right extmap (5) will work just fine */ + expected_caps = gst_caps_from_string ("application/x-rtp, " + "ssrc=(uint)456, extmap-5=" TWCC_EXTMAP_STR ""); + gst_harness_set_src_caps (h1, gst_caps_ref (expected_caps)); + caps = gst_pad_get_current_caps (GST_PAD_PEER (h1->srcpad)); + fail_unless (gst_caps_is_equal (expected_caps, caps)); + gst_caps_unref (caps); + gst_caps_unref (expected_caps); + + gst_harness_teardown (h); + gst_harness_teardown (h0); + gst_harness_teardown (h1); +} + +GST_END_TEST; + +static GstBuffer * +generate_test_buffer (guint seqnum, guint ssrc, guint8 twcc_ext_id) +{ + GstBuffer *buf; + GstRTPBuffer rtp = GST_RTP_BUFFER_INIT; + + buf = gst_rtp_buffer_new_allocate (0, 0, 0); + GST_BUFFER_PTS (buf) = seqnum * 20 * GST_MSECOND; + GST_BUFFER_DTS (buf) = GST_BUFFER_PTS (buf); + + gst_rtp_buffer_map (buf, GST_MAP_READWRITE, &rtp); + gst_rtp_buffer_set_payload_type (&rtp, 100); + gst_rtp_buffer_set_seq (&rtp, seqnum); + gst_rtp_buffer_set_timestamp (&rtp, seqnum * 160); + gst_rtp_buffer_set_ssrc (&rtp, ssrc); + + if (twcc_ext_id > 0) { + guint16 data; + GST_WRITE_UINT16_BE (&data, seqnum); + gst_rtp_buffer_add_extension_onebyte_header (&rtp, twcc_ext_id, + &data, sizeof (guint16)); + } + + gst_rtp_buffer_unmap (&rtp); + + return buf; +} + +static gint32 +get_twcc_seqnum (GstBuffer * buf, guint8 ext_id) +{ + GstRTPBuffer rtp = GST_RTP_BUFFER_INIT; + gint32 val = -1; + gpointer data; + + gst_rtp_buffer_map (buf, GST_MAP_READ, &rtp); + if (gst_rtp_buffer_get_extension_onebyte_header (&rtp, ext_id, + 0, &data, NULL)) { + val = GST_READ_UINT16_BE (data); + } + gst_rtp_buffer_unmap (&rtp); + + return val; +} + +static guint32 +get_ssrc (GstBuffer * buf) +{ + guint32 ssrc; + GstRTPBuffer rtp = GST_RTP_BUFFER_INIT; + gst_rtp_buffer_map (buf, GST_MAP_READ, &rtp); + ssrc = gst_rtp_buffer_get_ssrc (&rtp); + gst_rtp_buffer_unmap (&rtp); + return ssrc; +} + +GST_START_TEST (rtpfunnel_twcc_passthrough) +{ + GstHarness *h, *h0; + GstBuffer *buf; + guint16 offset = 65530; + guint packets = 40; + guint i; + + h = gst_harness_new_with_padnames ("rtpfunnel", NULL, "src"); + h0 = gst_harness_new_with_element (h->element, "sink_0", NULL); + gst_harness_set_src_caps_str (h0, "application/x-rtp, " + "ssrc=(uint)123, extmap-5=" TWCC_EXTMAP_STR ""); + + /* push some packets with twcc seqnum */ + for (i = 0; i < packets; i++) { + guint16 seqnum = i + offset; + fail_unless_equals_int (GST_FLOW_OK, + gst_harness_push (h0, generate_test_buffer (seqnum, 123, 5))); + } + + /* and verify the seqnums stays unchanged through the funnel */ + for (i = 0; i < packets; i++) { + guint16 seqnum = i + offset; + buf = gst_harness_pull (h); + fail_unless_equals_int (seqnum, get_twcc_seqnum (buf, 5)); + gst_buffer_unref (buf); + } + + gst_harness_teardown (h); + gst_harness_teardown (h0); +} + +GST_END_TEST; + +GST_START_TEST (rtpfunnel_twcc_mux) +{ + GstHarness *h, *h0, *h1; + GstBuffer *buf; + + h = gst_harness_new_with_padnames ("rtpfunnel", NULL, "src"); + h0 = gst_harness_new_with_element (h->element, "sink_0", NULL); + h1 = gst_harness_new_with_element (h->element, "sink_1", NULL); + gst_harness_set_src_caps_str (h0, "application/x-rtp, " + "ssrc=(uint)123, extmap-5=" TWCC_EXTMAP_STR ""); + gst_harness_set_src_caps_str (h1, "application/x-rtp, " + "ssrc=(uint)456, extmap-5=" TWCC_EXTMAP_STR ""); + + /* push buffers on both pads with different twcc-seqnums on them (500 and 60000) */ + fail_unless_equals_int (GST_FLOW_OK, + gst_harness_push (h0, generate_test_buffer (500, 123, 5))); + fail_unless_equals_int (GST_FLOW_OK, + gst_harness_push (h1, generate_test_buffer (60000, 321, 5))); + + /* verify they are muxed continuously (0 -> 1) */ + buf = gst_harness_pull (h); + fail_unless_equals_int (123, get_ssrc (buf)); + fail_unless_equals_int (0, get_twcc_seqnum (buf, 5)); + gst_buffer_unref (buf); + + buf = gst_harness_pull (h); + fail_unless_equals_int (321, get_ssrc (buf)); + fail_unless_equals_int (1, get_twcc_seqnum (buf, 5)); + gst_buffer_unref (buf); + + gst_harness_teardown (h); + gst_harness_teardown (h0); + gst_harness_teardown (h1); +} + +GST_END_TEST; + + +GST_START_TEST (rtpfunnel_twcc_passthrough_then_mux) +{ + GstHarness *h, *h0, *h1; + GstBuffer *buf; + guint offset0 = 500; + guint offset1 = 45678; + guint i; + + h = gst_harness_new_with_padnames ("rtpfunnel", NULL, "src"); + h0 = gst_harness_new_with_element (h->element, "sink_0", NULL); + gst_harness_set_src_caps_str (h0, "application/x-rtp, " + "ssrc=(uint)123, extmap-5=" TWCC_EXTMAP_STR ""); + + /* push one packet with twcc seqnum 100 on pad0 */ + fail_unless_equals_int (GST_FLOW_OK, + gst_harness_push (h0, generate_test_buffer (offset0, 123, 5))); + + /* add pad1 to the funnel, also with twcc */ + h1 = gst_harness_new_with_element (h->element, "sink_1", NULL); + gst_harness_set_src_caps_str (h1, "application/x-rtp, " + "ssrc=(uint)456, extmap-5=" TWCC_EXTMAP_STR ""); + + /* push one buffer on both pads, with pad1 starting at a different + offset */ + fail_unless_equals_int (GST_FLOW_OK, + gst_harness_push (h0, generate_test_buffer (offset0 + 1, 123, 5))); + fail_unless_equals_int (GST_FLOW_OK, + gst_harness_push (h1, generate_test_buffer (offset1, 321, 5))); + + /* and verify the seqnums are continuous for all 3 packets, using + the inital offset from pad0 */ + for (i = 0; i < 3; i++) { + guint16 seqnum = i + offset0; + buf = gst_harness_pull (h); + fail_unless_equals_int (seqnum, get_twcc_seqnum (buf, 5)); + gst_buffer_unref (buf); + } + + gst_harness_teardown (h); + gst_harness_teardown (h0); + gst_harness_teardown (h1); +} + +GST_END_TEST; + static Suite * rtpfunnel_suite (void) { @@ -187,6 +411,11 @@ rtpfunnel_suite (void) tcase_add_test (tc_chain, rtpfunnel_stress); + tcase_add_test (tc_chain, rtpfunnel_twcc_caps); + tcase_add_test (tc_chain, rtpfunnel_twcc_passthrough); + tcase_add_test (tc_chain, rtpfunnel_twcc_mux); + tcase_add_test (tc_chain, rtpfunnel_twcc_passthrough_then_mux); + return s; } -- 2.7.4