Add: Ability to drop old frames

Change the way pipeline TX handles the order of framebuffers. Track
per-frame sequence numbers so transport can dequeue the newest converted
frame instead of always taking the next one, needed as otherwise
we cannot drop frames, as they will not be in order.
(This also makes it easier for multi threaded applications to use
the api should improve performance with multiple threads from the api
only).

Add a new flag to the TX pipeline to enable droping frames when we jump
over epoch. Mostly will aid with user control pacing.

Implemented in a way that only drops one frame in a scenario where late
is detected. This creates limitation for extreamly high drops The stream
will still not adjust immidiately instead doing
send frame | drop frame | send frame | drop frame ...
until we catch up.
(Done mostly as big jumps over more than 2 frames could be undesirable
and if you have that big of a drop ... well this is the least of your
troubles).

Due to lack of consumer and producer print stats of the queue via
reporting the framebuffers statuses (removed the ACCURATE_FRAMEBUFF) as
it became redundant.

Reverts: 772621d

All above done in pipeline as the purpose of the api pipeline is to
smoothe the library tight timing requirements from the user
applications.

Fix the st20p wrong user timestamp reporting in frames snapped to
the next frame.

if times given by the application looks as follows
----|--frame-1--|--frame-2--|--frame-3--|--frame-4--|-...
epoch-1--|--epoch-2--|--epoch-3--|--epoch-4--|...

This causes frames to snap to the next epoch, resulting in incorrect
timestamp reporting where every frame appears to have a "wrong user
timestamp" due to the forward snapping behavior.

Fixes: 88c3566

Fix the usdt for st40p tx (only to add the spurn event).

Fixes: 921fa0d

Author: DawidWesierski4

include/st20_api.h

@@ -1169,6 +1169,13 @@ struct st20_tx_ops {
   int (*notify_frame_done)(void* priv, uint16_t frame_idx,
                            struct st20_tx_frame_meta* meta);
 
+  /**
+   * Optional. Callback triggered when a frame epoch is omitted/skipped in the lib.
+   * This occurs when the transmission timing falls behind schedule and an epoch
+   * must be skipped to maintain synchronization. (or in the user pacing mode
+   * when the user time is behind the lib sending time).
+   */
+  int (*notify_frame_late)(void* priv, uint64_t epoch_skipped);
   /**
    * Optional. The event callback when there is some event(vsync or others) happened for
    * this session. Only non-block method can be used in this callback as it run from lcore
@@ -1319,6 +1326,14 @@ struct st22_tx_ops {
   int (*notify_frame_done)(void* priv, uint16_t frame_idx,
                            struct st22_tx_frame_meta* meta);
 
+  /**
+   * Optional. Callback triggered when a frame epoch is omitted/skipped in the lib.
+   * This occurs when the transmission timing falls behind schedule and an epoch
+   * must be skipped to maintain synchronization. (or in the user pacing mode
+   * when the user time is behind the lib sending time).
+   */
+  int (*notify_frame_late)(void* priv, uint64_t epoch_skipped);
+
   /**
    * Optional. The event callback when there is some event(vsync or others) happened for
    * this session. Only non-block method can be used in this callback as it run from lcore

include/st30_api.h

@@ -383,6 +383,13 @@ struct st30_tx_ops {
   int (*notify_frame_done)(void* priv, uint16_t frame_idx,
                            struct st30_tx_frame_meta* meta);
 
+  /**
+   * Optional. Callback when frame done in the lib.
+   * And only non-block method can be used within this callback as it run from lcore
+   * tasklet routine.
+   */
+  int (*notify_frame_late)(void* priv, uint64_t epoch_skipped);
+
   /*
    * Optional. The size of fifo ring which used between the packet builder and pacing.
    * Leave to zero to use default value: the packet number within

include/st30_pipeline_api.h

@@ -52,6 +52,13 @@ enum st30p_tx_flag {
   /** Enable the st30p_tx_get_frame block behavior to wait until a frame becomes
    available or timeout(default: 1s, use st30p_tx_set_block_timeout to customize)*/
   ST30P_TX_FLAG_BLOCK_GET = (MTL_BIT32(15)),
+
+  /**
+   * Drop frames when the mtl reports late frames (transport can't keep up).
+   * When late frame is detected, next frame from pipeline is ommited.
+   * Untill we resume normal frame sending.
+   */
+  ST30P_TX_FLAG_DROP_WHEN_LATE = (MTL_BIT32(16)),
 };
 
 /** The structure info for st30 frame meta. */
@@ -131,6 +138,14 @@ struct st30p_tx_ops {
    * tasklet routine.
    */
   int (*notify_frame_done)(void* priv, struct st30_frame* frame);
+  /**
+   * Optional. Callback when frame timing issues occur.
+   * If ST30P_TX_FLAG_DROP_WHEN_LATE is enabled: triggered when a frame is dropped
+   * from the pipeline due to late transmission.
+   * If ST30P_TX_FLAG_DROP_WHEN_LATE is disabled: triggered when the transport
+   * layer reports late frame delivery.
+   */
+  int (*notify_frame_late)(void* priv, uint64_t epoch_skipped);
 
   /**
    * Optional. The rtp timestamp delta(us) to the start time of frame.

include/st40_api.h

@@ -337,6 +337,8 @@ struct st40_tx_ops {
   int (*notify_frame_done)(void* priv, uint16_t frame_idx,
                            struct st40_tx_frame_meta* meta);
 
+  int (*notify_frame_late)(void* priv, uint64_t epoch_skipped);
+
   /** Optional. UDP source port number, leave as 0 to use same port as dst */
   uint16_t udp_src_port[MTL_SESSION_PORT_MAX];
   /**

include/st40_pipeline_api.h

@@ -67,6 +67,12 @@ enum st40p_tx_flag {
    * lib will wait until timestamp is reached for each frame.
    */
   ST40P_TX_FLAG_USER_PACING = (MTL_BIT32(3)),
+  /**
+   * Drop frames when the mtl reports late frames (transport can't keep up).
+   * When late frame is detected, next frame from pipeline is ommited.
+   * Untill we resume normal frame sending.
+   */
+  ST40P_TX_FLAG_DROP_WHEN_LATE = (MTL_BIT32(7)),
   /**
    * Flag bit in flags of struct st40_tx_ops.
    * If enabled, lib will assign the rtp timestamp to the value in
@@ -91,10 +97,6 @@ enum st40p_tx_flag {
   /** Enable the st40p_tx_get_frame block behavior to wait until a frame becomes
    available or timeout(default: 1s, use st40p_tx_set_block_timeout to customize)*/
   ST40P_TX_FLAG_BLOCK_GET = (MTL_BIT32(15)),
-  /**
-   ** Enable verbose reporting of framebuffer statuses in statistics output
-   */
-  ST40P_TX_FLAG_ACCURATE_FRAMEBUFF_STATISTICS = (MTL_BIT32(25)),
 };
 
 /**
@@ -131,6 +133,15 @@ struct st40p_tx_ops {
    */
   int (*notify_frame_done)(void* priv, struct st40_frame_info* frame_info);
 
+  /**
+   * Optional. Callback when frame timing issues occur.
+   * If ST30P_TX_FLAG_DROP_WHEN_LATE is enabled: triggered when a frame is dropped
+   * from the pipeline due to late transmission.
+   * If ST30P_TX_FLAG_DROP_WHEN_LATE is disabled: triggered when the transport
+   * layer reports late frame delivery.
+   */
+  int (*notify_frame_late)(void* priv, uint64_t epoch_skipped);
+
   /**
    * Optional. tx destination mac address.
    * Valid if ST40P_TX_FLAG_USER_P(R)_MAC is enabled

include/st41_api.h

@@ -43,6 +43,13 @@ typedef struct st_rx_fastmetadata_session_handle_impl* st41_rx_handle;
  * lib will wait until timestamp is reached for each frame.
  */
 #define ST41_TX_FLAG_USER_PACING (MTL_BIT32(3))
+/**
+ * Flag bit in flags of struct st41_tx_ops.
+ * Drop all packets that are old to recover from overflow situations and user
+ * mismanagement. This flag works in conjunction with ST41_TX_FLAG_USER_PACING to handle
+ * overflow scenarios.
+ */
+#define ST41_TX_FLAG_DROP_OLD (MTL_BIT32(7))
 /**
  * Flag bit in flags of struct st41_tx_ops.
  * If enabled, lib will assign the rtp timestamp to the value in

include/st_pipeline_api.h

@@ -384,6 +384,12 @@ enum st22p_tx_flag {
    * lib will wait until timestamp is reached for each frame.
    */
   ST22P_TX_FLAG_USER_PACING = (MTL_BIT32(3)),
+  /**
+   * Drop frames when the mtl reports late frames (transport can't keep up).
+   * When late frame is detected, next frame from pipeline is ommited.
+   * Untill we resume normal frame sending.
+   */
+  ST22P_TX_FLAG_DROP_WHEN_LATE = (MTL_BIT32(12)),
   /**
    * If enabled, lib will assign the rtp timestamp to the value in
    * tx_frame_meta(ST10_TIMESTAMP_FMT_MEDIA_CLK is used)
@@ -413,6 +419,8 @@ enum st22p_tx_flag {
   /** Enable the st22p_tx_get_frame block behavior to wait until a frame becomes
    available or timeout(default: 1s, use st22p_tx_set_block_timeout to customize) */
   ST22P_TX_FLAG_BLOCK_GET = (MTL_BIT32(15)),
+  /** If enabled, drop old frames when user pacing is used */
+  ST22P_TX_FLAG_USER_TIMESTAMP_DROP_OLD = (MTL_BIT32(16)),
 };
 
 /** Bit define for flags of struct st20p_tx_ops. */
@@ -437,6 +445,12 @@ enum st20p_tx_flag {
    * aligned with virtual receiver read schedule.
    */
   ST20P_TX_FLAG_USER_PACING = (MTL_BIT32(3)),
+  /**
+   * Drop frames when the mtl reports late frames (transport can't keep up).
+   * When late frame is detected, next frame from pipeline is ommited.
+   * Untill we resume normal frame sending.
+   */
+  ST20P_TX_FLAG_DROP_WHEN_LATE = (MTL_BIT32(12)),
   /**
    * If enabled, lib will assign the rtp timestamp to the value of timestamp in
    * st_frame.timestamp (if needed the value will be converted to
@@ -875,12 +889,22 @@ struct st20p_tx_ops {
    */
   int (*notify_frame_available)(void* priv);
   /**
-   * Optional. Callback when frame done in the lib.
-   * And only non-block method can be used within this callback as it run from lcore
-   * tasklet routine.
+   * Optional. Callback when frame timing issues occur.
+   * If ST20P_TX_FLAG_DROP_WHEN_LATE is enabled: triggered when a frame is dropped
+   * from the pipeline due to late transmission.
+   * If ST20P_TX_FLAG_DROP_WHEN_LATE is disabled: triggered when the transport
+   * layer reports late frame delivery.
    */
   int (*notify_frame_done)(void* priv, struct st_frame* frame);
 
+  /**
+   * Optional. Callback when frame is late in the lib, if ST20P_TX_FLAG_DROP_WHEN_LATE is
+   * enabled then it is triggered when a frame is omitted in pipeline tx.
+   * If hte ST20P_TX_FLAG_DROP_WHEN_LATE is not enabled, then it is triggered when the
+   * transport layer reports a late frame.
+   */
+  int (*notify_frame_late)(void* priv, uint64_t epoch_skipped);
+
   /** Optional. Linesize for transport frame, only for non-convert mode */
   size_t transport_linesize;
 
@@ -1046,12 +1070,22 @@ struct st22p_tx_ops {
    */
   int (*notify_frame_available)(void* priv);
   /**
-   * Optional. Callback when frame done in the lib.
-   * And only non-block method can be used within this callback as it run from lcore
-   * tasklet routine.
+   * Optional. Callback when frame timing issues occur.
+   * If ST22P_TX_FLAG_DROP_WHEN_LATE is enabled: triggered when a frame is dropped
+   * from the pipeline due to late transmission.
+   * If ST22P_TX_FLAG_DROP_WHEN_LATE is disabled: triggered when the transport
+   * layer reports late frame delivery.
    */
   int (*notify_frame_done)(void* priv, struct st_frame* frame);
 
+  /**
+   * Optional. Callback when frame is late in the lib, if ST22P_TX_FLAG_DROP_WHEN_LATE is
+   * enabled then it is triggered when a frame is omitted in pipeline tx.
+   * If hte ST22P_TX_FLAG_DROP_WHEN_LATE is not enabled, then it is triggered when the
+   * transport layer reports a late frame.
+   */
+  int (*notify_frame_late)(void* priv, uint64_t epoch_skipped);
+
   /** Optional for ST22P_TX_FLAG_ENABLE_RTCP. RTCP info */
   struct st_tx_rtcp_ops rtcp;
   /**

lib/src/mt_usdt.h

@@ -106,6 +106,8 @@
   MT_DTRACE_PROBE4(st20p, tx_frame_put, idx, f_idx, va, stat)
 #define MT_USDT_ST20P_TX_FRAME_DONE(idx, f_idx, tmstamp) \
   MT_DTRACE_PROBE3(st20p, tx_frame_done, idx, f_idx, tmstamp)
+#define MT_USDT_ST20P_TX_FRAME_SPURN(idx, f_idx, tmstamp) \
+  MT_DTRACE_PROBE3(st20p, tx_frame_done, idx, f_idx, tmstamp)
 #define MT_USDT_ST20P_TX_FRAME_NEXT(idx, f_idx) \
   MT_DTRACE_PROBE2(st20p, tx_frame_next, idx, f_idx)
 #define MT_USDT_ST20P_TX_FRAME_DUMP(idx, file, va, sz) \
@@ -128,6 +130,8 @@
   MT_DTRACE_PROBE3(st30p, tx_frame_put, idx, f_idx, va)
 #define MT_USDT_ST30P_TX_FRAME_DONE(idx, f_idx, tmstamp) \
   MT_DTRACE_PROBE3(st30p, tx_frame_done, idx, f_idx, tmstamp)
+#define MT_USDT_ST30P_TX_FRAME_SPURN(idx, f_idx, tmstamp) \
+  MT_DTRACE_PROBE3(st30p, tx_frame_done, idx, f_idx, tmstamp)
 #define MT_USDT_ST30P_TX_FRAME_NEXT(idx, f_idx) \
   MT_DTRACE_PROBE2(st30p, tx_frame_next, idx, f_idx)
 #define MT_USDT_ST30P_TX_FRAME_DUMP(idx, file, frames) \
@@ -144,6 +148,17 @@
   MT_DTRACE_PROBE3(st30p, rx_frame_dump, idx, file, frames)
 #define MT_USDT_ST30P_RX_FRAME_DUMP_ENABLED() ST30P_RX_FRAME_DUMP_ENABLED()
 
+#define MT_USDT_ST40P_TX_FRAME_GET(idx, f_idx, va) \
+  MT_DTRACE_PROBE3(st40p, tx_frame_get, idx, f_idx, va)
+#define MT_USDT_ST40P_TX_FRAME_PUT(idx, f_idx, va) \
+  MT_DTRACE_PROBE3(st40p, tx_frame_put, idx, f_idx, va)
+#define MT_USDT_ST40P_TX_FRAME_DONE(idx, f_idx, tmstamp) \
+  MT_DTRACE_PROBE3(st40p, tx_frame_done, idx, f_idx, tmstamp)
+#define MT_USDT_ST40P_TX_FRAME_SPURN(idx, f_idx, tmstamp) \
+  MT_DTRACE_PROBE3(st30p, tx_frame_done, idx, f_idx, tmstamp)
+#define MT_USDT_ST40P_TX_FRAME_NEXT(idx, f_idx) \
+  MT_DTRACE_PROBE2(st40p, tx_frame_next, idx, f_idx)
+
 #define MT_USDT_ST20_TX_FRAME_NEXT(m_idx, s_idx, f_idx, va, tmstamp) \
   MT_DTRACE_PROBE5(st20, tx_frame_next, m_idx, s_idx, f_idx, va, tmstamp)
 #define MT_USDT_ST20_TX_FRAME_DONE(m_idx, s_idx, f_idx, tmstamp) \

lib/src/mt_usdt_provider.d

@@ -80,6 +80,7 @@ provider st20p {
   probe tx_frame_put(int idx, int f_idx, void* va, int stat);
   probe tx_frame_next(int idx, int f_idx);
   probe tx_frame_done(int idx, int f_idx, uint32_t tmstamp);
+  probe tx_frame_spurn(int idx, int f_idx, uint32_t tmstamp);
   /* attach to enable the frame dump at runtime */
   probe tx_frame_dump(int idx, char* dump_file, void* va, uint32_t data_size);
   /* rx */
@@ -94,6 +95,7 @@ provider st22 {
   /* tx */
   probe tx_frame_next(int m_idx, int s_idx, int f_idx, void* va, uint32_t tmstamp, uint32_t codestream_size);
   probe tx_frame_done(int m_idx, int s_idx, int f_idx, uint32_t tmstamp);
+  probe tx_frame_spurn(int m_idx, int s_idx, int f_idx, uint32_t tmstamp);
   /* attach to enable the frame dump at runtime */
   probe tx_frame_dump(int m_idx, int s_idx, char* dump_file, void* va, uint32_t data_size);
   /* rx */
@@ -110,6 +112,7 @@ provider st22p {
   probe tx_frame_put(int idx, int f_idx, void* va, int stat, uint32_t data_size);
   probe tx_frame_next(int idx, int f_idx);
   probe tx_frame_done(int idx, int f_idx, uint32_t tmstamp);
+  probe tx_frame_spurn(int idx, int f_idx, uint32_t tmstamp);
   /* attach to enable the frame dump at runtime */
   probe tx_frame_dump(int idx, char* dump_file, void* va, uint32_t data_size);
   /* rx */
@@ -132,6 +135,7 @@ provider st30p {
   probe tx_frame_put(int idx, int f_idx, void* va);
   probe tx_frame_next(int idx, int f_idx);
   probe tx_frame_done(int idx, int f_idx, uint32_t tmstamp);
+  probe tx_frame_spurn(int idx, int f_idx, uint32_t tmstamp);
   /* attach to enable the frame dump at runtime */
   probe tx_frame_dump(int idx, char* dump_file, int frames);
   /* rx */
@@ -143,6 +147,12 @@ provider st30p {
 }
 
 provider st40p {
+  /* tx */
+  probe tx_frame_get(int idx, int f_idx, void* va);
+  probe tx_frame_put(int idx, int f_idx, void* va);
+  probe tx_frame_next(int idx, int f_idx);
+  probe tx_frame_done(int idx, int f_idx, uint32_t tmstamp);
+  probe tx_frame_spurn(int idx, int f_idx, uint32_t tmstamp)
   /* rx */
   probe rx_frame_get(int idx, int f_idx, uint32_t meta_num);
   probe rx_frame_put(int idx, int f_idx, uint32_t meta_num);