Gst grace period

ecosystem/gstreamer_plugin/gst_mtl_common.c

@@ -189,6 +189,25 @@ gboolean gst_mtl_common_st_to_gst_sampling(enum st30_sampling st_sampling,
   }
 }
 
+gboolean gst_mtl_common_wait_pending_buffers(gint* pending_buffers, guint timeout_ms,
+                                             guint poll_sleep_us) {
+  gint64 deadline_us;
+
+  if (!pending_buffers) return TRUE;
+  if (g_atomic_int_get(pending_buffers) <= 0) return TRUE;
+  if (!timeout_ms) return FALSE;
+
+  if (!poll_sleep_us) poll_sleep_us = 1000;
+
+  deadline_us = g_get_monotonic_time() + ((gint64)timeout_ms * 1000);
+  while (g_atomic_int_get(pending_buffers) > 0) {
+    if (g_get_monotonic_time() >= deadline_us) return FALSE;
+    g_usleep(poll_sleep_us);
+  }
+
+  return TRUE;
+}
+
 void gst_mtl_common_init_general_arguments(GObjectClass* gobject_class) {
   g_object_class_install_property(
       gobject_class, PROP_GENERAL_LOG_LEVEL,

ecosystem/gstreamer_plugin/gst_mtl_common.h

@@ -115,4 +115,7 @@ mtl_handle gst_mtl_common_init_handle(GeneralArgs* generalArgs,
                                       gboolean force_to_initialize_new_instance);
 
 gint gst_mtl_common_deinit_handle(mtl_handle* handle);
+
+gboolean gst_mtl_common_wait_pending_buffers(gint* pending_buffers, guint timeout_ms,
+                                             guint poll_sleep_us);
 #endif /* __GST_MTL_COMMON_H__ */

ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.c

@@ -69,6 +69,21 @@
 
 #include "gst_mtl_st20p_tx.h"
 
+/*
+ * Grace period for finalize to wait for in-flight zero-copy frames to drain
+ * (the NIC frees the ext-buf mbufs, firing frame_done) before st20p_tx_free
+ * tears down the transport. Waiting here is the authoritative drain: the MTL
+ * lcore is still live (mtl_stop/uninit run later), so late completions are
+ * honored instead of abandoned with a non-zero refcnt. Aligned with the lib's
+ * default TX block timeout (NS_PER_S) so finalize never gives up before the
+ * transport's own get-frame block could resolve; bounded so a dead link cannot
+ * hang teardown. Overridable at compile time for tests.
+ */
+#ifndef GST_MTL_ST20P_TX_FINALIZE_GRACE_MS
+#define GST_MTL_ST20P_TX_FINALIZE_GRACE_MS 1000
+#endif
+#define GST_MTL_ST20P_TX_FINALIZE_POLL_US 1000
+
 GST_DEBUG_CATEGORY_STATIC(gst_mtl_st20p_tx_debug);
 #define GST_CAT_DEFAULT gst_mtl_st20p_tx_debug
 #ifndef GST_LICENSE
@@ -253,6 +268,8 @@ static void gst_mtl_st20p_tx_init(Gst_Mtl_St20p_Tx* sink) {
   GstElement* element = GST_ELEMENT(sink);
   GstPad* sinkpad;
 
+  g_atomic_int_set(&sink->pending_gst_buffers, 0);
+
   sinkpad = gst_element_get_static_pad(element, "sink");
   if (!sinkpad) {
     GST_ERROR_OBJECT(sink, "Failed to get sink pad from child element");
@@ -511,7 +528,13 @@ static GstFlowReturn gst_mtl_st20p_tx_chain(GstPad* pad, GstObject* parent,
   }
 
   if (sink->zero_copy) {
-    return gst_mtl_st20p_tx_zero_copy(sink, buf);
+    GstFlowReturn ret;
+
+    g_atomic_int_inc(&sink->pending_gst_buffers);
+    ret = gst_mtl_st20p_tx_zero_copy(sink, buf);
+    if (ret != GST_FLOW_OK) g_atomic_int_dec_and_test(&sink->pending_gst_buffers);
+
+    return ret;
   } else {
     return gst_mtl_st20p_tx_mem_copy(sink, buf);
   }
@@ -520,6 +543,14 @@ static GstFlowReturn gst_mtl_st20p_tx_chain(GstPad* pad, GstObject* parent,
 static void gst_mtl_st20p_tx_finalize(GObject* object) {
   Gst_Mtl_St20p_Tx* sink = GST_MTL_ST20P_TX(object);
 
+  if (!gst_mtl_common_wait_pending_buffers(&sink->pending_gst_buffers,
+                                           GST_MTL_ST20P_TX_FINALIZE_GRACE_MS,
+                                           GST_MTL_ST20P_TX_FINALIZE_POLL_US)) {
+    GST_WARNING_OBJECT(
+        sink, "Finalize timeout waiting for pending GstBuffers: %d still pending",
+        g_atomic_int_get(&sink->pending_gst_buffers));
+  }
+
   if (sink->async_session_create) {
     if (sink->session_thread) pthread_join(sink->session_thread, NULL);
     pthread_mutex_destroy(&sink->session_mutex);
@@ -533,6 +564,8 @@ static void gst_mtl_st20p_tx_finalize(GObject* object) {
     if (gst_mtl_common_deinit_handle(&sink->mtl_lib_handle))
       GST_ERROR("Failed to uninitialize MTL library");
   }
+
+  G_OBJECT_CLASS(parent_class)->finalize(object);
 }
 
 static gboolean plugin_init(GstPlugin* mtl_st20p_tx) {
@@ -588,6 +621,7 @@ static int gst_mtl_st20p_tx_frame_done(void* priv, struct st_frame* frame) {
 
   pthread_mutex_unlock(&parent->parent_mutex);
   gst_buffer_unref(parent->buf);
+  g_atomic_int_dec_and_test(&sink->pending_gst_buffers);
   pthread_mutex_destroy(&parent->parent_mutex);
   free(parent);
 

ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.h

@@ -58,6 +58,7 @@ struct _Gst_Mtl_St20p_Tx {
   GstVideoSink element;
   mtl_handle mtl_lib_handle;
   st20p_tx_handle tx_handle;
+  gint pending_gst_buffers;
   guint frame_size;
   gboolean zero_copy;
 

ecosystem/gstreamer_plugin/meson.build

@@ -26,6 +26,7 @@ gst_dep = dependency('gstreamer-1.0', version : '>=1.19', required : true)
 gstbase_dep = dependency('gstreamer-base-1.0', version : '>=1.19', required : true)
 gstreamer_video_dep = dependency('gstreamer-video-1.0', required : true)
 gstreamer_audio_dep = dependency('gstreamer-audio-1.0', required : true)
+glib_dep = dependency('glib-2.0', required : true)
 mtl_dep             = dependency('mtl', required : true)
 
 if run_command('test', '-d', '/usr/local/include/gstreamer-1.0', check: false).returncode() == 0
@@ -163,3 +164,4 @@ gst_mtl_st40p_rx = library('gstmtl_st40p_rx',
   include_directories: inc_dirs,
   c_args: plugin_c_args
 )
+

tests/unit/gstreamer/gst_mtl_st20p_tx_finalize_test.cpp

@@ -0,0 +1,87 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(c) 2026 Intel Corporation
+ *
+ * Graceful-shutdown contract for gst_mtl_st20p_tx_finalize():
+ *   - finalize must block until in-flight (pending) GstBuffers drain, so the
+ *     MTL session is not torn down while the DPDK lcore still owns them;
+ *   - if buffers never drain, finalize must still return after the bounded
+ *     grace window (shutdown must not hang);
+ *   - with nothing pending, finalize must return promptly.
+ *
+ * The static finalize is driven through a C harness that keeps the MTL
+ * handles NULL, so only the graceful drain path executes.
+ */
+
+#include <gtest/gtest.h>
+
+#include <thread>
+
+extern "C" {
+#include "gstreamer/gst_mtl_st20p_tx_harness.h"
+}
+
+namespace {
+/* Mirrors GST_MTL_ST20P_TX_FINALIZE_GRACE_MS in gst_mtl_st20p_tx.c. */
+constexpr gint64 kGraceMs = 200;
+
+gint64 now_ms() {
+  return g_get_monotonic_time() / 1000;
+}
+}  // namespace
+
+class St20pTxFinalizeTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    ut_st20p_tx_init();
+  }
+};
+
+/* A late frame_done that drains the last pending buffer mid-finalize must be
+ * waited for: finalize returns only once pending hits 0, well within grace. */
+TEST_F(St20pTxFinalizeTest, FinalizeWaitsForPendingDrain) {
+  Gst_Mtl_St20p_Tx* sink = ut_st20p_tx_new();
+  ASSERT_NE(sink, nullptr);
+  ut_st20p_tx_set_pending(sink, 1);
+
+  std::thread drainer([&] {
+    g_usleep(20 * 1000);
+    ut_st20p_tx_dec_pending(sink);
+  });
+
+  gint64 t0 = now_ms();
+  ut_st20p_tx_finalize(sink);
+  gint64 elapsed = now_ms() - t0;
+
+  drainer.join();
+
+  EXPECT_GE(elapsed, 15) << "finalize must block until the pending buffer drains";
+  EXPECT_LT(elapsed, kGraceMs) << "finalize must return as soon as buffers drain";
+  EXPECT_EQ(ut_st20p_tx_get_pending(sink), 0);
+}
+
+/* Negative case: buffers never drain. finalize must not hang — it returns after
+ * the bounded grace window with the buffers still outstanding. */
+TEST_F(St20pTxFinalizeTest, FinalizeTimesOutWhenBuffersNeverDrain) {
+  Gst_Mtl_St20p_Tx* sink = ut_st20p_tx_new();
+  ASSERT_NE(sink, nullptr);
+  ut_st20p_tx_set_pending(sink, 2);
+
+  gint64 t0 = now_ms();
+  ut_st20p_tx_finalize(sink);
+  gint64 elapsed = now_ms() - t0;
+
+  EXPECT_GE(elapsed, kGraceMs - 20) << "finalize must wait the full grace window";
+  EXPECT_EQ(ut_st20p_tx_get_pending(sink), 2) << "buffers never drained";
+}
+
+/* Nothing pending → finalize returns promptly (no grace wait). */
+TEST_F(St20pTxFinalizeTest, FinalizeFastWhenNothingPending) {
+  Gst_Mtl_St20p_Tx* sink = ut_st20p_tx_new();
+  ASSERT_NE(sink, nullptr);
+
+  gint64 t0 = now_ms();
+  ut_st20p_tx_finalize(sink);
+  gint64 elapsed = now_ms() - t0;
+
+  EXPECT_LT(elapsed, 50) << "no pending buffers must not trigger the grace wait";
+}