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 ) + 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']}" + )