pktref cleanups: fix memoryinfo statistics and improve code readability

src/packet.c

@@ -193,6 +193,37 @@ pktref_remove(struct th_pktref_queue *q, th_pktref_t *pr)
   }
 }
 
+/**
+ *
+ */
+th_pkt_t *
+pktref_get_first(struct th_pktref_queue *q)
+{
+  th_pktref_t *pr;
+  th_pkt_t *pkt;
+
+  pr = TAILQ_FIRST(q);
+  if (pr) {
+    pkt = pr->pr_pkt;
+    TAILQ_REMOVE(q, pr, pr_link);
+    free(pr);
+    memoryinfo_free(&pktref_memoryinfo, sizeof(*pr));
+    return pkt;
+  }
+  return NULL;
+}
+
+/**
+ *
+ */
+void
+pktref_insert_head(struct th_pktref_queue *q, th_pkt_t *pkt)
+{
+  th_pktref_t *pr;
+
+  pr = pktref_create(pkt);
+  TAILQ_INSERT_HEAD(q, pr, pr_link);
+}
 
 /**
  *

src/packet.h

@@ -105,6 +105,11 @@ void pktref_enqueue(struct th_pktref_queue *q, th_pkt_t *pkt);
 
 void pktref_remove(struct th_pktref_queue *q, th_pktref_t *pr);
 
+th_pkt_t *pktref_get_first(struct th_pktref_queue *q);
+
+void pktref_insert_head(struct th_pktref_queue *q, th_pkt_t *pkt);
+
+#define PKTREF_FOREACH(item, queue) TAILQ_FOREACH((item), (queue), pr_link)
 
 th_pkt_t *pkt_alloc(const void *data, size_t datalen, int64_t pts, int64_t dts);
 

src/plumbing/globalheaders.c

@@ -274,7 +274,6 @@ static void
 gh_hold(globalheaders_t *gh, streaming_message_t *sm)
 {
   th_pkt_t *pkt;
-  th_pktref_t *pr;
   streaming_start_component_t *ssc;
 
   switch(sm->sm_type) {
@@ -308,15 +307,12 @@ gh_hold(globalheaders_t *gh, streaming_message_t *sm)
     streaming_target_deliver2(gh->gh_output, sm);
 
     // Send all pending packets
-    while((pr = TAILQ_FIRST(&gh->gh_holdq)) != NULL) {
-      TAILQ_REMOVE(&gh->gh_holdq, pr, pr_link);
-      pkt = pr->pr_pkt;
+    while((pkt = pktref_get_first(&gh->gh_holdq)) != NULL) {
       if (pkt->pkt_payload) {
         sm = streaming_msg_create_pkt(pkt);
         streaming_target_deliver2(gh->gh_output, sm);
       }
       pkt_ref_dec(pkt);
-      free(pr);
     }
     gh->gh_passthru = 1;
     break;

src/plumbing/tsfix.c

@@ -287,13 +287,9 @@ static void
 tsfix_backlog(tsfix_t *tf)
 {
   th_pkt_t *pkt;
-  th_pktref_t *pr;
   tfstream_t *tfs;
 
-  while((pr = TAILQ_FIRST(&tf->tf_backlog)) != NULL) {
-    pkt = pr->pr_pkt;
-    TAILQ_REMOVE(&tf->tf_backlog, pr, pr_link);
-    free(pr);
+  while((pkt = pktref_get_first(&tf->tf_backlog)) != NULL) {
     tfs = tfs_find(tf, pkt);
     normalize_ts(tf, tfs, pkt, 0);
   }
@@ -311,7 +307,7 @@ tsfix_backlog_diff(tsfix_t *tf)
   tfstream_t *tfs;
   int64_t res = 0;
 
-  TAILQ_FOREACH(pr, &tf->tf_backlog, pr_link) {
+  PKTREF_FOREACH(pr, &tf->tf_backlog) {
     pkt = pr->pr_pkt;
     if (pkt->pkt_dts == PTS_UNSET) continue;
     if (pkt->pkt_dts >= tf->tf_tsref) continue;
@@ -331,15 +327,12 @@ tsfix_backlog_diff(tsfix_t *tf)
 static void
 recover_pts(tsfix_t *tf, tfstream_t *tfs, th_pkt_t *pkt)
 {
-  th_pktref_t *pr, *srch;
+  th_pktref_t *srch;
 
   pktref_enqueue(&tf->tf_ptsq, pkt);
 
-  while((pr = TAILQ_FIRST(&tf->tf_ptsq)) != NULL) {
+  while((pkt = pktref_get_first(&tf->tf_ptsq)) != NULL) {
 
-    pkt = pr->pr_pkt;
-    TAILQ_REMOVE(&tf->tf_ptsq, pr, pr_link);
-
     tfs = tfs_find(tf, pkt);
 
     switch(tfs->tfs_type) {
@@ -359,7 +352,7 @@ recover_pts(tsfix_t *tf, tfstream_t *tfs, th_pkt_t *pkt)
       case PKT_P_FRAME:
 	/* Presentation occures at DTS of next I or P frame,
 	   try to find it */
-	TAILQ_FOREACH(srch, &tf->tf_ptsq, pr_link)
+	PKTREF_FOREACH(srch, &tf->tf_ptsq)
 	  if (tfs_find(tf, srch->pr_pkt) == tfs &&
 	      srch->pr_pkt->pkt_frametype <= PKT_P_FRAME) {
 	    pkt->pkt_pts = srch->pr_pkt->pkt_dts;
@@ -370,7 +363,7 @@ recover_pts(tsfix_t *tf, tfstream_t *tfs, th_pkt_t *pkt)
 	  }
 	if (srch == NULL) {
 	  /* return packet back to tf_ptsq */
-	  TAILQ_INSERT_HEAD(&tf->tf_ptsq, pr, pr_link);
+	  pktref_insert_head(&tf->tf_ptsq, pkt);
 	  return; /* not arrived yet, wait */
         }
       }
@@ -380,7 +373,6 @@ recover_pts(tsfix_t *tf, tfstream_t *tfs, th_pkt_t *pkt)
       break;
     }
 
-    free(pr);
     normalize_ts(tf, tfs, pkt, 1);
   }
 }