rtptwcc: move TWCC-logic over to the TWCC-manager

Prevent cluttering up the rtpsession, and keeping things localized.

Also write TWCC-seqnums for *all* streams in the session if configured by
caps.

A while back WebRTC was not doing TWCC for audio, basically breaking the
whole idea of a "transport-wide seqnuencenumber" applying for all bundled
streams. However, they have since fixed this, and now it no longers
makes sense to be able to single out certain payloadtypes for
use with TWCC, rather just including them all.

This also makes using RTX, RED, FEC etc much simpler, as it will apply
to them all as they enter the rtpsession.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-good/-/merge_requests/927>

3 files changed, 140 insertions, 91 deletions

diff --git a/gst/rtpmanager/rtpsession.c b/gst/rtpmanager/rtpsession.c
index b8429a1a3..9bd537c93 100644
--- a/gst/rtpmanager/rtpsession.c
+++ b/gst/rtpmanager/rtpsession.c
@@ -117,9 +117,6 @@ enum
   else                                  \
    (avg) = ((val) + (15 * (avg))) >> 4;
 
-
-#define TWCC_EXTMAP_STR "http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01"
-
 /* GObject vmethods */
 static void rtp_session_finalize (GObject * object);
 static void rtp_session_set_property (GObject * object, guint prop_id,
@@ -2196,22 +2193,6 @@ clean_packet_info (RTPPacketInfo * pinfo)
     g_bytes_unref (pinfo->header_ext);
 }
 
-static gint32
-packet_info_get_twcc_seqnum (RTPPacketInfo * pinfo, guint8 ext_id)
-{
-  gint32 val = -1;
-  gpointer data;
-  guint size;
-
-  if (pinfo->header_ext &&
-      gst_rtp_buffer_get_extension_onebyte_header_from_bytes (pinfo->header_ext,
-          pinfo->header_ext_bit_pattern, ext_id, 0, &data, &size)) {
-    if (size == 2)
-      val = GST_READ_UINT16_BE (data);
-  }
-  return val;
-}
-
 static gboolean
 source_update_active (RTPSession * sess, RTPSource * source,
     gboolean prevactive)
@@ -2237,16 +2218,7 @@ source_update_active (RTPSession * sess, RTPSource * source,
 static void
 process_twcc_packet (RTPSession * sess, RTPPacketInfo * pinfo)
 {
-  gint32 twcc_seqnum;
-
-  if (sess->twcc_recv_ext_id == 0)
-    return;
-
-  twcc_seqnum = packet_info_get_twcc_seqnum (pinfo, sess->twcc_recv_ext_id);
-  if (twcc_seqnum == -1)
-    return;
-
-  if (rtp_twcc_manager_recv_packet (sess->twcc, twcc_seqnum, pinfo)) {
+  if (rtp_twcc_manager_recv_packet (sess->twcc, pinfo)) {
     RTP_SESSION_UNLOCK (sess);
 
     /* TODO: find a better rational for this number, and possibly tune it based
@@ -3151,29 +3123,6 @@ invalid_packet:
   }
 }
 
-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;
-}
-
 /**
  * rtp_session_update_send_caps:
  * @sess: an #RTPSession
@@ -3229,29 +3178,9 @@ rtp_session_update_send_caps (RTPSession * sess, GstCaps * caps)
     sess->internal_ssrc_from_caps_or_property = FALSE;
   }
 
-  sess->twcc_send_ext_id = _get_extmap_id_for_attribute (s, TWCC_EXTMAP_STR);
-  if (sess->twcc_send_ext_id > 0) {
-    GST_INFO ("TWCC enabled for send using extension id: %u",
-        sess->twcc_send_ext_id);
-  }
-}
-
-static void
-send_twcc_packet (RTPSession * sess, RTPPacketInfo * pinfo)
-{
-  gint32 twcc_seqnum;
-
-  if (sess->twcc_send_ext_id == 0)
-    return;
-
-  twcc_seqnum = packet_info_get_twcc_seqnum (pinfo, sess->twcc_send_ext_id);
-  if (twcc_seqnum == -1)
-    return;
-
-  rtp_twcc_manager_send_packet (sess->twcc, twcc_seqnum, pinfo);
+  rtp_twcc_manager_parse_send_ext_id (sess->twcc, s);
 }
 
-
 /**
  * rtp_session_send_rtp:
  * @sess: an #RTPSession
@@ -3286,7 +3215,7 @@ rtp_session_send_rtp (RTPSession * sess, gpointer data, gboolean is_list,
           current_time, running_time, -1))
     goto invalid_packet;
 
-  send_twcc_packet (sess, &pinfo);
+  rtp_twcc_manager_send_packet (sess->twcc, &pinfo);
 
   source = obtain_internal_source (sess, pinfo.ssrc, &created, current_time);
   if (created)
@@ -4947,10 +4876,5 @@ void
 rtp_session_update_recv_caps_structure (RTPSession * sess,
     const GstStructure * s)
 {
-  guint8 ext_id = _get_extmap_id_for_attribute (s, TWCC_EXTMAP_STR);
-  if (ext_id > 0) {
-    sess->twcc_recv_ext_id = ext_id;
-    GST_INFO ("TWCC enabled for recv using extension id: %u",
-        sess->twcc_recv_ext_id);
-  }
+  rtp_twcc_manager_parse_recv_ext_id (sess->twcc, s);
 }
diff --git a/gst/rtpmanager/rtptwcc.c b/gst/rtpmanager/rtptwcc.c
index f529f377c..b5fa79b5d 100644
--- a/gst/rtpmanager/rtptwcc.c
+++ b/gst/rtpmanager/rtptwcc.c
@@ -25,6 +25,8 @@
 GST_DEBUG_CATEGORY_EXTERN (rtp_session_debug);
 #define GST_CAT_DEFAULT rtp_session_debug
 
+#define TWCC_EXTMAP_STR "http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01"
+
 #define REF_TIME_UNIT (64 * GST_MSECOND)
 #define DELTA_UNIT (250 * GST_USECOND)
 #define MAX_TS_DELTA (0xff * DELTA_UNIT)
@@ -71,6 +73,10 @@ struct _RTPTWCCManager
 {
   GObject object;
 
+  guint8 send_ext_id;
+  guint8 recv_ext_id;
+  guint16 send_seqnum;
+
   guint mtu;
   guint max_packets_per_rtcp;
   GArray *recv_packets;
@@ -155,6 +161,51 @@ recv_packet_init (RecvPacket * packet, guint16 seqnum, RTPPacketInfo * pinfo)
   packet->ts = pinfo->running_time;
 }
 
+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;
+}
+
+void
+rtp_twcc_manager_parse_recv_ext_id (RTPTWCCManager * twcc,
+    const GstStructure * s)
+{
+  twcc->recv_ext_id = _get_extmap_id_for_attribute (s, TWCC_EXTMAP_STR);
+  if (twcc->recv_ext_id > 0) {
+    GST_INFO ("TWCC enabled for recv using extension id: %u",
+        twcc->recv_ext_id);
+  }
+}
+
+void
+rtp_twcc_manager_parse_send_ext_id (RTPTWCCManager * twcc,
+    const GstStructure * s)
+{
+  twcc->send_ext_id = _get_extmap_id_for_attribute (s, TWCC_EXTMAP_STR);
+  if (twcc->send_ext_id > 0) {
+    GST_INFO ("TWCC enabled for send using extension id: %u",
+        twcc->send_ext_id);
+  }
+}
+
 void
 rtp_twcc_manager_set_mtu (RTPTWCCManager * twcc, guint mtu)
 {
@@ -180,6 +231,63 @@ rtp_twcc_manager_get_feedback_interval (RTPTWCCManager * twcc)
   return twcc->feedback_interval;
 }
 
+static gboolean
+_get_twcc_seqnum_data (RTPPacketInfo * pinfo, guint8 ext_id, gpointer * data)
+{
+  gboolean ret = FALSE;
+  guint size;
+
+  if (pinfo->header_ext &&
+      gst_rtp_buffer_get_extension_onebyte_header_from_bytes (pinfo->header_ext,
+          pinfo->header_ext_bit_pattern, ext_id, 0, data, &size)) {
+    if (size == 2)
+      ret = TRUE;
+  }
+  return ret;
+}
+
+static void
+_set_twcc_seqnum_data (GstBuffer * buf, guint8 ext_id, guint16 seqnum)
+{
+  GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
+  gpointer data;
+
+  if (gst_rtp_buffer_map (buf, GST_MAP_READWRITE, &rtp)) {
+    if (gst_rtp_buffer_get_extension_onebyte_header (&rtp,
+            ext_id, 0, &data, NULL)) {
+      GST_WRITE_UINT16_BE (data, seqnum);
+    }
+    gst_rtp_buffer_unmap (&rtp);
+  }
+}
+
+static guint16
+rtp_twcc_manager_set_send_twcc_seqnum (RTPTWCCManager * twcc,
+    RTPPacketInfo * pinfo)
+{
+  guint16 seqnum = twcc->send_seqnum++;
+  pinfo->data = gst_buffer_make_writable (pinfo->data);
+  _set_twcc_seqnum_data (pinfo->data, twcc->send_ext_id, seqnum);
+  return seqnum;
+}
+
+static gint32
+rtp_twcc_manager_get_recv_twcc_seqnum (RTPTWCCManager * twcc,
+    RTPPacketInfo * pinfo)
+{
+  gint32 val = -1;
+  gpointer data;
+
+  if (twcc->recv_ext_id == 0)
+    return val;
+
+  if (_get_twcc_seqnum_data (pinfo, twcc->recv_ext_id, &data)) {
+    val = GST_READ_UINT16_BE (data);
+  }
+
+  return val;
+}
+
 static gint
 _twcc_seqnum_sort (gconstpointer a, gconstpointer b)
 {
@@ -621,13 +729,17 @@ _many_packets_some_lost (RTPTWCCManager * twcc, guint16 seqnum)
 }
 
 gboolean
-rtp_twcc_manager_recv_packet (RTPTWCCManager * twcc,
-    guint16 seqnum, RTPPacketInfo * pinfo)
+rtp_twcc_manager_recv_packet (RTPTWCCManager * twcc, RTPPacketInfo * pinfo)
 {
   gboolean send_feedback = FALSE;
   RecvPacket packet;
+  gint32 seqnum;
   gint diff;
 
+  seqnum = rtp_twcc_manager_get_recv_twcc_seqnum (twcc, pinfo);
+  if (seqnum == -1)
+    return FALSE;
+
   /* if this packet would exceed the capacity of our MTU, we create a feedback
      with the current packets, and start over with this one */
   if (_exceeds_max_packets (twcc, seqnum)) {
@@ -666,8 +778,10 @@ rtp_twcc_manager_recv_packet (RTPTWCCManager * twcc,
   recv_packet_init (&packet, seqnum, pinfo);
   g_array_append_val (twcc->recv_packets, packet);
   twcc->last_seqnum = seqnum;
-  GST_LOG ("Receive: twcc-seqnum: %u, marker: %d, ts: %" GST_TIME_FORMAT,
-      seqnum, pinfo->marker, GST_TIME_ARGS (pinfo->running_time));
+
+  GST_LOG ("Receive: twcc-seqnum: %u, pt: %u, marker: %d, ts: %"
+      GST_TIME_FORMAT, seqnum, pinfo->pt, pinfo->marker,
+      GST_TIME_ARGS (pinfo->running_time));
 
   if (!pinfo->marker)
     twcc->packet_count_no_marker++;
@@ -732,15 +846,22 @@ sent_packet_init (SentPacket * packet, guint16 seqnum, RTPPacketInfo * pinfo)
 }
 
 void
-rtp_twcc_manager_send_packet (RTPTWCCManager * twcc,
-    guint16 seqnum, RTPPacketInfo * pinfo)
+rtp_twcc_manager_send_packet (RTPTWCCManager * twcc, RTPPacketInfo * pinfo)
 {
   SentPacket packet;
+  guint16 seqnum;
+
+  if (twcc->send_ext_id == 0)
+    return;
+
+  seqnum = rtp_twcc_manager_set_send_twcc_seqnum (twcc, pinfo);
+
   sent_packet_init (&packet, seqnum, pinfo);
   g_array_append_val (twcc->sent_packets, packet);
 
-  GST_LOG ("Send: twcc-seqnum: %u, marker: %d, ts: %" GST_TIME_FORMAT,
-      seqnum, pinfo->marker, GST_TIME_ARGS (pinfo->running_time));
+
+  GST_LOG ("Send: twcc-seqnum: %u, pt: %u, marker: %d, ts: %" GST_TIME_FORMAT,
+      seqnum, pinfo->pt, pinfo->marker, GST_TIME_ARGS (pinfo->running_time));
 }
 
 static void
diff --git a/gst/rtpmanager/rtptwcc.h b/gst/rtpmanager/rtptwcc.h
index bb14dbad2..79b56877a 100644
--- a/gst/rtpmanager/rtptwcc.h
+++ b/gst/rtpmanager/rtptwcc.h
@@ -53,16 +53,20 @@ struct _RTPTWCCPacket
 
 RTPTWCCManager * rtp_twcc_manager_new (guint mtu);
 
+void rtp_twcc_manager_parse_recv_ext_id (RTPTWCCManager * twcc,
+    const GstStructure * s);
+void rtp_twcc_manager_parse_send_ext_id (RTPTWCCManager * twcc,
+    const GstStructure * s);
+
 void rtp_twcc_manager_set_mtu (RTPTWCCManager * twcc, guint mtu);
 void rtp_twcc_manager_set_feedback_interval (RTPTWCCManager * twcc,
     GstClockTime feedback_interval);
 GstClockTime rtp_twcc_manager_get_feedback_interval (RTPTWCCManager * twcc);
 
 gboolean rtp_twcc_manager_recv_packet (RTPTWCCManager * twcc,
-    guint16 seqnum, RTPPacketInfo * pinfo);
-
+    RTPPacketInfo * pinfo);
 void rtp_twcc_manager_send_packet (RTPTWCCManager * twcc,
-    guint16 seqnum, RTPPacketInfo * pinfo);
+    RTPPacketInfo * pinfo);
 
 GstBuffer * rtp_twcc_manager_get_feedback (RTPTWCCManager * twcc,
     guint32 sender_ssrc);