multiqueue: Add higher-resolution low/high-watermark properties

low/high-watermark are of type double, and given in range 0.0-1.0. This
makes it possible to set low/high watermarks with greater resolution,
which is useful with large multiqueue max sizes and watermarks like 0.5%.

Also adding a test to check the fill and watermark level behavior.

https://bugzilla.gnome.org/show_bug.cgi?id=770628

plugins/elements/gstmultiqueue.c

@@ -267,8 +267,8 @@ enum
 #define DEFAULT_EXTRA_SIZE_TIME 3 * GST_SECOND
 
 #define DEFAULT_USE_BUFFERING FALSE
-#define DEFAULT_LOW_PERCENT   10
+#define DEFAULT_LOW_WATERMARK  0.01
-#define DEFAULT_HIGH_PERCENT  99
+#define DEFAULT_HIGH_WATERMARK 0.99
 #define DEFAULT_SYNC_BY_RUNNING_TIME FALSE
 #define DEFAULT_USE_INTERLEAVE FALSE
 #define DEFAULT_UNLINKED_CACHE_TIME 250 * GST_MSECOND
@@ -285,6 +285,8 @@ enum
   PROP_USE_BUFFERING,
   PROP_LOW_PERCENT,
   PROP_HIGH_PERCENT,
+  PROP_LOW_WATERMARK,
+  PROP_HIGH_WATERMARK,
   PROP_SYNC_BY_RUNNING_TIME,
   PROP_USE_INTERLEAVE,
   PROP_UNLINKED_CACHE_TIME,
@@ -310,7 +312,12 @@ enum
  * range. Whenever "buffering_percent" is mentioned, it refers to the
  * percentage value that is relative to the low/high watermark. */
 
-#define MAX_BUFFERING_LEVEL 100
+/* Using a buffering level range of 0..1000000 to allow for a
+ * resolution in ppm (1 ppm = 0.0001%) */
+#define MAX_BUFFERING_LEVEL 1000000
+
+/* How much 1% makes up in the buffer level range */
+#define BUF_LEVEL_PERCENT_FACTOR ((MAX_BUFFERING_LEVEL) / 100)
 
 /* GstMultiQueuePad */
 
@@ -561,8 +568,10 @@ gst_multi_queue_class_init (GstMultiQueueClass * klass)
    */
   g_object_class_install_property (gobject_class, PROP_LOW_PERCENT,
       g_param_spec_int ("low-percent", "Low percent",
-          "Low threshold for buffering to start", 0, 100,
+          "Low threshold for buffering to start. Only used if use-buffering is True "
-          DEFAULT_LOW_PERCENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+          "(Deprecated: use low-watermark instead)",
+          0, 100, DEFAULT_LOW_WATERMARK * 100,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
   /**
    * GstMultiQueue:high-percent
    * 
@@ -570,8 +579,34 @@ gst_multi_queue_class_init (GstMultiQueueClass * klass)
    */
   g_object_class_install_property (gobject_class, PROP_HIGH_PERCENT,
       g_param_spec_int ("high-percent", "High percent",
-          "High threshold for buffering to finish", 0, 100,
+          "High threshold for buffering to finish. Only used if use-buffering is True "
-          DEFAULT_HIGH_PERCENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+          "(Deprecated: use high-watermark instead)",
+          0, 100, DEFAULT_HIGH_WATERMARK * 100,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+  /**
+   * GstMultiQueue:low-watermark
+   *
+   * Low threshold watermark for buffering to start.
+   *
+   * Since: 1.10
+   */
+  g_object_class_install_property (gobject_class, PROP_LOW_WATERMARK,
+      g_param_spec_double ("low-watermark", "Low watermark",
+          "Low threshold for buffering to start. Only used if use-buffering is True",
+          0.0, 1.0, DEFAULT_LOW_WATERMARK,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+  /**
+   * GstMultiQueue:high-watermark
+   *
+   * High threshold watermark for buffering to finish.
+   *
+   * Since: 1.10
+   */
+  g_object_class_install_property (gobject_class, PROP_HIGH_WATERMARK,
+      g_param_spec_double ("high-watermark", "High watermark",
+          "High threshold for buffering to finish. Only used if use-buffering is True",
+          0.0, 1.0, DEFAULT_HIGH_WATERMARK,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   /**
    * GstMultiQueue:sync-by-running-time
@@ -632,8 +667,8 @@ gst_multi_queue_init (GstMultiQueue * mqueue)
   mqueue->extra_size.time = DEFAULT_EXTRA_SIZE_TIME;
 
   mqueue->use_buffering = DEFAULT_USE_BUFFERING;
-  mqueue->low_watermark = DEFAULT_LOW_PERCENT;
+  mqueue->low_watermark = DEFAULT_LOW_WATERMARK * MAX_BUFFERING_LEVEL;
-  mqueue->high_watermark = DEFAULT_HIGH_PERCENT;
+  mqueue->high_watermark = DEFAULT_HIGH_WATERMARK * MAX_BUFFERING_LEVEL;
 
   mqueue->sync_by_running_time = DEFAULT_SYNC_BY_RUNNING_TIME;
   mqueue->use_interleave = DEFAULT_USE_INTERLEAVE;
@@ -748,7 +783,7 @@ gst_multi_queue_set_property (GObject * object, guint prop_id,
       recheck_buffering_status (mq);
       break;
     case PROP_LOW_PERCENT:
-      mq->low_watermark = g_value_get_int (value);
+      mq->low_watermark = g_value_get_int (value) * BUF_LEVEL_PERCENT_FACTOR;
       /* Recheck buffering status - the new low_watermark value might
        * be above the current fill level. If the old low_watermark one
        * was below the current level, this means that mq->buffering is
@@ -756,7 +791,15 @@ gst_multi_queue_set_property (GObject * object, guint prop_id,
       recheck_buffering_status (mq);
       break;
     case PROP_HIGH_PERCENT:
-      mq->high_watermark = g_value_get_int (value);
+      mq->high_watermark = g_value_get_int (value) * BUF_LEVEL_PERCENT_FACTOR;
+      recheck_buffering_status (mq);
+      break;
+    case PROP_LOW_WATERMARK:
+      mq->low_watermark = g_value_get_double (value) * MAX_BUFFERING_LEVEL;
+      recheck_buffering_status (mq);
+      break;
+    case PROP_HIGH_WATERMARK:
+      mq->high_watermark = g_value_get_double (value) * MAX_BUFFERING_LEVEL;
       recheck_buffering_status (mq);
       break;
     case PROP_SYNC_BY_RUNNING_TIME:
@@ -808,10 +851,18 @@ gst_multi_queue_get_property (GObject * object, guint prop_id,
       g_value_set_boolean (value, mq->use_buffering);
       break;
     case PROP_LOW_PERCENT:
-      g_value_set_int (value, mq->low_watermark);
+      g_value_set_int (value, mq->low_watermark / BUF_LEVEL_PERCENT_FACTOR);
       break;
     case PROP_HIGH_PERCENT:
-      g_value_set_int (value, mq->high_watermark);
+      g_value_set_int (value, mq->high_watermark / BUF_LEVEL_PERCENT_FACTOR);
+      break;
+    case PROP_LOW_WATERMARK:
+      g_value_set_double (value, mq->low_watermark /
+          (gdouble) MAX_BUFFERING_LEVEL);
+      break;
+    case PROP_HIGH_WATERMARK:
+      g_value_set_double (value, mq->high_watermark /
+          (gdouble) MAX_BUFFERING_LEVEL);
       break;
     case PROP_SYNC_BY_RUNNING_TIME:
       g_value_set_boolean (value, mq->sync_by_running_time);

tests/check/elements/multiqueue.c

@@ -976,6 +976,114 @@ GST_START_TEST (test_initial_fill_above_high_threshold)
 
 GST_END_TEST;
 
+GST_START_TEST (test_watermark_and_fill_level)
+{
+  /* This test checks the behavior of the fill level and
+   * the low/high watermarks. It also checks if the
+   * low/high-percent and low/high-watermark properties
+   * are coupled together properly. */
+  GstElement *pipe;
+  GstElement *mq, *fakesink;
+  GstPad *inputpad;
+  GstPad *mq_sinkpad;
+  GstPad *sinkpad;
+  GstSegment segment;
+  GThread *thread;
+  gint low_perc, high_perc;
+
+
+  /* Setup test pipeline with one multiqueue and one fakesink */
+
+  pipe = gst_pipeline_new ("testbin");
+  mq = gst_element_factory_make ("multiqueue", NULL);
+  fail_unless (mq != NULL);
+  gst_bin_add (GST_BIN (pipe), mq);
+
+  fakesink = gst_element_factory_make ("fakesink", NULL);
+  fail_unless (fakesink != NULL);
+  gst_bin_add (GST_BIN (pipe), fakesink);
+
+  /* Block fakesink sinkpad flow to ensure the queue isn't emptied
+   * by the prerolling sink */
+  sinkpad = gst_element_get_static_pad (fakesink, "sink");
+  gst_pad_add_probe (sinkpad, GST_PAD_PROBE_TYPE_BLOCK, block_probe, NULL,
+      NULL);
+  gst_object_unref (sinkpad);
+
+  g_object_set (mq,
+      "use-buffering", (gboolean) TRUE,
+      "max-size-bytes", (guint) 1000 * 1000,
+      "max-size-buffers", (guint) 0,
+      "max-size-time", (guint64) 0,
+      "extra-size-bytes", (guint) 0,
+      "extra-size-buffers", (guint) 0,
+      "extra-size-time", (guint64) 0,
+      "low-watermark", (gdouble) 0.01, "high-watermark", (gdouble) 0.10, NULL);
+
+  g_object_get (mq, "low-percent", &low_perc, "high-percent", &high_perc, NULL);
+
+  /* Check that low/high-watermark and low/high-percent are
+   * coupled properly. (low/high-percent are deprecated and
+   * exist for backwards compatibility.) */
+  fail_unless_equals_int (low_perc, 1);
+  fail_unless_equals_int (high_perc, 10);
+
+  gst_segment_init (&segment, GST_FORMAT_TIME);
+
+  inputpad = gst_pad_new ("dummysrc", GST_PAD_SRC);
+  gst_pad_set_query_function (inputpad, mq_dummypad_query);
+
+  mq_sinkpad = gst_element_get_request_pad (mq, "sink_%u");
+  fail_unless (mq_sinkpad != NULL);
+  fail_unless (gst_pad_link (inputpad, mq_sinkpad) == GST_PAD_LINK_OK);
+
+  gst_pad_set_active (inputpad, TRUE);
+
+  gst_pad_push_event (inputpad, gst_event_new_stream_start ("test"));
+  gst_pad_push_event (inputpad, gst_event_new_segment (&segment));
+
+  gst_object_unref (mq_sinkpad);
+
+  fail_unless (gst_element_link (mq, fakesink));
+
+  /* Start pipeline in paused state to ensure the sink remains
+   * in preroll mode and blocks */
+  gst_element_set_state (pipe, GST_STATE_PAUSED);
+
+  /* Feed data. queue will be filled to 8% (because it pushes 80000 bytes),
+   * which is below the high-threshold, provoking a buffering message. */
+  thread = g_thread_new ("push1", pad_push_datablock_thread, inputpad);
+  g_thread_join (thread);
+
+  /* Check for the buffering message; it should indicate 80% fill level
+   * (Note that the percentage from the message is normalized) */
+  CHECK_FOR_BUFFERING_MSG (pipe, 80);
+
+  /* Increase the buffer size and lower the watermarks to test
+   * if <1% watermarks are supported. */
+  g_object_set (mq,
+      "max-size-bytes", (guint) 20 * 1000 * 1000,
+      "low-watermark", (gdouble) 0.0001, "high-watermark", (gdouble) 0.005,
+      NULL);
+  /* First buffering message is posted after the max-size-bytes limit
+   * is set to 20000000 bytes & the low-watermark is set. Since the
+   * multiqueue contains 80000 bytes, and the high watermark still is
+   * 0.1 at this point, and the buffer level 80000 / 20000000 = 0.004 is
+   * normalized by 0.1: 0.004 / 0.1 => buffering percentage 4%. */
+  CHECK_FOR_BUFFERING_MSG (pipe, 4);
+  /* Second buffering message is posted after the high-watermark limit
+   * is set to 0.005. This time, the buffer level is normalized this way:
+   * 0.004 / 0.005 => buffering percentage 80%. */
+  CHECK_FOR_BUFFERING_MSG (pipe, 80);
+
+
+  gst_element_set_state (pipe, GST_STATE_NULL);
+  gst_object_unref (inputpad);
+  gst_object_unref (pipe);
+}
+
+GST_END_TEST;
+
 GST_START_TEST (test_high_threshold_change)
 {
   /* This test checks what happens if the high threshold is changed to a
@@ -1508,6 +1616,7 @@ multiqueue_suite (void)
 
   tcase_add_test (tc_chain, test_sparse_stream);
   tcase_add_test (tc_chain, test_initial_fill_above_high_threshold);
+  tcase_add_test (tc_chain, test_watermark_and_fill_level);
   tcase_add_test (tc_chain, test_high_threshold_change);
   tcase_add_test (tc_chain, test_low_threshold_change);
   tcase_add_test (tc_chain, test_limit_changes);