queue: add queue batch optionqueue-batch

The queue batch options makes the queue buffer data until it's filled before
pushing all the data until it's empty again. This avoids context switches
between reader and writer but also reduces paralelism.

2 files changed, 27 insertions, 4 deletions

diff --git a/plugins/elements/gstqueue.c b/plugins/elements/gstqueue.c
index 6989faf56..591430641 100644
--- a/plugins/elements/gstqueue.c
+++ b/plugins/elements/gstqueue.c
@@ -118,13 +118,16 @@ enum
   PROP_MIN_THRESHOLD_BYTES,
   PROP_MIN_THRESHOLD_TIME,
   PROP_LEAKY,
-  PROP_SILENT
+  PROP_SILENT,
+  PROP_BATCH,
+  PROP_LAST
 };
 
 /* default property values */
 #define DEFAULT_MAX_SIZE_BUFFERS  200   /* 200 buffers */
 #define DEFAULT_MAX_SIZE_BYTES    (10 * 1024 * 1024)    /* 10 MB       */
 #define DEFAULT_MAX_SIZE_TIME     GST_SECOND    /* 1 second    */
+#define DEFAULT_BATCH             FALSE
 
 #define GST_QUEUE_MUTEX_LOCK(q) G_STMT_START {                          \
   g_mutex_lock (q->qlock);                                              \
@@ -372,6 +375,11 @@ gst_queue_class_init (GstQueueClass * klass)
           "Don't emit queue signals", FALSE,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
+  g_object_class_install_property (gobject_class, PROP_BATCH,
+      g_param_spec_boolean ("batch", "Batch",
+          "Buffer min-threashold before pushing data", DEFAULT_BATCH,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
   gobject_class->finalize = gst_queue_finalize;
 
   /* Registering debug symbols for function pointers */
@@ -423,6 +431,7 @@ gst_queue_init (GstQueue * queue, GstQueueClass * g_class)
   gst_segment_init (&queue->sink_segment, GST_FORMAT_TIME);
   gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);
   queue->head_needs_discont = queue->tail_needs_discont = FALSE;
+  queue->batch = DEFAULT_BATCH;
 
   queue->leaky = GST_QUEUE_NO_LEAK;
   queue->srcresult = GST_FLOW_WRONG_STATE;
@@ -649,7 +658,6 @@ apply_buffer (GstQueue * queue, GstBuffer * buffer, GstSegment * segment,
   else
     queue->src_tainted = TRUE;
 
-
   /* calc diff with other end */
   update_time_level (queue);
 }
@@ -691,7 +699,9 @@ gst_queue_locked_enqueue_buffer (GstQueue * queue, gpointer item)
   apply_buffer (queue, buffer, &queue->sink_segment, TRUE, TRUE);
 
   g_queue_push_tail (queue->queue, item);
-  GST_QUEUE_SIGNAL_ADD (queue);
+
+  if (!queue->batch)
+    GST_QUEUE_SIGNAL_ADD (queue);
 }
 
 static inline void
@@ -783,7 +793,9 @@ gst_queue_locked_dequeue (GstQueue * queue, gboolean * is_buffer)
         item, GST_OBJECT_NAME (queue));
     item = NULL;
   }
-  GST_QUEUE_SIGNAL_DEL (queue);
+
+  if (!queue->batch)
+    GST_QUEUE_SIGNAL_DEL (queue);
 
   return item;
 
@@ -995,6 +1007,8 @@ gst_queue_chain (GstPad * pad, GstBuffer * buffer)
 
         /* don't leak. Instead, wait for space to be available */
         do {
+          if (queue->batch)
+            GST_QUEUE_SIGNAL_ADD (queue);
           /* for as long as the queue is filled, wait till an item was deleted. */
           GST_QUEUE_WAIT_DEL_CHECK (queue, out_flushing);
         } while (gst_queue_is_filled (queue));
@@ -1241,6 +1255,8 @@ gst_queue_loop (GstPad * pad)
 
     /* we recheck, the signal could have changed the thresholds */
     while (gst_queue_is_empty (queue)) {
+      if (queue->batch)
+        GST_QUEUE_SIGNAL_DEL (queue);
       GST_QUEUE_WAIT_ADD_CHECK (queue, out_flushing);
     }
 
@@ -1508,6 +1524,9 @@ gst_queue_set_property (GObject * object,
     case PROP_SILENT:
       queue->silent = g_value_get_boolean (value);
       break;
+    case PROP_BATCH:
+      queue->batch = g_value_get_boolean (value);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -1558,6 +1577,9 @@ gst_queue_get_property (GObject * object,
     case PROP_SILENT:
       g_value_set_boolean (value, queue->silent);
       break;
+    case PROP_BATCH:
+      g_value_set_boolean (value, queue->batch);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
diff --git a/plugins/elements/gstqueue.h b/plugins/elements/gstqueue.h
index c72bcb790..a55a87e33 100644
--- a/plugins/elements/gstqueue.h
+++ b/plugins/elements/gstqueue.h
@@ -117,6 +117,7 @@ struct _GstQueue {
   gboolean push_newsegment;
   
   gboolean silent;      /* don't emit signals */ 
+  gboolean batch;
 
   /* whether the first new segment has been applied to src */
   gboolean newseg_applied_to_src;