From: Timo Sirainen <timo.sirainen@dovecot.fi>
Date: Thu, 26 Oct 2017 13:37:49 +0000 (+0300)
Subject: lib: istream-seekable - Change to use i_stream_*alloc()
X-Git-Tag: 2.3.0.rc1~645
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=3e55775bc7a37ebc05e06c04cafb32eee9888e87;p=thirdparty%2Fdovecot%2Fcore.git

lib: istream-seekable - Change to use i_stream_*alloc()

This simplifies implementing the snapshotting.
---

diff --git a/src/lib/istream-seekable.c b/src/lib/istream-seekable.c
index 1fd14c72b8..c9703657ce 100644
--- a/src/lib/istream-seekable.c
+++ b/src/lib/istream-seekable.c
@@ -3,6 +3,7 @@
 #include "lib.h"
 #include "buffer.h"
 #include "str.h"
+#include "memarea.h"
 #include "read-full.h"
 #include "write-full.h"
 #include "safe-mkstemp.h"
@@ -20,11 +21,11 @@ struct seekable_istream {
 	char *temp_path;
 	uoff_t write_peak;
 	uoff_t size;
+	size_t buffer_peak;
 
 	int (*fd_callback)(const char **path_r, void *context);
 	void *context;
 
-	buffer_t *membuf;
 	struct istream **input, *cur_input;
 	struct istream *fd_input;
 	unsigned int cur_idx;
@@ -54,7 +55,7 @@ static void i_stream_seekable_destroy(struct iostream_private *stream)
 {
 	struct seekable_istream *sstream = (struct seekable_istream *)stream;
 
-	buffer_free(&sstream->membuf);
+	i_stream_free_buffer(&sstream->istream);
 	i_stream_unref(&sstream->fd_input);
 	unref_streams(sstream);
 
@@ -91,14 +92,15 @@ static int copy_to_temp_file(struct seekable_istream *sstream)
 		return -1;
 
 	/* copy our currently read buffer to it */
-	if (write_full(fd, sstream->membuf->data, sstream->membuf->used) < 0) {
+	i_assert(stream->pos <= sstream->buffer_peak);
+	if (write_full(fd, stream->buffer, sstream->buffer_peak) < 0) {
 		if (!ENOSPACE(errno))
 			i_error("istream-seekable: write_full(%s) failed: %m", path);
 		i_close_fd(&fd);
 		return -1;
 	}
 	sstream->temp_path = i_strdup(path);
-	sstream->write_peak = sstream->membuf->used;
+	sstream->write_peak = sstream->buffer_peak;
 
 	sstream->fd = fd;
 	sstream->fd_input = i_stream_create_fd_autoclose(&fd,
@@ -107,7 +109,6 @@ static int copy_to_temp_file(struct seekable_istream *sstream)
 		"(seekable temp-istream for: %s)", i_stream_get_name(&stream->istream)));
 
 	/* read back the data we just had in our buffer */
-	i_stream_seek(sstream->fd_input, stream->istream.v_offset);
 	for (;;) {
 		buffer = i_stream_get_data(sstream->fd_input, &size);
 		if (size >= stream->pos)
@@ -132,8 +133,7 @@ static int copy_to_temp_file(struct seekable_istream *sstream)
 	i_stream_set_max_buffer_size(sstream->fd_input,
 				     sstream->istream.max_buffer_size);
 	stream->buffer = buffer;
-	stream->pos = size;
-	buffer_free(&sstream->membuf);
+	i_stream_free_buffer(&sstream->istream);
 	return 0;
 }
 
@@ -181,15 +181,18 @@ static bool read_from_buffer(struct seekable_istream *sstream, ssize_t *ret_r)
 {
 	struct istream_private *stream = &sstream->istream;
 	const unsigned char *data;
-	size_t size, pos, offset;
-
-	i_assert(stream->skip == 0);
-
-	if (stream->istream.v_offset + stream->pos >= sstream->membuf->used) {
+	size_t size, avail_size;
+
+	if (stream->pos < sstream->buffer_peak) {
+		/* This could be the first read() or we could have already
+		   seeked backwards. */
+		i_assert(stream->pos == 0 && stream->skip == 0);
+		stream->skip = stream->istream.v_offset;
+		stream->pos = sstream->buffer_peak;
+		size = stream->pos - stream->skip;
+	} else {
 		/* need to read more */
-		if (sstream->membuf->used >= stream->max_buffer_size)
-			return FALSE;
-
+		i_assert(stream->pos == sstream->buffer_peak);
 		size = sstream->cur_input == NULL ? 0 :
 			i_stream_get_data_size(sstream->cur_input);
 		if (size == 0) {
@@ -202,17 +205,26 @@ static bool read_from_buffer(struct seekable_istream *sstream, ssize_t *ret_r)
 		/* we should have more now. */
 		data = i_stream_get_data(sstream->cur_input, &size);
 		i_assert(size > 0);
-		buffer_append(sstream->membuf, data, size);
+
+		/* change skip to 0 temporarily so i_stream_try_alloc() won't try to
+		   compress the buffer. */
+		size_t old_skip = stream->skip;
+		stream->skip = 0;
+		bool have_space = i_stream_try_alloc(stream, size, &avail_size);
+		stream->skip = old_skip;
+		if (!have_space)
+			return FALSE;
+
+		if (size > avail_size)
+			size = avail_size;
+		memcpy(stream->w_buffer + stream->pos, data, size);
+		stream->pos += size;
+		sstream->buffer_peak += size;
 		i_stream_skip(sstream->cur_input, size);
 	}
 
-	offset = stream->istream.v_offset;
-	stream->buffer = CONST_PTR_OFFSET(sstream->membuf->data, offset);
-	pos = sstream->membuf->used - offset;
-
-	*ret_r = pos - stream->pos;
+	*ret_r = size;
 	i_assert(*ret_r > 0);
-	stream->pos = pos;
 	return TRUE;
 }
 
@@ -221,21 +233,19 @@ static int i_stream_seekable_write_failed(struct seekable_istream *sstream)
 	struct istream_private *stream = &sstream->istream;
 	void *data;
 
-	i_assert(sstream->membuf == NULL);
+	i_assert(sstream->fd != -1);
 
-	sstream->membuf =
-		buffer_create_dynamic(default_pool, sstream->write_peak);
-	data = buffer_append_space_unsafe(sstream->membuf, sstream->write_peak);
+	stream->max_buffer_size = (size_t)-1;
+	data = i_stream_alloc(stream, sstream->write_peak);
 
 	if (pread_full(sstream->fd, data, sstream->write_peak, 0) < 0) {
 		i_error("istream-seekable: read(%s) failed: %m", sstream->temp_path);
-		buffer_free(&sstream->membuf);
+		memarea_unref(&stream->memarea);
 		return -1;
 	}
 	i_stream_destroy(&sstream->fd_input);
 	i_close_fd(&sstream->fd);
 
-	stream->max_buffer_size = (size_t)-1;
 	i_free_and_null(sstream->temp_path);
 	return 0;
 }
@@ -247,11 +257,7 @@ static ssize_t i_stream_seekable_read(struct istream_private *stream)
 	size_t size, pos;
 	ssize_t ret;
 
-	stream->buffer = CONST_PTR_OFFSET(stream->buffer, stream->skip);
-	stream->pos -= stream->skip;
-	stream->skip = 0;
-
-	if (sstream->membuf != NULL) {
+	if (sstream->fd == -1) {
 		if (read_from_buffer(sstream, &ret))
 			return ret;
 
@@ -262,15 +268,22 @@ static ssize_t i_stream_seekable_read(struct istream_private *stream)
 				i_unreached();
 			return ret;
 		}
-		i_assert(sstream->membuf == NULL);
+		i_assert(sstream->fd != -1);
 	}
 
+	stream->buffer = CONST_PTR_OFFSET(stream->buffer, stream->skip);
+	stream->pos -= stream->skip;
+	stream->skip = 0;
+
 	i_assert(stream->istream.v_offset + stream->pos <= sstream->write_peak);
 	if (stream->istream.v_offset + stream->pos == sstream->write_peak) {
 		/* need to read more */
-		ret = read_more(sstream);
-		if (ret == -1 || ret == 0)
-			return ret;
+		if (sstream->cur_input == NULL ||
+		    i_stream_get_data_size(sstream->cur_input) == 0) {
+			ret = read_more(sstream);
+			if (ret == -1 || ret == 0)
+				return ret;
+		}
 
 		/* save to our file */
 		data = i_stream_get_data(sstream->cur_input, &size);
@@ -352,9 +365,9 @@ i_stream_seekable_stat(struct istream_private *stream, bool exact)
 		stream->statbuf = *st;
 	} else {
 		/* buffer is completely in memory */
-		i_assert(sstream->membuf != NULL);
+		i_assert(sstream->fd == -1);
 
-		stream->statbuf.st_size = sstream->membuf->used;
+		stream->statbuf.st_size = stream->pos;
 	}
 	return 0;
 }
@@ -397,7 +410,6 @@ i_streams_merge(struct istream *input[], size_t max_buffer_size,
 	sstream = i_new(struct seekable_istream, 1);
 	sstream->fd_callback = fd_callback;
 	sstream->context = context;
-	sstream->membuf = buffer_create_dynamic(default_pool, BUF_INITIAL_SIZE);
         sstream->istream.max_buffer_size = max_buffer_size;
 	sstream->fd = -1;
 	sstream->size = (uoff_t)-1;
@@ -406,11 +418,6 @@ i_streams_merge(struct istream *input[], size_t max_buffer_size,
 	memcpy(sstream->input, input, sizeof(*input) * count);
 	sstream->cur_input = sstream->input[0];
 
-	/* initialize our buffer from first stream's pending data */
-	data = i_stream_get_data(sstream->cur_input, &size);
-	buffer_append(sstream->membuf, data, size);
-	i_stream_skip(sstream->cur_input, size);
-
 	sstream->istream.iostream.close = i_stream_seekable_close;
 	sstream->istream.iostream.destroy = i_stream_seekable_destroy;
 	sstream->istream.iostream.set_max_buffer_size =
@@ -423,7 +430,16 @@ i_streams_merge(struct istream *input[], size_t max_buffer_size,
 	sstream->istream.istream.readable_fd = FALSE;
 	sstream->istream.istream.blocking = blocking;
 	sstream->istream.istream.seekable = TRUE;
-	return i_stream_create(&sstream->istream, NULL, -1);
+	(void)i_stream_create(&sstream->istream, NULL, -1);
+
+	/* initialize our buffer from first stream's pending data */
+	data = i_stream_get_data(sstream->cur_input, &size);
+	if (size > 0) {
+		memcpy(i_stream_alloc(&sstream->istream, size), data, size);
+		sstream->buffer_peak = size;
+		i_stream_skip(sstream->cur_input, size);
+	}
+	return &sstream->istream.istream;
 }
 
 static bool inputs_are_seekable(struct istream *input[])