New locking for frame status

doc/design.md

@@ -342,7 +342,27 @@ Sample application code can be find in [tx_st20_pipeline_sample.c](../app/sample
 By default, the `st20p_tx_get_frame` and `st20p_rx_get_frame` functions operate in non-blocking mode, which means the function call will immediately return `NULL` if no frame is available.
 To switch to blocking mode, where the call will wait until a frame is ready for application use or one second timeout occurs, you must enable the `ST20P_TX_FLAG_BLOCK_GET` or `ST20P_RX_FLAG_BLOCK_GET` flag respectively during the session creation stage, and application can use `st20p_tx_wake_block`/`st20p_rx_wake_block` to wake up the waiting directly.
 
-#### 6.3.1. Ancillary (ST40) pipeline
+#### 6.3.1. Threading model and lock-free assumptions
+
+Each pipeline framebuffer carries a single `_Atomic` status field, and every stage transition (for example `FREE`→`IN_USER`, `READY`→`CONVERTED`, `IN_TRANSMITTING`→`FREE`) is performed with a C11 atomic load/store or compare-exchange rather than a mutex. This lock-free protocol is correct only under the following assumptions, which the get/put API contract implicitly relies on:
+
+- **One application thread per session and per direction.** For a given session, `st*p_tx_get_frame`/`st*p_tx_put_frame`
+  (and, on the receive side, `st*p_rx_get_frame`/`st*p_rx_put_frame`) must be called from the same thread, and each call
+  must be synchronous — it must return before the next get/put is issued. The atomic status field prevents state
+  corruption and double-claim between the application thread and the data-plane thread, but it does **not** make these
+  calls re-entrant or safe to invoke concurrently from multiple application threads. The producer/consumer cursor hints
+  (`framebuff_producer_idx`/`framebuff_consumer_idx`) and the get→process→put handoff assume this single-threaded,
+  sequential usage.
+- **Synchronous lower-layer callbacks.** The transport-layer callbacks driven from the data-plane tasklet — for example
+  `frame_ready`, `packet_convert`, `frame_done`, and `query_ext_frame` — are assumed to run synchronously and to
+  completion within their single owning tasklet thread. Under this assumption each pipeline stage has exactly one
+  producer thread and one consumer thread, so the atomic status field (with acquire/release ordering) is the only
+  cross-thread synchronization required. If any such callback were deferred to another thread or re-entered
+  concurrently, the acquire/release ordering would no longer be sufficient and the lock-free protocol would break.
+
+In short: the atomic status field synchronizes exactly one data-plane (tasklet) thread against one application thread. Multi-threaded or asynchronous use of the get/put APIs, or asynchronous lower-layer callbacks, are outside the model and are not supported.
+
+#### 6.3.2. Ancillary (ST40) pipeline
 
 The same pipeline abstraction is now available for ancillary-only flows through `st40_pipeline_api.h`. It keeps the zero-copy RFC 8331 payload path but hides scheduler registration, frame queues, and pacing details behind the familiar get/put contract so applications only need to implement:
 

lib/src/st2110/pipeline/st20_pipeline_rx.h

@@ -18,7 +18,7 @@ enum st20p_rx_frame_status {
 };
 
 struct st20p_rx_frame {
-  enum st20p_rx_frame_status stat;
+  _Atomic uint32_t stat;
   struct st_frame src; /* before converting */
   struct st_frame dst; /* converted */
   struct st20_convert_frame_meta convert_frame;
@@ -54,7 +54,6 @@ struct st20p_rx_ctx {
   uint16_t framebuff_convert_idx;
   uint16_t framebuff_consumer_idx;
   struct st20p_rx_frame* framebuffs;
-  pthread_mutex_t lock;
   int usdt_frame_cnt;
 
   /* for ST20P_RX_FLAG_BLOCK_GET */

lib/src/st2110/pipeline/st20_pipeline_tx.h

@@ -20,7 +20,7 @@ enum st20p_tx_frame_status {
 };
 
 struct st20p_tx_frame {
-  enum st20p_tx_frame_status stat;
+  _Atomic uint32_t stat;
   struct st_frame src; /* before converting */
   struct st_frame dst; /* converted */
   struct st20_convert_frame_meta convert_frame;
@@ -47,9 +47,8 @@ struct st20p_tx_ctx {
 
   st20_tx_handle transport;
   uint16_t framebuff_cnt;
-  uint32_t framebuff_sequence_number;
+  _Atomic uint32_t framebuff_sequence_number;
   struct st20p_tx_frame* framebuffs;
-  pthread_mutex_t lock;
   int usdt_frame_cnt;
 
   struct st20_convert_session_impl* convert_impl;
@@ -66,6 +65,7 @@ struct st20p_tx_ctx {
   pthread_cond_t block_wake_cond;
   pthread_mutex_t block_wake_mutex;
   uint64_t block_timeout_ns;
+  bool block_wake_pending;
 
   rte_atomic32_t stat_convert_fail;
   rte_atomic32_t stat_busy;

lib/src/st2110/pipeline/st22_pipeline_rx.c

@@ -94,6 +94,27 @@ static struct st22p_rx_frame* rx_st22p_next_available(
   return NULL;
 }
 
+/* Scan from idx_start for a framebuff in state `desired` and atomically claim
+ * it by transitioning it to `claimed`. A concurrent thread can win the race on
+ * the scanned candidate between the scan and the claim; on a lost race this
+ * keeps scanning instead of giving up, so the caller only sees NULL once every
+ * slot has genuinely been checked and found unavailable. */
+static struct st22p_rx_frame* rx_st22p_claim_available(
+    struct st22p_rx_ctx* ctx, uint16_t idx_start, enum st22p_rx_frame_status desired,
+    enum st22p_rx_frame_status claimed) {
+  struct st22p_rx_frame* framebuff;
+
+  while ((framebuff = rx_st22p_next_available(ctx, idx_start, desired))) {
+    uint32_t expected = desired;
+    if (atomic_compare_exchange_strong_explicit(&framebuff->stat, &expected, claimed,
+                                                memory_order_acq_rel,
+                                                memory_order_relaxed))
+      return framebuff;
+  }
+
+  return NULL;
+}
+
 static int rx_st22p_frame_ready(void* priv, void* frame,
                                 struct st22_rx_frame_meta* meta) {
   struct st22p_rx_ctx* ctx = priv;
@@ -102,13 +123,11 @@ static int rx_st22p_frame_ready(void* priv, void* frame,
 
   if (!ctx->ready) return -EBUSY; /* not ready */
 
-  mt_pthread_mutex_lock(&ctx->lock);
   framebuff =
       rx_st22p_next_available(ctx, ctx->framebuff_producer_idx, ST22P_RX_FRAME_FREE);
   /* not any free frame */
   if (!framebuff) {
     rte_atomic32_inc(&ctx->stat_busy);
-    mt_pthread_mutex_unlock(&ctx->lock);
     return -EBUSY;
   }
 
@@ -122,7 +141,6 @@ static int rx_st22p_frame_ready(void* priv, void* frame,
     if (ret < 0) {
       err("%s(%d), query_ext_frame for frame %u fail %d\n", __func__, ctx->idx,
           framebuff->idx, ret);
-      mt_pthread_mutex_unlock(&ctx->lock);
       return ret;
     }
 
@@ -139,7 +157,6 @@ static int rx_st22p_frame_ready(void* priv, void* frame,
     if (ret < 0) {
       err("%s(%d), ext_frame check frame %u fail %d\n", __func__, ctx->idx,
           framebuff->idx, ret);
-      mt_pthread_mutex_unlock(&ctx->lock);
       return ret;
     }
   }
@@ -170,14 +187,12 @@ static int rx_st22p_frame_ready(void* priv, void* frame,
     framebuff->stat = ST22P_RX_FRAME_DECODED;
     /* point to next */
     ctx->framebuff_producer_idx = rx_st22p_next_idx(ctx, framebuff->idx);
-    mt_pthread_mutex_unlock(&ctx->lock);
     rx_st22p_notify_frame_available(ctx);
     return 0;
   }
   framebuff->stat = ST22P_RX_FRAME_READY;
   /* point to next */
   ctx->framebuff_producer_idx = rx_st22p_next_idx(ctx, framebuff->idx);
-  mt_pthread_mutex_unlock(&ctx->lock);
 
   dbg("%s(%d), frame %u succ\n", __func__, ctx->idx, framebuff->idx);
   rx_st22p_decode_notify_frame_ready(ctx);
@@ -237,28 +252,25 @@ static struct st22_decode_frame_meta* rx_st22p_decode_get_frame(void* priv) {
 
   ctx->stat_decode_get_frame_try++;
 
-  mt_pthread_mutex_lock(&ctx->lock);
-  framebuff =
-      rx_st22p_next_available(ctx, ctx->framebuff_decode_idx, ST22P_RX_FRAME_READY);
+  /* Claim READY -> IN_DECODING. rx_st22p_claim_available() retries across the
+   * ring on a lost CAS race, so it only returns NULL once every slot has
+   * actually been checked -- unlike a scan-then-single-CAS-attempt, which could
+   * give up even while other READY frames remain. */
+  framebuff = rx_st22p_claim_available(ctx, ctx->framebuff_decode_idx,
+                                       ST22P_RX_FRAME_READY, ST22P_RX_FRAME_IN_DECODING);
   if (!framebuff && ctx->decode_block_get) { /* wait here for block mode */
-    mt_pthread_mutex_unlock(&ctx->lock);
     rx_st22p_decode_get_block_wait(ctx);
     /* get again */
-    mt_pthread_mutex_lock(&ctx->lock);
-    framebuff =
-        rx_st22p_next_available(ctx, ctx->framebuff_decode_idx, ST22P_RX_FRAME_READY);
+    framebuff = rx_st22p_claim_available(
+        ctx, ctx->framebuff_decode_idx, ST22P_RX_FRAME_READY, ST22P_RX_FRAME_IN_DECODING);
   }
   /* not any ready frame */
   if (!framebuff) {
-    mt_pthread_mutex_unlock(&ctx->lock);
     dbg("%s(%d), no ready frame\n", __func__, idx);
     goto out;
   }
-
-  framebuff->stat = ST22P_RX_FRAME_IN_DECODING;
   /* point to next */
   ctx->framebuff_decode_idx = rx_st22p_next_idx(ctx, framebuff->idx);
-  mt_pthread_mutex_unlock(&ctx->lock);
 
   ctx->stat_decode_get_frame_succ++;
   ret_frame = &framebuff->decode_frame;
@@ -580,32 +592,29 @@ struct st_frame* st22p_rx_get_frame(st22p_rx_handle handle) {
 
   ctx->stat_get_frame_try++;
 
-  mt_pthread_mutex_lock(&ctx->lock);
-  framebuff =
-      rx_st22p_next_available(ctx, ctx->framebuff_consumer_idx, ST22P_RX_FRAME_DECODED);
+  /* Claim DECODED->IN_USER. rx_st22p_claim_available() retries across the ring
+   * on a lost CAS race, so it only returns NULL once every slot has actually
+   * been checked -- unlike a scan-then-single-CAS-attempt, which could give up
+   * even while other DECODED frames remain (spurious failure under contention). */
+  framebuff = rx_st22p_claim_available(ctx, ctx->framebuff_consumer_idx,
+                                       ST22P_RX_FRAME_DECODED, ST22P_RX_FRAME_IN_USER);
   if (!framebuff && ctx->block_get) {
-    mt_pthread_mutex_unlock(&ctx->lock);
     mt_pthread_mutex_lock(&ctx->block_wake_mutex);
-    if (!__atomic_load_n(&ctx->lc_destroying, __ATOMIC_ACQUIRE))
+    if (!atomic_load_explicit(&ctx->lc_destroying, memory_order_acquire))
       mt_pthread_cond_timedwait_ns(&ctx->block_wake_cond, &ctx->block_wake_mutex,
                                    ctx->block_timeout_ns);
     mt_pthread_mutex_unlock(&ctx->block_wake_mutex);
-    if (__atomic_load_n(&ctx->lc_destroying, __ATOMIC_ACQUIRE)) goto out;
+    if (atomic_load_explicit(&ctx->lc_destroying, memory_order_acquire)) goto out;
     /* get again */
-    mt_pthread_mutex_lock(&ctx->lock);
-    framebuff =
-        rx_st22p_next_available(ctx, ctx->framebuff_consumer_idx, ST22P_RX_FRAME_DECODED);
+    framebuff = rx_st22p_claim_available(ctx, ctx->framebuff_consumer_idx,
+                                         ST22P_RX_FRAME_DECODED, ST22P_RX_FRAME_IN_USER);
   }
   /* not any decoded frame */
   if (!framebuff) {
-    mt_pthread_mutex_unlock(&ctx->lock);
     goto out;
   }
-
-  framebuff->stat = ST22P_RX_FRAME_IN_USER;
-  /* point to next */
+  /* point to next (best-effort hint; the CAS above is the real guard) */
   ctx->framebuff_consumer_idx = rx_st22p_next_idx(ctx, framebuff->idx);
-  mt_pthread_mutex_unlock(&ctx->lock);
 
   dbg("%s(%d), frame %u succ\n", __func__, idx, framebuff->idx);
   ctx->stat_get_frame_succ++;
@@ -765,7 +774,6 @@ st22p_rx_handle st22p_rx_create(mtl_handle mt, struct st22p_rx_ops* ops) {
   }
   rte_atomic32_set(&ctx->stat_decode_fail, 0);
   rte_atomic32_set(&ctx->stat_busy, 0);
-  mt_pthread_mutex_init(&ctx->lock, NULL);
 
   mt_pthread_mutex_init(&ctx->decode_block_wake_mutex, NULL);
   mt_pthread_cond_wait_init(&ctx->decode_block_wake_cond);
@@ -846,7 +854,6 @@ int st22p_rx_free(st22p_rx_handle handle) {
   }
   rx_st22p_uinit_dst_fbs(ctx);
 
-  mt_pthread_mutex_destroy(&ctx->lock);
   mt_pthread_mutex_destroy(&ctx->block_wake_mutex);
   mt_pthread_cond_destroy(&ctx->block_wake_cond);
   mt_pthread_mutex_destroy(&ctx->decode_block_wake_mutex);

lib/src/st2110/pipeline/st22_pipeline_rx.h

@@ -18,7 +18,7 @@ enum st22p_rx_frame_status {
 };
 
 struct st22p_rx_frame {
-  enum st22p_rx_frame_status stat;
+  _Atomic uint32_t stat;
   struct st_frame src; /* before decoding */
   struct st_frame dst; /* decoded */
   struct st22_decode_frame_meta decode_frame;
@@ -45,7 +45,6 @@ struct st22p_rx_ctx {
   uint16_t framebuff_decode_idx;
   uint16_t framebuff_consumer_idx;
   struct st22p_rx_frame* framebuffs;
-  pthread_mutex_t lock;
 
   /* for ST22P_RX_FLAG_BLOCK_GET */
   bool block_get;