Re: buffer management [was Re: libsoup-2.4 branch merged to trunk]

[Dan Winship <danw gnome org>]

Wouter Cloetens wrote:
> Patch attached to bug 513810.

I saw that. I didn't love it. It really violates the abstraction of
SoupBuffer.

> I discovered a third issue; GStreamer allows for downstream elements to
> allocate the buffers in which to write. This is especially useful if the
> next element in the pipeline does hardware buffer management to offload
> some operation to hardware.
> To manage that, I'd need to be able to configure a callback to do the
> actual allocation. That should be easy enough... Tomorrow's job.

That seems a lot nicer; we could make a new SoupBuffer subtype
representing data owned by another object, with a GDestroyNotify
attached to clean things up, and then the allocation callback would
allocate SoupBuffers, and everything should just work...

What do you think of the attached diff? I think this would let you solve
both of your issues (hardware buffer management, and the original
I-want-smaller-buffers-and-no-copies thing).

Index: libsoup/soup-message.h
===================================================================
--- libsoup/soup-message.h	(revision 1072)
+++ libsoup/soup-message.h	(working copy)
@@ -132,7 +132,16 @@
 						 guint              status_code, 
 						 const char        *reason_phrase);
 
+/* I/O */
+typedef SoupBuffer * (*SoupChunkAllocator)      (SoupMessage       *msg,
+						 gsize              max_len,
+						 gpointer           user_data);
 
+void           soup_message_set_chunk_allocator (SoupMessage       *msg,
+						 SoupChunkAllocator allocator,
+						 gpointer           user_data,
+						 GDestroyNotify     destroy_notify);
+
 void soup_message_wrote_informational (SoupMessage *msg);
 void soup_message_wrote_headers       (SoupMessage *msg);
 void soup_message_wrote_chunk         (SoupMessage *msg);
Index: libsoup/soup-message.c
===================================================================
--- libsoup/soup-message.c	(revision 1072)
+++ libsoup/soup-message.c	(working copy)
@@ -117,6 +117,8 @@
 	SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 
 	soup_message_io_cleanup (msg);
+	if (priv->chunk_allocator_dnotify)
+		priv->chunk_allocator_dnotify (priv->chunk_allocator_data);
 
 	if (priv->uri)
 		soup_uri_free (priv->uri);
@@ -1265,3 +1267,64 @@
 
 	return priv->io_status;
 }
+
+/**
+ * SoupChunkAllocator:
+ * @msg: the #SoupMessage the chunk is being allocated for
+ * @max_len: the maximum length that will be read, or 0.
+ * @user_data: the data passed to soup_message_set_chunk_allocator()
+ *
+ * The prototype for a chunk allocation callback. This should allocate
+ * a new #SoupBuffer and return it for the I/O layer to read message
+ * body data into.
+ *
+ * If @max_len is non-0, it indicates the maximum number of bytes that
+ * could be read, based on what is known about the message size. Note
+ * that this might be a very large number, and you should not simply
+ * try to allocate that many bytes blindly. If @max_len is 0, that
+ * means that libsoup does not know how many bytes remain to be read,
+ * and the allocator should return a buffer of a convenient size.
+ *
+ * Return value: the new buffer
+ **/
+
+/**
+ * soup_message_set_chunk_allocator:
+ * @msg: a #SoupMessage
+ * @allocator: the chunk allocator callback
+ * @user_data: data to pass to @allocator
+ * @destroy_notify: destroy notifier to free @user_data when @msg is
+ * destroyed
+ *
+ * Sets an alternate chunk-allocation function to use when reading
+ * @msg's body. Every time data is available to read, the I/O layer
+ * will call @allocator, which should return a #SoupBuffer. The I/O
+ * layer will then read data off the network into that buffer, and
+ * update the buffer's %length to indicate how much data it read. This
+ * buffer will then be processed normally (eg, it will be emitted in
+ * the #SoupMessage::got_chunk signal, and appended to the message
+ * body unless %SOUP_MESSAGE_OVERWRITE_CHUNKS is set).
+ *
+ * If @allocator returns %NULL, the message will be paused. It is up
+ * to the application to make sure that it is unpaused when it becomes
+ * possible to allocate a new buffer.
+ **/
+void
+soup_message_set_chunk_allocator (SoupMessage *msg,
+				  SoupChunkAllocator allocator,
+				  gpointer user_data,
+				  GDestroyNotify destroy_notify)
+{
+	SoupMessagePrivate *priv;
+
+	g_return_if_fail (SOUP_IS_MESSAGE (msg));
+
+	priv = SOUP_MESSAGE_GET_PRIVATE (msg);
+
+	if (priv->chunk_allocator_dnotify)
+		priv->chunk_allocator_dnotify (priv->chunk_allocator_data);
+
+	priv->chunk_allocator         = allocator;
+	priv->chunk_allocator_data    = user_data;
+	priv->chunk_allocator_dnotify = destroy_notify;
+}
Index: libsoup/soup-message-body.c
===================================================================
--- libsoup/soup-message-body.c	(revision 1072)
+++ libsoup/soup-message-body.c	(working copy)
@@ -43,6 +43,10 @@
  *
  * Describes how #SoupBuffer should use the data passed in by the
  * caller.
+ *
+ * See also soup_buffer_new_external(), which allows to you create a
+ * buffer and provide a callback which will be invoked to free its
+ * data.
  **/
 
 /**
@@ -62,12 +66,14 @@
 	SoupMemoryUse  use;
 	guint          refcount;
 
-	/* @other is used in subbuffers to store a reference to
-	 * the parent buffer, or in TEMPORARY buffers to store a
-	 * reference to a copy (see soup_buffer_copy()). Either
-	 * way, we hold a ref.
+	/* @extra_data is used in subbuffers to store a reference to
+	 * the parent buffer, in TEMPORARY buffers to store a
+	 * reference to a copy (see soup_buffer_copy()), and in
+	 * external buffers to store a pointer to the external data.
+	 * @destroy_notify will be called to free it.
 	 */
-	SoupBuffer    *other;
+	gpointer       extra_data;
+	GDestroyNotify destroy_notify;
 } SoupBufferPrivate;
 
 /**
@@ -116,16 +122,66 @@
 {
 	SoupBufferPrivate *priv;
 
+	/* Normally this is just a ref, but if @parent is TEMPORARY,
+	 * it will do an actual copy.
+	 */
+	parent = soup_buffer_copy (parent);
+
 	priv = g_slice_new0 (SoupBufferPrivate);
-	priv->other = soup_buffer_copy (parent);
-	priv->buffer.data = priv->other->data + offset;
+	priv->buffer.data = parent->data + offset;
 	priv->buffer.length = length;
 	priv->use = SOUP_MEMORY_STATIC;
+	priv->extra_data = parent;
+	priv->destroy_notify = (GDestroyNotify)soup_buffer_free;
 	priv->refcount = 1;
+
 	return (SoupBuffer *)priv;
 }
 
 /**
+ * soup_buffer_new_external:
+ * @data: data
+ * @length: length of @data
+ * @destroy_data: pointer to an object that owns @data
+ * @destroy_notify: a function to free the data when the buffer is freed
+ *
+ * Creates a new #SoupBuffer containing @length bytes from @data. When
+ * the #SoupBuffer is freed, it will call @destroy_notify, passing
+ * @destroy_data to it. You must ensure that @data will remain valid
+ * until @destroy_notify is called.
+ *
+ * For example, you could use this to create a buffer containing data
+ * returned from libxml:
+ *
+ * <informalexample><programlisting>
+ *	xmlDocDumpMemory (doc, &xmlbody, &len);
+ *	return soup_buffer_new_external (xmlbody, len, xmlbody,
+ *	                                 (GDestroyNotify)xmlFree);
+ * </programlisting></informalexample>
+ *
+ * In this example, @data and @destroy_data are the same, but in other
+ * cases they would be different (eg, @destroy_data would be a object,
+ * and @data would be one of the object's fields).
+ *
+ * Return value: the new #SoupBuffer.
+ **/
+SoupBuffer *
+soup_buffer_new_external (gconstpointer  data, gsize length,
+			  gpointer destroy_data, GDestroyNotify destroy_notify)
+{
+	SoupBufferPrivate *priv = g_slice_new0 (SoupBufferPrivate);
+
+	priv->buffer.data = data;
+	priv->buffer.length = length;
+	priv->use = SOUP_MEMORY_STATIC;
+	priv->extra_data = destroy_data;
+	priv->destroy_notify = destroy_notify;
+	priv->refcount = 1;
+
+	return (SoupBuffer *)priv;
+}
+
+/**
  * soup_buffer_copy:
  * @buffer: a #SoupBuffer
  *
@@ -151,14 +207,16 @@
 
 	/* For TEMPORARY buffers, we need to do a real copy the
 	 * first time, and then after that, we just keep returning
-	 * the copy. Use priv->other to store the copy.
+	 * the copy.
 	 */
 
-	if (!priv->other) {
-		priv->other = soup_buffer_new (SOUP_MEMORY_COPY,
-					       buffer->data, buffer->length);
+	if (!priv->extra_data) {
+		priv->extra_data = soup_buffer_new (SOUP_MEMORY_COPY,
+						    buffer->data,
+						    buffer->length);
+		priv->destroy_notify = (GDestroyNotify)soup_buffer_free;
 	}
-	return soup_buffer_copy (priv->other);
+	return soup_buffer_copy (priv->extra_data);
 }
 
 /**
@@ -177,8 +235,8 @@
 	if (!--priv->refcount) {
 		if (priv->use == SOUP_MEMORY_TAKE)
 			g_free ((gpointer)buffer->data);
-		if (priv->other)
-			soup_buffer_free (priv->other);
+		if (priv->destroy_notify)
+			priv->destroy_notify (priv->extra_data);
 		g_slice_free (SoupBufferPrivate, priv);
 	}
 }
Index: libsoup/soup-message-body.h
===================================================================
--- libsoup/soup-message-body.h	(revision 1072)
+++ libsoup/soup-message-body.h	(working copy)
@@ -31,6 +31,10 @@
 SoupBuffer *soup_buffer_new_subbuffer (SoupBuffer    *parent,
 				       gsize          offset,
 				       gsize          length);
+SoupBuffer *soup_buffer_new_external  (gconstpointer  data,
+				       gsize          length,
+				       gpointer       destroy_data,
+				       GDestroyNotify destroy_notify);
 
 SoupBuffer *soup_buffer_copy          (SoupBuffer    *buffer);
 void        soup_buffer_free          (SoupBuffer    *buffer);
Index: libsoup/soup-message-private.h
===================================================================
--- libsoup/soup-message-private.h	(revision 1072)
+++ libsoup/soup-message-private.h	(working copy)
@@ -22,6 +22,10 @@
 	gpointer           io_data;
 	SoupMessageIOStatus io_status;
 
+	SoupChunkAllocator chunk_allocator;
+	gpointer           chunk_allocator_data;
+	GDestroyNotify     chunk_allocator_dnotify;
+
 	guint              msg_flags;
 
 	SoupHTTPVersion    http_version;
Index: libsoup/soup-message-io.c
===================================================================
--- libsoup/soup-message-io.c	(revision 1072)
+++ libsoup/soup-message-io.c	(working copy)
@@ -266,27 +266,39 @@
 	SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 	SoupMessageIOData *io = priv->io_data;
 	SoupSocketIOStatus status;
-	guchar read_buf[RESPONSE_BLOCK_SIZE];
-	guint len = sizeof (read_buf);
+	guchar *stack_buf = NULL;
+	gsize len;
 	gboolean read_to_eof = (io->read_encoding == SOUP_ENCODING_EOF);
 	gsize nread;
 	GError *error = NULL;
 	SoupBuffer *buffer;
 
 	while (read_to_eof || io->read_length > 0) {
-		if (!read_to_eof)
-			len = MIN (len, io->read_length);
+		if (priv->chunk_allocator) {
+			buffer = priv->chunk_allocator (msg, io->read_length, priv->chunk_allocator_data);
+			if (!buffer) {
+				soup_message_io_pause (msg);
+				return FALSE;
+			}
+		} else {
+			if (!stack_buf)
+				stack_buf = alloca (RESPONSE_BLOCK_SIZE);
+			buffer = soup_buffer_new (SOUP_MEMORY_TEMPORARY,
+						  stack_buf,
+						  RESPONSE_BLOCK_SIZE);
+		}
 
-		status = soup_socket_read (io->sock, read_buf, len,
+		if (read_to_eof)
+			len = buffer->length;
+		else
+			len = MIN (buffer->length, io->read_length);
+
+		status = soup_socket_read (io->sock,
+					   (guchar *)buffer->data, len,
 					   &nread, NULL, &error);
 
-		switch (status) {
-		case SOUP_SOCKET_OK:
-			if (!nread)
-				break;
-
-			buffer = soup_buffer_new (SOUP_MEMORY_TEMPORARY,
-						  read_buf, nread);
+		if (status == SOUP_SOCKET_OK && nread) {
+			buffer->length = nread;
 			if (!(priv->msg_flags & SOUP_MESSAGE_OVERWRITE_CHUNKS))
 				soup_message_body_append_buffer (io->read_body, buffer);
 
@@ -296,6 +308,12 @@
 			soup_message_got_chunk (msg, buffer);
 			soup_buffer_free (buffer);
 			SOUP_MESSAGE_IO_RETURN_VAL_IF_CANCELLED_OR_PAUSED (FALSE);
+			return TRUE;
+		}
+
+		soup_buffer_free (buffer);
+		switch (status) {
+		case SOUP_SOCKET_OK:
 			break;
 
 		case SOUP_SOCKET_EOF: