mpegts input: fix TS packet / PID management for SAT>IP server subscr…

…iptions, fixes #3450

- this also fixes the wrong PID management (slave service close)

src/input/mpegts.h

@@ -147,6 +147,7 @@ struct mpegts_table_mux_cb
 typedef struct mpegts_pid_sub
 {
   RB_ENTRY(mpegts_pid_sub) mps_link;
+  LIST_ENTRY(mpegts_pid_sub) mps_raw_link;
   LIST_ENTRY(mpegts_pid_sub) mps_svcraw_link;
 #define MPS_NONE    0x00
 #define MPS_ALL     0x01
@@ -189,6 +190,7 @@ typedef struct mpegts_pid
   int                      mp_type; // mask for all subscribers
   int8_t                   mp_cc;
   RB_HEAD(,mpegts_pid_sub) mp_subs; // subscribers to pid
+  LIST_HEAD(,mpegts_pid_sub) mp_raw_subs;
   LIST_HEAD(,mpegts_pid_sub) mp_svc_subs;
   RB_ENTRY(mpegts_pid)     mp_link;
 } mpegts_pid_t;
@@ -788,6 +790,8 @@ int         mpegts_input_class_network_set  ( void *o, const void *p );
 htsmsg_t   *mpegts_input_class_network_enum ( void *o, const char *lang );
 char       *mpegts_input_class_network_rend ( void *o, const char *lang );
 
+int mpegts_mps_weight(elementary_stream_t *st);
+
 int mpegts_mps_cmp( mpegts_pid_sub_t *a, mpegts_pid_sub_t *b );
 
 void mpegts_network_register_builder
@@ -1037,14 +1041,15 @@ mpegts_service_find_e2
 mpegts_service_t *
 mpegts_service_find_by_pid ( mpegts_mux_t *mm, int pid );
 
+void mpegts_service_update_slave_pids ( mpegts_service_t *t, int del );
+
 static inline mpegts_service_t *mpegts_service_find_by_uuid(const char *uuid)
   { return idnode_find(uuid, &mpegts_service_class, NULL); }
 
 void mpegts_service_unref ( service_t *s );
 
 void mpegts_service_delete ( service_t *s, int delconf );
 
-
 /*
  * MPEG-TS event handler
  */

src/input/mpegts/mpegts_input.c

@@ -478,7 +478,9 @@ mpegts_input_open_pid
       }
     } else if (!RB_INSERT_SORTED(&mp->mp_subs, mps, mps_link, mpegts_mps_cmp)) {
       mp->mp_type |= type;
-      if (type & (MPS_SERVICE|MPS_RAW))
+      if (type & MPS_RAW)
+        LIST_INSERT_HEAD(&mp->mp_raw_subs, mps, mps_raw_link);
+      if (type & MPS_SERVICE)
         LIST_INSERT_HEAD(&mp->mp_svc_subs, mps, mps_svcraw_link);
       mpegts_mux_nice_name(mm, buf, sizeof(buf));
       tvhdebug("mpegts", "%s - open PID %04X (%d) [%d/%p]",
@@ -534,7 +536,9 @@ mpegts_input_close_pid
       mpegts_mux_nice_name(mm, buf, sizeof(buf));
       tvhdebug("mpegts", "%s - close PID %04X (%d) [%d/%p]",
                buf, mp->mp_pid, mp->mp_pid, type, owner);
-      if (type & (MPS_SERVICE|MPS_RAW))
+      if (type & MPS_RAW)
+        LIST_REMOVE(mps, mps_raw_link);
+      if (type & MPS_SERVICE)
         LIST_REMOVE(mps, mps_svcraw_link);
       RB_REMOVE(&mp->mp_subs, mps, mps_link);
       free(mps);
@@ -561,7 +565,7 @@ mpegts_input_update_pids
   /* nothing - override */
 }
 
-static int mps_weight(elementary_stream_t *st)
+int mpegts_mps_weight(elementary_stream_t *st)
 {
    if (SCT_ISVIDEO(st->es_type))
      return MPS_WEIGHT_VIDEO + MIN(st->es_index, 49);
@@ -645,7 +649,6 @@ mpegts_input_open_service
   elementary_stream_t *st;
   mpegts_apids_t *pids;
   mpegts_apid_t *p;
-  mpegts_service_t *s2;
   int i;
 
   /* Add to list */
@@ -654,39 +657,22 @@ mpegts_input_open_service
     LIST_INSERT_HEAD(&mm->mm_transports, ((service_t*)s), s_active_link);
     s->s_dvb_active_input = mi;
   }
-
   /* Register PIDs */
   pthread_mutex_lock(&s->s_stream_mutex);
   if (s->s_type == STYPE_STD) {
 
-    pids = mpegts_pid_alloc();
-
     mpegts_input_open_pid(mi, mm, s->s_pmt_pid, MPS_SERVICE, MPS_WEIGHT_PMT, s);
     mpegts_input_open_pid(mi, mm, s->s_pcr_pid, MPS_SERVICE, MPS_WEIGHT_PCR, s);
-    mpegts_pid_add(pids, s->s_pmt_pid, MPS_WEIGHT_PMT);
-    mpegts_pid_add(pids, s->s_pcr_pid, MPS_WEIGHT_PCR);
     if (s->s_scrambled_pass)
       mpegts_input_open_pid(mi, mm, DVB_CAT_PID, MPS_SERVICE, MPS_WEIGHT_CAT, s);
     /* Open only filtered components here */
     TAILQ_FOREACH(st, &s->s_filt_components, es_filt_link)
       if (s->s_scrambled_pass || st->es_type != SCT_CA) {
         st->es_pid_opened = 1;
-        mpegts_input_open_pid(mi, mm, st->es_pid, MPS_SERVICE, mps_weight(st), s);
+        mpegts_input_open_pid(mi, mm, st->es_pid, MPS_SERVICE, mpegts_mps_weight(st), s);
       }
 
-    /* Ensure that filtered PIDs are not send in ts_recv_raw */
-    TAILQ_FOREACH(st, &s->s_filt_components, es_filt_link)
-      if ((s->s_scrambled_pass || st->es_type != SCT_CA) &&
-          st->es_pid >= 0 && st->es_pid < 8192)
-        mpegts_pid_add(pids, st->es_pid, mps_weight(st));
-
-    LIST_FOREACH(s2, &s->s_masters, s_masters_link) {
-      pthread_mutex_lock(&s2->s_stream_mutex);
-      mpegts_pid_add_group(s2->s_slaves_pids, pids);
-      pthread_mutex_unlock(&s2->s_stream_mutex);
-    }
-
-    mpegts_pid_destroy(&pids);
+    mpegts_service_update_slave_pids(s, 0);
 
   } else {
     if ((pids = s->s_pids) != NULL) {
@@ -731,8 +717,6 @@ mpegts_input_close_service ( mpegts_input_t *mi, mpegts_service_t *s )
 {
   mpegts_mux_t *mm = s->s_dvb_mux;
   elementary_stream_t *st;
-  mpegts_apids_t *pids;
-  mpegts_service_t *s2;
 
   /* Close PMT/CAT tables */
   if (s->s_type == STYPE_STD) {
@@ -749,36 +733,23 @@ mpegts_input_close_service ( mpegts_input_t *mi, mpegts_service_t *s )
     LIST_REMOVE(((service_t*)s), s_active_link);
     s->s_dvb_active_input = NULL;
   }
-
   /* Close PID */
   pthread_mutex_lock(&s->s_stream_mutex);
   if (s->s_type == STYPE_STD) {
 
-    pids = mpegts_pid_alloc();
-
     mpegts_input_close_pid(mi, mm, s->s_pmt_pid, MPS_SERVICE, MPS_WEIGHT_PMT, s);
     mpegts_input_close_pid(mi, mm, s->s_pcr_pid, MPS_SERVICE, MPS_WEIGHT_PCR, s);
-    mpegts_pid_del(pids, s->s_pmt_pid, MPS_WEIGHT_PMT);
-    mpegts_pid_del(pids, s->s_pcr_pid, MPS_WEIGHT_PCR);
     if (s->s_scrambled_pass)
       mpegts_input_close_pid(mi, mm, DVB_CAT_PID, MPS_SERVICE, MPS_WEIGHT_CAT, s);
     /* Close all opened PIDs (the component filter may be changed at runtime) */
     TAILQ_FOREACH(st, &s->s_components, es_link) {
       if (st->es_pid_opened) {
         st->es_pid_opened = 0;
-        mpegts_input_close_pid(mi, mm, st->es_pid, MPS_SERVICE, mps_weight(st), s);
+        mpegts_input_close_pid(mi, mm, st->es_pid, MPS_SERVICE, mpegts_mps_weight(st), s);
       }
-      if (st->es_pid >= 0 && st->es_pid < 8192)
-        mpegts_pid_del(pids, st->es_pid, mps_weight(st));
-    }
-
-    LIST_FOREACH(s2, &s->s_masters, s_masters_link) {
-      pthread_mutex_lock(&s2->s_stream_mutex);
-      mpegts_pid_del_group(s2->s_slaves_pids, pids);
-      pthread_mutex_unlock(&s2->s_stream_mutex);
     }
 
-    mpegts_pid_destroy(&pids);
+    mpegts_service_update_slave_pids(s, 1);
 
   } else {
     mpegts_input_close_pids(mi, mm, s, 1);
@@ -1316,7 +1287,7 @@ mpegts_input_process
 
       /* Stream raw PIDs */
       if (type & MPS_RAW) {
-        LIST_FOREACH(mps, &mp->mp_svc_subs, mps_svcraw_link)
+        LIST_FOREACH(mps, &mp->mp_raw_subs, mps_raw_link)
           ts_recv_raw((mpegts_service_t *)mps->mps_owner, tsb, llen);
       }
 

src/input/mpegts/mpegts_mux.c

@@ -809,8 +809,10 @@ mpegts_mux_stop ( mpegts_mux_t *mm, int force, int reason )
         tvhdebug("mpegts", "%s - close PID %04X (%d) [%d/%p]", buf,
                  mp->mp_pid, mp->mp_pid, mps->mps_type, mps->mps_owner);
         RB_REMOVE(&mp->mp_subs, mps, mps_link);
-        if (mps->mps_type & (MPS_SERVICE|MPS_RAW|MPS_ALL))
+        if (mps->mps_type & (MPS_SERVICE|MPS_ALL))
           LIST_REMOVE(mps, mps_svcraw_link);
+        if (mps->mps_type & MPS_RAW)
+          LIST_REMOVE(mps, mps_raw_link);
         free(mps);
       }
     }

src/input/mpegts/mpegts_service.c

@@ -922,23 +922,61 @@ mpegts_service_raw_update_pids(mpegts_service_t *t, mpegts_apids_t *pids)
   return 0;
 }
 
+void
+mpegts_service_update_slave_pids ( mpegts_service_t *s, int del )
+{
+  mpegts_service_t *s2;
+  mpegts_apids_t *pids;
+  elementary_stream_t *st;
+
+  lock_assert(&s->s_stream_mutex);
+
+  pids = mpegts_pid_alloc();
+
+  mpegts_pid_add(pids, s->s_pmt_pid, MPS_WEIGHT_PMT);
+  mpegts_pid_add(pids, s->s_pcr_pid, MPS_WEIGHT_PCR);
+
+  /* Ensure that filtered PIDs are not send in ts_recv_raw */
+  TAILQ_FOREACH(st, &s->s_filt_components, es_filt_link)
+    if ((s->s_scrambled_pass || st->es_type != SCT_CA) &&
+        st->es_pid >= 0 && st->es_pid < 8192)
+      mpegts_pid_add(pids, st->es_pid, mpegts_mps_weight(st));
+
+  LIST_FOREACH(s2, &s->s_masters, s_masters_link) {
+    pthread_mutex_lock(&s2->s_stream_mutex);
+    if (!del)
+      mpegts_pid_add_group(s2->s_slaves_pids, pids);
+    else
+      mpegts_pid_del_group(s2->s_slaves_pids, pids);
+    pthread_mutex_unlock(&s2->s_stream_mutex);
+  }
+
+  mpegts_pid_destroy(&pids);
+}
+
 static int
 mpegts_service_link ( mpegts_service_t *master, mpegts_service_t *slave )
 {
+  pthread_mutex_lock(&slave->s_stream_mutex);
   pthread_mutex_lock(&master->s_stream_mutex);
   LIST_INSERT_HEAD(&slave->s_masters, master, s_masters_link);
   LIST_INSERT_HEAD(&master->s_slaves, slave, s_slaves_link);
   pthread_mutex_unlock(&master->s_stream_mutex);
+  mpegts_service_update_slave_pids(slave, 0);
+  pthread_mutex_unlock(&slave->s_stream_mutex);
   return 0;
 }
 
 static int
 mpegts_service_unlink ( mpegts_service_t *master, mpegts_service_t *slave )
 {
+  pthread_mutex_lock(&slave->s_stream_mutex);
+  mpegts_service_update_slave_pids(slave, 1);
   pthread_mutex_lock(&master->s_stream_mutex);
   LIST_SAFE_REMOVE(master, s_masters_link);
   LIST_SAFE_REMOVE(slave, s_slaves_link);
   pthread_mutex_unlock(&master->s_stream_mutex);
+  pthread_mutex_unlock(&slave->s_stream_mutex);
   return 0;
 }
 

src/input/mpegts/tsdemux.c

@@ -53,18 +53,15 @@ ts_recv_packet0
   (mpegts_service_t *t, elementary_stream_t *st, const uint8_t *tsb, int len)
 {
   mpegts_service_t *m;
-  int len2, off, pusi, cc, error, errors;
+  int len2, off, pusi, cc, pid, error, errors = 0;
   const uint8_t *tsb2;
 
   service_set_streaming_status_flags((service_t*)t, TSS_MUX_PACKETS);
 
-  if (!st) {
-    if(streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
-      ts_remux(t, tsb, len, 0);
-    return;
-  }
+  if (!st)
+    goto skip_cc;
 
-  for (errors = 0, tsb2 = tsb, len2 = len; len2 > 0; tsb2 += 188, len2 -= 188) {
+  for (tsb2 = tsb, len2 = len; len2 > 0; tsb2 += 188, len2 -= 188) {
 
     pusi    = (tsb2[1] >> 6) & 1; /* 0x40 */
     error   = (tsb2[1] >> 7) & 1; /* 0x80 */
@@ -105,13 +102,17 @@ ts_recv_packet0
 
   }
 
+skip_cc:
   if(streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
     ts_remux(t, tsb, len, errors);
 
   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_remux(m, tsb, len, errors);
+    if(streaming_pad_probe_type(&m->s_streaming_pad, SMT_MPEGTS)) {
+      pid = (tsb[1] & 0x1f) << 8 | tsb[2];
+      if (mpegts_pid_rexists(m->s_slaves_pids, pid))
+        ts_remux(m, tsb, len, errors);
+    }
     pthread_mutex_unlock(&m->s_stream_mutex);
   }
 }
@@ -124,14 +125,11 @@ ts_recv_skipped0
   (mpegts_service_t *t, elementary_stream_t *st, const uint8_t *tsb, int len)
 {
   mpegts_service_t *m;
-  int len2, cc;
+  int len2, cc, pid;
   const uint8_t *tsb2;
 
-  if (!st) {
-    if(streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
-      ts_skip(t, tsb, len);
-    return;
-  }
+  if (!st)
+    goto skip_cc;
 
   for (tsb2 = tsb, len2 = len; len2 > 0; tsb2 += 188, len2 -= 188) {
 
@@ -157,13 +155,17 @@ ts_recv_skipped0
      streaming_pad_probe_type(&t->s_streaming_pad, SMT_PACKET))
     skip_mpeg_ts((service_t*)t, st, tsb, len);
 
+skip_cc:
   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);
+    if(streaming_pad_probe_type(&m->s_streaming_pad, SMT_MPEGTS)) {
+      pid = (tsb[1] & 0x1f) << 8 | tsb[2];
+      if (mpegts_pid_rexists(t->s_slaves_pids, pid))
+        ts_skip(m, tsb, len);
+    }
     pthread_mutex_unlock(&m->s_stream_mutex);
   }
 }
@@ -303,10 +305,10 @@ ts_recv_raw(mpegts_service_t *t, const uint8_t *tsb, int len)
     service_set_streaming_status_flags((service_t*)t, TSS_PACKETS);
     t->s_streaming_live |= TSS_LIVE;
   }
-  if(!parent) {
-    if (streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS))
+  if (!parent) {
+    if (streaming_pad_probe_type(&t->s_streaming_pad, SMT_MPEGTS)) {
       ts_remux(t, tsb, len, 0);
-    else {
+    } else {
       /* No subscriber - set OK markers */
       service_set_streaming_status_flags((service_t*)t, TSS_PACKETS);
       t->s_streaming_live |= TSS_LIVE;