← Back to branch summary

~ubuntu-branches/ubuntu/trusty/gstreamer1.0/trusty

« back to all changes in this revision

Viewing changes to plugins/elements/gstqueue.c

  • Committer: Package Import Robot
  • Author(s): Sebastian Dröge
  • Date: 2012-08-08 18:12:33 UTC
  • mfrom: (1.1.3)
  • Revision ID: package-import@ubuntu.com-20120808181233-riejwxprfsxh1njl
Tags: 0.11.93-1
* New upstream release:
  + debian/libgstreamer.symbols:
    - Update symbols file.

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugins/elements/gstqueue.c
353
353
   *
354
354
   * Don't emit queue signals. Makes queues more lightweight if no signals are
355
355
   * needed.
356
 
   *
357
 
   * Since: 0.10.31
358
356
   */
359
357
  g_object_class_install_property (gobject_class, PROP_SILENT,
360
358
      g_param_spec_boolean ("silent", "Silent",
419
417
  g_cond_init (&queue->item_add);
420
418
  g_cond_init (&queue->item_del);
421
419
 
422
 
  g_queue_init (&queue->queue);
 
420
  gst_queue_array_init (&queue->queue, DEFAULT_MAX_SIZE_BUFFERS * 3 / 2);
423
421
 
424
422
  queue->sinktime = GST_CLOCK_TIME_NONE;
425
423
  queue->srctime = GST_CLOCK_TIME_NONE;
442
440
 
443
441
  GST_DEBUG_OBJECT (queue, "finalizing queue");
444
442
 
445
 
  while ((data = g_queue_pop_head (&queue->queue))) {
 
443
  while (!gst_queue_array_is_empty (&queue->queue)) {
 
444
    data = gst_queue_array_pop_head (&queue->queue);
 
445
    /* FIXME: if it's a query, shouldn't we unref that too? */
446
446
    if (!GST_IS_QUERY (data))
447
447
      gst_mini_object_unref (data);
448
448
  }
 
449
  gst_queue_array_clear (&queue->queue);
449
450
 
450
 
  g_queue_clear (&queue->queue);
451
451
  g_mutex_clear (&queue->qlock);
452
452
  g_cond_clear (&queue->item_add);
453
453
  g_cond_clear (&queue->item_del);
556
556
{
557
557
  GstMiniObject *data;
558
558
 
559
 
  while ((data = g_queue_pop_head (&queue->queue))) {
 
559
  while (!gst_queue_array_is_empty (&queue->queue)) {
 
560
    data = gst_queue_array_pop_head (&queue->queue);
560
561
    /* Then lose another reference because we are supposed to destroy that
561
562
       data when flushing */
562
563
    if (!GST_IS_QUERY (data))
588
589
  queue->cur_level.bytes += gst_buffer_get_size (buffer);
589
590
  apply_buffer (queue, buffer, &queue->sink_segment, TRUE, TRUE);
590
591
 
591
 
  g_queue_push_tail (&queue->queue, item);
 
592
  if (item)
 
593
    gst_queue_array_push_tail (&queue->queue, item);
592
594
  GST_QUEUE_SIGNAL_ADD (queue);
593
595
}
594
596
 
622
624
      break;
623
625
  }
624
626
 
625
 
  g_queue_push_tail (&queue->queue, item);
 
627
  if (item)
 
628
    gst_queue_array_push_tail (&queue->queue, item);
626
629
  GST_QUEUE_SIGNAL_ADD (queue);
627
630
}
628
631
 
632
635
{
633
636
  GstMiniObject *item;
634
637
 
635
 
  item = g_queue_pop_head (&queue->queue);
 
638
  item = gst_queue_array_pop_head (&queue->queue);
636
639
  if (item == NULL)
637
640
    goto no_item;
638
641
 
735
738
      queue->eos = FALSE;
736
739
      queue->unexpected = FALSE;
737
740
      gst_pad_start_task (queue->srcpad, (GstTaskFunction) gst_queue_loop,
738
 
          queue->srcpad);
 
741
          queue->srcpad, NULL);
739
742
      GST_QUEUE_MUTEX_UNLOCK (queue);
740
743
 
741
744
      STATUS (queue, pad, "after flush");
789
792
        GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
790
793
        GST_LOG_OBJECT (queue, "queuing query %p (%s)", query,
791
794
            GST_QUERY_TYPE_NAME (query));
792
 
        g_queue_push_tail (&queue->queue, query);
 
795
        gst_queue_array_push_tail (&queue->queue, query);
793
796
        GST_QUEUE_SIGNAL_ADD (queue);
794
797
        while (queue->queue.length != 0) {
795
798
          /* for as long as the queue has items, we know the query is
817
820
static gboolean
818
821
gst_queue_is_empty (GstQueue * queue)
819
822
{
 
823
  GstMiniObject *head;
 
824
 
820
825
  if (queue->queue.length == 0)
821
826
    return TRUE;
822
827
 
 
828
  /* Only consider the queue empty if the minimum thresholds
 
829
   * are not reached and data is at the queue head. Otherwise
 
830
   * we would block forever on serialized queries.
 
831
   */
 
832
  head = queue->queue.array[queue->queue.head];
 
833
  if (!GST_IS_BUFFER (head) && !GST_IS_BUFFER_LIST (head))
 
834
    return FALSE;
 
835
 
823
836
  /* It is possible that a max size is reached before all min thresholds are.
824
837
   * Therefore, only consider it empty if it is not filled. */
825
838
  return ((queue->min_threshold.buffers > 0 &&
1203
1216
  GstQueue *queue = GST_QUEUE (parent);
1204
1217
  gboolean res;
1205
1218
 
1206
 
  res = gst_pad_query_default (pad, parent, query);
 
1219
  switch (GST_QUERY_TYPE (query)) {
 
1220
    case GST_QUERY_SCHEDULING:{
 
1221
      gst_query_add_scheduling_mode (query, GST_PAD_MODE_PUSH);
 
1222
      res = TRUE;
 
1223
      break;
 
1224
    }
 
1225
    default:
 
1226
      res = gst_pad_query_default (pad, parent, query);
 
1227
      break;
 
1228
  }
 
1229
 
1207
1230
  if (!res)
1208
1231
    return FALSE;
1209
1232
 
 
1233
  /* Adjust peer response for data contained in queue */
1210
1234
  switch (GST_QUERY_TYPE (query)) {
1211
1235
    case GST_QUERY_POSITION:
1212
1236
    {
1314
1338
        queue->eos = FALSE;
1315
1339
        queue->unexpected = FALSE;
1316
1340
        result =
1317
 
            gst_pad_start_task (pad, (GstTaskFunction) gst_queue_loop, pad);
 
1341
            gst_pad_start_task (pad, (GstTaskFunction) gst_queue_loop, pad,
 
1342
            NULL);
1318
1343
        GST_QUEUE_MUTEX_UNLOCK (queue);
1319
1344
      } else {
1320
1345
        /* step 1, unblock loop function */