Fix multiqueue leaking buffers and events when downstream or the queue are flushing...

Original commit message from CVS:
* libs/gst/base/gstdataqueue.c:
* libs/gst/base/gstdataqueue.h:
* plugins/elements/gstmultiqueue.c: (gst_single_queue_push_one),
(gst_multi_queue_item_new), (gst_multi_queue_chain),
(gst_multi_queue_sink_event):
* tests/check/elements/multiqueue.c: (multiqueue_suite):
Fix multiqueue leaking buffers and events when downstream or the
queue are flushing. Make refcounting assumptions explicit and
document them (shouldn't break existing code that uses it other than
maybe leak miniobjects, but that already happens anyway). Add unit
test for the most common flushing case. Fixes #423700.

diff --git a/ChangeLog b/ChangeLog
index f810346..6c66d79 100644 (file)
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,17 @@
+2007-06-06  Tim-Philipp Müller  <tim at centricular dot net>
+
+       * libs/gst/base/gstdataqueue.c:
+       * libs/gst/base/gstdataqueue.h:
+       * plugins/elements/gstmultiqueue.c: (gst_single_queue_push_one),
+       (gst_multi_queue_item_new), (gst_multi_queue_chain),
+       (gst_multi_queue_sink_event):
+       * tests/check/elements/multiqueue.c: (multiqueue_suite):
+         Fix multiqueue leaking buffers and events when downstream or the
+         queue are flushing. Make refcounting assumptions explicit and
+         document them (shouldn't break existing code that uses it other than
+         maybe leak miniobjects, but that already happens anyway). Add unit
+         test for the most common flushing case. Fixes #423700.
+         
 2007-06-06  Sebastian Dröge  <slomo@circular-chaos.org>
 
        * libs/gst/controller/gstcontroller.c:

diff --git a/libs/gst/base/gstdataqueue.c b/libs/gst/base/gstdataqueue.c
index 5c009d4..a3bfde2 100644 (file)
--- a/libs/gst/base/gstdataqueue.c
+++ b/libs/gst/base/gstdataqueue.c
@@ -397,6 +397,11 @@ gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing)
  * available, OR the @queue is set to flushing state.
  * MT safe.
  *
+ * Note that this function has slightly different semantics than gst_pad_push()
+ * and gst_pad_push_event(): this function only takes ownership of @item and
+ * the #GstMiniObject contained in @item if the push was successful. If FALSE
+ * is returned, the caller is responsible for freeing @item and its contents.
+ *
  * Returns: #TRUE if the @item was successfully pushed on the @queue.
  */
 gboolean

diff --git a/libs/gst/base/gstdataqueue.h b/libs/gst/base/gstdataqueue.h
index b3cd87b..503e44c 100644 (file)
--- a/libs/gst/base/gstdataqueue.h
+++ b/libs/gst/base/gstdataqueue.h
@@ -48,7 +48,9 @@ typedef struct _GstDataQueueItem GstDataQueueItem;
  * @duration: the duration in #GstClockTime of the miniobject. Can not be
  * #GST_CLOCK_TIME_NONE.
  * @visible: #TRUE if @object should be considered as a visible object.
- * @destroy: The #GDestroyNotify to use to free the #GstDataQueueItem.
+ * @destroy: The #GDestroyNotify function to use to free the #GstDataQueueItem.
+ * This function should also drop the reference to @object the owner of the
+ * #GstDataQueueItem is assumed to hold.
  *
  * Structure used by #GstDataQueue. You can supply a different structure, as
  * long as the top of the structure is identical to this structure.

diff --git a/plugins/elements/gstmultiqueue.c b/plugins/elements/gstmultiqueue.c
index 3aca642..d4208af 100644 (file)
--- a/plugins/elements/gstmultiqueue.c
+++ b/plugins/elements/gstmultiqueue.c
@@ -427,7 +427,10 @@ gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
     GstMiniObject * object)
 {
   if (GST_IS_BUFFER (object)) {
-    sq->srcresult = gst_pad_push (sq->srcpad, GST_BUFFER (object));
+    GstBuffer *buf;
+
+    buf = gst_buffer_ref (GST_BUFFER_CAST (object));
+    sq->srcresult = gst_pad_push (sq->srcpad, buf);
 
     if ((sq->srcresult != GST_FLOW_OK)
         && (sq->srcresult != GST_FLOW_NOT_LINKED)) {
@@ -437,7 +440,10 @@ gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
       gst_pad_pause_task (sq->srcpad);
     }
   } else if (GST_IS_EVENT (object)) {
-    if (GST_EVENT_TYPE (object) == GST_EVENT_EOS) {
+    GstEvent *event;
+
+    event = gst_event_ref (GST_EVENT_CAST (object));
+    if (GST_EVENT_TYPE (event) == GST_EVENT_EOS) {
       sq->srcresult = GST_FLOW_UNEXPECTED;
 
       GST_DEBUG_OBJECT (mq, "GstSingleQueue %d : pausing queue, got EOS",
@@ -445,7 +451,7 @@ gst_single_queue_push_one (GstMultiQueue * mq, GstSingleQueue * sq,
       gst_data_queue_set_flushing (sq->queue, TRUE);
       gst_pad_pause_task (sq->srcpad);
     }
-    gst_pad_push_event (sq->srcpad, GST_EVENT (object));
+    gst_pad_push_event (sq->srcpad, event);
   } else {
     g_warning ("Unexpected object in singlequeue %d (refcounting problem?)",
         sq->id);
@@ -461,13 +467,14 @@ gst_multi_queue_item_destroy (GstMultiQueueItem * item)
   g_free (item);
 }
 
+/* takes ownership of passed mini object! */
 static GstMultiQueueItem *
 gst_multi_queue_item_new (GstMiniObject * object)
 {
   GstMultiQueueItem *item;
 
   item = g_new0 (GstMultiQueueItem, 1);
-  item->object = gst_mini_object_ref (object);
+  item->object = object;
   item->destroy = (GDestroyNotify) gst_multi_queue_item_destroy;
 
   if (GST_IS_BUFFER (object)) {
@@ -621,7 +628,6 @@ gst_multi_queue_chain (GstPad * pad, GstBuffer * buffer)
     GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
         sq->id, gst_flow_get_name (sq->srcresult));
     gst_multi_queue_item_destroy (item);
-    gst_buffer_unref (buffer);
     ret = sq->srcresult;
   }
 
@@ -718,7 +724,6 @@ gst_multi_queue_sink_event (GstPad * pad, GstEvent * event)
     GST_LOG_OBJECT (mq, "SingleQueue %d : exit because task paused, reason: %s",
         sq->id, gst_flow_get_name (sq->srcresult));
     gst_multi_queue_item_destroy (item);
-    gst_event_unref (event);
   }
 
 done:

diff --git a/tests/check/elements/multiqueue.c b/tests/check/elements/multiqueue.c
index ddf01ae..e6826e2 100644 (file)
--- a/tests/check/elements/multiqueue.c
+++ b/tests/check/elements/multiqueue.c
@@ -259,11 +259,7 @@ multiqueue_suite (void)
   suite_add_tcase (s, tc_chain);
   tcase_add_test (tc_chain, test_simple_create_destroy);
   tcase_add_test (tc_chain, test_simple_pipeline);
-
-  if (0) {
-    /* FIXME: this leaks buffers, disabled for now */
-    tcase_add_test (tc_chain, test_simple_shutdown_while_running);
-  }
+  tcase_add_test (tc_chain, test_simple_shutdown_while_running);
 
   /* FIXME: test_request_pads() needs some more fixes, see comments there */
   tcase_add_test (tc_chain, test_request_pads);