st2110: add ssrc support

CHANGELOG.md

@@ -17,6 +17,8 @@
 * backend/kernel: add multi thread support for both TX and RX.
 * convert: add interlace support.
 * rtcp: add retransmit packet support for st20 sessions, see STxx_RX_FLAG_ENABLE_RTCP.
+* st2110: add ssrc support.
+* rss: add multi-core support.
 
 ## Changelog for 23.08
 

include/st20_api.h

@@ -972,9 +972,12 @@ struct st20_tx_ops {
   enum st_fps fps;
   /** Mandatory. Session resolution format */
   enum st20_fmt fmt;
-  /** Mandatory. 7 bits payload type define in RFC3550 */
+  /** Mandatory. 7 bits payload type defined in RFC3550 */
   uint8_t payload_type;
 
+  /** Optional. Synchronization source defined in RFC3550, if zero the session will assign
+   * a random value */
+  uint32_t ssrc;
   /** Optional. Name */
   const char* name;
   /** Optional. Private data to the cb functions(get_next_frame and others) */
@@ -1118,6 +1121,9 @@ struct st22_tx_ops {
   /** Mandatory. 7 bits payload type define in RFC3550 */
   uint8_t payload_type;
 
+  /** Optional. Synchronization source defined in RFC3550, if zero the session will assign
+   * a random value */
+  uint32_t ssrc;
   /** Optional. Name */
   const char* name;
   /** Optional. Private data to the cb functions(get_next_frame and others) */
@@ -1286,6 +1292,9 @@ struct st20_rx_ops {
   /** Mandatory. 7 bits payload type define in RFC3550 */
   uint8_t payload_type;
 
+  /** Optional. Synchronization source defined in RFC3550, RX session will check the
+   * incoming RTP packets match the ssrc. Leave to zero to disable the ssrc check */
+  uint32_t ssrc;
   /** Optional. Not in use now as RX support all pacing type, reserved for future */
   enum st21_pacing pacing;
   /** Optional. Not in use now as RX support all packing type, reserved for future */
@@ -1434,6 +1443,9 @@ struct st22_rx_ops {
   /** Mandatory. 7 bits payload type define in RFC3550 */
   uint8_t payload_type;
 
+  /** Optional. Synchronization source defined in RFC3550, RX session will check the
+   * incoming RTP packets match the ssrc. Leave to zero to disable the ssrc check */
+  uint32_t ssrc;
   /** Optional. Name */
   const char* name;
   /** Optional. Private data to the cb functions(notify_frame_ready and others) */

include/st20_redundant_api.h

@@ -88,6 +88,10 @@ struct st20r_rx_ops {
   bool interlaced;
   /** 7 bits payload type define in RFC3550 */
   uint8_t payload_type;
+  /** Optional. Synchronization source defined in RFC3550, RX session will check the
+   * incoming RTP packets match the ssrc. Leave to zero to disable the ssrc check */
+  uint32_t ssrc;
+
   /** flags, value in ST20R_RX_FLAG_* */
   uint32_t flags;
   /**

include/st30_api.h

@@ -271,6 +271,9 @@ struct st30_tx_ops {
   /** Mandatory. 7 bits payload type define in RFC3550 */
   uint8_t payload_type;
 
+  /** Optional. Synchronization source defined in RFC3550, if zero the session will assign
+   * a random value */
+  uint32_t ssrc;
   /** Optional. name */
   const char* name;
   /** Optional. private data to the callback function */
@@ -368,6 +371,9 @@ struct st30_rx_ops {
   /** Mandatory. 7 bits payload type define in RFC3550 */
   uint8_t payload_type;
 
+  /** Optional. Synchronization source defined in RFC3550, RX session will check the
+   * incoming RTP packets match the ssrc. Leave to zero to disable the ssrc check */
+  uint32_t ssrc;
   /** Optional. name */
   const char* name;
   /** Optional. private data to the callback function */

include/st40_api.h

@@ -242,6 +242,9 @@ struct st40_tx_ops {
   /** Mandatory. 7 bits payload type define in RFC3550 */
   uint8_t payload_type;
 
+  /** Optional. Synchronization source defined in RFC3550, if zero the session will assign
+   * a random value */
+  uint32_t ssrc;
   /** Optional. name */
   const char* name;
   /** Optional. private data to the callback function */
@@ -307,6 +310,9 @@ struct st40_rx_ops {
   /** Mandatory. 7 bits payload type define in RFC3550 */
   uint8_t payload_type;
 
+  /** Optional. Synchronization source defined in RFC3550, RX session will check the
+   * incoming RTP packets match the ssrc. Leave to zero to disable the ssrc check */
+  uint32_t ssrc;
   /** Optional. name */
   const char* name;
   /** Optional. private data to the callback function */

include/st_pipeline_api.h

@@ -705,6 +705,9 @@ struct st_tx_port {
   uint8_t payload_type;
   /** Optional. UDP source port number, leave as 0 to use same port as dst */
   uint16_t udp_src_port[MTL_SESSION_PORT_MAX];
+  /** Optional. Synchronization source defined in RFC3550, if zero the session will assign
+   * a random value */
+  uint32_t ssrc;
 };
 
 /** The structure info for st rx port, used in creating session. */
@@ -719,6 +722,9 @@ struct st_rx_port {
   uint16_t udp_port[MTL_SESSION_PORT_MAX];
   /** Mandatory. 7 bits payload type define in RFC3550 */
   uint8_t payload_type;
+  /** Optional. Synchronization source defined in RFC3550, RX session will check the
+   * incoming RTP packets match the ssrc. Leave to zero to disable the ssrc check */
+  uint32_t ssrc;
 };
 
 /** The structure describing how to create a tx st2110-20 pipeline session. */

lib/src/st2110/pipeline/st20_pipeline_rx.c

@@ -421,6 +421,7 @@ static int rx_st20p_create_transport(struct mtl_main_impl* impl, struct st20p_rx
   ops_rx.interlaced = ops->interlaced;
   ops_rx.linesize = ops->transport_linesize;
   ops_rx.payload_type = ops->port.payload_type;
+  ops_rx.ssrc = ops->port.ssrc;
   ops_rx.type = ST20_TYPE_FRAME_LEVEL;
   ops_rx.framebuff_cnt = ops->framebuff_cnt;
   ops_rx.notify_frame_ready = rx_st20p_frame_ready;

lib/src/st2110/pipeline/st20_pipeline_tx.c

@@ -263,6 +263,7 @@ static int tx_st20p_create_transport(struct mtl_main_impl* impl, struct st20p_tx
   ops_tx.interlaced = ops->interlaced;
   ops_tx.linesize = ops->transport_linesize;
   ops_tx.payload_type = ops->port.payload_type;
+  ops_tx.ssrc = ops->port.ssrc;
   ops_tx.type = ST20_TYPE_FRAME_LEVEL;
   ops_tx.framebuff_cnt = ops->framebuff_cnt;
   ops_tx.get_next_frame = tx_st20p_next_frame;

lib/src/st2110/pipeline/st22_pipeline_rx.c

@@ -252,6 +252,7 @@ static int rx_st22p_create_transport(struct mtl_main_impl* impl, struct st22p_rx
   ops_rx.height = ops->height;
   ops_rx.fps = ops->fps;
   ops_rx.payload_type = ops->port.payload_type;
+  ops_rx.ssrc = ops->port.ssrc;
   ops_rx.type = ST22_TYPE_FRAME_LEVEL;
   ops_rx.pack_type = ops->pack_type;
   ops_rx.framebuff_cnt = ops->framebuff_cnt;

lib/src/st2110/pipeline/st22_pipeline_tx.c

@@ -258,6 +258,7 @@ static int tx_st22p_create_transport(struct mtl_main_impl* impl, struct st22p_tx
   ops_tx.height = ops->height;
   ops_tx.fps = ops->fps;
   ops_tx.payload_type = ops->port.payload_type;
+  ops_tx.ssrc = ops->port.ssrc;
   ops_tx.type = ST22_TYPE_FRAME_LEVEL;
   ops_tx.pack_type = ops->pack_type;
   ops_tx.framebuff_cnt = ops->framebuff_cnt;

lib/src/st2110/redundant/st20_redundant_rx.c

@@ -166,6 +166,7 @@ static int rx_st20r_create_transport(struct st20r_rx_ctx* ctx, struct st20r_rx_o
   ops_rx.fmt = ops->fmt;
   ops_rx.interlaced = ops->interlaced;
   ops_rx.payload_type = ops->payload_type;
+  ops_rx.ssrc = ops->ssrc;
   ops_rx.type = ST20_TYPE_FRAME_LEVEL;
   ops_rx.framebuff_cnt = ops->framebuff_cnt;
   ops_rx.notify_frame_ready = rx_st20r_frame_ready;

lib/src/st2110/st_rx_ancillary_session.c

@@ -98,6 +98,13 @@ static int rx_ancillary_session_handle_pkt(struct mtl_main_impl* impl,
     s->st40_stat_pkts_wrong_hdr_dropped++;
     return -EINVAL;
   }
+  if (ops->ssrc) {
+    uint32_t ssrc = ntohl(rtp->ssrc);
+    if (ssrc != ops->ssrc) {
+      s->st40_stat_pkts_wrong_hdr_dropped++;
+      return -EINVAL;
+    }
+  }
 
   /* set if it is first pkt */
   if (unlikely(s->latest_seq_id == -1)) s->latest_seq_id = seq_id - 1;

lib/src/st2110/st_rx_audio_session.c

@@ -399,6 +399,13 @@ static int rx_audio_session_handle_frame_pkt(struct mtl_main_impl* impl,
     s->st30_stat_pkts_wrong_hdr_dropped++;
     return -EINVAL;
   }
+  if (ops->ssrc) {
+    uint32_t ssrc = ntohl(rtp->ssrc);
+    if (ssrc != ops->ssrc) {
+      s->st30_stat_pkts_wrong_hdr_dropped++;
+      return -EINVAL;
+    }
+  }
 
   if (pkt_len != s->pkt_len) {
     dbg("%s(%d,%d), drop as pkt_len mismatch now %u expect %u\n", __func__, s->idx,
@@ -494,6 +501,13 @@ static int rx_audio_session_handle_rtp_pkt(struct mtl_main_impl* impl,
     s->st30_stat_pkts_wrong_hdr_dropped++;
     return -EINVAL;
   }
+  if (ops->ssrc) {
+    uint32_t ssrc = ntohl(rtp->ssrc);
+    if (ssrc != ops->ssrc) {
+      s->st30_stat_pkts_wrong_hdr_dropped++;
+      return -EINVAL;
+    }
+  }
 
   /* set first seq_id - 1 */
   if (unlikely(s->latest_seq_id == -1)) s->latest_seq_id = seq_id - 1;

lib/src/st2110/st_rx_video_session.c

@@ -1764,6 +1764,15 @@ static int rv_handle_frame_pkt(struct st_rx_video_session_impl* s, struct rte_mb
     s->stat_pkts_wrong_hdr_dropped++;
     return -EINVAL;
   }
+  if (ops->ssrc) {
+    uint32_t ssrc = ntohl(rtp->base.ssrc);
+    dbg("%s(%d,%d): expect ssrc %u actual %u\n", __func__, s->idx, s_port, ops->ssrc,
+        ssrc);
+    if (ssrc != ops->ssrc) {
+      s->stat_pkts_wrong_hdr_dropped++;
+      return -EINVAL;
+    }
+  }
   if (mbuf_next && mbuf_next->data_len) {
     /* for some reason mbuf splits into 2 segments (1024 bytes + left bytes) */
     /* todo: payload needs to be copied from 2 places */
@@ -2004,6 +2013,13 @@ static int rv_handle_rtp_pkt(struct st_rx_video_session_impl* s, struct rte_mbuf
     s->stat_pkts_wrong_hdr_dropped++;
     return -EINVAL;
   }
+  if (ops->ssrc) {
+    uint32_t ssrc = ntohl(rtp->ssrc);
+    if (ssrc != ops->ssrc) {
+      s->stat_pkts_wrong_hdr_dropped++;
+      return -EINVAL;
+    }
+  }
 
   /* find the target slot by tmstamp */
   struct st_rx_video_slot_impl* slot = rv_rtp_slot_by_tmstamp(s, tmstamp);
@@ -2146,6 +2162,13 @@ static int rv_handle_st22_pkt(struct st_rx_video_session_impl* s, struct rte_mbu
     s->stat_pkts_wrong_hdr_dropped++;
     return -EINVAL;
   }
+  if (ops->ssrc) {
+    uint32_t ssrc = ntohl(rtp->base.ssrc);
+    if (ssrc != ops->ssrc) {
+      s->stat_pkts_wrong_hdr_dropped++;
+      return -EINVAL;
+    }
+  }
 
   if (rtp->kmode) {
     s->stat_pkts_wrong_hdr_dropped++;
@@ -2300,6 +2323,13 @@ static int rv_handle_hdr_split_pkt(struct st_rx_video_session_impl* s,
     s->stat_pkts_wrong_hdr_dropped++;
     return -EINVAL;
   }
+  if (ops->ssrc) {
+    uint32_t ssrc = ntohl(rtp->base.ssrc);
+    if (ssrc != ops->ssrc) {
+      s->stat_pkts_wrong_hdr_dropped++;
+      return -EINVAL;
+    }
+  }
   if (!hdr_split->mbuf_pool_ready) {
     s->stat_pkts_no_slot++;
     return -EINVAL;
@@ -2747,6 +2777,13 @@ static int rv_handle_detect_pkt(struct st_rx_video_session_impl* s, struct rte_m
     s->stat_pkts_wrong_hdr_dropped++;
     return -EINVAL;
   }
+  if (ops->ssrc) {
+    uint32_t ssrc = ntohl(rtp->base.ssrc);
+    if (ssrc != ops->ssrc) {
+      s->stat_pkts_wrong_hdr_dropped++;
+      return -EINVAL;
+    }
+  }
 
   /* detect continuous bit */
   if (extra_rtp) detector->single_line = false;
@@ -4535,6 +4572,7 @@ st22_rx_handle st22_rx_create(mtl_handle mt, struct st22_rx_ops* ops) {
   st20_ops.fps = ops->fps;
   st20_ops.fmt = ST20_FMT_YUV_422_10BIT;
   st20_ops.payload_type = ops->payload_type;
+  st20_ops.ssrc = ops->ssrc;
   st20_ops.rtp_ring_size = ops->rtp_ring_size;
   st20_ops.notify_rtp_ready = ops->notify_rtp_ready;
   st20_ops.framebuff_cnt = ops->framebuff_cnt;

lib/src/st2110/st_tx_ancillary_session.c

@@ -186,15 +186,16 @@ static int tx_ancillary_session_init_hdr(struct mtl_main_impl* impl,
   rtp->base.payload_type = st_is_valid_payload_type(ops->payload_type)
                                ? ops->payload_type
                                : ST_RANCRTP_PAYLOAD_TYPE_ANCILLARY;
-  rtp->base.ssrc = htonl(s->idx + 0x323450);
+  uint32_t ssrc = ops->ssrc ? ops->ssrc : s->idx + 0x323450;
+  rtp->base.ssrc = htonl(ssrc);
   s->st40_seq_id = 0;
   s->st40_ext_seq_id = 0;
 
   info("%s(%d,%d), ip %u.%u.%u.%u port %u:%u\n", __func__, idx, s_port, dip[0], dip[1],
        dip[2], dip[3], s->st40_src_port[s_port], s->st40_dst_port[s_port]);
-  info("%s(%d), mac: %02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx\n", __func__, idx,
+  info("%s(%d), mac: %02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx, ssrc %u\n", __func__, idx,
        d_addr->addr_bytes[0], d_addr->addr_bytes[1], d_addr->addr_bytes[2],
-       d_addr->addr_bytes[3], d_addr->addr_bytes[4], d_addr->addr_bytes[5]);
+       d_addr->addr_bytes[3], d_addr->addr_bytes[4], d_addr->addr_bytes[5], ssrc);
   return 0;
 }
 

lib/src/st2110/st_tx_audio_session.c

@@ -187,14 +187,16 @@ static int tx_audio_session_init_hdr(struct mtl_main_impl* impl,
   rtp->payload_type = st_is_valid_payload_type(ops->payload_type)
                           ? ops->payload_type
                           : ST_RARTP_PAYLOAD_TYPE_PCM_AUDIO;
-  rtp->ssrc = htonl(s->idx + 0x223450);
+  uint32_t ssrc = ops->ssrc ? ops->ssrc : s->idx + 0x223450;
+  rtp->ssrc = htonl(ssrc);
+
   s->st30_seq_id = 0;
 
   info("%s(%d,%d), ip %u.%u.%u.%u port %u:%u\n", __func__, idx, s_port, dip[0], dip[1],
        dip[2], dip[3], s->st30_src_port[s_port], s->st30_dst_port[s_port]);
-  info("%s(%d), mac: %02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx\n", __func__, idx,
+  info("%s(%d), mac: %02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx, ssrc %u\n", __func__, idx,
        d_addr->addr_bytes[0], d_addr->addr_bytes[1], d_addr->addr_bytes[2],
-       d_addr->addr_bytes[3], d_addr->addr_bytes[4], d_addr->addr_bytes[5]);
+       d_addr->addr_bytes[3], d_addr->addr_bytes[4], d_addr->addr_bytes[5], ssrc);
   return 0;
 }
 

lib/src/st2110/st_tx_video_session.c

@@ -799,7 +799,8 @@ static int tv_init_hdr(struct mtl_main_impl* impl, struct st_tx_video_session_im
   rtp->base.payload_type = st_is_valid_payload_type(ops->payload_type)
                                ? ops->payload_type
                                : ST_RVRTP_PAYLOAD_TYPE_RAW_VIDEO;
-  rtp->base.ssrc = htonl(s->idx + 0x123450);
+  uint32_t ssrc = ops->ssrc ? ops->ssrc : s->idx + 0x123450;
+  rtp->base.ssrc = htonl(ssrc);
   rtp->row_length = htons(s->st20_pkt_len);
   rtp->row_number = 0;
   rtp->row_offset = 0;
@@ -817,9 +818,9 @@ static int tv_init_hdr(struct mtl_main_impl* impl, struct st_tx_video_session_im
 
   info("%s(%d,%d), ip %u.%u.%u.%u port %u:%u\n", __func__, idx, s_port, dip[0], dip[1],
        dip[2], dip[3], s->st20_src_port[s_port], s->st20_dst_port[s_port]);
-  info("%s(%d), mac: %02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx\n", __func__, idx,
+  info("%s(%d), mac: %02hhx:%02hhx:%02hhx:%02hhx:%02hhx:%02hhx, ssrc %u\n", __func__, idx,
        d_addr->addr_bytes[0], d_addr->addr_bytes[1], d_addr->addr_bytes[2],
-       d_addr->addr_bytes[3], d_addr->addr_bytes[4], d_addr->addr_bytes[5]);
+       d_addr->addr_bytes[3], d_addr->addr_bytes[4], d_addr->addr_bytes[5], ssrc);
   return 0;
 }
 
@@ -4228,6 +4229,7 @@ st22_tx_handle st22_tx_create(mtl_handle mt, struct st22_tx_ops* ops) {
   st20_ops.fmt = ST20_FMT_YUV_422_10BIT;
   st20_ops.framebuff_cnt = ops->framebuff_cnt;
   st20_ops.payload_type = ops->payload_type;
+  st20_ops.ssrc = ops->ssrc;
   st20_ops.rtp_ring_size = ops->rtp_ring_size;
   st20_ops.rtp_frame_total_pkts = ops->rtp_frame_total_pkts;
   st20_ops.rtp_pkt_size = ops->rtp_pkt_size;

tests/src/st20p_test.cpp

@@ -609,6 +609,7 @@ struct st20p_rx_digest_test_para {
   bool rtcp;
   enum st20_packing packing;
   enum st21_pacing pacing;
+  uint32_t ssrc;
 };
 
 static void test_st20p_init_rx_digest_para(struct st20p_rx_digest_test_para* para) {
@@ -634,6 +635,7 @@ static void test_st20p_init_rx_digest_para(struct st20p_rx_digest_test_para* par
   para->rtcp = false;
   para->packing = ST20_PACKING_BPM;
   para->pacing = ST21_PACING_NARROW;
+  para->ssrc = 0;
 }
 
 static void st20p_rx_digest_test(enum st_fps fps[], int width[], int height[],
@@ -730,6 +732,7 @@ static void st20p_rx_digest_test(enum st_fps fps[], int width[], int height[],
              ctx->para.port[MTL_PORT_P]);
     ops_tx.port.udp_port[MTL_SESSION_PORT_P] = ST20P_TEST_UDP_PORT + i * 2;
     ops_tx.port.payload_type = ST20P_TEST_PAYLOAD_TYPE;
+    ops_tx.port.ssrc = para->ssrc;
     ops_tx.width = width[i];
     ops_tx.height = height[i];
     ops_tx.fps = fps[i];
@@ -948,6 +951,7 @@ static void st20p_rx_digest_test(enum st_fps fps[], int width[], int height[],
              ctx->para.port[MTL_PORT_R]);
     ops_rx.port.udp_port[MTL_SESSION_PORT_P] = ST20P_TEST_UDP_PORT + i * 2;
     ops_rx.port.payload_type = ST20P_TEST_PAYLOAD_TYPE;
+    ops_rx.port.ssrc = para->ssrc;
     ops_rx.width = width[i];
     ops_rx.height = height[i];
     ops_rx.fps = fps[i];
@@ -1130,6 +1134,7 @@ TEST(St20p, digest_1080i_s2) {
   para.check_fps = false;
   para.interlace = true;
   para.device = ST_PLUGIN_DEVICE_TEST_INTERNAL;
+  para.ssrc = 54321;
 
   st20p_rx_digest_test(fps, width, height, tx_fmt, t_fmt, rx_fmt, &para);
 }