factor out subscription byte count updates to separate functions. Thi…

…s has the added benefit of ensuring that all updates are done atomically (previously only a minority of them were).

src/dvr/dvr_rec.c

@@ -915,7 +915,7 @@ dvr_thread(void *aux)
         }
       }
       if (pb)
-        atomic_add(&ts->ths_bytes_out, pktbuf_len(pb));
+        subscription_add_bytes_out(ts, pktbuf_len(pb));
     }
 
     streaming_queue_remove(sq, sm);

src/epggrab/module/eit.c

@@ -597,8 +597,8 @@ _eit_callback
   /* Statistics */
   ths = mpegts_mux_find_subscription_by_name(mm, "epggrab");
   if (ths) {
-    ths->ths_bytes_in += len;
-    ths->ths_bytes_out += len;
+    subscription_add_bytes_in(ths, len);
+    subscription_add_bytes_out(ths, len);
   }
 
   /* Validate */

src/epggrab/module/opentv.c

@@ -563,8 +563,8 @@ opentv_table_callback
   /* Statistics */
   ths = mpegts_mux_find_subscription_by_name(mt->mt_mux, "epggrab");
   if (ths) {
-    ths->ths_bytes_in += len;
-    ths->ths_bytes_out += len;
+    subscription_add_bytes_in(ths, len);
+    subscription_add_bytes_out(ths, len);
   }
 
 
@@ -642,8 +642,8 @@ opentv_bat_callback
   /* Statistics */
   ths = mpegts_mux_find_subscription_by_name(mt->mt_mux, "epggrab");
   if (ths) {
-    ths->ths_bytes_in += len;
-    ths->ths_bytes_out += len;
+    subscription_add_bytes_in(ths, len);
+    subscription_add_bytes_out(ths, len);
   }
 
   r = dvb_bat_callback(mt, buf, len, tableid);

src/htsp_server.c

@@ -3415,7 +3415,7 @@ htsp_stream_deliver(htsp_subscription_t *hs, th_pkt_t *pkt)
   payloadlen = pktbuf_len(pkt->pkt_payload);
   htsmsg_add_binptr(m, "payload", pktbuf_ptr(pkt->pkt_payload), payloadlen);
   htsp_send_subscription(htsp, m, pkt->pkt_payload, hs, payloadlen);
-  atomic_add(&hs->hs_s->ths_bytes_out, payloadlen);
+  subscription_add_bytes_out(hs->hs_s, payloadlen);
 
   if(hs->hs_last_report != dispatch_clock) {
 

src/satip/rtp.c

@@ -189,7 +189,7 @@ satip_rtp_thread(void *aux)
     switch (sm->sm_type) {
     case SMT_MPEGTS:
       pb = sm->sm_data;
-      atomic_add(&subs->ths_bytes_out, pktbuf_len(pb));
+      subscription_add_bytes_out(subs, pktbuf_len(pb));
       pthread_mutex_lock(&rtp->lock);
       r = satip_rtp_loop(rtp, pktbuf_ptr(pb), pktbuf_len(pb));
       pthread_mutex_unlock(&rtp->lock);

src/subscriptions.c

@@ -457,11 +457,11 @@ subscription_input_direct(void *opauqe, streaming_message_t *sm)
     th_pkt_t *pkt = sm->sm_data;
     s->ths_total_err += pkt->pkt_err;
     if (pkt->pkt_payload)
-      s->ths_bytes_in += pkt->pkt_payload->pb_size;
+      subscription_add_bytes_in(s, pkt->pkt_payload->pb_size);
   } else if(sm->sm_type == SMT_MPEGTS) {
     pktbuf_t *pb = sm->sm_data;
     s->ths_total_err += pb->pb_err;
-    s->ths_bytes_in += pb->pb_size;
+    subscription_add_bytes_in(s, pb->pb_size);
   }
 
   /* Pass to output */
@@ -951,6 +951,22 @@ subscription_done(void)
  * Subscription control
  * *************************************************************************/
 
+/**
+ * Update incoming byte count
+ */
+void subscription_add_bytes_in(th_subscription_t *s, size_t in)
+{
+  atomic_add(&s->ths_bytes_in, in);
+}
+
+/**
+ * Update outgoing byte count
+ */
+void subscription_add_bytes_out(th_subscription_t *s, size_t out)
+{
+  atomic_add(&s->ths_bytes_out, out);
+}
+
 /**
  * Change weight
  */

src/subscriptions.h

@@ -201,6 +201,9 @@ void subscription_unlink_service(th_subscription_t *s, int reason);
 
 void subscription_dummy_join(const char *id, int first);
 
+void subscription_add_bytes_in(th_subscription_t *s, size_t in);
+
+void subscription_add_bytes_out(th_subscription_t *s, size_t out);
 
 static inline int subscriptions_active(void)
   { return LIST_FIRST(&subscriptions) != NULL; }

src/webui/webui.c

@@ -349,7 +349,7 @@ http_stream_run(http_connection_t *hc, profile_chain_t *prch,
           pb = ((th_pkt_t*)sm->sm_data)->pkt_payload;
         else
           pb = sm->sm_data;
-        atomic_add(&s->ths_bytes_out, pktbuf_len(pb));
+        subscription_add_bytes_out(s, pktbuf_len(pb));
         muxer_write_pkt(mux, sm->sm_type, sm->sm_data);
         sm->sm_data = NULL;
       }
@@ -1500,8 +1500,8 @@ page_dvrfile(http_connection_t *hc, const char *remain, void *opaque)
       }
       content_len -= r;
       if (sub) {
-        sub->ths_bytes_in += r;
-        sub->ths_bytes_out += r;
+        subscription_add_bytes_in(sub, r);
+        subscription_add_bytes_out(sub, r);
       }
     }
   }