Add: Consistent redundant stream logging

DawidWesierski4 · DawidWesierski4 · commit 9ed19e670b7b · 2025-10-28T11:50:12.000-04:00
Unify redundant packet logging with regular packet reporting.
Use global session ID for redundant checking across
st40/st41 sessions, similar to timestamp validation
to make things as similar to eachoder as possible.

Change API to report out-of-order packets in user port statistics.

Note: Out-of-order per port tracking not implemented for st20.
diff --git a/include/st_api.h b/include/st_api.h
@@ -271,6 +271,8 @@ struct st_rx_port_stats {
   uint64_t incomplete_frames;
   /** Total number of received packets which are not valid. */
   uint64_t err_packets;
+  /** Total number of out-of-order packets received */
+  uint64_t out_of_order_packets;
 };
 
 /**
diff --git a/lib/src/mt_util.h b/lib/src/mt_util.h
@@ -115,20 +115,6 @@ static inline bool mt_seq32_greater(uint32_t a, uint32_t b) {
   return (diff & 0x80000000u) == 0 && diff != 0;
 }
 
-static inline bool st_rx_seq_redundant_drop(uint16_t new_id, int* sessions_redundant,
-                                            enum mtl_session_port s_port, int num_port) {
-  for (int i = MTL_SESSION_PORT_P; i < num_port; i++) {
-    if (i == s_port) continue;
-
-    uint16_t old_id = sessions_redundant[i];
-    if (mt_seq16_greater(old_id, new_id)) {
-      return true;
-    }
-  }
-
-  return false;
-}
-
 struct rte_mbuf* mt_build_pad(struct mtl_main_impl* impl, struct rte_mempool* mempool,
                               enum mtl_port port, uint16_t ether_type, uint16_t len);
 
@@ -346,4 +332,4 @@ static inline void mt_stat_u64_update(struct mt_stat_u64* stat, uint64_t new) {
   stat->cnt++;
 }
 
-#endif
+#endif
diff --git a/lib/src/st2110/st_header.h b/lib/src/st2110/st_header.h
@@ -690,6 +690,7 @@ struct st_rx_video_session_impl {
   int stat_pkts_enqueue_fallback; /* for pkt lcore */
   int stat_pkts_offset_dropped;
   int stat_pkts_out_of_order;
+  int stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_MAX];
   int stat_pkts_redundant_dropped;
   int stat_pkts_wrong_pt_dropped;
   int stat_pkts_wrong_ssrc_dropped;
@@ -1016,6 +1017,7 @@ struct st_rx_audio_session_impl {
   uint32_t st30_pkt_size; /* size for each pkt which include the header */
   int st30_total_pkts;    /* total pkts in one frame */
   int st30_pkt_idx;       /* pkt index in current frame */
+  int session_seq_id;     /* global session seq id to track continuity across redundant */
   int latest_seq_id[MTL_SESSION_PORT_MAX]; /* latest seq id */
 
   uint32_t first_pkt_rtp_ts; /* rtp time stamp for the first pkt */
@@ -1033,6 +1035,7 @@ struct st_rx_audio_session_impl {
   int stat_pkts_dropped;
   int stat_pkts_redundant;
   int stat_pkts_out_of_order;
+  int stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_MAX];
   int stat_slot_get_frame_fail;
   int stat_pkts_wrong_pt_dropped;
   int stat_pkts_wrong_ssrc_dropped;
@@ -1182,7 +1185,7 @@ struct st_rx_ancillary_session_impl {
 
   uint16_t st40_dst_port[MTL_SESSION_PORT_MAX]; /* udp port */
   bool mcast_joined[MTL_SESSION_PORT_MAX];
-
+  int session_seq_id; /* global session seq id to track continuity across redundant */
   int latest_seq_id[MTL_SESSION_PORT_MAX]; /* latest seq id */
 
   struct mt_rtcp_rx* rtcp_rx[MTL_SESSION_PORT_MAX];
@@ -1193,6 +1196,7 @@ struct st_rx_ancillary_session_impl {
   int stat_pkts_dropped;
   int stat_pkts_redundant;
   int stat_pkts_out_of_order;
+  int stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_MAX];
   int stat_pkts_enqueue_fail;
   int stat_pkts_wrong_pt_dropped;
   int stat_pkts_wrong_ssrc_dropped;
@@ -1354,7 +1358,7 @@ struct st_rx_fastmetadata_session_impl {
 
   uint16_t st41_dst_port[MTL_SESSION_PORT_MAX]; /* udp port */
   bool mcast_joined[MTL_SESSION_PORT_MAX];
-
+  int session_seq_id; /* global session seq id to track continuity across redundant */
   int latest_seq_id[MTL_SESSION_PORT_MAX]; /* latest seq id */
 
   struct mt_rtcp_rx* rtcp_rx[MTL_SESSION_PORT_MAX];
@@ -1364,6 +1368,7 @@ struct st_rx_fastmetadata_session_impl {
   rte_atomic32_t stat_frames_received;
   int stat_pkts_redundant;
   int stat_pkts_out_of_order;
+  int stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_MAX];
   int stat_pkts_enqueue_fail;
   int stat_pkts_wrong_pt_dropped;
   int stat_pkts_wrong_ssrc_dropped;
diff --git a/lib/src/st2110/st_rx_ancillary_session.c b/lib/src/st2110/st_rx_ancillary_session.c
@@ -131,37 +131,46 @@ static int rx_ancillary_session_handle_pkt(struct mtl_main_impl* impl,
   }
   /* 0b00: progressive or not specified, do nothing */
 
-  if (unlikely(s->latest_seq_id[MTL_PORT_P] == -1))
-    s->latest_seq_id[MTL_PORT_P] = seq_id - 1;
-  if (unlikely(s->latest_seq_id[MTL_PORT_R] == -1))
-    s->latest_seq_id[MTL_PORT_R] = seq_id - 1;
+  if (unlikely(s->latest_seq_id[s_port] == -1)) s->latest_seq_id[s_port] = seq_id - 1;
+  if (unlikely(s->session_seq_id == -1)) s->session_seq_id = seq_id - 1;
   if (unlikely(s->tmstamp == -1)) s->tmstamp = tmstamp - 1;
 
+  /* not a big deal as long as stream is continous */
   if (seq_id != (uint16_t)(s->latest_seq_id[s_port] + 1)) {
     dbg("%s(%d,%d), non-continuous seq now %u last %d\n", __func__, s->idx, s_port,
         seq_id, s->latest_seq_id[s_port]);
-    ST_SESSION_STAT_INC(s, port_user_stats.common, stat_pkts_out_of_order);
+    s->port_user_stats.common.port[s_port].out_of_order_packets++;
+    s->stat_pkts_out_of_order_per_port[s_port]++;
   }
+  s->latest_seq_id[s_port] = seq_id;
 
   /* in ancillary we assume packet is redundant when the seq_id is old (it's possible to
   get multiple packets with the same timestamp)) */
   if ((mt_seq32_greater(s->tmstamp, tmstamp)) ||
-      st_rx_seq_redundant_drop(seq_id, s->latest_seq_id, s_port, s->ops.num_port)) {
-    if (mt_seq32_greater(s->tmstamp, tmstamp)) {
-      dbg("%s(%d,%d), drop as pkt tmstamp %u is old\n", __func__, s->idx, s_port,
-          tmstamp);
+      !mt_seq16_greater(seq_id, s->session_seq_id)) {
+    if (!mt_seq16_greater(seq_id, s->session_seq_id)) {
+      dbg("%s(%d,%d), redundant seq now %u session last %d\n", __func__, s->idx, s_port,
+          seq_id, s->session_seq_id);
     } else {
-      dbg("%s(%d,%d), drop as pkt seq %d is old\n", __func__, s->idx, s_port, seq_id);
+      dbg("%s(%d,%d), redundant tmstamp now %u session last %u\n", __func__, s->idx,
+          s_port, tmstamp, s->tmstamp);
     }
 
     ST_SESSION_STAT_INC(s, port_user_stats, stat_pkts_redundant);
-    s->latest_seq_id[s_port] = seq_id;
     return -EIO;
   }
 
+  /* hole in seq id packets going into the session check if the seq_id of the session is
+   * consistent */
+  if (seq_id != (uint16_t)(s->session_seq_id + 1)) {
+    dbg("%s(%d,%d), session seq_id %u out of order %d\n", __func__, s->idx, s_port,
+        seq_id, s->session_seq_id);
+    s->stat_pkts_out_of_order++;
+    ST_SESSION_STAT_INC(s, port_user_stats.common, stat_pkts_out_of_order);
+  }
+
   /* update seq id */
-  s->latest_seq_id[s_port] = seq_id;
-  s->tmstamp = tmstamp;
+  s->session_seq_id = seq_id;
 
   /* enqueue to packet ring to let app to handle */
   int ret = rte_ring_sp_enqueue(s->packet_ring, (void*)mbuf);
@@ -417,6 +426,7 @@ static int rx_ancillary_session_attach(struct mtl_main_impl* impl,
     s->st40_dst_port[i] = (ops->udp_port[i]) ? (ops->udp_port[i]) : (30000 + idx * 2);
   }
 
+  s->session_seq_id = -1;
   s->latest_seq_id[MTL_SESSION_PORT_P] = -1;
   s->latest_seq_id[MTL_SESSION_PORT_R] = -1;
   s->tmstamp = -1;
@@ -462,23 +472,32 @@ static void rx_ancillary_session_stat(struct st_rx_ancillary_session_impl* s) {
 
   rte_atomic32_set(&s->stat_frames_received, 0);
 
-  notice("RX_ANC_SESSION(%d:%s): fps %f frames %d pkts %d\n", idx, s->ops_name, framerate,
-         frames_received, s->stat_pkts_received);
-  s->stat_pkts_received = 0;
-  s->stat_last_time = cur_time_ns;
-
   if (s->stat_pkts_redundant) {
-    notice("RX_ANC_SESSION(%d): redundant pkts %d\n", idx, s->stat_pkts_redundant);
+    notice("RX_ANC_SESSION(%d:%s): fps %f frames %d pkts %d (redundant %d)\n", idx,
+           s->ops_name, framerate, frames_received, s->stat_pkts_received,
+           s->stat_pkts_redundant);
     s->stat_pkts_redundant = 0;
+  } else {
+    notice("RX_ANC_SESSION(%d:%s): fps %f frames %d pkts %d\n", idx, s->ops_name,
+           framerate, frames_received, s->stat_pkts_received);
   }
+  s->stat_pkts_received = 0;
+  s->stat_last_time = cur_time_ns;
+
   if (s->stat_pkts_dropped) {
     notice("RX_ANC_SESSION(%d): dropped pkts %d\n", idx, s->stat_pkts_dropped);
     s->stat_pkts_dropped = 0;
   }
   if (s->stat_pkts_out_of_order) {
-    warn("RX_ANC_SESSION(%d): out of order pkts %d\n", idx, s->stat_pkts_out_of_order);
+    warn("RX_ANC_SESSION(%d): out of order pkts %d (%d:%d)\n", idx,
+         s->stat_pkts_out_of_order,
+         s->stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_P],
+         s->stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_R]);
     s->stat_pkts_out_of_order = 0;
+    s->stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_P] = 0;
+    s->stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_R] = 0;
   }
+
   if (s->stat_pkts_wrong_pt_dropped) {
     notice("RX_ANC_SESSION(%d): wrong hdr payload_type dropped pkts %d\n", idx,
            s->stat_pkts_wrong_pt_dropped);
@@ -546,6 +565,8 @@ static int rx_ancillary_session_update_src(struct mtl_main_impl* impl,
     s->st40_dst_port[i] = (ops->udp_port[i]) ? (ops->udp_port[i]) : (30000 + idx * 2);
   }
   /* reset seq id */
+
+  s->session_seq_id = -1;
   s->latest_seq_id[MTL_SESSION_PORT_P] = -1;
   s->latest_seq_id[MTL_SESSION_PORT_R] = -1;
   s->tmstamp = -1;
diff --git a/lib/src/st2110/st_rx_audio_session.c b/lib/src/st2110/st_rx_audio_session.c
@@ -292,32 +292,42 @@ static int rx_audio_session_handle_frame_pkt(struct mtl_main_impl* impl,
     s->first_pkt_rtp_ts = tmstamp;
   }
 
+  if (unlikely(s->latest_seq_id[s_port] == -1)) s->latest_seq_id[s_port] = seq_id - 1;
+  if (unlikely(s->session_seq_id == -1)) s->session_seq_id = seq_id - 1;
   if (unlikely(s->tmstamp == -1)) s->tmstamp = tmstamp - 1;
 
-  /* all packets need to have increasing tmstamp */
+  /* redundant stream seq_id out of order is not a big deal as long as stream is continous
+   */
+  if (seq_id != (uint16_t)(s->latest_seq_id[s_port] + 1)) {
+    dbg("%s(%d,%d), non-continuous seq now %u last %d\n", __func__, s->idx, s_port,
+        seq_id, s->latest_seq_id[s_port]);
+    s->port_user_stats.common.port[s_port].out_of_order_packets++;
+    s->stat_pkts_out_of_order_per_port[s_port]++;
+  }
+  s->latest_seq_id[s_port] = seq_id;
+
+  /* all packets need to have increasing timestamp */
   if (!mt_seq32_greater(tmstamp, s->tmstamp)) {
     dbg("%s(%d,%d), drop as pkt seq_id %u (%u) or tmstamp %u (%ld) is old\n", __func__,
         s->idx, s_port, seq_id, s->latest_seq_id[s_port], tmstamp, s->tmstamp);
+    s->stat_pkts_redundant++;
     ST_SESSION_STAT_INC(s, port_user_stats, stat_pkts_redundant);
-    s->latest_seq_id[s_port] = seq_id;
     return -EIO;
   }
   s->tmstamp = tmstamp;
 
-  if (unlikely(s->latest_seq_id[MTL_PORT_P] == -1))
-    s->latest_seq_id[MTL_PORT_P] = seq_id - 1;
-  if (unlikely(s->latest_seq_id[MTL_PORT_R] == -1))
-    s->latest_seq_id[MTL_PORT_R] = seq_id - 1;
-
-  if (seq_id != (uint16_t)(s->latest_seq_id[s_port] + 1)) {
-    dbg("%s(%d,%d), non-continuous seq now %u last %d\n", __func__, s->idx, s_port,
-        seq_id, s->latest_seq_id[s_port]);
+  /* hole in seq id packets going into the session check if the seq_id of the session is
+   * consistent */
+  if (seq_id != (uint16_t)(s->session_seq_id + 1)) {
+    dbg("%s(%d,%d), session seq_id %u out of order %d\n", __func__, s->idx, s_port,
+        seq_id, s->session_seq_id);
+    s->stat_pkts_out_of_order++;
     ST_SESSION_STAT_INC(s, port_user_stats.common, stat_pkts_out_of_order);
   }
 
   /* The package is accepted and goes into the frame */
-  /* update seq id */
-  s->latest_seq_id[s_port] = seq_id;
+
+  s->session_seq_id = seq_id;
 
   if (!s->st30_cur_frame) {
     s->st30_cur_frame = rx_audio_session_get_frame(s);
@@ -423,6 +433,7 @@ static int rx_audio_session_handle_rtp_pkt(struct mtl_main_impl* impl,
 
   uint16_t seq_id = ntohs(rtp->seq_number);
   uint8_t payload_type = rtp->payload_type;
+  uint32_t tmstamp = ntohl(rtp->tmstamp);
 
   if (ops->payload_type && (payload_type != ops->payload_type)) {
     dbg("%s(%d,%d), get payload_type %u but expect %u\n", __func__, s->idx, s_port,
@@ -441,23 +452,40 @@ static int rx_audio_session_handle_rtp_pkt(struct mtl_main_impl* impl,
   }
 
   if (unlikely(s->latest_seq_id[s_port] == -1)) s->latest_seq_id[s_port] = seq_id - 1;
+  if (unlikely(s->session_seq_id == -1)) s->session_seq_id = seq_id - 1;
+  if (unlikely(s->tmstamp == -1)) s->tmstamp = tmstamp - 1;
+
+  /* redundant stream seq_id out of order is not a big deal as long as stream is continous
+   */
   if (seq_id != (uint16_t)(s->latest_seq_id[s_port] + 1)) {
     dbg("%s(%d,%d), non-continuous seq now %u last %d\n", __func__, s->idx, s_port,
         seq_id, s->latest_seq_id[s_port]);
-    ST_SESSION_STAT_INC(s, port_user_stats.common, stat_pkts_out_of_order);
+    s->port_user_stats.common.port[s_port].out_of_order_packets++;
+    s->stat_pkts_out_of_order_per_port[s_port]++;
   }
+  s->latest_seq_id[s_port] = seq_id;
 
-  /* Only works for redundant streams basicly drop packets that are old */
-  if (st_rx_seq_redundant_drop(seq_id, s->latest_seq_id, s_port, s->ops.num_port)) {
-    dbg("%s(%d,%d), drop as pkt seq %d is old\n", __func__, s->idx, s_port, seq_id);
-
+  /* all packets need to have increasing timestamp */
+  if (!mt_seq32_greater(tmstamp, s->tmstamp)) {
+    dbg("%s(%d,%d), drop as pkt seq_id %u (%u) or tmstamp %u (%ld) is old\n", __func__,
+        s->idx, s_port, seq_id, s->latest_seq_id[s_port], tmstamp, s->tmstamp);
+    s->stat_pkts_redundant++;
     ST_SESSION_STAT_INC(s, port_user_stats, stat_pkts_redundant);
-    s->latest_seq_id[s_port] = seq_id;
     return -EIO;
   }
+  s->tmstamp = tmstamp;
 
-  /* update seq id */
-  s->latest_seq_id[s_port] = seq_id;
+  /* hole in seq id packets going into the session check if the seq_id of the session is
+   * consistent */
+  if (seq_id != (uint16_t)(s->session_seq_id + 1)) {
+    dbg("%s(%d,%d), session seq_id %u out of order %d\n", __func__, s->idx, s_port,
+        seq_id, s->session_seq_id);
+    s->stat_pkts_out_of_order++;
+    ST_SESSION_STAT_INC(s, port_user_stats.common, stat_pkts_out_of_order);
+  }
+
+  /* The package is accepted and goes into the frame */
+  s->session_seq_id = seq_id;
 
   /* enqueue the packet ring to app */
   int ret = rte_ring_sp_enqueue(s->st30_rtps_ring, (void*)mbuf);
@@ -605,10 +633,6 @@ static int rx_audio_session_tasklet(struct st_rx_audio_session_impl* s) {
     rv = mt_rxq_burst(s->rxq[s_port], &mbuf[0], ST_RX_AUDIO_BURST_SIZE);
     if (!rv) continue;
 
-    if (s_port == 1) {
-      dbg("DEBUG");
-    }
-
     rx_audio_session_handle_mbuf(&s->priv[s_port], &mbuf[0], rv);
     rte_pktmbuf_free_bulk(&mbuf[0], rv);
     if (s->enable_timing_parser && s->tp) {
@@ -821,6 +845,7 @@ static int rx_audio_session_attach(struct mtl_main_impl* impl,
   s->st30_pkt_idx = 0;
   s->st30_frame_size = ops->framebuff_size;
 
+  s->session_seq_id = -1;
   s->latest_seq_id[MTL_SESSION_PORT_P] = -1;
   s->latest_seq_id[MTL_SESSION_PORT_R] = -1;
   s->tmstamp = -1;
@@ -897,21 +922,28 @@ static void rx_audio_session_stat(struct st_rx_audio_sessions_mgr* mgr,
 
   rte_atomic32_set(&s->stat_frames_received, 0);
 
-  notice("RX_AUDIO_SESSION(%d,%d:%s): fps %f frames %d pkts %d\n", m_idx, idx,
-         s->ops_name, framerate, frames_received, s->stat_pkts_received);
-  s->stat_pkts_received = 0;
-  s->stat_last_time = cur_time_ns;
-
   if (s->stat_pkts_redundant) {
-    notice("RX_AUDIO_SESSION(%d,%d): redundant pkts %d\n", m_idx, idx,
+    notice("RX_AUDIO_SESSION(%d,%d:%s): fps %f frames %d pkts %d (redundant %d)\n", m_idx,
+           idx, s->ops_name, framerate, frames_received, s->stat_pkts_received,
            s->stat_pkts_redundant);
     s->stat_pkts_redundant = 0;
+  } else {
+    info("RX_AUDIO_SESSION(%d,%d:%s): fps %f frames %d pkts %d\n", m_idx, idx,
+         s->ops_name, framerate, frames_received, s->stat_pkts_received);
   }
+
+  s->stat_pkts_received = 0;
+  s->stat_last_time = cur_time_ns;
   if (s->stat_pkts_out_of_order) {
-    warn("RX_AUDIO_SESSION(%d,%d): out of order pkts %d\n", m_idx, idx,
-         s->stat_pkts_out_of_order);
+    warn("RX_AUDIO_SESSION(%d): out of order pkts %d (%d:%d)\n", idx,
+         s->stat_pkts_out_of_order,
+         s->stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_P],
+         s->stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_R]);
     s->stat_pkts_out_of_order = 0;
+    s->stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_P] = 0;
+    s->stat_pkts_out_of_order_per_port[MTL_SESSION_PORT_R] = 0;
   }
+
   if (s->stat_pkts_dropped) {
     notice("RX_AUDIO_SESSION(%d,%d): dropped pkts %d\n", m_idx, idx,
            s->stat_pkts_dropped);
@@ -989,6 +1021,8 @@ static int rx_audio_session_update_src(struct mtl_main_impl* impl,
     s->st30_dst_port[i] = (ops->udp_port[i]) ? (ops->udp_port[i]) : (20000 + idx * 2);
   }
   /* reset seq id */
+
+  s->session_seq_id = -1;
   s->latest_seq_id[MTL_SESSION_PORT_P] = -1;
   s->latest_seq_id[MTL_SESSION_PORT_R] = -1;
   s->tmstamp = -1;
diff --git a/lib/src/st2110/st_rx_fastmetadata_session.c b/lib/src/st2110/st_rx_fastmetadata_session.c
diff --git a/lib/src/st2110/st_rx_video_session.c b/lib/src/st2110/st_rx_video_session.c
