Commit 177d0fa1 authored by Jan Schmidt's avatar Jan Schmidt Committed by Tim-Philipp Müller

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 #518
parent 0f91868e
Pipeline #157167 passed with stages
in 51 minutes and 21 seconds
......@@ -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;
}
......
......@@ -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);
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment