multiqueue: Don't do an infinite loop in the loop function

Instead return after every iteration, which makes sure that the
stream lock is released for a short time after every iteration,
task state changes are checked, etc and this allows the task
to be stopped properly.

diff --git a/plugins/elements/gstmultiqueue.c b/plugins/elements/gstmultiqueue.c
index be653fbd36d6fdf1b7d64247148a046e513e4517..bd54f56ce312ee43fd6ec140669dd04316efd390 100644 (file)
--- a/plugins/elements/gstmultiqueue.c
+++ b/plugins/elements/gstmultiqueue.c
@@ -1022,95 +1022,93 @@ gst_multi_queue_loop (GstPad * pad)
   sq = (GstSingleQueue *) gst_pad_get_element_private (pad);
   mq = sq->mqueue;
 
-  do {
-    GST_DEBUG_OBJECT (mq, "SingleQueue %d : trying to pop an object", sq->id);
-
-    /* Get something from the queue, blocking until that happens, or we get
-     * flushed */
-    if (!(gst_data_queue_pop (sq->queue, &sitem)))
-      goto out_flushing;
-
-    item = (GstMultiQueueItem *) sitem;
-    newid = item->posid;
-
-    /* steal the object and destroy the item */
-    object = gst_multi_queue_item_steal_object (item);
-    gst_multi_queue_item_destroy (item);
-
-    GST_LOG_OBJECT (mq, "SingleQueue %d : newid:%d , oldid:%d",
-        sq->id, newid, oldid);
-
-    /* If we're not-linked, we do some extra work because we might need to
-     * wait before pushing. If we're linked but there's a gap in the IDs,
-     * or it's the first loop, or we just passed the previous highid, 
-     * we might need to wake some sleeping pad up, so there's extra work 
-     * there too */
-    if (sq->srcresult == GST_FLOW_NOT_LINKED ||
-        (oldid == G_MAXUINT32) || (newid != (oldid + 1)) ||
-        oldid > mq->highid) {
-      GST_LOG_OBJECT (mq, "CHECKING sq->srcresult: %s",
-          gst_flow_get_name (sq->srcresult));
-
-      GST_MULTI_QUEUE_MUTEX_LOCK (mq);
+  GST_DEBUG_OBJECT (mq, "SingleQueue %d : trying to pop an object", sq->id);
+
+  /* Get something from the queue, blocking until that happens, or we get
+   * flushed */
+  if (!(gst_data_queue_pop (sq->queue, &sitem)))
+    goto out_flushing;
+
+  item = (GstMultiQueueItem *) sitem;
+  newid = item->posid;
+
+  /* steal the object and destroy the item */
+  object = gst_multi_queue_item_steal_object (item);
+  gst_multi_queue_item_destroy (item);
+
+  GST_LOG_OBJECT (mq, "SingleQueue %d : newid:%d , oldid:%d",
+      sq->id, newid, oldid);
+
+  /* If we're not-linked, we do some extra work because we might need to
+   * wait before pushing. If we're linked but there's a gap in the IDs,
+   * or it's the first loop, or we just passed the previous highid, 
+   * we might need to wake some sleeping pad up, so there's extra work 
+   * there too */
+  if (sq->srcresult == GST_FLOW_NOT_LINKED ||
+      (oldid == G_MAXUINT32) || (newid != (oldid + 1)) || oldid > mq->highid) {
+    GST_LOG_OBJECT (mq, "CHECKING sq->srcresult: %s",
+        gst_flow_get_name (sq->srcresult));
 
-      /* Update the nextid so other threads know when to wake us up */
-      sq->nextid = newid;
+    GST_MULTI_QUEUE_MUTEX_LOCK (mq);
 
-      /* Update the oldid (the last ID we output) for highid tracking */
-      if (oldid != G_MAXUINT32)
-        sq->oldid = oldid;
+    /* Update the nextid so other threads know when to wake us up */
+    sq->nextid = newid;
 
-      if (sq->srcresult == GST_FLOW_NOT_LINKED) {
-        /* Go to sleep until it's time to push this buffer */
+    /* Update the oldid (the last ID we output) for highid tracking */
+    if (oldid != G_MAXUINT32)
+      sq->oldid = oldid;
 
-        /* Recompute the highid */
-        compute_high_id (mq);
-        while (newid > mq->highid && sq->srcresult == GST_FLOW_NOT_LINKED) {
-          GST_DEBUG_OBJECT (mq, "queue %d sleeping for not-linked wakeup with "
-              "newid %u and highid %u", sq->id, newid, mq->highid);
+    if (sq->srcresult == GST_FLOW_NOT_LINKED) {
+      /* Go to sleep until it's time to push this buffer */
 
+      /* Recompute the highid */
+      compute_high_id (mq);
+      while (newid > mq->highid && sq->srcresult == GST_FLOW_NOT_LINKED) {
+        GST_DEBUG_OBJECT (mq, "queue %d sleeping for not-linked wakeup with "
+            "newid %u and highid %u", sq->id, newid, mq->highid);
 
-          /* Wake up all non-linked pads before we sleep */
-          wake_up_next_non_linked (mq);
 
-          mq->numwaiting++;
-          g_cond_wait (sq->turn, mq->qlock);
-          mq->numwaiting--;
+        /* Wake up all non-linked pads before we sleep */
+        wake_up_next_non_linked (mq);
 
-          GST_DEBUG_OBJECT (mq, "queue %d woken from sleeping for not-linked "
-              "wakeup with newid %u and highid %u", sq->id, newid, mq->highid);
-        }
+        mq->numwaiting++;
+        g_cond_wait (sq->turn, mq->qlock);
+        mq->numwaiting--;
 
-        /* Re-compute the high_id in case someone else pushed */
-        compute_high_id (mq);
-      } else {
-        compute_high_id (mq);
-        /* Wake up all non-linked pads */
-        wake_up_next_non_linked (mq);
+        GST_DEBUG_OBJECT (mq, "queue %d woken from sleeping for not-linked "
+            "wakeup with newid %u and highid %u", sq->id, newid, mq->highid);
       }
-      /* We're done waiting, we can clear the nextid */
-      sq->nextid = 0;
 
-      GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
+      /* Re-compute the high_id in case someone else pushed */
+      compute_high_id (mq);
+    } else {
+      compute_high_id (mq);
+      /* Wake up all non-linked pads */
+      wake_up_next_non_linked (mq);
     }
+    /* We're done waiting, we can clear the nextid */
+    sq->nextid = 0;
 
-    GST_LOG_OBJECT (mq, "BEFORE PUSHING sq->srcresult: %s",
-        gst_flow_get_name (sq->srcresult));
+    GST_MULTI_QUEUE_MUTEX_UNLOCK (mq);
+  }
 
-    /* Try to push out the new object */
-    result = gst_single_queue_push_one (mq, sq, object);
-    sq->srcresult = result;
+  GST_LOG_OBJECT (mq, "BEFORE PUSHING sq->srcresult: %s",
+      gst_flow_get_name (sq->srcresult));
 
-    if (result != GST_FLOW_OK && result != GST_FLOW_NOT_LINKED
-        && result != GST_FLOW_UNEXPECTED)
-      goto out_flushing;
+  /* Try to push out the new object */
+  result = gst_single_queue_push_one (mq, sq, object);
+  sq->srcresult = result;
 
-    GST_LOG_OBJECT (mq, "AFTER PUSHING sq->srcresult: %s",
-        gst_flow_get_name (sq->srcresult));
+  if (result != GST_FLOW_OK && result != GST_FLOW_NOT_LINKED
+      && result != GST_FLOW_UNEXPECTED)
+    goto out_flushing;
 
-    oldid = newid;
-  }
-  while (TRUE);
+  GST_LOG_OBJECT (mq, "AFTER PUSHING sq->srcresult: %s",
+      gst_flow_get_name (sq->srcresult));
+
+  oldid = newid;
+
+  return;
 
 out_flushing:
   {