descrambler/tsdemux: propagate descramble errors to other layers, fix…

…es #3165

src/descrambler.h

@@ -62,7 +62,9 @@ typedef struct th_descrambler {
 } th_descrambler_t;
 
 typedef struct th_descrambler_runtime {
+  struct service *dr_service;
   tvhcsa_t dr_csa;
+  uint32_t dr_skip:1;
   uint32_t dr_quick_ecm:1;
   uint32_t dr_key:1;
   uint32_t dr_key_first:1;

src/descrambler/descrambler.c

@@ -49,6 +49,9 @@ static void
 descrambler_data_destroy(th_descrambler_runtime_t *dr, th_descrambler_data_t *dd)
 {
   if (dd) {
+    if (dr->dr_skip)
+      ts_skip_packet2((mpegts_service_t *)dr->dr_service,
+                      dd->dd_sbuf.sb_data, dd->dd_sbuf.sb_ptr);
     dr->dr_queue_total -= dd->dd_sbuf.sb_ptr;
     TAILQ_REMOVE(&dr->dr_queue, dd, dd_link);
     sbuf_free(&dd->dd_sbuf);
@@ -100,6 +103,8 @@ descrambler_data_cut(th_descrambler_runtime_t *dr, int len)
 
   while (len > 0 && (dd = TAILQ_FIRST(&dr->dr_queue)) != NULL) {
     if (len < dd->dd_sbuf.sb_ptr) {
+      if (dr->dr_skip)
+        ts_skip_packet2((mpegts_service_t *)dr->dr_service, dd->dd_sbuf.sb_data, len);
       sbuf_cut(&dd->dd_sbuf, len);
       dr->dr_queue_total -= len;
       break;
@@ -231,9 +236,11 @@ descrambler_service_start ( service_t *t )
   ((mpegts_service_t *)t)->s_dvb_mux->mm_descrambler_flush = 0;
   if (t->s_descramble == NULL) {
     t->s_descramble = dr = calloc(1, sizeof(th_descrambler_runtime_t));
+    dr->dr_service = t;
     TAILQ_INIT(&dr->dr_queue);
     dr->dr_key_index = 0xff;
     dr->dr_key_interval = 10;
+    dr->dr_skip = 0;
     tvhcsa_init(&dr->dr_csa);
   }
   caclient_start(t);
@@ -680,6 +687,7 @@ descrambler_descramble ( service_t *t,
         key_update(dr, ki, dispatch_clock);
       }
     }
+    dr->dr_skip = 1;
     dr->dr_csa.csa_descramble(&dr->dr_csa, (mpegts_service_t *)t, tsb, len);
     service_reset_streaming_status_flags(t, TSS_NO_ACCESS);
     return 1;

src/input/mpegts/tsdemux.c

@@ -43,6 +43,7 @@
 #define TS_REMUX_BUFSIZE (188 * 100)
 
 static void ts_remux(mpegts_service_t *t, const uint8_t *tsb, int len, int errors);
+static void ts_skip(mpegts_service_t *t, const uint8_t *tsb, int len);
 
 /**
  * Continue processing of transport stream packets
@@ -115,6 +116,58 @@ ts_recv_packet0
   }
 }
 
+/**
+ * Continue processing of skipped packets
+ */
+static void
+ts_recv_skipped0
+  (mpegts_service_t *t, elementary_stream_t *st, const uint8_t *tsb, int len)
+{
+  mpegts_service_t *m;
+  int len2, cc;
+  const uint8_t *tsb2;
+
+  if (!st) {
+    if(streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
+      ts_skip(t, tsb, len);
+    return;
+  }
+
+  for (tsb2 = tsb, len2 = len; len2 > 0; tsb2 += 188, len2 -= 188) {
+
+    /* Check CC */
+
+    if(tsb2[3] & 0x10) {
+      cc = tsb2[3] & 0xf;
+      if(st->es_cc != -1 && cc != st->es_cc) {
+        /* Let the hardware to stabilize and don't flood the log */
+        if (t->s_start_time + 1 < dispatch_clock &&
+            tvhlog_limit(&st->es_cc_log, 10))
+          tvhwarn("TS", "%s Continuity counter error (total %zi)",
+                        service_component_nicename(st), st->es_cc_log.count);
+        avgstat_add(&t->s_cc_errors, 1, dispatch_clock);
+        avgstat_add(&st->es_cc_errors, 1, dispatch_clock);
+      }
+      st->es_cc = (cc + 1) & 0xf;
+    }
+
+  }
+
+  if(st->es_type != SCT_CA &&
+     streaming_pad_probe_type(&t->s_streaming_pad, SMT_PACKET))
+    skip_mpeg_ts((service_t*)t, st, tsb, len);
+
+  if(streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
+    ts_skip(t, tsb, len);
+
+  LIST_FOREACH(m, &t->s_masters, s_masters_link) {
+    pthread_mutex_lock(&m->s_stream_mutex);
+    if(streaming_pad_probe_type(&m->s_streaming_pad, SMT_MPEGTS))
+      ts_skip(m, tsb, len);
+    pthread_mutex_unlock(&m->s_stream_mutex);
+  }
+}
+
 /**
  * Process service stream packets, extract PCR and optionally descramble
  */
@@ -228,6 +281,23 @@ ts_recv_packet2(mpegts_service_t *t, const uint8_t *tsb, int len)
   }
 }
 
+/*
+ * Process transport stream packets, simple version
+ */
+void
+ts_skip_packet2(mpegts_service_t *t, const uint8_t *tsb, int len)
+{
+  elementary_stream_t *st;
+  int pid, len2;
+
+  for ( ; len > 0; tsb += len2, len -= len2 ) {
+    len2 = mpegts_word_count(tsb, len, 0xFF9FFFD0);
+    pid = (tsb[1] & 0x1f) << 8 | tsb[2];
+    if((st = service_stream_find((service_t*)t, pid)) != NULL)
+      ts_recv_skipped0(t, st, tsb, len2);
+  }
+}
+
 /*
  *
  */
@@ -263,20 +333,10 @@ ts_recv_raw(mpegts_service_t *t, const uint8_t *tsb, int len)
  *
  */
 static void
-ts_remux(mpegts_service_t *t, const uint8_t *src, int len, int errors)
+ts_flush(mpegts_service_t *t, sbuf_t *sb)
 {
   streaming_message_t sm;
   pktbuf_t *pb;
-  sbuf_t *sb = &t->s_tsbuf;
-
-  if (sb->sb_data == NULL)
-    sbuf_init_fixed(sb, TS_REMUX_BUFSIZE);
-
-  sbuf_append(sb, src, len);
-  sb->sb_err += errors;
-
-  if(dispatch_clock == t->s_tsbuf_last && sb->sb_ptr < TS_REMUX_BUFSIZE)
-    return;
 
   t->s_tsbuf_last = dispatch_clock;
 
@@ -295,6 +355,48 @@ ts_remux(mpegts_service_t *t, const uint8_t *src, int len, int errors)
   sbuf_reset(sb, 2*TS_REMUX_BUFSIZE);
 }
 
+/**
+ *
+ */
+static void
+ts_remux(mpegts_service_t *t, const uint8_t *src, int len, int errors)
+{
+  sbuf_t *sb = &t->s_tsbuf;
+
+  if (sb->sb_data == NULL)
+    sbuf_init_fixed(sb, TS_REMUX_BUFSIZE);
+
+  sbuf_append(sb, src, len);
+  sb->sb_err += errors;
+
+  if(dispatch_clock == t->s_tsbuf_last && sb->sb_ptr < TS_REMUX_BUFSIZE)
+    return;
+
+  ts_flush(t, sb);
+}
+
+/**
+ *
+ */
+static void
+ts_skip(mpegts_service_t *t, const uint8_t *src, int len)
+{
+  sbuf_t *sb = &t->s_tsbuf;
+
+  if (len < 188)
+    return;
+
+  if (sb->sb_data == NULL)
+    sbuf_init_fixed(sb, TS_REMUX_BUFSIZE);
+
+  sb->sb_err += len / 188;
+
+  if(dispatch_clock == t->s_tsbuf_last && sb->sb_err < (TS_REMUX_BUFSIZE / 188))
+    return;
+
+  ts_flush(t, sb);
+}
+
 /*
  * Attempt to re-sync a ts stream (3 valid sync's in a row)
  */

src/input/mpegts/tsdemux.h

@@ -26,6 +26,8 @@ int ts_recv_packet1
 
 void ts_recv_packet2(struct mpegts_service *t, const uint8_t *tsb, int len);
 
+void ts_skip_packet2(struct mpegts_service *t, const uint8_t *tsb, int len);
+
 void ts_recv_raw(struct mpegts_service *t, const uint8_t *tsb, int len);
 
 #endif /* TSDEMUX_H */

src/parsers.h

@@ -25,8 +25,8 @@ void parse_mpeg_ts(struct service *t, struct elementary_stream *st,
 		   const uint8_t *data, 
 		   int len, int start, int err);
 
-void parse_mpeg_ps(struct service *t, struct elementary_stream *st,
-		   uint8_t *data, int len);
+void skip_mpeg_ts(struct service *t, struct elementary_stream *st,
+                  const uint8_t *data, int len);
 
 void parser_enqueue_packet(struct service *t, struct elementary_stream *st,
 			   th_pkt_t *pkt);

src/parsers/parsers.c

@@ -170,6 +170,18 @@ parse_mpeg_ts(service_t *t, elementary_stream_t *st, const uint8_t *data,
   }
 }
 
+/**
+ * Skip raw mpeg data
+ */
+void
+skip_mpeg_ts(service_t *t, elementary_stream_t *st, const uint8_t *data, int len)
+{
+  if(len >= 188)
+    sbuf_err(&st->es_buf, len / 188);
+  if(st->es_buf.sb_err > 1000)
+    parser_deliver_error(t, st);
+}
+
 /**
  * Parse AAC LATM and ADTS
  */