Fix: st40_pipeline_tx.c meta_num handling

We process the meta_num incorrectly in
st40_pipeline_tx.c, leading to potential issues
with frame metadata handling.
Fix it by overwriting the anc_frame meta_num and
treating the framebuff from pipeline anc size
as the only source of truth.

Fix the inproper handling of the CAPS negotiation
signal, for now just ignore consecutive signals

TODO: Checking and reinitialization of the session.

Add the ability for the st40 gstreamer plugin to
parse multiple ancillary packages, including
support for different payload types and
dynamic reconfiguration.

Add the ability for st20 video plugin to work
without metadata with assumed bit configuration.

Add the st40p meta insertion into the metadata
element

Test the changes by adding the ability for the
pytest framework to generate data from both the
Rx simplified rfc8331 and the Tx rfc8331
custom formats (ancgenerator)
For more advanced anc frames testing.

Add the ability for the testing framework in
gstreamer to run user pacing scanrios
TODO: Add tests for user pacing
TODO: Add lipsync for gstreamer

Fix: fix the way we handle killing and signal
handling in gstreamer engine.

Fix: error reporting for paths in testing framework
For easier debugging and maintenance.

reverts: fc4c95a

Author: DawidWesierski4

.gitignore

@@ -55,6 +55,7 @@ dkms.conf
 .*
 *.log
 *.log_*
+tests/validation/logs
 logs_*
 *.patch
 !patches/**/*.patch

ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.c

@@ -446,6 +446,11 @@ static gboolean gst_mtl_st20p_tx_sink_event(GstPad* pad, GstObject* parent,
 
   switch (GST_EVENT_TYPE(event)) {
     case GST_EVENT_CAPS:
+      if (sink->session_capabilites_set) {
+        GST_WARNING("Capabilities already set, ignoring");
+        return gst_pad_event_default(pad, parent, event);
+      }
+
       gst_event_parse_caps(event, &caps);
       if (sink->async_session_create) {
         thread_data = malloc(sizeof(GstMtlSt20pTxThreadData));
@@ -460,6 +465,7 @@ static gboolean gst_mtl_st20p_tx_sink_event(GstPad* pad, GstObject* parent,
           return FALSE;
         }
       }
+      sink->session_capabilites_set = TRUE;
       ret = gst_pad_event_default(pad, parent, event);
       break;
     case GST_EVENT_EOS:
@@ -562,12 +568,8 @@ static GstFlowReturn gst_mtl_st20p_tx_zero_copy(Gst_Mtl_St20p_Tx* sink, GstBuffe
   GstSt20pTxExternalDataParent* parent;
   struct st_frame* frame;
   struct st_ext_frame ext_frame;
-  GstVideoMeta* video_meta = gst_buffer_get_video_meta(buf);
+  GstVideoMeta* video_meta = NULL;
   gint buffer_n = gst_buffer_n_memory(buf);
-  if (!video_meta) {
-    g_print("Failed to get video meta from buffer\n");
-    return GST_FLOW_ERROR;
-  }
 
   parent = malloc(sizeof(GstSt20pTxExternalDataParent));
   if (!parent) {
@@ -615,11 +617,30 @@ static GstFlowReturn gst_mtl_st20p_tx_zero_copy(Gst_Mtl_St20p_Tx* sink, GstBuffe
       frame->tfmt = ST10_TIMESTAMP_FMT_TAI;
     }
 
-    for (int i = 0; i < video_meta->n_planes; i++) {
-      ext_frame.addr[i] = child->map_info.data + video_meta->offset[i];
-      ext_frame.linesize[i] = video_meta->stride[i];
-      ext_frame.iova[i] = 0;
+    video_meta = gst_buffer_get_video_meta(buf);
+    if (video_meta) {
+      for (int i = 0; i < video_meta->n_planes; i++) {
+        ext_frame.addr[i] = child->map_info.data + video_meta->offset[i];
+        ext_frame.linesize[i] = video_meta->stride[i];
+        ext_frame.iova[i] = 0;
+      }
+
+    } else {
+      ext_frame.addr[0] = child->map_info.data;
+      ext_frame.iova[0] = 0;
+      ext_frame.linesize[0] = st_frame_least_linesize(frame->fmt, frame->width, 0);
+      guint8 planes = st_frame_fmt_planes(frame->fmt);
+
+      /* Assume video planes are stored contiguously in memory */
+      for (gint plane = 1; plane < planes; plane++) {
+        ext_frame.linesize[plane] =
+            st_frame_least_linesize(frame->fmt, frame->width, plane);
+        ext_frame.addr[plane] = (guint8*)ext_frame.addr[plane - 1] +
+                                ext_frame.linesize[plane - 1] * frame->height;
+        ext_frame.iova[plane] = 0;
+      }
     }
+
     ext_frame.size = child->map_info.size;
     ext_frame.opaque = child;
     frame->opaque = NULL;

ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.h

@@ -62,6 +62,7 @@ struct _Gst_Mtl_St20p_Tx {
   gboolean zero_copy;
 
   gboolean session_ready;
+  gboolean session_capabilites_set;
   pthread_mutex_t session_mutex;
   pthread_t session_thread;
 

ecosystem/gstreamer_plugin/gst_mtl_st30p_tx.c

@@ -455,6 +455,11 @@ static gboolean gst_mtl_st30p_tx_sink_event(GstPad* pad, GstObject* parent,
 
   switch (GST_EVENT_TYPE(event)) {
     case GST_EVENT_CAPS:
+      if (sink->session_capabilites_set) {
+        GST_WARNING("Capabilities already set, ignoring");
+        return gst_pad_event_default(pad, parent, event);
+      }
+
       gst_event_parse_caps(event, &caps);
       if (sink->async_session_create) {
         thread_data = malloc(sizeof(GstMtlSt30pTxThreadData));
@@ -470,6 +475,7 @@ static gboolean gst_mtl_st30p_tx_sink_event(GstPad* pad, GstObject* parent,
         }
       }
 
+      sink->session_capabilites_set = TRUE;
       ret = gst_pad_event_default(pad, parent, event);
       break;
     case GST_EVENT_EOS:

ecosystem/gstreamer_plugin/gst_mtl_st30p_tx.h

@@ -61,6 +61,7 @@ struct _Gst_Mtl_St30p_Tx {
   guint frame_size;
 
   gboolean session_ready;
+  gboolean session_capabilites_set;
   pthread_mutex_t session_mutex;
   pthread_t session_thread;
 

ecosystem/gstreamer_plugin/gst_mtl_st40_rx.c

@@ -96,6 +96,7 @@ GST_DEBUG_CATEGORY_STATIC(gst_mtl_st40_rx_debug);
 enum {
   PROP_ST40_RX_BUFFER_SIZE = PROP_GENERAL_MAX,
   PROP_ST40_RX_TIMEOUT_MBUF_GET,
+  PROP_ST40_RX_INCLUDE_METADATA_IN_BUFFER,
   PROP_MAX
 };
 
@@ -108,8 +109,6 @@ G_DEFINE_TYPE_WITH_CODE(Gst_Mtl_St40_Rx, gst_mtl_st40_rx, GST_TYPE_BASE_SRC,
                         GST_DEBUG_CATEGORY_INIT(gst_mtl_st40_rx_debug, "mtl_st40_rx", 0,
                                                 "MTL St2110 st40 transmission src"));
 
-#define IS_POWER_OF_2(x) (((x) & ((x)-1)) == 0)
-
 GST_ELEMENT_REGISTER_DEFINE(mtl_st40_rx, "mtl_st40_rx", GST_RANK_NONE,
                             GST_TYPE_MTL_ST40_RX);
 
@@ -131,6 +130,8 @@ static GstFlowReturn gst_mtl_st40_rx_fill_buffer(Gst_Mtl_St40_Rx* src, GstBuffer
                                                  void* usrptr);
 static guint gst_mtl_st40_rx_parse_port_arguments(struct st40_rx_ops* ops_rx,
                                                   SessionPortArgs* portArgs);
+static struct st40_rfc8331_payload_hdr* gst_mtl_st40_rx_shift_payload_hdr(
+  struct st40_rfc8331_payload_hdr* payload_hdr, int udw_size);
 
 static gint gst_mtl_st40_rx_mbuff_available(void* priv) {
   Gst_Mtl_St40_Rx* src = (Gst_Mtl_St40_Rx*)priv;
@@ -179,6 +180,12 @@ static void gst_mtl_st40_rx_class_init(Gst_Mtl_St40_RxClass* klass) {
       g_param_spec_uint("timeout", "Timeout for Mbuf",
                         "Timeout in seconds for getting mbuf", 0, G_MAXUINT, 10,
                         G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property(
+  gobject_class, PROP_ST40_RX_INCLUDE_METADATA_IN_BUFFER,
+  g_param_spec_boolean("include-metadata-in-buffer", "Include Metadata in Buffer",
+           "Whether to include metadata in the output buffer", FALSE,
+           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 }
 
 static guint gst_mtl_st40_rx_parse_port_arguments(struct st40_rx_ops* ops_rx,
@@ -319,6 +326,9 @@ static void gst_mtl_st40_rx_set_property(GObject* object, guint prop_id,
     case PROP_ST40_RX_TIMEOUT_MBUF_GET:
       self->timeout_mbuf_get_seconds = g_value_get_uint(value);
       break;
+    case PROP_ST40_RX_INCLUDE_METADATA_IN_BUFFER:
+      self->include_metadata_in_buffer = g_value_get_boolean(value);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
       break;
@@ -342,6 +352,9 @@ static void gst_mtl_st40_rx_get_property(GObject* object, guint prop_id, GValue*
     case PROP_ST40_RX_TIMEOUT_MBUF_GET:
       g_value_set_uint(value, src->timeout_mbuf_get_seconds);
       break;
+    case PROP_ST40_RX_INCLUDE_METADATA_IN_BUFFER:
+      g_value_set_boolean(value, src->include_metadata_in_buffer);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
       break;
@@ -372,42 +385,100 @@ static void* gst_mtl_st40_rx_get_mbuf_with_timeout(Gst_Mtl_St40_Rx* src,
   return mbuf;
 }
 
+static struct st40_rfc8331_payload_hdr* gst_mtl_st40_rx_shift_payload_hdr(
+  struct st40_rfc8331_payload_hdr* payload_hdr, int udw_size) {
+  int package_size;
+  int payload_len;
+
+  package_size = ((3 + udw_size + 1) * 10) / 8;
+  package_size = (4 - package_size % 4) + package_size;
+  payload_len = sizeof(struct st40_rfc8331_payload_hdr) - 4 + package_size;
+
+  return (struct st40_rfc8331_payload_hdr*)((uint8_t*)payload_hdr + payload_len);
+}
+
 static GstFlowReturn gst_mtl_st40_rx_fill_buffer(Gst_Mtl_St40_Rx* src, GstBuffer** buffer,
                                                  void* usrptr) {
   struct st40_rfc8331_rtp_hdr* hdr;
-  struct st40_rfc8331_payload_hdr *payload_hdr, payload_hdr_swapped;
+  struct st40_rfc8331_payload_hdr *payload_hdr;
   GstMapInfo dest_info;
-  guint16 data, fill_size;
-  gint udw_size;
+  guint16 data;
+  gint udw_size, anc_count;
+  guint buffer_size = 0, meta_offset;
 
   hdr = (struct st40_rfc8331_rtp_hdr*)usrptr;
+  anc_count = hdr->anc_count;
+
+
+  if (anc_count == 0 || anc_count > ST40_RFC8331_PAYLOAD_MAX_ANCILLARY_COUNT) {
+    GST_ERROR("Ancillary data count: %d (must be between 1 and %d)", anc_count, ST40_RFC8331_PAYLOAD_MAX_ANCILLARY_COUNT);
+    return GST_FLOW_ERROR;
+  }
+
+  /* local is fine anc_count will not be bigger than 20*/
+  guint8* anc_data[anc_count];
+  guint8 anc_data_count[anc_count];
 
   payload_hdr = (struct st40_rfc8331_payload_hdr*)(&hdr[1]);
+  for (int i = 0; i < anc_count; i++) {
+    payload_hdr->swapped_first_hdr_chunk = ntohl(payload_hdr->swapped_first_hdr_chunk);
+    payload_hdr->swapped_second_hdr_chunk = ntohl(payload_hdr->swapped_second_hdr_chunk);
+    if (!st40_check_parity_bits(payload_hdr->second_hdr_chunk.did)) {
+      GST_ERROR("Parity check failed for DID");
+      return GST_FLOW_ERROR;
+    }
+    if (!st40_check_parity_bits(payload_hdr->second_hdr_chunk.sdid)) {
+      GST_ERROR("Parity check failed for SDID");
+      return GST_FLOW_ERROR;
+    }
+    if (!st40_check_parity_bits(payload_hdr->second_hdr_chunk.data_count)) {
+      GST_ERROR("Parity check failed for Data Count");
+      return GST_FLOW_ERROR;
+    }
 
-  hdr->swapped_first_hdr_chunk = ntohl(hdr->swapped_first_hdr_chunk);
+    udw_size = payload_hdr->second_hdr_chunk.data_count & 0xff;
+    anc_data_count[i] = udw_size;
+    meta_offset = 0;
+    buffer_size += udw_size;
+
+    if (src->include_metadata_in_buffer) {
+      buffer_size += ST40_BYTE_SIZE_OF_PAYLOAD_METADATA;
+      anc_data[i] = malloc(udw_size + ST40_BYTE_SIZE_OF_PAYLOAD_METADATA);
+      anc_data[i][meta_offset++] = payload_hdr->first_hdr_chunk.c;
+      anc_data[i][meta_offset++] = payload_hdr->first_hdr_chunk.line_number;
+      anc_data[i][meta_offset++] = payload_hdr->first_hdr_chunk.horizontal_offset;
+      anc_data[i][meta_offset++] = payload_hdr->first_hdr_chunk.s;
+      anc_data[i][meta_offset++] = payload_hdr->first_hdr_chunk.stream_num;
+      anc_data[i][meta_offset++] = payload_hdr->second_hdr_chunk.did & 0xff;
+      anc_data[i][meta_offset++] = payload_hdr->second_hdr_chunk.sdid & 0xff;
+      anc_data[i][meta_offset++] = payload_hdr->second_hdr_chunk.data_count & 0xff;
+    } else {
+      anc_data[i] = malloc(udw_size);
+    }
 
-  payload_hdr_swapped.swapped_second_hdr_chunk =
-      ntohl(payload_hdr->swapped_second_hdr_chunk);
-  udw_size = payload_hdr_swapped.second_hdr_chunk.data_count & 0xff;
+    payload_hdr->swapped_second_hdr_chunk = htonl(payload_hdr->swapped_second_hdr_chunk);
+    if (!anc_data[i]) {
+      GST_ERROR("Failed to allocate memory for ancillary data");
+      for (int j = 0; j < i; j++)
+        free(anc_data[j]);
+      return GST_FLOW_ERROR;
+    }
 
-  if (udw_size == 0) {
-    GST_ERROR("Ancillary data size is 0");
-    return GST_FLOW_ERROR;
-  } else if (src->udw_size == 0) {
-    src->udw_size = udw_size;
-    src->anc_data = (char*)malloc(udw_size);
-  } else if (src->udw_size != udw_size) {
-    GST_INFO("Size of received ancillary data has changed");
-    if (src->anc_data) {
-      free(src->anc_data);
-      src->anc_data = NULL;
+    for (int d = 0; d < udw_size; d++) {
+      data = st40_get_udw(d + 3, (uint8_t*)&payload_hdr->second_hdr_chunk);
+      if (!st40_check_parity_bits(data)) {
+        GST_ERROR("Ancillary data parity bits check failed, data=0x%03x", data & 0x3FF);
+        for (int j = 0; j < i; j++)
+          free(anc_data[j]);
+        return GST_FLOW_ERROR;
+      }
+      anc_data[i][d + meta_offset] = data & 0xff;
     }
 
-    src->udw_size = udw_size;
-    src->anc_data = (char*)malloc(udw_size);
+    payload_hdr = gst_mtl_st40_rx_shift_payload_hdr(payload_hdr, udw_size);
   }
 
-  *buffer = gst_buffer_new_allocate(NULL, src->udw_size, NULL);
+  *buffer = gst_buffer_new_allocate(NULL, buffer_size, NULL);
   if (!*buffer) {
     GST_ERROR("Failed to allocate space for the buffer");
     return GST_FLOW_ERROR;
@@ -418,21 +489,21 @@ static GstFlowReturn gst_mtl_st40_rx_fill_buffer(Gst_Mtl_St40_Rx* src, GstBuffer
     return GST_FLOW_ERROR;
   }
 
-  for (int i = 0; i < udw_size; i++) {
-    data = st40_get_udw(i + 3, (uint8_t*)&payload_hdr->second_hdr_chunk);
-    if (!st40_check_parity_bits(data)) {
-      GST_ERROR("Ancillary data parity bits check failed");
-      return GST_FLOW_ERROR;
-    }
-    src->anc_data[i] = data & 0xff;
+  guint fill_size = 0;
+  for (int i = 0; i < anc_count; i++) {
+    fill_size += gst_buffer_fill(*buffer, fill_size, anc_data[i], anc_data_count[i] + meta_offset);
+    free(anc_data[i]);
+  }
+  /* Print the whole buffer for debugging */
+  for (guint i = 0; i < buffer_size; i++) {
+    printf("%02x ", dest_info.data[i]);
   }
+  printf("\n");
 
-  fill_size = gst_buffer_fill(*buffer, 0, src->anc_data, udw_size);
   gst_buffer_unmap(*buffer, &dest_info);
 
-  if (fill_size != src->udw_size) {
-    GST_ERROR("Failed to fill buffer");
-    return GST_FLOW_ERROR;
+  if (fill_size != buffer_size) {
+    GST_ERROR("Failed to fill buffer (buffer size = %d, fill size = %d)", buffer_size, fill_size);;
   }
 
   return GST_FLOW_OK;
@@ -478,8 +549,6 @@ static GstFlowReturn gst_mtl_st40_rx_create(GstBaseSrc* basesrc, guint64 offset,
 static void gst_mtl_st40_rx_finalize(GObject* object) {
   Gst_Mtl_St40_Rx* src = GST_MTL_ST40_RX(object);
 
-  if (src->anc_data) free(src->anc_data);
-
   if (src->rx_handle) {
     if (st40_rx_free(src->rx_handle)) {
       GST_ERROR("Failed to free rx handle");

ecosystem/gstreamer_plugin/gst_mtl_st40_rx.h

@@ -47,6 +47,16 @@
 #ifndef __GST_MTL_ST40_RX_H__
 #define __GST_MTL_ST40_RX_H__
 
+#define IS_POWER_OF_2(x) (((x) & ((x)-1)) == 0)
+#define ST40_RFC8331_PAYLOAD_MAX_ANCILLARY_COUNT 20
+/**
+ * ST40_SIZE_OF_PAYLOAD_METADATA:
+ * Defines the size (in bytes) of the payload metadata for ST40 streams.
+ * This value is set to 8 to accommodate the following fields:
+ * c | line_number | horizontal_offset | s | stream_num | did | sdid | data_count
+ */
+#define ST40_BYTE_SIZE_OF_PAYLOAD_METADATA 8
+
 #include <gst/base/gstbasesrc.h>
 
 #include "gst_mtl_common.h"
@@ -70,8 +80,7 @@ struct _Gst_Mtl_St40_Rx {
   SessionPortArgs portArgs; /* imtl session device */
   guint timeout_mbuf_get_seconds;
   guint16 mbuff_size;
-  guint16 udw_size;
-  char* anc_data;
+  gboolean include_metadata_in_buffer;
 };
 
 G_END_DECLS

ecosystem/gstreamer_plugin/gst_mtl_st40p_tx.c

@@ -567,7 +567,7 @@ static GstFlowReturn gst_mtl_st40p_tx_parse_8331_anc_words(
     data_count = payload_header.second_hdr_chunk.data_count & 0xff;
 
     /* data count * 10 bits + 10 bit checksum*/
-    udw_byte_size = (data_count * UDW_WORD_BIT_SIZE) + UDW_WORD_BIT_SIZE;
+    udw_byte_size = (data_count * UDW_WORD_BIT_SIZE) + UDW_WORD_BIT_SIZE - 2;
     /* round up to the nearest byte */
     udw_byte_size = (udw_byte_size + 7) / 8;