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 1fd14c72b85496ed6d0ab7b27735c6070c2382bf..c9703657ceff0e51f8b6257ec71ba036babe0229 100644 (file)
--- 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[])