ext/gio/gstgiobasesrc.*: Try to read the requested number of bytes, even if the first read returns less than requeste...

Original commit message from CVS:
* ext/gio/gstgiobasesrc.c: (gst_gio_base_src_finalize),
(gst_gio_base_src_create):
* ext/gio/gstgiobasesrc.h:
Try to read the requested number of bytes, even if the first
read returns less than requested, until nothing is read anymore
or we have the requested amount of bytes. This fixes playback of
files via Samba as Samba only allows to read 64k at once.
Implement a caching algorithm that makes sure that we read at
least 4k of data every time. Some elements will try to read a few
bytes, then seek, read again a few bytes and so on and this is
painfully slow as every operation has to go over DBus if GVfs is
used as backend.
Fixes bug #536849 and #536848.
* ext/gio/gstgiosrc.c: (gst_gio_src_class_init),
(gst_gio_src_check_get_range):
Override check_get_range() to blacklist http/https URIs
and whitelist file URIs. More to be added on demand.

ChangeLog

@@ -1,3 +1,26 @@
+2008-06-11  Sebastian Dröge  <slomo@circular-chaos.org>
+
+	* ext/gio/gstgiobasesrc.c: (gst_gio_base_src_finalize),
+	(gst_gio_base_src_create):
+	* ext/gio/gstgiobasesrc.h:
+	Try to read the requested number of bytes, even if the first
+	read returns less than requested, until nothing is read anymore
+	or we have the requested amount of bytes. This fixes playback of
+	files via Samba as Samba only allows to read 64k at once.
+
+	Implement a caching algorithm that makes sure that we read at
+	least 4k of data every time. Some elements will try to read a few
+	bytes, then seek, read again a few bytes and so on and this is
+	painfully slow as every operation has to go over DBus if GVfs is
+	used as backend.
+
+	Fixes bug #536849 and #536848.
+
+	* ext/gio/gstgiosrc.c: (gst_gio_src_class_init),
+	(gst_gio_src_check_get_range):
+	Override check_get_range() to blacklist http/https URIs
+	and whitelist file URIs. More to be added on demand.
+
 2008-06-05  Jan Schmidt  <jan.schmidt@sun.com>
 
 	* configure.ac:

ext/gio/gstgiobasesrc.c

@@ -37,14 +37,21 @@ GST_BOILERPLATE (GstGioBaseSrc, gst_gio_base_src, GstBaseSrc,
     GST_TYPE_BASE_SRC);
 
 static void gst_gio_base_src_finalize (GObject * object);
+
 static gboolean gst_gio_base_src_start (GstBaseSrc * base_src);
+
 static gboolean gst_gio_base_src_stop (GstBaseSrc * base_src);
+
 static gboolean gst_gio_base_src_get_size (GstBaseSrc * base_src,
     guint64 * size);
 static gboolean gst_gio_base_src_is_seekable (GstBaseSrc * base_src);
+
 static gboolean gst_gio_base_src_unlock (GstBaseSrc * base_src);
+
 static gboolean gst_gio_base_src_unlock_stop (GstBaseSrc * base_src);
+
 static gboolean gst_gio_base_src_check_get_range (GstBaseSrc * base_src);
+
 static GstFlowReturn gst_gio_base_src_create (GstBaseSrc * base_src,
     guint64 offset, guint size, GstBuffer ** buf);
 
@@ -64,7 +71,9 @@ static void
 gst_gio_base_src_class_init (GstGioBaseSrcClass * klass)
 {
   GObjectClass *gobject_class;
+
   GstElementClass *gstelement_class;
+
   GstBaseSrcClass *gstbasesrc_class;
 
   gobject_class = (GObjectClass *) klass;
@@ -107,6 +116,11 @@ gst_gio_base_src_finalize (GObject * object)
     src->stream = NULL;
   }
 
+  if (src->cache) {
+    gst_buffer_unref (src->cache);
+    src->cache = NULL;
+  }
+
   GST_CALL_PARENT (G_OBJECT_CLASS, finalize, (object));
 }
 
@@ -132,7 +146,9 @@ static gboolean
 gst_gio_base_src_stop (GstBaseSrc * base_src)
 {
   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
+
   gboolean success;
+
   GError *err = NULL;
 
   if (G_IS_INPUT_STREAM (src->stream)) {
@@ -167,6 +183,7 @@ gst_gio_base_src_get_size (GstBaseSrc * base_src, guint64 * size)
 
   if (G_IS_FILE_INPUT_STREAM (src->stream)) {
     GFileInfo *info;
+
     GError *err = NULL;
 
     info = g_file_input_stream_query_info (G_FILE_INPUT_STREAM (src->stream),
@@ -193,9 +210,13 @@ gst_gio_base_src_get_size (GstBaseSrc * base_src, guint64 * size)
 
   if (GST_GIO_STREAM_IS_SEEKABLE (src->stream)) {
     goffset old;
+
     goffset stream_size;
+
     gboolean ret;
+
     GSeekable *seekable = G_SEEKABLE (src->stream);
+
     GError *err = NULL;
 
     old = g_seekable_tell (seekable);
@@ -248,6 +269,7 @@ static gboolean
 gst_gio_base_src_is_seekable (GstBaseSrc * base_src)
 {
   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
+
   gboolean seekable;
 
   seekable = GST_GIO_STREAM_IS_SEEKABLE (src->stream);
@@ -295,56 +317,114 @@ gst_gio_base_src_create (GstBaseSrc * base_src, guint64 offset, guint size,
     GstBuffer ** buf_return)
 {
   GstGioBaseSrc *src = GST_GIO_BASE_SRC (base_src);
+
   GstBuffer *buf;
-  gssize read;
+
-  gboolean success, eos;
   GstFlowReturn ret = GST_FLOW_OK;
-  GError *err = NULL;
 
   g_return_val_if_fail (G_IS_INPUT_STREAM (src->stream), GST_FLOW_ERROR);
 
-  if (G_UNLIKELY (offset != src->position)) {
+  /* If we have the requested part in our cache take a subbuffer of that,
-    if (!GST_GIO_STREAM_IS_SEEKABLE (src->stream))
+   * otherwise fill the cache again with at least 4096 bytes from the
-      return GST_FLOW_NOT_SUPPORTED;
+   * requested offset and return a subbuffer of that.
+   *
+   * We need caching because every read/seek operation will need to go
+   * over DBus if our backend is GVfs and this is painfully slow. */
+  if (src->cache && offset >= GST_BUFFER_OFFSET (src->cache) &&
+      offset + size <= GST_BUFFER_OFFSET_END (src->cache)) {
 
-    ret = gst_gio_seek (src, G_SEEKABLE (src->stream), offset, src->cancel);
+    GST_DEBUG_OBJECT (src, "Creating subbuffer from cached buffer: offset %"
+        G_GUINT64_FORMAT " length %u", offset, size);
 
-    if (ret == GST_FLOW_OK)
+    buf = gst_buffer_create_sub (src->cache,
-      src->position = offset;
+        offset - GST_BUFFER_OFFSET (src->cache), size);
-    else
-      return ret;
-  }
 
-  buf = gst_buffer_new_and_alloc (size);
-
-  GST_LOG_OBJECT (src, "reading %u bytes from offset %" G_GUINT64_FORMAT,
-      size, offset);
-
-  read =
-      g_input_stream_read (G_INPUT_STREAM (src->stream), GST_BUFFER_DATA (buf),
-      size, src->cancel, &err);
-
-  success = (read >= 0);
-  eos = (size > 0 && read == 0);
-
-  if (!success && !gst_gio_error (src, "g_input_stream_read", &err, &ret)) {
-    GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
-        ("Could not read from stream: %s", err->message));
-    g_clear_error (&err);
-  }
-
-  if (success && !eos) {
-    src->position += read;
     GST_BUFFER_OFFSET (buf) = offset;
-    GST_BUFFER_SIZE (buf) = read;
+    GST_BUFFER_OFFSET_END (buf) = offset + size;
-    *buf_return = buf;
+    GST_BUFFER_SIZE (buf) = size;
   } else {
-    /* !success || eos */
+    guint cachesize = MAX (4096, size);
-    gst_buffer_unref (buf);
+
+    gssize read, res;
+
+    gboolean success, eos;
+
+    GError *err = NULL;
+
+    if (src->cache) {
+      gst_buffer_unref (src->cache);
+      src->cache = NULL;
+    }
+
+    if (G_UNLIKELY (offset != src->position)) {
+      if (!GST_GIO_STREAM_IS_SEEKABLE (src->stream))
+        return GST_FLOW_NOT_SUPPORTED;
+
+      GST_DEBUG_OBJECT (src, "Seeking to position %" G_GUINT64_FORMAT, offset);
+      ret = gst_gio_seek (src, G_SEEKABLE (src->stream), offset, src->cancel);
+
+      if (ret == GST_FLOW_OK)
+        src->position = offset;
+      else
+        return ret;
+    }
+
+    src->cache = gst_buffer_new_and_alloc (cachesize);
+
+    GST_LOG_OBJECT (src, "Reading %u bytes from offset %" G_GUINT64_FORMAT,
+        cachesize, offset);
+
+    /* GIO sometimes gives less bytes than requested although
+     * it's not at the end of file. SMB for example only
+     * supports reads up to 64k. So we loop here until we get at
+     * at least the requested amount of bytes or a read returns
+     * nothing. */
+    read = 0;
+    while (size - read > 0 && (res =
+            g_input_stream_read (G_INPUT_STREAM (src->stream),
+                GST_BUFFER_DATA (src->cache) + read, cachesize - read,
+                src->cancel, &err)) > 0) {
+      read += res;
+    }
+
+    success = (read >= 0);
+    eos = (cachesize > 0 && read == 0);
+
+    if (!success && !gst_gio_error (src, "g_input_stream_read", &err, &ret)) {
+      GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
+          ("Could not read from stream: %s", err->message));
+      g_clear_error (&err);
+    }
+
+    if (success && !eos) {
+      src->position += read;
+      GST_BUFFER_SIZE (src->cache) = read;
+
+      GST_BUFFER_OFFSET (src->cache) = offset;
+      GST_BUFFER_OFFSET_END (src->cache) = offset + read;
+
+      GST_DEBUG_OBJECT (src, "Read successful");
+      GST_DEBUG_OBJECT (src, "Creating subbuffer from new "
+          "cached buffer: offset %" G_GUINT64_FORMAT " length %u", offset,
+          size);
+
+      buf = gst_buffer_create_sub (src->cache, 0, MIN (size, read));
+
+      GST_BUFFER_OFFSET (buf) = offset;
+      GST_BUFFER_OFFSET_END (buf) = offset + MIN (size, read);
+      GST_BUFFER_SIZE (buf) = MIN (size, read);
+    } else {
+      GST_DEBUG_OBJECT (src, "Read not successful");
+      gst_buffer_unref (src->cache);
+      src->cache = NULL;
+      buf = NULL;
+    }
+
+    if (eos)
+      ret = GST_FLOW_UNEXPECTED;
   }
 
-  if (eos)
+  *buf_return = buf;
-    ret = GST_FLOW_UNEXPECTED;
 
   return ret;
 }
@@ -353,6 +433,7 @@ void
 gst_gio_base_src_set_stream (GstGioBaseSrc * src, GInputStream * stream)
 {
   gboolean success;
+
   GError *err = NULL;
 
   g_return_if_fail (G_IS_INPUT_STREAM (stream));

ext/gio/gstgiobasesrc.h

@@ -49,6 +49,8 @@ struct _GstGioBaseSrc
   GCancellable *cancel;
   guint64 position;
   GInputStream *stream;
+
+  GstBuffer *cache;
 };
 
 struct _GstGioBaseSrcClass 

ext/gio/gstgiosrc.c

@@ -72,6 +72,7 @@
 #endif
 
 #include "gstgiosrc.h"
+#include <string.h>
 
 GST_DEBUG_CATEGORY_STATIC (gst_gio_src_debug);
 #define GST_CAT_DEFAULT gst_gio_src_debug
@@ -87,12 +88,16 @@ GST_BOILERPLATE_FULL (GstGioSrc, gst_gio_src, GstGioBaseSrc,
     GST_TYPE_GIO_BASE_SRC, gst_gio_uri_handler_do_init);
 
 static void gst_gio_src_finalize (GObject * object);
+
 static void gst_gio_src_set_property (GObject * object, guint prop_id,
     const GValue * value, GParamSpec * pspec);
 static void gst_gio_src_get_property (GObject * object, guint prop_id,
     GValue * value, GParamSpec * pspec);
+
 static gboolean gst_gio_src_start (GstBaseSrc * base_src);
 
+static gboolean gst_gio_src_check_get_range (GstBaseSrc * base_src);
+
 static void
 gst_gio_src_base_init (gpointer gclass)
 {
@@ -111,7 +116,9 @@ static void
 gst_gio_src_class_init (GstGioSrcClass * klass)
 {
   GObjectClass *gobject_class;
+
   GstElementClass *gstelement_class;
+
   GstBaseSrcClass *gstbasesrc_class;
 
   gobject_class = (GObjectClass *) klass;
@@ -138,6 +145,8 @@ gst_gio_src_class_init (GstGioSrcClass * klass)
           G_TYPE_FILE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_gio_src_start);
+  gstbasesrc_class->check_get_range =
+      GST_DEBUG_FUNCPTR (gst_gio_src_check_get_range);
 }
 
 static void
@@ -247,13 +256,53 @@ gst_gio_src_get_property (GObject * object, guint prop_id,
   }
 }
 
+static gboolean
+gst_gio_src_check_get_range (GstBaseSrc * base_src)
+{
+  GstGioSrc *src = GST_GIO_SRC (base_src);
+
+  gchar *scheme;
+
+  if (src->file == NULL)
+    goto done;
+
+  scheme = g_file_get_uri_scheme (src->file);
+  if (scheme == NULL)
+    goto done;
+
+  if (strcmp (scheme, "file") == 0) {
+    GST_LOG_OBJECT (src, "local URI, assuming random access is possible");
+    g_free (scheme);
+    return TRUE;
+  } else if (strcmp (scheme, "http") == 0 || strcmp (scheme, "https") == 0) {
+    GST_LOG_OBJECT (src, "blacklisted protocol '%s', "
+        "no random access possible", scheme);
+    g_free (scheme);
+    return FALSE;
+  }
+
+  g_free (scheme);
+
+done:
+
+  GST_DEBUG_OBJECT (src, "undecided about random access, asking base class");
+
+  return GST_CALL_PARENT_WITH_DEFAULT (GST_BASE_SRC_CLASS,
+      check_get_range, (base_src), FALSE);
+}
+
+
 static gboolean
 gst_gio_src_start (GstBaseSrc * base_src)
 {
   GstGioSrc *src = GST_GIO_SRC (base_src);
+
   GError *err = NULL;
+
   GInputStream *stream;
+
   GCancellable *cancel = GST_GIO_BASE_SRC (src)->cancel;
+
   gchar *uri = NULL;
 
   if (src->file == NULL) {