baseparse: Fix upstream read caching

When running in pull mode (for e.g. mp3 reading),
baseparse currently reads 64KB from upstream, then mp3parse
consumes typically around 417/418 bytes of it. Then
on the next loop, it will read a full fresh 64KB again,
which is a big waste.

Fix the read loop to use the available cache buffer first
before going for more data, until the cache drops to < 1KB.

Fixes https://gitlab.freedesktop.org/gstreamer/gstreamer/issues/518

2 files changed, 53 insertions, 19 deletions

diff --git a/libs/gst/base/gstbaseparse.c b/libs/gst/base/gstbaseparse.c
index a233959c31..e2ec6ae0f3 100644
--- a/libs/gst/base/gstbaseparse.c
+++ b/libs/gst/base/gstbaseparse.c
@@ -3307,6 +3307,22 @@ done:
   return ret;
 }
 
+/* Return the number of bytes available in the cached
+ * read buffer, if any */
+static guint
+gst_base_parse_get_cached_available (GstBaseParse * parse)
+{
+  if (parse->priv->cache != NULL) {
+    gint64 cache_offset = GST_BUFFER_OFFSET (parse->priv->cache);
+    gint cache_size = gst_buffer_get_size (parse->priv->cache);
+
+    if (parse->priv->offset >= cache_offset
+        && parse->priv->offset < cache_offset + cache_size)
+      return cache_size - (parse->priv->offset - cache_offset); /* Size of the cache minus consumed */
+  }
+  return 0;
+}
+
 /* pull @size bytes at current offset,
  * i.e. at least try to and possibly return a shorter buffer if near the end */
 static GstFlowReturn
@@ -3328,6 +3344,9 @@ gst_base_parse_pull_range (GstBaseParse * parse, guint size,
       *buffer = gst_buffer_copy_region (parse->priv->cache, GST_BUFFER_COPY_ALL,
           parse->priv->offset - cache_offset, size);
       GST_BUFFER_OFFSET (*buffer) = parse->priv->offset;
+      GST_LOG_OBJECT (parse,
+          "Satisfying read request of %u bytes from cached buffer with offset %"
+          G_GINT64_FORMAT, size, cache_offset);
       return GST_FLOW_OK;
     }
     /* not enough data in the cache, free cache and get a new one */
@@ -3336,9 +3355,13 @@ gst_base_parse_pull_range (GstBaseParse * parse, guint size,
   }
 
   /* refill the cache */
+  size = MAX (64 * 1024, size);
+  GST_LOG_OBJECT (parse,
+      "Reading cache buffer of %u bytes from offset %" G_GINT64_FORMAT,
+      size, parse->priv->offset);
   ret =
-      gst_pad_pull_range (parse->sinkpad, parse->priv->offset, MAX (size,
-          64 * 1024), &parse->priv->cache);
+      gst_pad_pull_range (parse->sinkpad, parse->priv->offset, size,
+      &parse->priv->cache);
   if (ret != GST_FLOW_OK) {
     parse->priv->cache = NULL;
     return ret;
@@ -3355,6 +3378,8 @@ gst_base_parse_pull_range (GstBaseParse * parse, guint size,
     return GST_FLOW_OK;
   }
 
+  GST_BUFFER_OFFSET (parse->priv->cache) = parse->priv->offset;
+
   *buffer =
       gst_buffer_copy_region (parse->priv->cache, GST_BUFFER_COPY_ALL, 0, size);
   GST_BUFFER_OFFSET (*buffer) = parse->priv->offset;
@@ -3441,9 +3466,12 @@ gst_base_parse_scan_frame (GstBaseParse * parse, GstBaseParseClass * klass)
 
   /* let's make this efficient for all subclass once and for all;
    * maybe it does not need this much, but in the latter case, we know we are
-   * in pull mode here and might as well try to read and supply more anyway
-   * (so does the buffer caching mechanism) */
-  fsize = 64 * 1024;
+   * in pull mode here and might as well try to read and supply more anyway,
+   * so start with the cached buffer, or if that's shrunk below 1024 bytes,
+   * pull a new cache buffer */
+  fsize = gst_base_parse_get_cached_available (parse);
+  if (fsize < 1024)
+    fsize = 64 * 1024;
 
   while (TRUE) {
     min_size = MAX (parse->priv->min_frame_size, fsize);
@@ -3471,7 +3499,8 @@ gst_base_parse_scan_frame (GstBaseParse * parse, GstBaseParseClass * klass)
           GST_ERROR_OBJECT (parse, "Failed to detect format but draining");
           return GST_FLOW_ERROR;
         } else {
-          fsize += 64 * 1024;
+          /* Double our frame size, or increment by at most 64KB */
+          fsize += MIN (fsize, 64 * 1024);
           gst_buffer_unref (buffer);
           continue;
         }
@@ -3502,18 +3531,20 @@ gst_base_parse_scan_frame (GstBaseParse * parse, GstBaseParseClass * klass)
       GST_LOG_OBJECT (parse, "frame finished, breaking loop");
       break;
     }
-    /* nothing flushed, no skip and draining, so nothing left to do */
-    if (!skip && parse->priv->drain) {
-      GST_LOG_OBJECT (parse, "no activity or result when draining; "
-          "breaking loop and marking EOS");
-      ret = GST_FLOW_EOS;
-      break;
-    }
-    /* otherwise, get some more data
-     * note that is checked this does not happen indefinitely */
     if (!skip) {
+      if (parse->priv->drain) {
+        /* nothing flushed, no skip and draining, so nothing left to do */
+        GST_LOG_OBJECT (parse, "no activity or result when draining; "
+            "breaking loop and marking EOS");
+        ret = GST_FLOW_EOS;
+        break;
+      }
+      /* otherwise, get some more data
+       * note that is checked this does not happen indefinitely */
       GST_LOG_OBJECT (parse, "getting some more data");
-      fsize += 64 * 1024;
+
+      /* Double our frame size, or increment by at most 64KB */
+      fsize += MIN (fsize, 64 * 1024);
     }
     parse->priv->drain = FALSE;
   }
diff --git a/tests/check/libs/baseparse.c b/tests/check/libs/baseparse.c
index 576cae37a7..53e058ffa0 100644
--- a/tests/check/libs/baseparse.c
+++ b/tests/check/libs/baseparse.c
@@ -464,12 +464,13 @@ _sink_chain_pull_short_read (GstPad * pad, GstObject * parent,
   GstMapInfo map;
   gint buffer_size = 0;
   gint compare_result = 0;
+  gint64 result_offset = GST_BUFFER_OFFSET (buffer);
 
   gst_buffer_map (buffer, &map, GST_MAP_READ);
   buffer_size = map.size;
 
-  fail_unless (current_offset + buffer_size <= raw_buffer_size);
-  compare_result = memcmp (map.data, &raw_buffer[current_offset], buffer_size);
+  fail_unless (result_offset + buffer_size <= raw_buffer_size);
+  compare_result = memcmp (map.data, &raw_buffer[result_offset], buffer_size);
   fail_unless_equals_int (compare_result, 0);
 
   gst_buffer_unmap (buffer, &map);
@@ -549,7 +550,9 @@ GST_START_TEST (parser_pull_short_read)
   g_main_loop_run (loop);
   fail_unless_equals_int (have_eos, TRUE);
   fail_unless_equals_int (have_data, TRUE);
-  fail_unless_equals_int (buffer_pull_count, buffer_count);
+  /* If the parser asked upstream for buffers more times than buffers were
+   * produced, then something is wrong */
+  fail_unless (buffer_pull_count <= buffer_count);
 
   gst_element_set_state (parsetest, GST_STATE_NULL);