From be9d737c0cf5c3974804430e1b7f49a0d059fe60 Mon Sep 17 00:00:00 2001 From: "Wesierski, Dawid" Date: Tue, 16 Jun 2026 02:58:06 -0400 Subject: [PATCH 1/2] Add: Grace period for ST20p TX finalize to drain pending GstBuffers The zero-copy ST20p TX path hands GstBuffer-backed external frames to MTL and reclaims them in the notify_frame_done callback once the NIC is done with each frame. On shutdown, gst_mtl_st20p_tx_finalize freed the MTL session immediately, so frames the NIC still owned were abandoned and the data-plane tripped its "frame refcnt not zero" guard, leaking the GstBuffer references. Track the in-flight zero-copy buffers in an atomic pending_gst_buffers counter: increment it when a buffer enters the zero-copy chain (rolled back if the push fails) and decrement it in frame_done. finalize now blocks on gst_mtl_common_wait_pending_buffers, polling until the counter drains or a bounded grace period elapses (default 1s, aligned with the lib's TX block timeout), and warns instead of hanging teardown if it times out. The wait runs while the MTL lcore is still live so late completions are honoured. finalize also chains up to the parent class, which the previous implementation omitted. Signed-off-by: Wesierski, Dawid --- ecosystem/gstreamer_plugin/gst_mtl_common.c | 19 ++++++++++ ecosystem/gstreamer_plugin/gst_mtl_common.h | 3 ++ ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.c | 36 ++++++++++++++++++- ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.h | 1 + ecosystem/gstreamer_plugin/meson.build | 2 ++ 5 files changed, 60 insertions(+), 1 deletion(-) diff --git a/ecosystem/gstreamer_plugin/gst_mtl_common.c b/ecosystem/gstreamer_plugin/gst_mtl_common.c index 53d739a0e..e6951926d 100644 --- a/ecosystem/gstreamer_plugin/gst_mtl_common.c +++ b/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, diff --git a/ecosystem/gstreamer_plugin/gst_mtl_common.h b/ecosystem/gstreamer_plugin/gst_mtl_common.h index f4bf0c24f..801a78f48 100644 --- a/ecosystem/gstreamer_plugin/gst_mtl_common.h +++ b/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__ */ \ No newline at end of file diff --git a/ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.c b/ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.c index 9b52c8fa6..3224d98e0 100644 --- a/ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.c +++ b/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); diff --git a/ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.h b/ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.h index bf1bcb9a4..c84895fec 100644 --- a/ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.h +++ b/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; diff --git a/ecosystem/gstreamer_plugin/meson.build b/ecosystem/gstreamer_plugin/meson.build index 422eabea1..8e9cf74af 100644 --- a/ecosystem/gstreamer_plugin/meson.build +++ b/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 ) + From 6a1e515e97f5818272f0d05d2d8f1fd03948350d Mon Sep 17 00:00:00 2001 From: "Wesierski, Dawid" Date: Tue, 16 Jun 2026 02:58:12 -0400 Subject: [PATCH 2/2] Test: Add unit and validation tests for ST20p TX finalize grace period Adds comprehensive coverage for the zero-copy buffer drain feature: 1. Unit tests: gst_mtl_st20p_tx_refcnt_test and gst_mtl_st20p_tx_finalize_test validate the atomic counter logic and the finalize blocking behavior using a mock harness. 2. Validation: Adds an acceptance test under tests/validation/gstreamer that streams v210 and I422_10LE zero-copy pipelines between real VFs. Asserts that TX reaches EOS and shutdown concludes without "refcnt not zero" warnings in the logs. 3. Engine: Updates GstreamerApp.py with a finalize-grace helper and optional queue-cap parameters to stay within iavf interrupt mapping thresholds on single-host VF loops. Signed-off-by: Wesierski, Dawid --- .../gst_mtl_st20p_tx_finalize_test.cpp | 87 ++++++++ .../unit/gstreamer/gst_mtl_st20p_tx_harness.c | 197 ++++++++++++++++++ .../unit/gstreamer/gst_mtl_st20p_tx_harness.h | 76 +++++++ .../gst_mtl_st20p_tx_refcnt_test.cpp | 130 ++++++++++++ tests/unit/meson.build | 41 ++++ tests/validation/mtl_engine/GstreamerApp.py | 120 +++++++++++ .../tests/single/gstreamer/refcnt/__init__.py | 0 .../gstreamer/refcnt/test_st20p_tx_refcnt.py | 124 +++++++++++ 8 files changed, 775 insertions(+) create mode 100644 tests/unit/gstreamer/gst_mtl_st20p_tx_finalize_test.cpp create mode 100644 tests/unit/gstreamer/gst_mtl_st20p_tx_harness.c create mode 100644 tests/unit/gstreamer/gst_mtl_st20p_tx_harness.h create mode 100644 tests/unit/gstreamer/gst_mtl_st20p_tx_refcnt_test.cpp create mode 100644 tests/validation/tests/single/gstreamer/refcnt/__init__.py create mode 100644 tests/validation/tests/single/gstreamer/refcnt/test_st20p_tx_refcnt.py diff --git a/tests/unit/gstreamer/gst_mtl_st20p_tx_finalize_test.cpp b/tests/unit/gstreamer/gst_mtl_st20p_tx_finalize_test.cpp new file mode 100644 index 000000000..ae9fac868 --- /dev/null +++ b/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 + +#include + +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"; +} diff --git a/tests/unit/gstreamer/gst_mtl_st20p_tx_harness.c b/tests/unit/gstreamer/gst_mtl_st20p_tx_harness.c new file mode 100644 index 000000000..3a8300b6c --- /dev/null +++ b/tests/unit/gstreamer/gst_mtl_st20p_tx_harness.c @@ -0,0 +1,197 @@ +/* SPDX-License-Identifier: BSD-3-Clause + * Copyright(c) 2026 Intel Corporation + * + * Pulls in the production ST20p TX sink so the static finalize symbol is + * reachable. MTL handles are kept NULL, so finalize's st20p_tx_free() and + * gst_mtl_common_deinit_handle() branches are skipped and only the graceful + * pending-GstBuffer wait runs — exactly the path under test. + */ + +#include + +/* Keep the timeout-path tests fast and deterministic: override the production + * finalize grace (1s) before pulling in the plugin. Exercises the same drain + * logic on a unit-friendly timescale. */ +#define GST_MTL_ST20P_TX_FINALIZE_GRACE_MS 200 + +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wunused-function" +#include "gst_mtl_st20p_tx.c" +#pragma GCC diagnostic pop + +#include "gstreamer/gst_mtl_st20p_tx_harness.h" + +void ut_st20p_tx_init(void) { + gst_init(NULL, NULL); +} + +Gst_Mtl_St20p_Tx* ut_st20p_tx_new(void) { + Gst_Mtl_St20p_Tx* sink = g_object_new(GST_TYPE_MTL_ST20P_TX, NULL); + if (!sink) return NULL; + + /* Keep teardown confined to the graceful drain: no real session/library. */ + sink->tx_handle = NULL; + sink->mtl_lib_handle = NULL; + sink->async_session_create = FALSE; + g_atomic_int_set(&sink->pending_gst_buffers, 0); + return sink; +} + +void ut_st20p_tx_set_pending(Gst_Mtl_St20p_Tx* sink, gint pending) { + g_atomic_int_set(&sink->pending_gst_buffers, pending); +} + +gint ut_st20p_tx_get_pending(Gst_Mtl_St20p_Tx* sink) { + return g_atomic_int_get(&sink->pending_gst_buffers); +} + +void ut_st20p_tx_dec_pending(Gst_Mtl_St20p_Tx* sink) { + g_atomic_int_dec_and_test(&sink->pending_gst_buffers); +} + +void ut_st20p_tx_finalize(Gst_Mtl_St20p_Tx* sink) { + gst_mtl_st20p_tx_finalize(G_OBJECT(sink)); +} + +/* --- Fake MTL transport --------------------------------------------------- + * Faithfully models the slot-refcnt contract of st_tx_video_session.c without a + * NIC. Each slot mirrors one st_frame_trans: refcnt is taken when the frame is + * handed to the (fake) NIC and dropped when its packet mbufs are "freed". */ + +#define UT_FAKE_SLOTS 4 + +struct ut_fake_transport { + gint refcnt[UT_FAKE_SLOTS]; /* mirrors st20_frames[i].refcnt */ + struct st_frame frame[UT_FAKE_SLOTS]; /* st_frame handed to the app; opaque=child */ + int bound[UT_FAKE_SLOTS]; /* slot holds an in-flight ext buffer */ + int stuck_frames; /* slots left refcnt!=0 at free */ + int refcnt_violation; /* "frame refcnt not zero" guard tripped */ +}; + +static struct ut_fake_transport g_ut_fake; + +void ut_st20p_tx_fake_reset(void) { + memset(&g_ut_fake, 0, sizeof(g_ut_fake)); +} + +void ut_st20p_tx_fake_attach(Gst_Mtl_St20p_Tx* sink) { + sink->tx_handle = (st20p_tx_handle)&g_ut_fake; + sink->mtl_lib_handle = NULL; + sink->zero_copy = TRUE; + sink->frame_size = 1024; + g_atomic_int_set(&sink->pending_gst_buffers, 0); +} + +/* Models the transport get_next_frame refcnt guard + * (st_tx_video_session.c: "frame %u refcnt not zero"). */ +static int ut_fake_acquire(int slot) { + if (g_atomic_int_get(&g_ut_fake.refcnt[slot]) != 0) { + g_printerr("frame %d refcnt not zero %d\n", slot, + g_atomic_int_get(&g_ut_fake.refcnt[slot])); + g_ut_fake.refcnt_violation = 1; + return -1; + } + g_atomic_int_set(&g_ut_fake.refcnt[slot], 1); + return 0; +} + +int ut_st20p_tx_put_ext_buffer(Gst_Mtl_St20p_Tx* sink) { + GstSt20pTxExternalDataParent* parent; + GstSt20pTxExternalDataChild* child; + GstBuffer* buf; + int slot = -1; + + for (int i = 0; i < UT_FAKE_SLOTS; i++) { + if (!g_ut_fake.bound[i]) { + slot = i; + break; + } + } + if (slot < 0) return -1; + + buf = gst_buffer_new_allocate(NULL, sink->frame_size, NULL); + if (!buf) return -1; + + parent = malloc(sizeof(*parent)); + parent->buf = buf; + parent->child_count = 1; + pthread_mutex_init(&parent->parent_mutex, NULL); + + child = malloc(sizeof(*child)); + memset(child, 0, sizeof(*child)); + child->parent = parent; + child->gst_buffer_memory = gst_buffer_peek_memory(buf, 0); + if (!gst_memory_map(child->gst_buffer_memory, &child->map_info, GST_MAP_READ)) { + free(child); + pthread_mutex_destroy(&parent->parent_mutex); + gst_buffer_unref(buf); + free(parent); + return -1; + } + + /* chain(): one in-flight GstBuffer */ + g_atomic_int_inc(&sink->pending_gst_buffers); + + /* put_ext_frame + tasklet acquire bind the ext buffer to a transport slot */ + g_ut_fake.frame[slot].opaque = child; + g_ut_fake.bound[slot] = 1; + ut_fake_acquire(slot); + return slot; +} + +void ut_st20p_tx_transport_reacquire(Gst_Mtl_St20p_Tx* sink, int slot) { + (void)sink; + if (slot < 0 || slot >= UT_FAKE_SLOTS) return; + ut_fake_acquire(slot); +} + +void ut_st20p_tx_transport_complete(Gst_Mtl_St20p_Tx* sink, int slot) { + if (slot < 0 || slot >= UT_FAKE_SLOTS) return; + /* tv_frame_free_cb: the NIC freed every packet mbuf. Drop the transport + * refcnt first, then notify the app. Ordering refcnt-before-notify keeps the + * pending counter authoritative for a concurrent finalize: once pending hits + * 0 (decremented inside frame_done) the slot is already free, so st20p_tx_free + * can never observe a half-completed frame. */ + g_atomic_int_set(&g_ut_fake.refcnt[slot], 0); + g_ut_fake.bound[slot] = 0; + gst_mtl_st20p_tx_frame_done(sink, &g_ut_fake.frame[slot]); +} + +int ut_st20p_tx_fake_stuck_frames(void) { + int n = 0; + for (int i = 0; i < UT_FAKE_SLOTS; i++) + if (g_atomic_int_get(&g_ut_fake.refcnt[i]) != 0) n++; + return n; +} + +int ut_st20p_tx_fake_refcnt_violation(void) { + return g_ut_fake.refcnt_violation; +} + +/* --- Fakes for the libmtl symbols the teardown path executes ---------------- + * The real symbols live in libmtl but need a NIC; -Wl,--allow-multiple-definition + * (set in unit_link_args) lets these in-tree models win at link time because the + * executable's own objects precede the archive. */ +int st20p_tx_notify_ext_frame_free(st20p_tx_handle handle, struct st_frame* frame) { + (void)handle; + (void)frame; /* slot lifetime is modeled by the ut_* helpers */ + return 0; +} + +int st20p_tx_free(st20p_tx_handle handle) { + struct ut_fake_transport* t = (struct ut_fake_transport*)handle; + if (!t) return 0; + /* Models st20p_tx_free's bounded flush: it does NOT wait for IN_TRANSMITTING + * frames to truly drain (the 50ms WA in tx_st20p_framebuffs_flush), so a frame + * the NIC never completed is abandoned with refcnt still set — exactly the + * stuck-frame / "frame are not free, refcnt" symptom. */ + for (int i = 0; i < UT_FAKE_SLOTS; i++) { + if (g_atomic_int_get(&t->refcnt[i]) != 0) { + g_printerr("frame %d are not free, refcnt %d\n", i, + g_atomic_int_get(&t->refcnt[i])); + t->stuck_frames++; + t->refcnt_violation = 1; + } + } + return 0; +} diff --git a/tests/unit/gstreamer/gst_mtl_st20p_tx_harness.h b/tests/unit/gstreamer/gst_mtl_st20p_tx_harness.h new file mode 100644 index 000000000..3ff4c760e --- /dev/null +++ b/tests/unit/gstreamer/gst_mtl_st20p_tx_harness.h @@ -0,0 +1,76 @@ +/* SPDX-License-Identifier: BSD-3-Clause + * Copyright(c) 2026 Intel Corporation + * + * C harness that exposes the ST20p TX sink's static finalize path + * (gst_mtl_st20p_tx_finalize) to the unit tests, so its graceful-shutdown + * grace period can be exercised without real MTL hardware. + * + * The sink is created with its MTL handles left NULL, so finalize's + * st20p_tx_free() / gst_mtl_common_deinit_handle() branches are skipped and + * only the pending-GstBuffer drain runs — the behaviour under test. + */ + +#ifndef UT_GST_MTL_ST20P_TX_HARNESS_H +#define UT_GST_MTL_ST20P_TX_HARNESS_H + +#include + +#ifdef __cplusplus +extern "C" { +#endif + +/* Opaque to the tests; the real struct lives in the production header. */ +typedef struct _Gst_Mtl_St20p_Tx Gst_Mtl_St20p_Tx; + +/* One-time GStreamer init. */ +void ut_st20p_tx_init(void); + +/* Create a sink whose MTL handles are NULL, confining finalize to the + * graceful pending-buffer wait. Returns NULL on failure. */ +Gst_Mtl_St20p_Tx* ut_st20p_tx_new(void); + +void ut_st20p_tx_set_pending(Gst_Mtl_St20p_Tx* sink, gint pending); +gint ut_st20p_tx_get_pending(Gst_Mtl_St20p_Tx* sink); +void ut_st20p_tx_dec_pending(Gst_Mtl_St20p_Tx* sink); + +/* Invoke the real (static) finalize — the graceful-shutdown unit under test. */ +void ut_st20p_tx_finalize(Gst_Mtl_St20p_Tx* sink); + +/* --- Fake MTL transport --------------------------------------------------- + * Models the st20_frames[] slot-refcnt contract from st_tx_video_session.c so + * the zero-copy teardown path can be exercised without a NIC. A transport slot + * is refcnt=1 from the moment a frame is handed to the (fake) NIC until the + * frame's packet mbufs are "freed" — mirroring how the real driver drives + * notify_frame_done. */ + +/* Reset all fake-transport state between tests. */ +void ut_st20p_tx_fake_reset(void); + +/* Point sink->tx_handle at the fake transport and put the sink in zero-copy + * mode with a small frame size; MTL lib handle stays NULL. */ +void ut_st20p_tx_fake_attach(Gst_Mtl_St20p_Tx* sink); + +/* Model chain()+zero_copy()+tasklet-acquire for one single-memory GstBuffer: + * allocate+map a real GstBuffer, build the plugin's parent/child structs, + * increment pending, bind to a transport slot and take the refcnt (0->1). + * Returns the slot index (>=0) or -1 on failure. */ +int ut_st20p_tx_put_ext_buffer(Gst_Mtl_St20p_Tx* sink); + +/* Model the tasklet's get_next_frame acquire on an already-bound slot (the + * slot-reuse-before-drain window). Trips the "frame refcnt not zero" guard if + * the slot is still referenced. */ +void ut_st20p_tx_transport_reacquire(Gst_Mtl_St20p_Tx* sink, int slot); + +/* Model tv_frame_free_cb after the NIC freed every packet mbuf: fire the real + * app frame_done (unmap + unref + pending--), then drop the transport refcnt. */ +void ut_st20p_tx_transport_complete(Gst_Mtl_St20p_Tx* sink, int slot); + +/* Observers. */ +int ut_st20p_tx_fake_stuck_frames(void); /* slots left refcnt!=0 */ +int ut_st20p_tx_fake_refcnt_violation(void); /* "refcnt not zero" guard tripped */ + +#ifdef __cplusplus +} +#endif + +#endif /* UT_GST_MTL_ST20P_TX_HARNESS_H */ diff --git a/tests/unit/gstreamer/gst_mtl_st20p_tx_refcnt_test.cpp b/tests/unit/gstreamer/gst_mtl_st20p_tx_refcnt_test.cpp new file mode 100644 index 000000000..6f78cbd14 --- /dev/null +++ b/tests/unit/gstreamer/gst_mtl_st20p_tx_refcnt_test.cpp @@ -0,0 +1,130 @@ +/* SPDX-License-Identifier: BSD-3-Clause + * Copyright(c) 2026 Intel Corporation + * + * Reproduces the "frame %u refcnt not zero" / stuck-frame symptom of the ST20p + * TX zero-copy path at unit scope. + * + * The literal message is emitted by the data-plane tasklet + * (st_tx_video_session.c:1883), which needs a NIC + lcore and therefore cannot + * run in UnitTest. Instead the harness models the transport's per-slot refcnt + * contract faithfully and drives the REAL plugin release path + * (gst_mtl_st20p_tx_frame_done) and the REAL static finalize through it, so the + * teardown ordering that leads to stuck frames is exercised without hardware. + * + * Negative terminology: each test asserts that the failure CONDITION is + * reproduced (the guard trips / the frame stays referenced / the buffer leaks). + */ + +#include +#include +#include + +extern "C" { +#include "gstreamer/gst_mtl_st20p_tx_harness.h" +} + +class St20pTxRefcntContractNegativeTest : public ::testing::Test { + protected: + void SetUp() override { + ut_st20p_tx_init(); + ut_st20p_tx_fake_reset(); + } +}; + +/* A transport slot reused before its previous TX drained trips the + * "frame refcnt not zero" guard — the steady-state root cause. */ +TEST_F(St20pTxRefcntContractNegativeTest, ReproducesRefcntNotZeroOnSlotReuse) { + Gst_Mtl_St20p_Tx* sink = ut_st20p_tx_new(); + ASSERT_NE(sink, nullptr); + ut_st20p_tx_fake_attach(sink); + + int slot = ut_st20p_tx_put_ext_buffer(sink); /* refcnt 0->1, pending 1 */ + ASSERT_GE(slot, 0); + ASSERT_EQ(ut_st20p_tx_get_pending(sink), 1); + + /* Reuse the same transport slot before frame_done drained it (the + * tv_frame_free_cb window: pipeline slot FREE but transport refcnt still 1). */ + ut_st20p_tx_transport_reacquire(sink, slot); + + EXPECT_TRUE(ut_st20p_tx_fake_refcnt_violation()) + << "reusing a transport slot while refcnt!=0 must trip 'frame refcnt not zero'"; + + /* Drain so the test does not leak. */ + ut_st20p_tx_transport_complete(sink, slot); + EXPECT_EQ(ut_st20p_tx_get_pending(sink), 0); +} + +/* Tearing down while a zero-copy frame is still in flight (frame_done never + * delivered) leaves the transport frame referenced and leaks the GstBuffer — + * the user's shutdown symptom. */ +TEST_F(St20pTxRefcntContractNegativeTest, ReproducesStuckFrameAndBufferLeakOnTeardown) { + Gst_Mtl_St20p_Tx* sink = ut_st20p_tx_new(); + ASSERT_NE(sink, nullptr); + ut_st20p_tx_fake_attach(sink); + + int slot = ut_st20p_tx_put_ext_buffer(sink); /* in flight: refcnt 1, pending 1 */ + ASSERT_GE(slot, 0); + ASSERT_EQ(ut_st20p_tx_get_pending(sink), 1); + + /* App tears down before the NIC finished TX. The grace wait times out, then + * st20p_tx_free runs without the frame having drained. */ + ut_st20p_tx_finalize(sink); + + EXPECT_GE(ut_st20p_tx_fake_stuck_frames(), 1) + << "transport frame still refcnt!=0 at free (reproduces stuck frame)"; + EXPECT_EQ(ut_st20p_tx_get_pending(sink), 1) + << "GstBuffer leaked: frame_done never reclaimed it"; +} + +/* Option A: finalize's authoritative drain. When the in-flight frame completes + * shortly after teardown begins (the common case — NIC finishes TX), finalize + * waits for frame_done to reclaim it before st20p_tx_free, so the transport is + * freed with no frame still referenced and the GstBuffer is not leaked. */ +class St20pTxFinalizeDrainTest : public ::testing::Test { + protected: + void SetUp() override { + ut_st20p_tx_init(); + ut_st20p_tx_fake_reset(); + } +}; + +namespace { +struct DrainCtx { + Gst_Mtl_St20p_Tx* sink; + int slot; +}; + +void* drain_complete_thread(void* arg) { + DrainCtx* c = static_cast(arg); + /* NIC frees the frame's mbufs shortly after teardown starts, well inside the + * grace window. */ + g_usleep(20 * 1000); + ut_st20p_tx_transport_complete(c->sink, c->slot); + return nullptr; +} +} // namespace + +TEST_F(St20pTxFinalizeDrainTest, FinalizeWaitsForInFlightFrameThenFreesClean) { + Gst_Mtl_St20p_Tx* sink = ut_st20p_tx_new(); + ASSERT_NE(sink, nullptr); + ut_st20p_tx_fake_attach(sink); + + int slot = ut_st20p_tx_put_ext_buffer(sink); /* in flight: refcnt 1, pending 1 */ + ASSERT_GE(slot, 0); + ASSERT_EQ(ut_st20p_tx_get_pending(sink), 1); + + DrainCtx ctx = {sink, slot}; + pthread_t th; + ASSERT_EQ(pthread_create(&th, nullptr, drain_complete_thread, &ctx), 0); + + ut_st20p_tx_finalize(sink); /* blocks on the grace wait until the frame drains */ + + pthread_join(th, nullptr); + + EXPECT_EQ(ut_st20p_tx_get_pending(sink), 0) + << "finalize must wait for the frame to drain before freeing"; + EXPECT_EQ(ut_st20p_tx_fake_stuck_frames(), 0) + << "no transport frame left referenced at free"; + EXPECT_FALSE(ut_st20p_tx_fake_refcnt_violation()) + << "clean drain must not trip the refcnt guard"; +} diff --git a/tests/unit/meson.build b/tests/unit/meson.build index 268b4bef3..f89c60133 100644 --- a/tests/unit/meson.build +++ b/tests/unit/meson.build @@ -61,3 +61,44 @@ executable('UnitTest', unit_sources, dependencies : [mtl_internal_dep, dpdk_dep, libm_dep, gtest_dep, libpthread_dep, usdt_provider_header_dep], install : false) + +gstreamer_unit_include_dirs = unit_include_dirs + [ + include_directories('../../ecosystem/gstreamer_plugin'), +] + +gstreamer_unit_c_args = unit_c_args + [ + '-Wno-redundant-decls', + '-Wno-discarded-qualifiers', + '-Wno-declaration-after-statement', + '-Wno-missing-prototypes', + '-Wno-missing-declarations', + '-Wno-stringop-truncation', +] + +gst_dep = dependency('gstreamer-1.0', required : false) +gstbase_dep = dependency('gstreamer-base-1.0', required : false) +gstreamer_video_dep = dependency('gstreamer-video-1.0', required : false) +gstreamer_audio_dep = dependency('gstreamer-audio-1.0', required : false) +glib_dep = dependency('glib-2.0', required : false) + +if gst_dep.found() and gstbase_dep.found() and gstreamer_video_dep.found() and gstreamer_audio_dep.found() and glib_dep.found() + gstreamer_unit_sources = [ + '../../ecosystem/gstreamer_plugin/gst_mtl_common.c', + 'gstreamer/gst_mtl_st20p_tx_harness.c', + 'gstreamer/gst_mtl_st20p_tx_finalize_test.cpp', + 'gstreamer/gst_mtl_st20p_tx_refcnt_test.cpp', + 'main.cpp', + ] + + executable('UnitTestGstreamer', gstreamer_unit_sources, + c_args : gstreamer_unit_c_args, + cpp_args : unit_cpp_args, + link_args : unit_link_args, + include_directories : gstreamer_unit_include_dirs, + dependencies : [mtl_internal_dep, dpdk_dep, libm_dep, gtest_dep, libpthread_dep, + usdt_provider_header_dep, gst_dep, gstbase_dep, + gstreamer_video_dep, gstreamer_audio_dep, glib_dep], + install : false) +else + message('Skipping UnitTestGstreamer: required gstreamer/glib dependencies not found') +endif diff --git a/tests/validation/mtl_engine/GstreamerApp.py b/tests/validation/mtl_engine/GstreamerApp.py index 887277ebb..6be1342fd 100755 --- a/tests/validation/mtl_engine/GstreamerApp.py +++ b/tests/validation/mtl_engine/GstreamerApp.py @@ -14,6 +14,21 @@ logger = logging.getLogger(__name__) +# Log markers that prove the ST20P TX zero-copy refcount / finalize grace-period +# contract was broken. Both MUST be absent from the TX output after a clean +# shutdown: +# - "refcnt not zero": the MTL TX data-plane guard fired because the transport +# freed a frame slot whose mbufs were still owned by the NIC, i.e. finalize +# tore the session down before the in-flight zero-copy GstBuffers drained +# (lib/src/st2110/st_tx_video_session.c). +# - "Finalize timeout waiting for pending GstBuffers": the plugin's bounded +# grace wait expired with pending_gst_buffers still > 0 +# (ecosystem/gstreamer_plugin/gst_mtl_st20p_tx.c). +ST20P_TX_REFCNT_VIOLATION_MARKERS = ( + "refcnt not zero", + "Finalize timeout waiting for pending GstBuffers", +) + def capture_stdout(process, process_name): """ @@ -206,6 +221,7 @@ def setup_gstreamer_st20p_tx_pipeline( tx_queues: int, tx_framebuff_num: int = None, tx_fps: int = None, + rx_queues: int = None, ): connection_params = create_connection_params( dev_port=nic_port_list, @@ -254,6 +270,12 @@ def setup_gstreamer_st20p_tx_pipeline( pipeline_command.extend(["mtl_st20p_tx", f"tx-queues={tx_queues}"]) + # A TX-only session needs only a minimal RX queue count (ARP/control); capping + # it keeps the VF below the iavf "large VF" queue-pair threshold on PFs that + # reject large-VF interrupt mapping. + if rx_queues is not None: + pipeline_command.append(f"rx-queues={rx_queues}") + if tx_framebuff_num is not None: pipeline_command.append(f"tx-framebuff-num={tx_framebuff_num}") @@ -280,6 +302,7 @@ def setup_gstreamer_st20p_rx_pipeline( rx_queues: int, rx_framebuff_num: int = None, rx_fps: int = None, + tx_queues: int = None, ): connection_params = create_connection_params( dev_port=nic_port_list, @@ -311,6 +334,11 @@ def setup_gstreamer_st20p_rx_pipeline( if rx_framebuff_num is not None: pipeline_command.append(f"rx-framebuff-num={rx_framebuff_num}") + # An RX-only session needs only a minimal TX queue count (ARP/control); capping + # it keeps the VF below the iavf "large VF" queue-pair threshold. + if tx_queues is not None: + pipeline_command.append(f"tx-queues={tx_queues}") + pipeline_command.extend(["!", "filesink", f"location={output_path}"]) pipeline_command.append(f"--gst-plugin-path={setup_gstreamer_plugins_paths(build)}") @@ -1016,6 +1044,98 @@ def compare_files( return False +def execute_st20p_tx_finalize_grace( + build: str, + tx_command: list, + rx_command: list, + host, + tx_run_timeout: int = 90, + rx_lead_time: int = 4, + gst_debug: str = "mtl_st20p_tx:4", +) -> dict: + """Stream a finite ST20P zero-copy clip and drive TX through finalize. + + RX is started first so ARP resolves and TX actually paces frames onto the + wire, leaving real in-flight zero-copy frames at end-of-stream. TX then runs + until its finite filesrc reaches EOS; gst-launch tears the pipeline down on + EOS, which invokes gst_mtl_st20p_tx_finalize() and its pending_gst_buffers + grace drain. The captured TX stdout (stderr merged) is returned so the caller + can assert the refcount drained cleanly. + + ``GST_DEBUG`` is raised for the plugin category only, so the finalize-timeout + warning is visible; the MTL "refcnt not zero" guard is an err() that prints + regardless. + + :return: dict with keys ``tx_output`` (str), ``tx_return_code`` (int|None), + ``tx_clean_exit`` (bool) and ``violations`` (list[str]). + """ + rx_process = None + tx_process = None + tx_output = "" + tx_return_code = None + + try: + logger.info("Starting RX pipeline (receiver / link partner)...") + rx_process = run( + " ".join(rx_command), + cwd=build, + timeout=tx_run_timeout + 60, + testcmd=True, + host=host, + background=True, + ) + time.sleep(rx_lead_time) + + tx_cmd = " ".join(tx_command) + if gst_debug: + tx_cmd = f"GST_DEBUG={gst_debug} {tx_cmd}" + + logger.info("Starting TX pipeline (zero-copy)...") + tx_process = run( + tx_cmd, + cwd=build, + timeout=tx_run_timeout + 60, + testcmd=True, + host=host, + background=True, + ) + + # gst-launch exits 0 only on a clean EOS-driven teardown, which is the + # path that runs the finalize grace drain. Wait for that to happen on + # its own instead of killing TX. + logger.info(f"Waiting up to {tx_run_timeout}s for TX to reach EOS...") + try: + tx_process.wait(timeout=tx_run_timeout) + except Exception as e: + logger.warning(f"TX did not exit within {tx_run_timeout}s: {e}") + + tx_return_code = getattr(tx_process, "return_code", None) + tx_output = capture_stdout(tx_process, "TX") or "" + finally: + for proc, name in ((tx_process, "TX"), (rx_process, "RX")): + if proc and is_process_running(proc): + logger.info(f"Killing leftover {name} process") + try: + proc.kill() + except Exception as e: + logger.warning(f"Failed to kill {name} process: {e}") + + for line in tx_output.splitlines(): + logger.info(f"TX Output: {line}") + + violations = [m for m in ST20P_TX_REFCNT_VIOLATION_MARKERS if m in tx_output] + logger.info( + f"TX finalize result: return_code={tx_return_code} violations={violations}" + ) + + return { + "tx_output": tx_output, + "tx_return_code": tx_return_code, + "tx_clean_exit": tx_return_code == 0, + "violations": violations, + } + + def video_format_change(file_format): if file_format in ["YUV422PLANAR10LE", "YUV_422_10bit"]: return "I422_10LE" diff --git a/tests/validation/tests/single/gstreamer/refcnt/__init__.py b/tests/validation/tests/single/gstreamer/refcnt/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/validation/tests/single/gstreamer/refcnt/test_st20p_tx_refcnt.py b/tests/validation/tests/single/gstreamer/refcnt/test_st20p_tx_refcnt.py new file mode 100644 index 000000000..14df83aad --- /dev/null +++ b/tests/validation/tests/single/gstreamer/refcnt/test_st20p_tx_refcnt.py @@ -0,0 +1,124 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright(c) 2026 Intel Corporation + +"""GStreamer ST20P TX finalize / zero-copy refcount acceptance tests. + +Validates the finalize grace period added to ``gst_mtl_st20p_tx``: on shutdown +the element must wait for the in-flight zero-copy GstBuffers (tracked by the +``pending_gst_buffers`` refcount) to drain before tearing down the MTL session, +so the transport never frees a frame slot the NIC still owns. + +The ``v210`` / ``I422_10LE`` inputs always select the zero-copy path +(``st_frame_fmt_to_transport()`` maps neither to the hardcoded +``ST20_FMT_YUV_422_10BIT`` transport format, so ``sink->zero_copy`` is true), +and that is the only path that arms the refcount. A clean EOS-driven shutdown +after real streaming must therefore leave NO ``refcnt not zero`` (the MTL +data-plane guard) and NO finalize-timeout warning in the TX log. + +Unit-level coverage of the same contract lives in +``tests/unit/gstreamer/gst_mtl_st20p_tx_refcnt_test.cpp`` and +``gst_mtl_st20p_tx_finalize_test.cpp``; these tests exercise it end-to-end on a +real VF pair. +""" + +import mtl_engine.media_creator as media_create +import pytest +from common.nicctl import InterfaceSetup +from mtl_engine import GstreamerApp +from mtl_engine.media_files import gstreamer_formats + +# The GStreamer ST20P TX element hardcodes transport_fmt = ST20_FMT_YUV_422_10BIT +# and sets zero_copy = (transport_fmt != st_frame_fmt_to_transport(input_fmt)). +# Both catalogued GStreamer inputs (v210 -> ST_FRAME_FMT_V210, I422_10LE -> +# ST_FRAME_FMT_YUV422PLANAR10LE) fall through st_frame_fmt_to_transport()'s +# switch to ST20_FMT_MAX, so zero_copy is always true for them -- and that is the +# only path that arms the pending_gst_buffers refcount. These are exactly the +# proven-good 1920x1080@60 params the video_format suite streams, so we reuse the +# catalog instead of re-deriving (v210 in particular requires width % 6 == 0 and +# is broken at 720p per SDBQ-1971). +ZERO_COPY_FORMATS = gstreamer_formats + +WORK_DIR = "/tmp/mtl_gst_st20p_refcnt" + + +@pytest.mark.nightly +@pytest.mark.parametrize("fmt_key", list(ZERO_COPY_FORMATS.keys())) +def test_st20p_tx_finalize_refcnt_drains( + hosts, + mtl_path, + setup_interfaces: InterfaceSetup, + test_config, + fmt_key, +): + """A zero-copy ST20P TX pipeline must finalize without leaking refcounts. + + Streams a short synthetic clip TX -> RX on a single host, lets TX reach EOS + on its own (triggering the finalize grace drain), and asserts the TX log is + free of the refcount-violation markers. + """ + spec = ZERO_COPY_FORMATS[fmt_key] + host = list(hosts.values())[0] + interfaces_list = setup_interfaces.get_interfaces_list_single( + test_config.get("interface_type", "VF") + ) + + host.connection.execute_command(f"mkdir -p {WORK_DIR}", expected_return_codes=None) + + gst_format = GstreamerApp.video_format_change(spec["format"]) + input_file_path = media_create.create_video_file( + width=spec["width"], + height=spec["height"], + framerate=spec["fps"], + format=gst_format, + media_path=WORK_DIR, + output_path=f"refcnt_{fmt_key}.yuv", + duration=2, + host=host, + ) + + tx_config = GstreamerApp.setup_gstreamer_st20p_tx_pipeline( + build=mtl_path, + nic_port_list=interfaces_list[0], + input_path=input_file_path, + width=spec["width"], + height=spec["height"], + framerate=spec["fps"], + format=gst_format, + tx_payload_type=112, + tx_queues=4, + rx_queues=1, + ) + + # RX only needs to pull frames so TX paces normally; discard the payload. + rx_config = GstreamerApp.setup_gstreamer_st20p_rx_pipeline( + build=mtl_path, + nic_port_list=interfaces_list[1], + output_path="/dev/null", + width=spec["width"], + height=spec["height"], + framerate=spec["fps"], + format=gst_format, + rx_payload_type=112, + rx_queues=4, + tx_queues=1, + ) + + try: + result = GstreamerApp.execute_st20p_tx_finalize_grace( + build=mtl_path, + tx_command=tx_config, + rx_command=rx_config, + host=host, + ) + finally: + media_create.remove_file(input_file_path, host=host) + + assert result["tx_clean_exit"], ( + "TX gst-launch did not exit cleanly on EOS " + f"(return_code={result['tx_return_code']}); the finalize grace path may " + "not have run" + ) + assert not result["violations"], ( + "ST20P TX zero-copy refcount / finalize grace-period was violated on " + f"shutdown: {result['violations']}" + )