gst: fix aravissrc teardown deadlock under incomplete frames

gst/gstaravis.c

@@ -47,6 +47,8 @@
 
 #define GST_ARAVIS_DEFAULT_N_BUFFERS		50
 #define GST_ARAVIS_BUFFER_TIMEOUT_DEFAULT	2000000
+/* Poll the buffer wait in slices this size so unlock() is noticed promptly. */
+#define GST_ARAVIS_BUFFER_POLL_SLICE_US		100000
 
 GST_DEBUG_CATEGORY_STATIC (aravis_debug);
 #define GST_CAT_DEFAULT aravis_debug
@@ -488,6 +490,10 @@ gst_aravis_start (GstBaseSrc *src)
         else
                 GST_LOG_OBJECT (gst_aravis, "Open first available camera");
 
+	/* Clear a stale flush from a previous cycle (unlock_stop is not guaranteed
+	 * on the way to NULL) so a reused element streams again. */
+	g_atomic_int_set (&gst_aravis->flushing, 0);
+
 	GST_OBJECT_LOCK (gst_aravis);
 	if (gst_aravis->camera == NULL)
 		result = gst_aravis_init_camera (gst_aravis, &notify, &error);
@@ -560,6 +566,28 @@ gst_aravis_get_times (GstBaseSrc * basesrc, GstBuffer * buffer,
 	}
 }
 
+/* Without unlock(), a create() blocked waiting for a buffer (no complete frames
+ * arriving) can't be interrupted, deadlocking the state change to NULL. */
+static gboolean
+gst_aravis_unlock (GstBaseSrc *src)
+{
+	GstAravis *gst_aravis = GST_ARAVIS (src);
+
+	g_atomic_int_set (&gst_aravis->flushing, 1);
+
+	return TRUE;
+}
+
+static gboolean
+gst_aravis_unlock_stop (GstBaseSrc *src)
+{
+	GstAravis *gst_aravis = GST_ARAVIS (src);
+
+	g_atomic_int_set (&gst_aravis->flushing, 0);
+
+	return TRUE;
+}
+
 static GstFlowReturn
 gst_aravis_create (GstPushSrc * push_src, GstBuffer ** buffer)
 {
@@ -571,18 +599,47 @@ gst_aravis_create (GstPushSrc * push_src, GstBuffer ** buffer)
 	guint64 timestamp_ns;
 	gboolean base_src_does_timestamp;
 	ArvBuffer *arv_buffer = NULL;
+	ArvStream *stream;
+	guint64 timeout_us;
+	gint64 deadline;
 
 	gst_aravis = GST_ARAVIS (push_src);
 	base_src_does_timestamp = gst_base_src_get_do_timestamp(GST_BASE_SRC(push_src));
 
+	/* Ref the stream and drop GST_OBJECT_LOCK: holding it across the buffer
+	 * wait would deadlock the state change that tears the element down. */
 	GST_OBJECT_LOCK (gst_aravis);
+	stream = gst_aravis->stream != NULL ? g_object_ref (gst_aravis->stream) : NULL;
+	timeout_us = gst_aravis->buffer_timeout_us;
+	GST_OBJECT_UNLOCK (gst_aravis);
+
+	if (stream == NULL)
+		return GST_FLOW_FLUSHING;
 
+	/* Poll for a complete buffer until the deadline, recycling the non-SUCCESS
+	 * buffers that incomplete frames produce instead of spinning on them. */
+	deadline = g_get_monotonic_time () + timeout_us;
 	do {
-		if (arv_buffer) arv_stream_push_buffer (gst_aravis->stream, arv_buffer);
-		arv_buffer = arv_stream_timeout_pop_buffer (gst_aravis->stream, gst_aravis->buffer_timeout_us);
-	} while (arv_buffer != NULL && arv_buffer_get_status (arv_buffer) != ARV_BUFFER_STATUS_SUCCESS);
+		gint64 now = g_get_monotonic_time ();
+		guint64 remaining = now < deadline ? (guint64) (deadline - now) : 0;
+
+		if (g_atomic_int_get (&gst_aravis->flushing)) {
+			if (arv_buffer != NULL)
+				arv_stream_push_buffer (stream, arv_buffer);
+			g_object_unref (stream);
+			return GST_FLOW_FLUSHING;
+		}
 
-	if (arv_buffer == NULL)
+		if (arv_buffer != NULL)
+			arv_stream_push_buffer (stream, arv_buffer);
+
+		arv_buffer = arv_stream_timeout_pop_buffer (stream, MIN (remaining, GST_ARAVIS_BUFFER_POLL_SLICE_US));
+
+		if (arv_buffer != NULL && arv_buffer_get_status (arv_buffer) == ARV_BUFFER_STATUS_SUCCESS)
+			break;
+	} while (g_get_monotonic_time () < deadline);
+
+	if (arv_buffer == NULL || arv_buffer_get_status (arv_buffer) != ARV_BUFFER_STATUS_SUCCESS)
 		goto error;
 
 	buffer_data = (char *) arv_buffer_get_data (arv_buffer, &buffer_size);
@@ -612,6 +669,7 @@ gst_aravis_create (GstPushSrc * push_src, GstBuffer ** buffer)
 	}
 
 	if (!base_src_does_timestamp) {
+		GST_OBJECT_LOCK (gst_aravis);
 		if (gst_aravis->timestamp_offset == 0) {
 			gst_aravis->timestamp_offset = timestamp_ns;
 			gst_aravis->last_timestamp = timestamp_ns;
@@ -621,15 +679,18 @@ gst_aravis_create (GstPushSrc * push_src, GstBuffer ** buffer)
 		GST_BUFFER_DURATION (*buffer) = timestamp_ns - gst_aravis->last_timestamp;
 
 		gst_aravis->last_timestamp = timestamp_ns;
+		GST_OBJECT_UNLOCK (gst_aravis);
 	}
 
-	arv_stream_push_buffer (gst_aravis->stream, arv_buffer);
-	GST_OBJECT_UNLOCK (gst_aravis);
+	arv_stream_push_buffer (stream, arv_buffer);
+	g_object_unref (stream);
 
 	return GST_FLOW_OK;
 
 error:
-	GST_OBJECT_UNLOCK (gst_aravis);
+	if (arv_buffer != NULL)
+		arv_stream_push_buffer (stream, arv_buffer);
+	g_object_unref (stream);
 	return GST_FLOW_ERROR;
 }
 
@@ -1197,6 +1258,8 @@ gst_aravis_class_init (GstAravisClass * klass)
 	gstbasesrc_class->fixate = GST_DEBUG_FUNCPTR (gst_aravis_fixate_caps);
 	gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_aravis_start);
 	gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_aravis_stop);
+	gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_aravis_unlock);
+	gstbasesrc_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_aravis_unlock_stop);
 	gstbasesrc_class->query = GST_DEBUG_FUNCPTR (gst_aravis_query);
 
 	gstbasesrc_class->get_times = GST_DEBUG_FUNCPTR (gst_aravis_get_times);

gst/gstaravis.h

@@ -85,6 +85,8 @@ struct _GstAravis {
 	char *trigger_source;
 
 	char *features;
+
+	volatile gint flushing;	/* atomic; set by unlock() to break create()'s buffer wait */
 };
 
 struct _GstAravisClass {

meson.build

@@ -119,6 +119,14 @@ else # not Linux
 endif
 
 subdir ('src')
+
+gst_enabled = false
+gst_option = get_option ('gst-plugin')
+gst_deps = aravis_dependencies + [dependency ('gstreamer-base-1.0', required: gst_option),
+                                  dependency ('gstreamer-app-1.0', required: gst_option)]
+subdir('gst', if_found: gst_deps)
+
+# built after gst so the test suite can exercise the GStreamer plugin
 subdir ('tests')
 
 viewer_enabled = false
@@ -130,12 +138,6 @@ viewer_deps = aravis_dependencies + [dependency ('gtk+-3.0', version: '>=3.12',
 subdir ('po', if_found: viewer_deps)
 subdir ('viewer', if_found: viewer_deps)
 
-gst_enabled = false
-gst_option = get_option ('gst-plugin')
-gst_deps = aravis_dependencies + [dependency ('gstreamer-base-1.0', required: gst_option),
-                                  dependency ('gstreamer-app-1.0', required: gst_option)]
-subdir('gst', if_found: gst_deps)
-
 doc_deps = dependency ('gi-docgen', version:'>= 2021.1', fallback: ['gi-docgen', 'dummy_dep'], required:get_option('documentation'))
 subdir('docs', if_found: doc_deps)
 

tests/arvgstteardowntest.c

@@ -0,0 +1,109 @@
+/* Aravis - Digital camera library
+ *
+ * SPDX-License-Identifier: LGPL-2.1-or-later
+ */
+
+/* Regression test for the aravissrc teardown deadlock: a GigE Vision stream
+ * with packet loss delivers a steady flow of incomplete (non-SUCCESS) buffers,
+ * which used to wedge gst_aravis_create() with no way to interrupt it, so
+ * set_state(NULL) never returned. Assert teardown completes within a bound. */
+
+#include <arv.h>
+#include <gst/gst.h>
+
+/* High enough that multi-packet frames essentially never complete. */
+#define TEARDOWN_TEST_LOST_RATIO	0.2
+
+static ArvGvFakeCamera *simulator = NULL;
+
+typedef struct {
+	GstElement *pipeline;
+	GMutex mutex;
+	GCond cond;
+	gboolean done;
+} TeardownData;
+
+static gpointer
+set_state_null_thread (gpointer user_data)
+{
+	TeardownData *data = user_data;
+
+	/* The call that used to deadlock. */
+	gst_element_set_state (data->pipeline, GST_STATE_NULL);
+
+	g_mutex_lock (&data->mutex);
+	data->done = TRUE;
+	g_cond_signal (&data->cond);
+	g_mutex_unlock (&data->mutex);
+
+	return NULL;
+}
+
+static void
+teardown_under_packet_loss_test (void)
+{
+	GstElement *pipeline;
+	GError *error = NULL;
+	TeardownData data = {0};
+	GThread *thread;
+	gint64 deadline;
+	gboolean torn_down;
+
+	pipeline = gst_parse_launch ("aravissrc camera-name=Aravis-GVTest ! fakesink sync=false",
+				     &error);
+	g_assert_no_error (error);
+	g_assert_nonnull (pipeline);
+
+	gst_element_set_state (pipeline, GST_STATE_PLAYING);
+	/* Let the streaming task settle into create()'s buffer wait. */
+	g_usleep (G_USEC_PER_SEC);
+
+	data.pipeline = pipeline;
+	g_mutex_init (&data.mutex);
+	g_cond_init (&data.cond);
+
+	thread = g_thread_new ("set-state-null", set_state_null_thread, &data);
+
+	deadline = g_get_monotonic_time () + 5 * G_USEC_PER_SEC;
+	g_mutex_lock (&data.mutex);
+	while (!data.done) {
+		if (!g_cond_wait_until (&data.cond, &data.mutex, deadline))
+			break;
+	}
+	torn_down = data.done;
+	g_mutex_unlock (&data.mutex);
+
+	/* Aborts here pre-fix (deadlock) instead of hanging the whole binary. */
+	g_assert_true (torn_down);
+
+	g_thread_join (thread);
+	gst_object_unref (pipeline);
+	g_mutex_clear (&data.mutex);
+	g_cond_clear (&data.cond);
+}
+
+int
+main (int argc, char **argv)
+{
+	int result;
+
+	gst_init (&argc, &argv);
+	g_test_init (&argc, &argv, NULL);
+
+	arv_set_fake_camera_genicam_filename (GENICAM_FILENAME);
+
+	simulator = arv_gv_fake_camera_new ("127.0.0.1", "GVTest");
+	g_assert (ARV_IS_GV_FAKE_CAMERA (simulator));
+	g_object_set (simulator, "gvsp-lost-ratio", TEARDOWN_TEST_LOST_RATIO, NULL);
+
+	arv_update_device_list ();
+
+	g_test_add_func ("/gstreamer/teardown-under-packet-loss", teardown_under_packet_loss_test);
+
+	result = g_test_run ();
+
+	g_object_unref (simulator);
+	arv_shutdown ();
+
+	return result;
+}

tests/meson.build

@@ -11,7 +11,8 @@ if get_option('tests')
       '--error-exitcode=1'
     ],
     env:['DEBUGINFOD_URLS=', 'ARV_TEST_IGNORE_BUFFER='],
-    exclude_suites:['python'],
+    # 'gstreamer' excluded like 'python': GStreamer's global allocations aren't worth suppressing
+    exclude_suites:['python', 'gstreamer'],
     timeout_multiplier:5
   )
 
@@ -38,6 +39,19 @@ if get_option('tests')
 		test (t[0], exe, suite: t[1], timeout: 60)
 	endforeach
 
+	# Regression test for the aravissrc teardown deadlock.
+	if gst_enabled
+		gst_teardown_test = executable ('arv-gst-teardown-test', 'arvgstteardowntest.c',
+			c_args: ['-DGENICAM_FILENAME="@0@/src/arv-fake-camera.xml"'.format (meson.project_source_root ())],
+			link_with: aravis_library,
+			dependencies: gst_deps + [dependency ('gstreamer-1.0')],
+			include_directories: [library_inc])
+		test ('gst-teardown', gst_teardown_test,
+			suite: 'gstreamer',
+			env: ['GST_PLUGIN_PATH=' + meson.project_build_root() / 'gst'],
+			timeout: 30)
+	endif
+
         py_script_config_data = configuration_data ()
         py_script_config_data.set ('GI_TYPELIB_PATH', meson.project_build_root() / 'src')
         py_script_config_data.set ('LD_LIBRARY_PATH', meson.project_build_root() / 'src')