tsparse: Handle backward and discont timestamps better.

Assume that small backward PCR jumps are just from upstream packet
mis-ordering and don't reset timestamp tracking state - assuming that
things will be OK again shortly.

Make the threshold for detecting discont between sequential buffers
configurable and match the smoothing-latency setting on tsparse
to better cope with data bursts.

1 files changed, 40 insertions, 16 deletions

diff --git a/gst/mpegtsdemux/mpegtspacketizer.c b/gst/mpegtsdemux/mpegtspacketizer.c
index 55eae074d..ca62bcaf7 100644
--- a/gst/mpegtsdemux/mpegtspacketizer.c
+++ b/gst/mpegtsdemux/mpegtspacketizer.c
@@ -50,8 +50,8 @@ G_DEFINE_TYPE_EXTENDED (MpegTSPacketizer2, mpegts_packetizer, G_TYPE_OBJECT, 0,
 
 static void mpegts_packetizer_dispose (GObject * object);
 static void mpegts_packetizer_finalize (GObject * object);
-static GstClockTime calculate_skew (MpegTSPCR * pcr, guint64 pcrtime,
-    GstClockTime time);
+static GstClockTime calculate_skew (MpegTSPacketizer2 * packetizer,
+    MpegTSPCR * pcr, guint64 pcrtime, GstClockTime time);
 static void _close_current_group (MpegTSPCR * pcrtable);
 static void record_pcr (MpegTSPacketizer2 * packetizer, MpegTSPCR * pcrtable,
     guint64 pcr, guint64 offset);
@@ -272,6 +272,7 @@ mpegts_packetizer_init (MpegTSPacketizer2 * packetizer)
   packetizer->nb_seen_offsets = 0;
   packetizer->refoffset = -1;
   packetizer->last_in_time = GST_CLOCK_TIME_NONE;
+  packetizer->pcr_discont_threshold = GST_SECOND;
 }
 
 static void
@@ -397,7 +398,8 @@ mpegts_packetizer_parse_adaptation_field_control (MpegTSPacketizer2 *
     if (packetizer->calculate_skew
         && GST_CLOCK_TIME_IS_VALID (packetizer->last_in_time)) {
       pcrtable = get_pcr_table (packetizer, packet->pid);
-      calculate_skew (pcrtable, packet->pcr, packetizer->last_in_time);
+      calculate_skew (packetizer, pcrtable, packet->pcr,
+          packetizer->last_in_time);
     }
     if (packetizer->calculate_offset) {
       if (!pcrtable)
@@ -1283,7 +1285,8 @@ mpegts_packetizer_resync (MpegTSPCR * pcr, GstClockTime time,
  * Returns: @time adjusted with the clock skew.
  */
 static GstClockTime
-calculate_skew (MpegTSPCR * pcr, guint64 pcrtime, GstClockTime time)
+calculate_skew (MpegTSPacketizer2 * packetizer,
+    MpegTSPCR * pcr, guint64 pcrtime, GstClockTime time)
 {
   guint64 send_diff, recv_diff;
   gint64 delta;
@@ -1321,7 +1324,8 @@ calculate_skew (MpegTSPCR * pcr, guint64 pcrtime, GstClockTime time)
       send_diff = gstpcrtime - pcr->base_pcrtime;
     } else if (GST_CLOCK_TIME_IS_VALID (time)
         && pcr->last_pcrtime - gstpcrtime > 15 * GST_SECOND) {
-      /* Assume a reset */
+      /* Time jumped backward by > 15 seconds, and we have a timestamp
+       * to use to close the discont. Assume a reset */
       GST_DEBUG ("PCR reset");
       /* Calculate PCR we would have expected for the given input time,
        * essentially applying the reverse correction process
@@ -1348,10 +1352,22 @@ calculate_skew (MpegTSPCR * pcr, guint64 pcrtime, GstClockTime time)
           " corrected pcr time %" GST_TIME_FORMAT,
           GST_TIME_ARGS (pcr->pcroffset), GST_TIME_ARGS (gstpcrtime));
     } else {
-      GST_WARNING ("backward timestamps at server but no timestamps");
+      /* Small jumps backward, assume some arrival jitter and skip it */
       send_diff = 0;
-      /* at least try to get a new timestamp.. */
-      pcr->base_time = GST_CLOCK_TIME_NONE;
+
+      if (pcr->last_pcrtime - gstpcrtime < GST_SECOND) {
+        GST_WARNING
+            ("(small) backward timestamps at server or no buffer timestamps. Ignoring.");
+        /* This will trigger the no_skew logic before but leave other state
+         * intact */
+        time = GST_CLOCK_TIME_NONE;
+      } else {
+        /* A bigger backward step than packet out-of-order can account for. Reset base PCR time
+         * to be resynched the next time we see a PCR */
+        GST_WARNING
+            ("backward timestamps at server or no buffer timestamps. Resync base PCR");
+        pcr->base_pcrtime = GST_CLOCK_TIME_NONE;
+      }
     }
   } else
     send_diff = gstpcrtime - pcr->base_pcrtime;
@@ -1396,7 +1412,7 @@ calculate_skew (MpegTSPCR * pcr, guint64 pcrtime, GstClockTime time)
   /* if the difference between the sender timeline and the receiver timeline
    * changed too quickly we have to resync because the server likely restarted
    * its timestamps. */
-  if (ABS (delta - pcr->skew) > GST_SECOND) {
+  if (ABS (delta - pcr->skew) > packetizer->pcr_discont_threshold) {
     GST_WARNING ("delta - skew: %" GST_TIME_FORMAT " too big, reset skew",
         GST_TIME_ARGS (delta - pcr->skew));
     mpegts_packetizer_resync (pcr, time, gstpcrtime, TRUE);
@@ -1595,8 +1611,8 @@ _reevaluate_group_pcr_offset (MpegTSPCR * pcrtable, PCROffsetGroup * group)
         GST_DEBUG ("Previous group bitrate (%" G_GUINT64_FORMAT " / %"
             GST_TIME_FORMAT ") : %" G_GUINT64_FORMAT,
             current->pending[current->last].offset,
-            GST_TIME_ARGS (PCRTIME_TO_GSTTIME (current->pending[current->
-                        last].pcr)), prevbr);
+            GST_TIME_ARGS (PCRTIME_TO_GSTTIME (current->pending[current->last].
+                    pcr)), prevbr);
       } else if (prev->values[prev->last_value].offset) {
         prevoffset = prev->values[prev->last_value].offset + prev->first_offset;
         prevpcr = prev->values[prev->last_value].pcr + prev->first_pcr;
@@ -1608,8 +1624,8 @@ _reevaluate_group_pcr_offset (MpegTSPCR * pcrtable, PCROffsetGroup * group)
         GST_DEBUG ("Previous group bitrate (%" G_GUINT64_FORMAT " / %"
             GST_TIME_FORMAT ") : %" G_GUINT64_FORMAT,
             prev->values[prev->last_value].offset,
-            GST_TIME_ARGS (PCRTIME_TO_GSTTIME (prev->values[prev->
-                        last_value].pcr)), prevbr);
+            GST_TIME_ARGS (PCRTIME_TO_GSTTIME (prev->values[prev->last_value].
+                    pcr)), prevbr);
       } else {
         GST_DEBUG ("Using overall bitrate");
         prevoffset = prev->values[prev->last_value].offset + prev->first_offset;
@@ -1918,9 +1934,8 @@ record_pcr (MpegTSPacketizer2 * packetizer, MpegTSPCR * pcrtable,
           group->first_offset,
           GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->pcr_offset)));
       GST_DEBUG ("Last PCR: +%" GST_TIME_FORMAT " offset: +%" G_GUINT64_FORMAT,
-          GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->values[group->
-                      last_value].pcr)),
-          group->values[group->last_value].offset);
+          GST_TIME_ARGS (PCRTIME_TO_GSTTIME (group->values[group->last_value].
+                  pcr)), group->values[group->last_value].offset);
       /* Check if before group */
       if (offset < group->first_offset) {
         GST_DEBUG ("offset is before that group");
@@ -2417,6 +2432,15 @@ mpegts_packetizer_set_reference_offset (MpegTSPacketizer2 * packetizer,
 }
 
 void
+mpegts_packetizer_set_pcr_discont_threshold (MpegTSPacketizer2 * packetizer,
+    GstClockTime threshold)
+{
+  PACKETIZER_GROUP_LOCK (packetizer);
+  packetizer->pcr_discont_threshold = threshold;
+  PACKETIZER_GROUP_UNLOCK (packetizer);
+}
+
+void
 mpegts_packetizer_set_current_pcr_offset (MpegTSPacketizer2 * packetizer,
     GstClockTime offset, guint16 pcr_pid)
 {