remove pkt_merge_header()

- rename pkt_header to pkt_meta
- unify logic for the metadata processing
- add meta field to components to the htsp stream start message

diff --git a/src/htsp_server.c b/src/htsp_server.c
index 36a586b0ccff8096f24ccd92e58a0a384cfea37d..76482aef90f457511ced62d96308f99028b34e38 100644 (file)
--- a/src/htsp_server.c
+++ b/src/htsp_server.c
@@ -68,7 +68,7 @@
 
 static void *htsp_server, *htsp_server_2;
 
-#define HTSP_PROTO_VERSION 16
+#define HTSP_PROTO_VERSION 17
 
 #define HTSP_ASYNC_OFF  0x00
 #define HTSP_ASYNC_ON   0x01
@@ -211,6 +211,8 @@ typedef struct htsp_subscription {
 
   int hs_first;
 
+  int hs_merge_meta_compomentidx;
+
 } htsp_subscription_t;
 
 
@@ -2912,6 +2914,21 @@ const static char frametypearray[PKT_NTYPES] = {
   [PKT_B_FRAME] = 'B',
 };
 
+static th_pkt_t *merge_pkt(th_pkt_t *pkt, size_t payloadlen)
+{
+  th_pkt_t *n = pkt_alloc(NULL, 0, 0, 0);
+  *n = *pkt;
+  n->pkt_refcount = 1;
+  n->pkt_meta = NULL;
+  n->pkt_payload = pktbuf_alloc(NULL, payloadlen);
+  memcpy(pktbuf_ptr(n->pkt_payload),
+         pktbuf_ptr(pkt->pkt_meta), pktbuf_len(pkt->pkt_meta));
+  memcpy(pktbuf_ptr(n->pkt_payload) + pktbuf_len(pkt->pkt_meta),
+         pktbuf_ptr(pkt->pkt_payload), pktbuf_len(pkt->pkt_payload));
+  pkt_ref_dec(pkt);
+  return n;
+}
+
 /**
  * Build a htsmsg from a th_pkt and enqueue it on our HTSP service
  */
@@ -2923,6 +2940,7 @@ htsp_stream_deliver(htsp_subscription_t *hs, th_pkt_t *pkt)
   htsp_connection_t *htsp = hs->hs_htsp;
   int64_t ts;
   int qlen = hs->hs_q.hmq_payload;
+  size_t payloadlen;
 
   if(!htsp_is_stream_enabled(hs, pkt->pkt_componentindex)) {
     pkt_ref_dec(pkt);
@@ -2962,16 +2980,22 @@ htsp_stream_deliver(htsp_subscription_t *hs, th_pkt_t *pkt)
   uint32_t dur = hs->hs_90khz ? pkt->pkt_duration : ts_rescale(pkt->pkt_duration, 1000000);
   htsmsg_add_u32(m, "duration", dur);
   
-  pkt = pkt_merge_header(pkt);
-
+  if (pkt->pkt_meta &&
+      hs->hs_merge_meta_compomentidx == pkt->pkt_componentindex) {
+    payloadlen = pktbuf_len(pkt->pkt_meta) + pktbuf_len(pkt->pkt_payload);
+    pkt = merge_pkt(pkt, payloadlen);
+    /* do it only once */
+    hs->hs_merge_meta_compomentidx = -1;
+  } else {
+    payloadlen = pktbuf_len(pkt->pkt_payload);
+  }
   /**
    * Since we will serialize directly we use 'binptr' which is a binary
    * object that just points to data, thus avoiding a copy.
    */
-  htsmsg_add_binptr(m, "payload", pktbuf_ptr(pkt->pkt_payload),
-                   pktbuf_len(pkt->pkt_payload));
-  htsp_send(htsp, m, pkt->pkt_payload, &hs->hs_q, pktbuf_len(pkt->pkt_payload));
-  atomic_add(&hs->hs_s->ths_bytes_out, pktbuf_len(pkt->pkt_payload));
+  htsmsg_add_binptr(m, "payload", pktbuf_ptr(pkt->pkt_payload), payloadlen);
+  htsp_send(htsp, m, pkt->pkt_payload, &hs->hs_q, payloadlen);
+  atomic_add(&hs->hs_s->ths_bytes_out, payloadlen);
 
   if(hs->hs_last_report != dispatch_clock) {
 
@@ -3041,9 +3065,11 @@ htsp_subscription_start(htsp_subscription_t *hs, const streaming_start_t *ss)
 
   tvhdebug("htsp", "%s - subscription start", hs->hs_htsp->htsp_logname);
 
+  hs->hs_merge_meta_compomentidx = -1;
+
   for(i = 0; i < ss->ss_num_components; i++) {
     const streaming_start_component_t *ssc = &ss->ss_components[i];
-    if (ssc->ssc_type == SCT_MPEG2VIDEO || ssc->ssc_type == SCT_H264) {
+    if (SCT_ISVIDEO(ssc->ssc_type)) {
       if (ssc->ssc_width == 0 || ssc->ssc_height == 0) {
         hs->hs_wait_for_video = 1;
         return;
@@ -3073,7 +3099,7 @@ htsp_subscription_start(htsp_subscription_t *hs, const streaming_start_t *ss)
       htsmsg_add_u32(c, "ancillary_id", ssc->ssc_ancillary_id);
     }
 
-    if(ssc->ssc_type == SCT_MPEG2VIDEO || ssc->ssc_type == SCT_H264) {
+    if(SCT_ISVIDEO(ssc->ssc_type)) {
       if(ssc->ssc_width)
         htsmsg_add_u32(c, "width", ssc->ssc_width);
       if(ssc->ssc_height)
@@ -3096,7 +3122,18 @@ htsp_subscription_start(htsp_subscription_t *hs, const streaming_start_t *ss)
         htsmsg_add_u32(c, "rate", ssc->ssc_sri);
     }
 
+    if (ssc->ssc_gh)
+      htsmsg_add_binptr(m, "meta", pktbuf_ptr(ssc->ssc_gh),
+                       pktbuf_len(ssc->ssc_gh));
+
     htsmsg_add_msg(streams, NULL, c);
+
+    if (ssc->ssc_type == SCT_H264 && hs->hs_htsp->htsp_version < 17) {
+      if (hs->hs_merge_meta_compomentidx < 0)
+        hs->hs_merge_meta_compomentidx = ssc->ssc_index;
+      else
+        tvherror("htsp", "multiple H264 video streams?");
+    }
   }
   
   htsmsg_add_msg(m, "streams", streams);

diff --git a/src/muxer/muxer_libav.c b/src/muxer/muxer_libav.c
index ea2d8d65872054500a25fe63027e1c7a4f46d9d5..726a58312a0e5e01895afea78182f1a7e65ed87f 100644 (file)
--- a/src/muxer/muxer_libav.c
+++ b/src/muxer/muxer_libav.c
@@ -387,9 +387,6 @@ lav_muxer_write_pkt(muxer_t *m, streaming_message_type_t smt, void *data)
 
     av_init_packet(&packet);
 
-    if(st->codec->codec_id == AV_CODEC_ID_MPEG2VIDEO)
-      pkt = pkt_merge_header(pkt);
-
     if(lm->lm_h264_filter && st->codec->codec_id == AV_CODEC_ID_H264) {
       if(av_bitstream_filter_filter(lm->lm_h264_filter,
                                    st->codec, 

diff --git a/src/muxer/tvh/mkmux.c b/src/muxer/tvh/mkmux.c
index 0c6d6c506b11e606ec1a71b2fc0d5db725dc1f90..0e21ed6e6fa362c813a792ba03c0687a7ea6e5f2 100644 (file)
--- a/src/muxer/tvh/mkmux.c
+++ b/src/muxer/tvh/mkmux.c
@@ -47,7 +47,6 @@ TAILQ_HEAD(mk_chapter_queue, mk_chapter);
 typedef struct mk_track {
   int index;
   int enabled;
-  int merge;
   int type;
   int tracknum;
   int disabled;
@@ -266,7 +265,6 @@ mk_build_tracks(mk_mux_t *mkm, const streaming_start_t *ss)
     case SCT_MPEG2VIDEO:
       tracktype = 1;
       codec_id = "V_MPEG2";
-      mkm->tracks[i].merge = 1;
       break;
 
     case SCT_H264:
@@ -1183,9 +1181,6 @@ mk_mux_write_pkt(mk_mux_t *mkm, th_pkt_t *pkt)
   if(mark)
     mk_mux_insert_chapter(mkm);
 
-  if(t->merge)
-    pkt = pkt_merge_header(pkt);
-
   mk_write_frame_i(mkm, t, pkt);
 
   pkt_ref_dec(pkt);

diff --git a/src/packet.c b/src/packet.c
index 41527364501dadaf30188f8755d81836197e367f..3dd617decb222d1f2bd3d706a2c3e88c4b71886c 100644 (file)
--- a/src/packet.c
+++ b/src/packet.c
@@ -32,11 +32,9 @@
 static void
 pkt_destroy(th_pkt_t *pkt)
 {
-  if(pkt->pkt_payload != NULL)
-    pktbuf_ref_dec(pkt->pkt_payload);
+  pktbuf_ref_dec(pkt->pkt_payload);
+  pktbuf_ref_dec(pkt->pkt_meta);
 
-  if(pkt->pkt_header != NULL)
-    pktbuf_ref_dec(pkt->pkt_header);
   free(pkt);
 }
 
@@ -129,41 +127,6 @@ pktref_remove(struct th_pktref_queue *q, th_pktref_t *pr)
 }
 
 
-/**
- *
- */
-th_pkt_t *
-pkt_merge_header(th_pkt_t *pkt)
-{
-  th_pkt_t *n;
-  size_t s;
-
-  if(pkt->pkt_header == NULL)
-    return pkt;
-
-  n = malloc(sizeof(th_pkt_t));
-  *n = *pkt;
-
-  n->pkt_refcount = 1;
-  n->pkt_header = NULL;
-
-  s = pktbuf_len(pkt->pkt_payload) + pktbuf_len(pkt->pkt_header);
-  n->pkt_payload = pktbuf_alloc(NULL, s);
-
-  memcpy(pktbuf_ptr(n->pkt_payload),
-        pktbuf_ptr(pkt->pkt_header),
-        pktbuf_len(pkt->pkt_header));
-
-  memcpy(pktbuf_ptr(n->pkt_payload) + pktbuf_len(pkt->pkt_header),
-        pktbuf_ptr(pkt->pkt_payload),
-        pktbuf_len(pkt->pkt_payload));
-
-  pkt_ref_dec(pkt);
-  return n;
-}
-
-
-
 /**
  *
  */
@@ -175,11 +138,8 @@ pkt_copy_shallow(th_pkt_t *pkt)
 
   n->pkt_refcount = 1;
 
-  if(n->pkt_header)
-    pktbuf_ref_inc(n->pkt_header);
-
-  if(n->pkt_payload)
-    pktbuf_ref_inc(n->pkt_payload);
+  pktbuf_ref_inc(n->pkt_meta);
+  pktbuf_ref_inc(n->pkt_payload);
 
   return n;
 }
@@ -196,21 +156,29 @@ pktref_create(th_pkt_t *pkt)
   return pr;
 }
 
+/*
+ *
+ */
 
-
-void 
+void
 pktbuf_ref_dec(pktbuf_t *pb)
 {
-  if((atomic_add(&pb->pb_refcount, -1)) == 1) {
-    free(pb->pb_data);
-    free(pb);
+  if (pb) {
+    if((atomic_add(&pb->pb_refcount, -1)) == 1) {
+      free(pb->pb_data);
+      free(pb);
+    }
   }
 }
 
-void
+pktbuf_t *
 pktbuf_ref_inc(pktbuf_t *pb)
 {
-  atomic_add(&pb->pb_refcount, 1);
+  if (pb) {
+    atomic_add(&pb->pb_refcount, 1);
+    return pb;
+  }
+  return NULL;
 }
 
 pktbuf_t *

diff --git a/src/packet.h b/src/packet.h
index 2aecce7b8e3fbe78fbac3ecbb21b07d094a0afd1..b5c2bcada97f9f6f31fac546943c8ce5ab6bcb9e 100644 (file)
--- a/src/packet.h
+++ b/src/packet.h
@@ -1,5 +1,5 @@
 /*
- *  Packet nanagement
+ *  Packet management
  *  Copyright (C) 2008 Andreas Öman
  *
  *  This program is free software: you can redistribute it and/or modify
@@ -19,6 +19,9 @@
 #ifndef PACKET_H_
 #define PACKET_H_
 
+/**
+ * Packet buffer
+ */
 
 typedef struct pktbuf {
   int pb_refcount;
@@ -26,8 +29,6 @@ typedef struct pktbuf {
   size_t pb_size;
 } pktbuf_t;
 
-
-
 /**
  * Packets
  */
@@ -62,8 +63,8 @@ typedef struct th_pkt {
   uint16_t pkt_aspect_num;
   uint16_t pkt_aspect_den;
 
+  pktbuf_t *pkt_meta;
   pktbuf_t *pkt_payload;
-  pktbuf_t *pkt_header;
 
 } th_pkt_t;
 
@@ -96,21 +97,23 @@ void pktref_remove(struct th_pktref_queue *q, th_pktref_t *pr);
 
 th_pkt_t *pkt_alloc(const void *data, size_t datalen, int64_t pts, int64_t dts);
 
-th_pkt_t *pkt_merge_header(th_pkt_t *pkt);
-
 th_pkt_t *pkt_copy_shallow(th_pkt_t *pkt);
 
 th_pktref_t *pktref_create(th_pkt_t *pkt);
 
+/*
+ *
+ */
+
 void pktbuf_ref_dec(pktbuf_t *pb);
 
-void pktbuf_ref_inc(pktbuf_t *pb);
+pktbuf_t *pktbuf_ref_inc(pktbuf_t *pb);
 
 pktbuf_t *pktbuf_alloc(const void *data, size_t size);
 
 pktbuf_t *pktbuf_make(void *data, size_t size);
 
-#define pktbuf_len(pb) ((pb)->pb_size)
-#define pktbuf_ptr(pb) ((pb)->pb_data)
+static inline size_t   pktbuf_len(pktbuf_t *pb) { return pb->pb_size; }
+static inline uint8_t *pktbuf_ptr(pktbuf_t *pb) { return pb->pb_data; }
 
 #endif /* PACKET_H_ */

diff --git a/src/parsers/parser_avc.c b/src/parsers/parser_avc.c
index bd0b911ed7c826834e7162996bba55ea89e8977b..3efacf09d42c49ff198c6ef509cfbe6acf2d8bb4 100644 (file)
--- a/src/parsers/parser_avc.c
+++ b/src/parsers/parser_avc.c
@@ -213,24 +213,24 @@ avc_convert_pkt(th_pkt_t *src)
   th_pkt_t *pkt = malloc(sizeof(th_pkt_t));
   *pkt = *src;
   pkt->pkt_refcount = 1;
-  pkt->pkt_header = NULL;
+  pkt->pkt_meta = NULL;
   pkt->pkt_payload = NULL;
 
-  if (src->pkt_header) {
+  if (src->pkt_meta) {
     sbuf_t headers;
     sbuf_init(&headers);
     
-    isom_write_avcc(&headers, pktbuf_ptr(src->pkt_header),
-                   pktbuf_len(src->pkt_header));
-    pkt->pkt_header = pktbuf_make(headers.sb_data, headers.sb_ptr);
+    isom_write_avcc(&headers, pktbuf_ptr(src->pkt_meta),
+                   pktbuf_len(src->pkt_meta));
+    pkt->pkt_meta = pktbuf_make(headers.sb_data, headers.sb_ptr);
   }
 
   sbuf_t payload;
   sbuf_init(&payload);
 
-  if(src->pkt_header)
-    avc_parse_nal_units(&payload, pktbuf_ptr(src->pkt_header),
-                       pktbuf_len(src->pkt_header));
+  if(src->pkt_meta)
+    avc_parse_nal_units(&payload, pktbuf_ptr(src->pkt_meta),
+                       pktbuf_len(src->pkt_meta));
 
   avc_parse_nal_units(&payload, pktbuf_ptr(src->pkt_payload),
                      pktbuf_len(src->pkt_payload));

diff --git a/src/parsers/parsers.c b/src/parsers/parsers.c
index 8a7fc8b0cec57dbd88255cde670a93525c4e4ce3..33a5c728b33240353653dcdf0dc8836043f9d80f 100644 (file)
--- a/src/parsers/parsers.c
+++ b/src/parsers/parsers.c
@@ -1094,22 +1094,30 @@ parse_mpeg2video(service_t *t, elementary_stream_t *st, size_t len,
     if(next_startcode == 0x100 || next_startcode > 0x1af) {
       /* Last picture slice (because next not a slice) */
       th_pkt_t *pkt = st->es_curpkt;
+      size_t metalen = 0;
       if(pkt == NULL) {
         /* no packet, may've been discarded by sanity checks here */
         return 1;
       }
 
       if(st->es_global_data) {
-        pkt->pkt_header = pktbuf_make(st->es_global_data,
-                                      st->es_global_data_len);
+        pkt->pkt_meta = pktbuf_make(st->es_global_data,
+                                    metalen = st->es_global_data_len);
         st->es_global_data = NULL;
         st->es_global_data_len = 0;
       }
 
-      pkt->pkt_payload = pktbuf_make(st->es_buf.sb_data,
-                                     st->es_buf.sb_ptr - 4);
+      if (metalen) {
+        pkt->pkt_payload = pktbuf_alloc(NULL, metalen + st->es_buf.sb_ptr - 4);
+        memcpy(pktbuf_ptr(pkt->pkt_payload), pktbuf_ptr(pkt->pkt_meta), metalen);
+        memcpy(pktbuf_ptr(pkt->pkt_payload) + metalen, st->es_buf.sb_data, st->es_buf.sb_ptr - 4);
+        sbuf_reset(&st->es_buf, 16000);
+      } else {
+        pkt->pkt_payload = pktbuf_make(st->es_buf.sb_data,
+                                       st->es_buf.sb_ptr - 4);
+        sbuf_steal_data(&st->es_buf);
+      }
       pkt->pkt_duration = st->es_frame_duration;
-      sbuf_steal_data(&st->es_buf);
 
       parser_deliver(t, st, pkt, st->es_buf.sb_err);
       st->es_curpkt = NULL;
@@ -1226,8 +1234,8 @@ parse_h264(service_t *t, elementary_stream_t *st, size_t len,
     if(pkt != NULL) {
       
       if(st->es_global_data) {
-        pkt->pkt_header = pktbuf_make(st->es_global_data,
-                                      st->es_global_data_len);
+        pkt->pkt_meta = pktbuf_make(st->es_global_data,
+                                    st->es_global_data_len);
         st->es_global_data = NULL;
         st->es_global_data_len = 0;
       }
@@ -1374,18 +1382,16 @@ parser_deliver(service_t *t, elementary_stream_t *st, th_pkt_t *pkt, int error)
     t->s_current_pts = pkt->pkt_pts;
 
   tvhtrace("parser",
-           "pkt stream %2d %-12s type %c dts %10"PRId64" pts %10"PRId64
+           "pkt stream %2d %-12s type %c"
+           " dts %10"PRId64" (%10"PRId64") pts %10"PRId64" (%10"PRId64")"
            " dur %10d len %10zu",
            st->es_index,
            streaming_component_type2txt(st->es_type),
            pkt_frametype_to_char(pkt->pkt_frametype),
-#if 1
            ts_rescale(pkt->pkt_pts, 1000000),
-           ts_rescale(pkt->pkt_dts, 1000000),
-#else
            pkt->pkt_dts,
+           ts_rescale(pkt->pkt_dts, 1000000),
            pkt->pkt_pts,
-#endif
            pkt->pkt_duration,
            pktbuf_len(pkt->pkt_payload));
 

diff --git a/src/plumbing/globalheaders.c b/src/plumbing/globalheaders.c
index 5571a0194c0bfadb1b47148f304cd23a660e475d..6994b650012b9d9d24264f250883009ec28b9771 100644 (file)
--- a/src/plumbing/globalheaders.c
+++ b/src/plumbing/globalheaders.c
@@ -77,8 +77,8 @@ apply_header(streaming_start_component_t *ssc, th_pkt_t *pkt)
   if(ssc->ssc_gh != NULL)
     return;
 
-  if(pkt->pkt_header != NULL) {
-    ssc->ssc_gh = pkt->pkt_header;
+  if(pkt->pkt_meta != NULL) {
+    ssc->ssc_gh = pkt->pkt_meta;
     pktbuf_ref_inc(ssc->ssc_gh);
     return;
   }

diff --git a/src/plumbing/transcoding.c b/src/plumbing/transcoding.c
index 874f53081b191acc67e131a3542088a819aa808e..e931503363c9c01735215f813cc41f385dd63b6d 100644 (file)
--- a/src/plumbing/transcoding.c
+++ b/src/plumbing/transcoding.c
@@ -386,8 +386,6 @@ transcoder_stream_audio(transcoder_t *t, transcoder_stream_t *ts, th_pkt_t *pkt)
     as->aud_dec_pts += (pkt->pkt_pts - as->aud_dec_pts);
   }
 
-  pkt = pkt_merge_header(pkt);
-
   av_init_packet(&packet);
   packet.data     = pktbuf_ptr(pkt->pkt_payload);
   packet.size     = pktbuf_len(pkt->pkt_payload);
@@ -740,7 +738,7 @@ transcoder_stream_audio(transcoder_t *t, transcoder_stream_t *ts, th_pkt_t *pkt)
         create_adts_header(pkt->pkt_payload, n->pkt_sri, octx->channels);
 
       if (octx->extradata_size)
-       n->pkt_header = pktbuf_alloc(octx->extradata, octx->extradata_size);
+       n->pkt_meta = pktbuf_alloc(octx->extradata, octx->extradata_size);
 
       tvhtrace("transcode", "%04X: deliver audio (pts = %" PRIi64 ", delay = %i)",
                shortid(t), n->pkt_pts, octx->delay);
@@ -823,7 +821,7 @@ Minimal of 12 bytes.
      }
     }
 
-    n->pkt_header = pktbuf_alloc(data, header_size);
+    n->pkt_meta = pktbuf_alloc(data, header_size);
   }
 }
 
@@ -886,7 +884,7 @@ send_video_packet(transcoder_t *t, transcoder_stream_t *ts, th_pkt_t *pkt,
   }
 
   if (octx->extradata_size)
-    n->pkt_header = pktbuf_alloc(octx->extradata, octx->extradata_size);
+    n->pkt_meta = pktbuf_alloc(octx->extradata, octx->extradata_size);
   else {
     if (octx->codec_id == AV_CODEC_ID_MPEG2VIDEO)
       extract_mpeg2_global_data(n, epkt->data, epkt->size);
@@ -933,8 +931,6 @@ transcoder_stream_video(transcoder_t *t, transcoder_stream_t *ts, th_pkt_t *pkt)
     }
   }
 
-  pkt = pkt_merge_header(pkt);
-
   av_init_packet(&packet);
   packet.data     = pktbuf_ptr(pkt->pkt_payload);
   packet.size     = pktbuf_len(pkt->pkt_payload);

diff --git a/src/timeshift/timeshift_reader.c b/src/timeshift/timeshift_reader.c
index bb79db995756a47ead1230f921a1fe2c4a5fc9b8..58d744ebaf171abdc29c70e7ad8a9dcdd09c6698 100644 (file)
--- a/src/timeshift/timeshift_reader.c
+++ b/src/timeshift/timeshift_reader.c
@@ -149,10 +149,10 @@ static ssize_t _read_msg ( int fd, streaming_message_t **sm )
       }
       if (type == SMT_PACKET) {
         th_pkt_t *pkt = data;
-        pkt->pkt_payload  = pkt->pkt_header = NULL;
+        pkt->pkt_payload  = pkt->pkt_meta = NULL;
         pkt->pkt_refcount = 0;
         *sm = streaming_msg_create_pkt(pkt);
-        r   = _read_pktbuf(fd, &pkt->pkt_header);
+        r   = _read_pktbuf(fd, &pkt->pkt_meta);
         if (r < 0) {
           streaming_msg_free(*sm);
           return r;

diff --git a/src/timeshift/timeshift_writer.c b/src/timeshift/timeshift_writer.c
index d534410a4a9b4d3c49a00b77104734b13b364ec8..40a0a6970ab5ecc8979044407954959368c7976a 100644 (file)
--- a/src/timeshift/timeshift_writer.c
+++ b/src/timeshift/timeshift_writer.c
@@ -116,7 +116,7 @@ ssize_t timeshift_write_packet ( int fd, int64_t time, th_pkt_t *pkt )
   ssize_t ret = 0, err;
   ret = err = _write_msg(fd, SMT_PACKET, time, pkt, sizeof(th_pkt_t));
   if (err <= 0) return err;
-  err = _write_pktbuf(fd, pkt->pkt_header);
+  err = _write_pktbuf(fd, pkt->pkt_meta);
   if (err <= 0) return err;
   ret += err;
   err = _write_pktbuf(fd, pkt->pkt_payload);