pipeline: add runtime update src and dst support (#629)

Signed-off-by: Frank Du <[email protected]>

doc/design.md

@@ -336,7 +336,8 @@ Enabling redundancy is straightforward with the setup parameters in the `st**_tx
 
 In iMTL, each field is treated as an individual frame, with fields being transmitted separately over the network, to avoid the need for new APIs specifically for fields. It is the application’s responsibility to recombine the fields into a full frame.
 
-For instance, with a 1080i50 format, you should use the following session parameters when creating a session: `interlaced: true, width: 1920, height: 1080, fps: ST_FPS_P50`. It is important to note that the height parameter should reflect the full frame height of 1080, not half height. However, keep in mind that the buffer for each field (or frame) will only contain data for 540 lines.
+For instance, with a 1080i50 format, you should use the following session parameters when creating a session: `interlaced: true, width: 1920, height: 1080, fps: ST_FPS_P50`. It is important to note that the height parameter should reflect the full frame height of 1080, not half height, FPS is fields per second.
+However, keep in mind that the buffer for each field (or frame) will only contain data for 540 lines.
 
 For transmission (TX), users can specify whether the current field is the first or second by using the `second_field` flag within the `struct st20_tx_frame_meta`. Similarly, for reception (RX), applications can determine the field order with the `second_field` flag present in the `struct st20_rx_frame_meta`.
 In the case of pipeline mode, the bool `second_field` within the `struct st_frame` also communicates field information between the application and IMTL.
@@ -383,7 +384,26 @@ The ST**_TX_FLAG_USER_TIMESTAMP flag is provided to enable applications to use t
 IMTL includes an array of built-in SIMD converters, providing high-performance data processing. The implementation details for these converters are available in the IMTL library source files [st_avx512.c](../lib/src/st2110/st_avx512.c) and [st_avx512_vbmi.c](../lib/src/st2110/st_avx512_vbmi.c).
 The API for these converters is publicly documented in the header file [st_convert_api.h](../include/st_convert_api.h). For more comprehensive information and instructions on using these converters, please refer to the [convert guide](./convert.md).
 
-### 6.13 Ecosystem
+### 6.13 Runtime update source and destination
+
+To offer significant flexibility in switch/forward scenarios, it is advantageous for a session to be able to dynamically change the source or destination address at runtime, thereby obviating the need for applications to recreate a session. The IMTL API below provides the capability to reconfigure the target address during runtime:
+
+```bash
+  st20_tx_update_destination
+  st20_rx_update_source
+  st22_tx_update_destination
+  st22_rx_update_source
+  st30_tx_update_destination
+  st30_rx_update_source
+  st40_tx_update_destination
+  st40_rx_update_source
+  st20p_tx_update_destination
+  st20p_rx_update_source
+  st22p_tx_update_destination
+  st22p_rx_update_source
+```
+
+### 6.14 Ecosystem
 
 The IMTL provides comprehensive documentation that includes reference code, demonstrating the seamless integration with popular media frameworks. Currently, IMTL supports the following plugins and SDKs:
 
@@ -393,6 +413,6 @@ OBS plugin: Streamline live streaming workflow in OBS (Open Broadcaster Software
 
 Intel® Media SDK: Leverage IMTL's robust capabilities within Intel® Media SDK projects to unlock advanced media functionalities on Intel platforms. Detail please refer to [Intel®_Media_SDK_guide](../ecosystem/msdk/).
 
-### 6.14 Sample code
+### 6.15 Sample code
 
 In addition to the built-in RxTxApp, IMTL also provides numerous sample codes that demonstrate how to construct simple test programs using its APIs. For more details, please refer to [sample](../app/sample/). We also provide some very useful forward application demo, detail can be found at [fwd](../app/sample/fwd/).

doc/performance.md

@@ -39,3 +39,25 @@ config | max-sessions | comment
 4320p59_4Tx | 8 | 4port on 4NIC
 4320p59_Tx + Rx​[Tx (Socket0), Rx (Socket1)] | 2TX+2RX | 2port on 2NIC
 4320p59_2Tx + 2Rx​[Tx (Socket0, Socket1), Rx (Socket1, Socket0)] | 4TX+4RX | 4port on 4NIC
+
+## 3. Measure DDR performance
+
+Intel® Memory Latency Checker (Intel® MLC) <https://www.intel.com/content/www/us/en/download/736633/intel-memory-latency-checker-intel-mlc.html> is a tool used to measure memory latencies and b/w, and how they change with increasing load on the system.
+
+The performance of IMTL is bound by memory throughput, with dependencies on DDR bandwidth and Last Level Cache (LLC). If you experience issues related to core density, it is recommended to run `Intel® Memory Latency Checker (Intel® MLC)` to diagnose any memory system bottlenecks, such as unpopulated DDR DIMM slots.
+
+Below is the output from Intel® MLC Bandwidths on our reference setup:
+
+```bash
+./Linux/mlc
+
+Measuring Peak Injection Memory Bandwidths for the system
+Bandwidths are in MB/sec (1 MB/sec = 1,000,000 Bytes/sec)
+Using all the threads from each core if Hyper-threading is enabled
+Using traffic with the following read-write ratios
+ALL Reads        :      223526.3
+3:1 Reads-Writes :      205940.2
+2:1 Reads-Writes :      203939.4
+1:1 Reads-Writes :      193479.6
+Stream-triad like:      182056.7
+```

include/st_api.h

@@ -131,9 +131,9 @@ MTL_PACK(struct st_rfc3550_rtp_hdr {
  */
 struct st_tx_dest_info {
   /** destination IP address of sender */
-  uint8_t dip_addr[MTL_PORT_MAX][MTL_IP_ADDR_LEN];
+  uint8_t dip_addr[MTL_SESSION_PORT_MAX][MTL_IP_ADDR_LEN];
   /** UDP port number */
-  uint16_t udp_port[MTL_PORT_MAX];
+  uint16_t udp_port[MTL_SESSION_PORT_MAX];
 };
 
 /**
@@ -142,9 +142,11 @@ struct st_tx_dest_info {
  */
 struct st_rx_source_info {
   /** source IP address of sender */
-  uint8_t sip_addr[MTL_PORT_MAX][MTL_IP_ADDR_LEN];
+  uint8_t sip_addr[MTL_SESSION_PORT_MAX][MTL_IP_ADDR_LEN];
   /** UDP port number */
-  uint16_t udp_port[MTL_PORT_MAX];
+  uint16_t udp_port[MTL_SESSION_PORT_MAX];
+  /** Optional. source filter IP address of multicast */
+  uint8_t mcast_sip_addr[MTL_SESSION_PORT_MAX][MTL_IP_ADDR_LEN];
 };
 
 /**

include/st_pipeline_api.h

@@ -1311,6 +1311,19 @@ void* st22p_tx_get_fb_addr(st22p_tx_handle handle, uint16_t idx);
  */
 size_t st22p_tx_frame_size(st22p_tx_handle handle);
 
+/**
+ * Online update the destination info for the tx st2110-22(pipeline) session.
+ *
+ * @param handle
+ *   The handle to the tx st2110-22(pipeline) session.
+ * @param dst
+ *   The pointer to the tx st2110-22(pipeline) destination info.
+ * @return
+ *   - 0: Success.
+ *   - <0: Error code.
+ */
+int st22p_tx_update_destination(st22p_tx_handle handle, struct st_tx_dest_info* dst);
+
 /**
  * Create one rx st2110-22 pipeline session.
  *
@@ -1417,6 +1430,19 @@ int st22p_rx_pcapng_dump(st22p_rx_handle handle, uint32_t max_dump_packets, bool
  */
 int st22p_rx_get_queue_meta(st22p_rx_handle handle, struct st_queue_meta* meta);
 
+/**
+ * Online update the source info for the rx st2110-22(pipeline) session.
+ *
+ * @param handle
+ *   The handle to the rx st2110-22(pipeline) session.
+ * @param src
+ *   The pointer to the rx st2110-22(pipeline) source info.
+ * @return
+ *   - 0: Success.
+ *   - <0: Error code.
+ */
+int st22p_rx_update_source(st22p_rx_handle handle, struct st_rx_source_info* src);
+
 /**
  * Create one tx st2110-20 pipeline session.
  *
@@ -1549,6 +1575,19 @@ int st20p_tx_get_port_stats(st20p_tx_handle handle, enum mtl_session_port port,
  */
 int st20p_tx_reset_port_stats(st20p_tx_handle handle, enum mtl_session_port port);
 
+/**
+ * Online update the destination info for the tx st2110-20(pipeline) session.
+ *
+ * @param handle
+ *   The handle to the tx st2110-20(pipeline) session.
+ * @param dst
+ *   The pointer to the tx st2110-20(pipeline) destination info.
+ * @return
+ *   - 0: Success.
+ *   - <0: Error code.
+ */
+int st20p_tx_update_destination(st20p_tx_handle handle, struct st_tx_dest_info* dst);
+
 /**
  * Create one rx st2110-20 pipeline session.
  *
@@ -1695,6 +1734,19 @@ int st20p_rx_get_port_stats(st20p_rx_handle handle, enum mtl_session_port port,
  */
 int st20p_rx_reset_port_stats(st20p_rx_handle handle, enum mtl_session_port port);
 
+/**
+ * Online update the source info for the rx st2110-20(pipeline) session.
+ *
+ * @param handle
+ *   The handle to the rx st2110-20(pipeline) session.
+ * @param src
+ *   The pointer to the rx st2110-20(pipeline) source info.
+ * @return
+ *   - 0: Success.
+ *   - <0: Error code.
+ */
+int st20p_rx_update_source(st20p_rx_handle handle, struct st_rx_source_info* src);
+
 /**
  * Convert color format from source frame to destination frame.
  *

lib/src/mt_sch.c

@@ -379,8 +379,8 @@ static int sch_stat(void* priv) {
 
   if (!mt_sch_is_active(sch)) return 0;
 
-  notice("SCH(%d:%s): tasklets %d max idx %d\n", idx, sch->name, num_tasklet,
-         sch->max_tasklet_idx);
+  notice("SCH(%d:%s): tasklets %d max idx %d, lcore %u\n", idx, sch->name, num_tasklet,
+         sch->max_tasklet_idx, sch->lcore);
   if (mt_user_tasklet_time_measure(sch->parent)) {
     for (int i = 0; i < num_tasklet; i++) {
       tasklet = sch->tasklet[i];

lib/src/st2110/pipeline/st20_pipeline_rx.c

@@ -930,3 +930,15 @@ int st20p_rx_reset_port_stats(st20p_rx_handle handle, enum mtl_session_port port
 
   return st20_rx_reset_port_stats(ctx->transport, port);
 }
+
+int st20p_rx_update_source(st20p_rx_handle handle, struct st_rx_source_info* src) {
+  struct st20p_rx_ctx* ctx = handle;
+  int cidx = ctx->idx;
+
+  if (ctx->type != MT_ST20_HANDLE_PIPELINE_RX) {
+    err("%s(%d), invalid type %d\n", __func__, cidx, ctx->type);
+    return 0;
+  }
+
+  return st20_rx_update_source(ctx->transport, src);
+}

lib/src/st2110/pipeline/st20_pipeline_tx.c

@@ -815,3 +815,15 @@ int st20p_tx_reset_port_stats(st20p_tx_handle handle, enum mtl_session_port port
 
   return st20_tx_reset_port_stats(ctx->transport, port);
 }
+
+int st20p_tx_update_destination(st20p_tx_handle handle, struct st_tx_dest_info* dst) {
+  struct st20p_tx_ctx* ctx = handle;
+  int cidx = ctx->idx;
+
+  if (ctx->type != MT_ST20_HANDLE_PIPELINE_TX) {
+    err("%s(%d), invalid type %d\n", __func__, cidx, ctx->type);
+    return 0;
+  }
+
+  return st20_tx_update_destination(ctx->transport, dst);
+}

lib/src/st2110/pipeline/st22_pipeline_rx.c

@@ -644,3 +644,15 @@ int st22p_rx_pcapng_dump(st22p_rx_handle handle, uint32_t max_dump_packets, bool
 
   return st22_rx_pcapng_dump(ctx->transport, max_dump_packets, sync, meta);
 }
+
+int st22p_rx_update_source(st22p_rx_handle handle, struct st_rx_source_info* src) {
+  struct st22p_rx_ctx* ctx = handle;
+  int cidx = ctx->idx;
+
+  if (ctx->type != MT_ST22_HANDLE_PIPELINE_RX) {
+    err("%s(%d), invalid type %d\n", __func__, cidx, ctx->type);
+    return -EIO;
+  }
+
+  return st22_rx_update_source(ctx->transport, src);
+}

lib/src/st2110/pipeline/st22_pipeline_tx.c

@@ -692,3 +692,15 @@ size_t st22p_tx_frame_size(st22p_tx_handle handle) {
 
   return ctx->src_size;
 }
+
+int st22p_tx_update_destination(st22p_tx_handle handle, struct st_tx_dest_info* dst) {
+  struct st22p_tx_ctx* ctx = handle;
+  int cidx = ctx->idx;
+
+  if (ctx->type != MT_ST22_HANDLE_PIPELINE_TX) {
+    err("%s(%d), invalid type %d\n", __func__, cidx, ctx->type);
+    return 0;
+  }
+
+  return st22_tx_update_destination(ctx->transport, dst);
+}

lib/src/st2110/st_rx_ancillary_session.c

@@ -439,6 +439,7 @@ static int rx_ancillary_session_update_src(struct mtl_main_impl* impl,
   /* update ip and port */
   for (int i = 0; i < num_port; i++) {
     memcpy(ops->sip_addr[i], src->sip_addr[i], MTL_IP_ADDR_LEN);
+    memcpy(ops->mcast_sip_addr[i], src->mcast_sip_addr[i], MTL_IP_ADDR_LEN);
     ops->udp_port[i] = src->udp_port[i];
     s->st40_dst_port[i] = (ops->udp_port[i]) ? (ops->udp_port[i]) : (30000 + idx * 2);
   }

lib/src/st2110/st_rx_audio_session.c

@@ -882,6 +882,7 @@ static int rx_audio_session_update_src(struct mtl_main_impl* impl,
   /* update ip and port */
   for (int i = 0; i < num_port; i++) {
     memcpy(ops->sip_addr[i], src->sip_addr[i], MTL_IP_ADDR_LEN);
+    memcpy(ops->mcast_sip_addr[i], src->mcast_sip_addr[i], MTL_IP_ADDR_LEN);
     ops->udp_port[i] = src->udp_port[i];
     s->st30_dst_port[i] = (ops->udp_port[i]) ? (ops->udp_port[i]) : (20000 + idx * 2);
   }

lib/src/st2110/st_rx_video_session.c

@@ -3386,6 +3386,7 @@ static int rv_update_src(struct st_rx_video_sessions_mgr* mgr,
   /* update ip and port */
   for (int i = 0; i < num_port; i++) {
     memcpy(ops->sip_addr[i], src->sip_addr[i], MTL_IP_ADDR_LEN);
+    memcpy(ops->mcast_sip_addr[i], src->mcast_sip_addr[i], MTL_IP_ADDR_LEN);
     ops->udp_port[i] = src->udp_port[i];
     s->st20_dst_port[i] = (ops->udp_port[i]) ? (ops->udp_port[i]) : (10000 + idx * 2);
   }