src/lib/istream-chain.c

	istream-chain.c revision 2c42748505ef4aed83ff59b34e50ed5606900c86
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen/* Copyright (c) 2003-2013 Dovecot authors, see the included COPYING file */
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
2fbc2a7c65d30e46803195ebb4547176b85c22c7Timo Sirainen#include "lib.h"
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen#include "llist.h"
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen#include "istream-private.h"
2fbc2a7c65d30e46803195ebb4547176b85c22c7Timo Sirainen#include "istream-chain.h"
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
3c9783956dea385b322cd7fa6bf8c98c17a907a0Timo Sirainenstruct chain_istream;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainenstruct istream_chain_link {
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    struct istream_chain_link *prev, *next;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    struct istream *stream;
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainen    bool eof;
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainen};
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainenstruct istream_chain {
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    struct istream_chain_link *head, *tail;
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainen
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainen    struct chain_istream *stream;
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainen};
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainen
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainenstruct chain_istream {
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainen    struct istream_private istream;
a84eb0599fa1d796206eaed65c4e3239f0799276Timo Sirainen
e25885d4c7c4b392c66bbf26a9b892362d90f001Timo Sirainen    size_t prev_stream_left, prev_skip;
e25885d4c7c4b392c66bbf26a9b892362d90f001Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    struct istream_chain chain;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen};
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainenstatic void ATTR_NULL(2)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Siraineni_stream_chain_append_internal(struct istream_chain *chain,
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen                   struct istream *stream)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen{
35136dd2baf8dc30e4e754294ed81ff48e8c1e64Timo Sirainen    struct istream_chain_link *link;
1108376e39a19912e8394e64e19b1bc6f6691cf6Timo Sirainen
1108376e39a19912e8394e64e19b1bc6f6691cf6Timo Sirainen    if (stream == NULL && chain->tail != NULL && chain->tail->stream == NULL)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        return;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    link = i_new(struct istream_chain_link, 1);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    link->stream = stream;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    link->eof = stream == NULL;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    if (stream != NULL)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        i_stream_ref(stream);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    if (chain->head == NULL && stream != NULL) {
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        if (chain->stream->istream.max_buffer_size == 0) {
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen            chain->stream->istream.max_buffer_size =
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen                stream->real_stream->max_buffer_size;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        } else {
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen            i_stream_set_max_buffer_size(stream,
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen                chain->stream->istream.max_buffer_size);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        }
2cfe9983ce7a6280636ee12beccc2e865111967bTimo Sirainen    }
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    DLLIST2_APPEND(&chain->head, &chain->tail, link);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen}
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainenvoid i_stream_chain_append(struct istream_chain *chain, struct istream *stream)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen{
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    i_stream_chain_append_internal(chain, stream);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen}
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainenvoid i_stream_chain_append_eof(struct istream_chain *chain)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen{
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    i_stream_chain_append_internal(chain, NULL);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen}
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainenstatic void
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Siraineni_stream_chain_set_max_buffer_size(struct iostream_private *stream,
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen                    size_t max_size)
2e1e493b248dec0127b1eabeea5a8bc330378fcdTimo Sirainen{
b7828b34f6d81cdea62761932a1da1a444a29bcdTimo Sirainen    struct chain_istream *cstream = (struct chain_istream *)stream;
2e1e493b248dec0127b1eabeea5a8bc330378fcdTimo Sirainen    struct istream_chain_link *link = cstream->chain.head;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    cstream->istream.max_buffer_size = max_size;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    while (link != NULL) {
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        if (link->stream != NULL)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen            i_stream_set_max_buffer_size(link->stream, max_size);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        link = link->next;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    }
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen}
59151b71059df1190acd75d8717ed04a7920c862Timo Sirainen
59151b71059df1190acd75d8717ed04a7920c862Timo Sirainenstatic void i_stream_chain_destroy(struct iostream_private *stream)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen{
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    struct chain_istream *cstream = (struct chain_istream *)stream;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    struct istream_chain_link *link = cstream->chain.head;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    while (link != NULL) {
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        struct istream_chain_link *next = link->next;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        if (link->stream != NULL)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen            i_stream_unref(&link->stream);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen        i_free(link);
f8a86fdfb0048f9c87bf223373b35416ceb5856bTimo Sirainen        link = next;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    }
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    i_free(cstream->istream.w_buffer);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen}
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainenstatic void i_stream_chain_read_next(struct chain_istream *cstream)
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen{
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen    struct istream_chain_link *link = cstream->chain.head;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    struct istream *prev_input;
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen    const unsigned char *data;
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen    size_t data_size, size, cur_data_pos;
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    i_assert(link != NULL && link->stream != NULL);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    i_assert(link->stream->eof);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen    prev_input = link->stream;
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    data = i_stream_get_data(prev_input, &data_size);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    DLLIST2_REMOVE(&cstream->chain.head, &cstream->chain.tail, link);
6c2c5f20760b06bfb4a40b0ee2ef5ab016bc41f0Timo Sirainen    i_free(link);

    /* a) we have more streams, b) we have EOF, c) we need to wait
       for more streams */
    link = cstream->chain.head;
    if (link != NULL && link->stream != NULL)
        i_stream_seek(link->stream, 0);

    if (cstream->prev_stream_left > 0) {
        /* we've already buffered some of the prev_input. continue
           appending the rest to it. */
        cur_data_pos = cstream->istream.pos -
            (cstream->istream.skip + cstream->prev_stream_left);
        i_assert(cur_data_pos <= data_size);
        data += cur_data_pos;
        data_size -= cur_data_pos;
    } else {
        cstream->istream.pos = 0;
        cstream->istream.skip = 0;
        cstream->prev_stream_left = 0;
    }

    /* we already verified that the data size is less than the
       maximum buffer size */
    if (data_size > 0) {
        if (!i_stream_try_alloc(&cstream->istream, data_size, &size))
            i_unreached();
        i_assert(size >= data_size);
    }
    memcpy(cstream->istream.w_buffer + cstream->istream.pos,
           data, data_size);
    cstream->istream.pos += data_size;
    cstream->prev_stream_left += data_size;

    i_stream_skip(prev_input, i_stream_get_data_size(prev_input));
    i_stream_unref(&prev_input);
}

static ssize_t i_stream_chain_read(struct istream_private *stream)
{
    struct chain_istream *cstream = (struct chain_istream *)stream;
    struct istream_chain_link *link = cstream->chain.head;
    const unsigned char *data;
    size_t size, data_size, cur_data_pos, new_pos, bytes_skipped;
    size_t new_bytes_count;
    ssize_t ret;

    if (link != NULL && link->eof) {
        stream->istream.eof = TRUE;
        return -1;
    }

    i_assert(stream->skip >= cstream->prev_skip);
    bytes_skipped = stream->skip - cstream->prev_skip;

    if (cstream->prev_stream_left == 0) {
        /* no need to worry about buffers, skip everything */
    } else if (bytes_skipped < cstream->prev_stream_left) {
        /* we're still skipping inside buffer */
        cstream->prev_stream_left -= bytes_skipped;
        bytes_skipped = 0;
    } else {
        /* done with the buffer */
        bytes_skipped -= cstream->prev_stream_left;
        cstream->prev_stream_left = 0;
    }
    stream->pos -= bytes_skipped;
    stream->skip -= bytes_skipped;
    stream->buffer += bytes_skipped;
    cstream->prev_skip = stream->skip;

    if (link == NULL) {
        i_assert(bytes_skipped == 0);
        return 0;
    }
    i_stream_skip(link->stream, bytes_skipped);

    i_assert(stream->pos >= stream->skip + cstream->prev_stream_left);
    cur_data_pos = stream->pos - (stream->skip + cstream->prev_stream_left);

    data = i_stream_get_data(link->stream, &data_size);
    if (data_size > cur_data_pos)
        ret = 0;
    else {
        /* need to read more */
        i_assert(cur_data_pos == data_size);
        ret = i_stream_read(link->stream);
        if (ret == -2 || ret == 0)
            return ret;

        if (ret == -1) {
            if (link->stream->stream_errno != 0) {
                io_stream_set_error(&stream->iostream,
                    "read(%s) failed: %s",
                    i_stream_get_name(link->stream),
                    i_stream_get_error(link->stream));
                stream->istream.stream_errno =
                    link->stream->stream_errno;
                return -1;
            }
            /* EOF of this stream, go to next stream */
            i_stream_chain_read_next(cstream);
            cstream->prev_skip = stream->skip;
            return i_stream_chain_read(stream);
        }
        /* we read something */
        data = i_stream_get_data(link->stream, &data_size);
    }

    if (cstream->prev_stream_left == 0) {
        /* we can point directly to the current stream's buffers */
        stream->buffer = data;
        stream->pos -= stream->skip;
        stream->skip = 0;
        new_pos = data_size;
    } else if (data_size == cur_data_pos) {
        /* nothing new read */
        i_assert(ret == 0 || ret == -1);
        stream->buffer = stream->w_buffer;
        new_pos = stream->pos;
    } else {
        /* we still have some of the previous stream left. merge the
           new data with it. */
        i_assert(data_size > cur_data_pos);
        new_bytes_count = data_size - cur_data_pos;
        if (!i_stream_try_alloc(stream, new_bytes_count, &size)) {
            stream->buffer = stream->w_buffer;
            return -2;
        }
        stream->buffer = stream->w_buffer;

        if (new_bytes_count > size)
            new_bytes_count = size;
        memcpy(stream->w_buffer + stream->pos,
               data + cur_data_pos, new_bytes_count);
        new_pos = stream->pos + new_bytes_count;
    }

    ret = new_pos > stream->pos ? (ssize_t)(new_pos - stream->pos) :
        (ret == 0 ? 0 : -1);
    stream->pos = new_pos;
    cstream->prev_skip = stream->skip;
    return ret;
}

struct istream *i_stream_create_chain(struct istream_chain **chain_r)
{
    struct chain_istream *cstream;

    cstream = i_new(struct chain_istream, 1);
    cstream->chain.stream = cstream;
    cstream->istream.max_buffer_size = 256;

    cstream->istream.iostream.destroy = i_stream_chain_destroy;
    cstream->istream.iostream.set_max_buffer_size =
        i_stream_chain_set_max_buffer_size;

    cstream->istream.read = i_stream_chain_read;

    cstream->istream.istream.readable_fd = FALSE;
    cstream->istream.istream.blocking = FALSE;
    cstream->istream.istream.seekable = FALSE;

    *chain_r = &cstream->chain;
    return i_stream_create(&cstream->istream, NULL, -1);
}