Fix: add block wake pending flag to ST20 and ST30 RX contexts for improved synch

Author: moleksy

lib/src/st2110/pipeline/st20_pipeline_rx.c

@@ -28,6 +28,7 @@ static uint16_t rx_st20p_next_idx(struct st20p_rx_ctx* ctx, uint16_t idx) {
 static void rx_st20p_block_wake(struct st20p_rx_ctx* ctx) {
   /* notify block */
   mt_pthread_mutex_lock(&ctx->block_wake_mutex);
+  ctx->block_wake_pending = true;
   mt_pthread_cond_signal(&ctx->block_wake_cond);
   mt_pthread_mutex_unlock(&ctx->block_wake_mutex);
 }
@@ -757,14 +758,23 @@ static int rx_st20p_get_converter(struct mtl_main_impl* impl, struct st20p_rx_ct
 static int rx_st20p_stat(void* priv) {
   struct st20p_rx_ctx* ctx = priv;
   struct st20p_rx_frame* framebuff = ctx->framebuffs;
+  uint16_t producer_idx;
+  uint16_t consumer_idx;
+  enum st20p_rx_frame_status producer_stat;
+  enum st20p_rx_frame_status consumer_stat;
 
   if (!ctx->ready) return -EBUSY; /* not ready */
 
-  uint16_t producer_idx = ctx->framebuff_producer_idx;
-  uint16_t consumer_idx = ctx->framebuff_consumer_idx;
+  mt_pthread_mutex_lock(&ctx->lock);
+  producer_idx = ctx->framebuff_producer_idx;
+  consumer_idx = ctx->framebuff_consumer_idx;
+  producer_stat = framebuff[producer_idx].stat;
+  consumer_stat = framebuff[consumer_idx].stat;
+  mt_pthread_mutex_unlock(&ctx->lock);
+
   notice("RX_st20p(%d,%s), p(%d:%s) c(%d:%s)\n", ctx->idx, ctx->ops_name, producer_idx,
-         rx_st20p_stat_name(framebuff[producer_idx].stat), consumer_idx,
-         rx_st20p_stat_name(framebuff[consumer_idx].stat));
+         rx_st20p_stat_name(producer_stat), consumer_idx,
+         rx_st20p_stat_name(consumer_stat));
 
   notice("RX_st20p(%d), frame get try %d succ %d, put %d\n", ctx->idx,
          ctx->stat_get_frame_try, ctx->stat_get_frame_succ, ctx->stat_put_frame);
@@ -810,8 +820,12 @@ static int st20p_rx_get_block_wait(struct st20p_rx_ctx* ctx) {
   dbg("%s(%d), start\n", __func__, ctx->idx);
   /* wait on the block cond */
   mt_pthread_mutex_lock(&ctx->block_wake_mutex);
-  mt_pthread_cond_timedwait_ns(&ctx->block_wake_cond, &ctx->block_wake_mutex,
-                               ctx->block_timeout_ns);
+  while (!ctx->block_wake_pending) {
+    int ret = mt_pthread_cond_timedwait_ns(&ctx->block_wake_cond, &ctx->block_wake_mutex,
+                                           ctx->block_timeout_ns);
+    if (ret) break;
+  }
+  ctx->block_wake_pending = false;
   mt_pthread_mutex_unlock(&ctx->block_wake_mutex);
   dbg("%s(%d), end\n", __func__, ctx->idx);
   return 0;
@@ -997,6 +1011,7 @@ st20p_rx_handle st20p_rx_create(mtl_handle mt, struct st20p_rx_ops* ops) {
   mt_pthread_mutex_init(&ctx->block_wake_mutex, NULL);
   mt_pthread_cond_wait_init(&ctx->block_wake_cond);
   ctx->block_timeout_ns = NS_PER_S;
+  ctx->block_wake_pending = false;
   if (ops->flags & ST20P_RX_FLAG_BLOCK_GET) ctx->block_get = true;
 
   /* copy ops */

lib/src/st2110/pipeline/st20_pipeline_rx.h

@@ -52,6 +52,7 @@ struct st20p_rx_ctx {
   pthread_cond_t block_wake_cond;
   pthread_mutex_t block_wake_mutex;
   uint64_t block_timeout_ns;
+  bool block_wake_pending;
 
   struct st20_convert_session_impl* convert_impl;
   struct st_frame_converter* internal_converter;

lib/src/st2110/pipeline/st30_pipeline_rx.c

@@ -28,6 +28,7 @@ static uint16_t rx_st30p_next_idx(struct st30p_rx_ctx* ctx, uint16_t idx) {
 static void rx_st30p_block_wake(struct st30p_rx_ctx* ctx) {
   /* notify block */
   mt_pthread_mutex_lock(&ctx->block_wake_mutex);
+  ctx->block_wake_pending = true;
   mt_pthread_cond_signal(&ctx->block_wake_cond);
   mt_pthread_mutex_unlock(&ctx->block_wake_mutex);
 }
@@ -194,14 +195,23 @@ static int rx_st30p_init_fbs(struct st30p_rx_ctx* ctx, struct st30p_rx_ops* ops)
 static int rx_st30p_stat(void* priv) {
   struct st30p_rx_ctx* ctx = priv;
   struct st30p_rx_frame* framebuff = ctx->framebuffs;
+  uint16_t producer_idx;
+  uint16_t consumer_idx;
+  enum st30p_rx_frame_status producer_stat;
+  enum st30p_rx_frame_status consumer_stat;
 
   if (!ctx->ready) return -EBUSY; /* not ready */
 
-  uint16_t producer_idx = ctx->framebuff_producer_idx;
-  uint16_t consumer_idx = ctx->framebuff_consumer_idx;
+  mt_pthread_mutex_lock(&ctx->lock);
+  producer_idx = ctx->framebuff_producer_idx;
+  consumer_idx = ctx->framebuff_consumer_idx;
+  producer_stat = framebuff[producer_idx].stat;
+  consumer_stat = framebuff[consumer_idx].stat;
+  mt_pthread_mutex_unlock(&ctx->lock);
+
   notice("RX_st30p(%d,%s), p(%d:%s) c(%d:%s)\n", ctx->idx, ctx->ops_name, producer_idx,
-         rx_st30p_stat_name(framebuff[producer_idx].stat), consumer_idx,
-         rx_st30p_stat_name(framebuff[consumer_idx].stat));
+         rx_st30p_stat_name(producer_stat), consumer_idx,
+         rx_st30p_stat_name(consumer_stat));
 
   notice("RX_st30p(%d), frame get try %d succ %d, put %d\n", ctx->idx,
          ctx->stat_get_frame_try, ctx->stat_get_frame_succ, ctx->stat_put_frame);
@@ -221,8 +231,12 @@ static int rx_st30p_get_block_wait(struct st30p_rx_ctx* ctx) {
   dbg("%s(%d), start\n", __func__, ctx->idx);
   /* wait on the block cond */
   mt_pthread_mutex_lock(&ctx->block_wake_mutex);
-  mt_pthread_cond_timedwait_ns(&ctx->block_wake_cond, &ctx->block_wake_mutex,
-                               ctx->block_timeout_ns);
+  while (!ctx->block_wake_pending) {
+    int ret = mt_pthread_cond_timedwait_ns(&ctx->block_wake_cond, &ctx->block_wake_mutex,
+                                           ctx->block_timeout_ns);
+    if (ret) break;
+  }
+  ctx->block_wake_pending = false;
   mt_pthread_mutex_unlock(&ctx->block_wake_mutex);
   dbg("%s(%d), end\n", __func__, ctx->idx);
   return 0;
@@ -437,6 +451,7 @@ st30p_rx_handle st30p_rx_create(mtl_handle mt, struct st30p_rx_ops* ops) {
   mt_pthread_mutex_init(&ctx->block_wake_mutex, NULL);
   mt_pthread_cond_wait_init(&ctx->block_wake_cond);
   ctx->block_timeout_ns = NS_PER_S;
+  ctx->block_wake_pending = false;
   if (ops->flags & ST30P_RX_FLAG_BLOCK_GET) ctx->block_get = true;
 
   /* copy ops */

lib/src/st2110/pipeline/st30_pipeline_rx.h

@@ -49,6 +49,7 @@ struct st30p_rx_ctx {
   pthread_cond_t block_wake_cond;
   pthread_mutex_t block_wake_mutex;
   uint64_t block_timeout_ns;
+  bool block_wake_pending;
 
   /* get frame stat */
   int stat_get_frame_try;