/src/dovecot/src/lib/istream.c

Source
/* Copyright (c) Dovecot authors, see top-level COPYING file */

#include "lib.h"
#include "ioloop.h"
#include "array.h"
#include "str.h"
#include "memarea.h"
#include "istream-private.h"

static bool i_stream_is_buffer_invalid(const struct istream_private *stream);

void i_stream_set_name(struct istream *stream, const char *name)
{
  i_free(stream->real_stream->iostream.name);
  stream->real_stream->iostream.name = i_strdup(name);
}

const char *i_stream_get_name(struct istream *stream)
{
  i_assert(stream != NULL);
  while (stream->real_stream->iostream.name == NULL) {
    stream = stream->real_stream->parent;
    if (stream == NULL)
      return "";
  }
  return stream->real_stream->iostream.name;
}

static void i_stream_close_full(struct istream *stream, bool close_parents)
{
  io_stream_close(&stream->real_stream->iostream, close_parents);
  stream->closed = TRUE;

  if (stream->stream_errno == 0)
    stream->stream_errno = EPIPE;
}

void i_stream_destroy(struct istream **stream)
{
  if (*stream == NULL)
    return;

  i_stream_close_full(*stream, FALSE);
  i_stream_unref(stream);
}

void i_stream_ref(struct istream *stream)
{
  io_stream_ref(&stream->real_stream->iostream);
}

void i_stream_unref(struct istream **_stream)
{
  struct istream *stream = *_stream;
  struct istream_private *rstream;

  if (stream == NULL)
    return;

  *_stream = NULL;
  rstream = stream->real_stream;

  if (rstream->iostream.refcount > 1) {
    if (!io_stream_unref(&rstream->iostream))
      i_unreached();
  } else {
    /* The snapshot may contain pointers to the parent istreams.
       Free it before io_stream_unref() frees the parents. */
    i_stream_snapshot_free(&rstream->prev_snapshot);

    if (io_stream_unref(&rstream->iostream))
      i_unreached();
    str_free(&rstream->line_str);
    i_stream_unref(&rstream->parent);
    io_stream_free(&rstream->iostream);
  }
}

#undef i_stream_add_destroy_callback
void i_stream_add_destroy_callback(struct istream *stream,
           istream_callback_t *callback, void *context)
{
  io_stream_add_destroy_callback(&stream->real_stream->iostream,
               callback, context);
}

void i_stream_remove_destroy_callback(struct istream *stream,
              void (*callback)())
{
  io_stream_remove_destroy_callback(&stream->real_stream->iostream,
            callback);
}

int i_stream_get_fd(struct istream *stream)
{
  struct istream_private *_stream = stream->real_stream;

  return _stream->fd;
}

void i_stream_copy_fd(struct istream *dest, struct istream *source)
{
  int fd = i_stream_get_fd(source);

  i_assert(fd != -1);
  i_assert(dest->real_stream->fd == -1);
  dest->real_stream->fd = fd;
  dest->readable_fd = source->readable_fd;
}

void i_stream_set_error(struct istream *stream, int stream_errno,
      const char *fmt, ...)
{
  va_list args;

  va_start(args, fmt);
  stream->stream_errno = stream_errno;
  io_stream_set_verror(&stream->real_stream->iostream, fmt, args);
  va_end(args);
}

const char *i_stream_get_error(struct istream *stream)
{
  struct istream *s;

  /* we'll only return errors for streams that have stream_errno set or
     that have reached EOF. we might be returning unintended error
     otherwise. */
  if (stream->stream_errno == 0)
    return stream->eof ? "EOF" : "<no error>";

  for (s = stream; s != NULL; s = s->real_stream->parent) {
    if (s->stream_errno == 0)
      break;
    if (s->real_stream->iostream.error != NULL)
      return s->real_stream->iostream.error;
  }
  return strerror(stream->stream_errno);
}

const char *i_stream_get_disconnect_reason(struct istream *stream)
{
  return io_stream_get_disconnect_reason(stream, NULL);
}

void i_stream_close(struct istream *stream)
{
  if (stream != NULL)
    i_stream_close_full(stream, TRUE);
}

void i_stream_set_init_buffer_size(struct istream *stream, size_t size)
{
  stream->real_stream->init_buffer_size = size;
}

void i_stream_set_max_buffer_size(struct istream *stream, size_t max_size)
{
  io_stream_set_max_buffer_size(&stream->real_stream->iostream, max_size);
}

size_t i_stream_get_max_buffer_size(struct istream *stream)
{
  size_t max_size = 0;

  do {
    if (max_size < stream->real_stream->max_buffer_size)
      max_size = stream->real_stream->max_buffer_size;
    stream = stream->real_stream->parent;
  } while (stream != NULL);
  return max_size;
}

void i_stream_set_return_partial_line(struct istream *stream, bool set)
{
  stream->real_stream->return_nolf_line = set;
}

void i_stream_set_persistent_buffers(struct istream *stream, bool set)
{
  do {
    stream->real_stream->nonpersistent_buffers = !set;
    stream = stream->real_stream->parent;
  } while (stream != NULL);
}

void i_stream_set_blocking(struct istream *stream, bool blocking)
{
  int prev_fd = -1;

  do {
    stream->blocking = blocking;
    if (stream->real_stream->fd != -1 &&
        stream->real_stream->fd != prev_fd) {
      fd_set_nonblock(stream->real_stream->fd, !blocking);
      prev_fd = stream->real_stream->fd;
    }
    stream = stream->real_stream->parent;
  } while (stream != NULL);
}

static void i_stream_update(struct istream_private *stream)
{
  if (stream->parent == NULL)
    stream->access_counter++;
  else {
    stream->access_counter =
      stream->parent->real_stream->access_counter;
    stream->parent_expected_offset = stream->parent->v_offset;
  }
}

static bool snapshot_has_memarea(struct istream_snapshot *snapshot,
         struct memarea *memarea)
{
  if (snapshot->old_memarea == memarea)
    return TRUE;
  if (snapshot->prev_snapshot != NULL)
    return snapshot_has_memarea(snapshot->prev_snapshot, memarea);
  return FALSE;
}

struct istream_snapshot *
i_stream_default_snapshot(struct istream_private *stream,
        struct istream_snapshot *prev_snapshot)
{
  struct istream_snapshot *snapshot;

  if (stream->memarea != NULL) {
    if (prev_snapshot != NULL) {
      if (snapshot_has_memarea(prev_snapshot, stream->memarea))
        return prev_snapshot;
    }
    /* This stream has a memarea. Reference it, so we can later on
       rollback if needed. */
    snapshot = i_new(struct istream_snapshot, 1);
    snapshot->old_memarea = stream->memarea;
    snapshot->prev_snapshot = prev_snapshot;
    memarea_ref(snapshot->old_memarea);
    return snapshot;
  }
  if (stream->parent == NULL) {
    if (stream->nonpersistent_buffers) {
      /* Assume that memarea would be used normally, but
         now it's NULL because the buffer is empty and
         empty buffers are freed. */
      i_assert(stream->skip == stream->pos);
      return prev_snapshot;
    }
    i_panic("%s is missing istream.snapshot() implementation",
      i_stream_get_name(&stream->istream));
  }
  struct istream_private *_parent_stream =
    stream->parent->real_stream;
  return _parent_stream->snapshot(_parent_stream, prev_snapshot);
}

void i_stream_snapshot_free(struct istream_snapshot **_snapshot)
{
  struct istream_snapshot *snapshot = *_snapshot;

  if (*_snapshot == NULL)
    return;
  *_snapshot = NULL;

  i_stream_snapshot_free(&snapshot->prev_snapshot);
  if (snapshot->free != NULL)
    snapshot->free(snapshot);
  else {
    if (snapshot->old_memarea != NULL)
      memarea_unref(&snapshot->old_memarea);
    i_stream_unref(&snapshot->istream);
    i_free(snapshot);
  }
}

static struct istream_snapshot *
i_stream_noop_snapshot(struct istream_private *stream ATTR_UNUSED,
           struct istream_snapshot *prev_snapshot)
{
  return prev_snapshot;
}

static struct istream_private *
i_stream_get_io_parent(struct istream_private *_stream)
{
  if (_stream->parent != NULL)
    return _stream->parent->real_stream;
  if (_stream->io_parent != NULL)
    return _stream->io_parent->real_stream;
  return NULL;
}

static bool i_stream_is_io_pending_until_read(struct istream_private *_stream)
{
  struct istream_private *parent;

  while (!_stream->io_pending_until_read &&
         (parent = i_stream_get_io_parent(_stream)) != NULL)
    _stream = parent;
  return _stream->io_pending_until_read;
}

ssize_t i_stream_read(struct istream *stream)
{
  struct istream_private *_stream = stream->real_stream;
  ssize_t ret;
#ifdef DEBUG
  unsigned char prev_buf[4];
  const unsigned char *prev_data = _stream->buffer;
  size_t prev_skip = _stream->skip, prev_pos = _stream->pos;
  bool invalid = i_stream_is_buffer_invalid(_stream);

  i_assert(prev_skip <= prev_pos);
  size_t prev_size = prev_pos - prev_skip;
  if (invalid)
    ;
  else if (prev_size > 4) {
    memcpy(prev_buf, prev_data + prev_skip, 2);
    memcpy(prev_buf+2, prev_data + prev_pos - 2, 2);
  } else if (prev_size > 0) {
    memcpy(prev_buf, prev_data + prev_skip, prev_size);
  }
#endif

  if (_stream->skip != _stream->pos || _stream->prev_snapshot != NULL) {
    _stream->prev_snapshot =
      _stream->snapshot(_stream, _stream->prev_snapshot);
  }
  ret = i_stream_read_memarea(stream);
  if (ret > 0)
    i_stream_snapshot_free(&_stream->prev_snapshot);
#ifdef DEBUG
  else if (!invalid) {
    i_assert((_stream->pos - _stream->skip) == (prev_pos - prev_skip) ||
       prev_pos == prev_skip);
    if (prev_data == NULL)
      i_assert(prev_pos == prev_skip);
    else if (prev_pos - prev_skip <= 4)
      i_assert(memcmp(prev_buf, prev_data + prev_skip, prev_pos - prev_skip) == 0);
    else {
      i_assert(memcmp(prev_buf, prev_data + prev_skip, 2) == 0);
      i_assert(memcmp(prev_buf+2, prev_data + prev_pos - 2, 2) == 0);
    }
  }
#endif
  if (!_stream->istream.eof &&
      i_stream_is_io_pending_until_read(_stream)) {
    /* One of the parent istreams still has IO pending, because its
       read() wasn't called. Set IO back to pending to prevent
       hangs. */
    i_stream_set_input_pending(stream, TRUE);
  }
  return ret;
}

/* The istreams whose read() is currently running, innermost first. Used for
   verifying that istreams declare the istreams they read. */
struct istream_reader {
  const struct istream_reader *prev;
  struct istream *stream;
  /* The ioloop that was running when read() was called. A nested
     io_loop_run() (e.g. dict_wait()) runs unrelated istreams, which
     this istream isn't reading. */
  struct ioloop *ioloop;
};
static const struct istream_reader *istream_cur_reader = NULL;

static void i_stream_verify_reader(struct istream *stream)
{
  struct istream_private *reader;

  if (istream_cur_reader == NULL || istream_cur_reader->stream == stream)
    return;
  if (istream_cur_reader->ioloop != current_ioloop) {
    /* Running in a nested ioloop, so this istream is unrelated to
       the istream whose read() is running. */
    return;
  }
  if (stream->blocking) {
    /* Blocking istreams are read until they're finished, so their
       ioloop IO doesn't matter. read() implementations may create
       and drain such istreams internally (e.g. a temp file). */
    return;
  }
  if (istream_cur_reader->stream->blocking) {
    /* A blocking istream is also read until it's finished, so it
       never depends on an ioloop IO to wake it up, and it doesn't
       matter which istreams it reads. */
    return;
  }

  reader = istream_cur_reader->stream->real_stream;
  if (stream == reader->parent)
    return;
  if (reader->io_parent != NULL &&
      i_stream_get_root_io(stream) ==
      i_stream_get_root_io(reader->io_parent)) {
    /* The declared istream, or another istream reading from the
       same one (e.g. istream-decompress reads both its input and
       the decompressing istream created on top of it). */
    return;
  }
  if (reader->hidden_inputs == ISTREAM_HIDDEN_INPUTS_UNCHECKED ||
      reader->hidden_inputs == ISTREAM_HIDDEN_INPUTS_PANIC) {
    /* Reads istreams it doesn't declare, so there's nothing to
       compare against. */
    return;
  }

  i_panic("istream %s reads undeclared istream %s "
    "(missing istream_private.io_parent?)",
    i_stream_get_name(&reader->istream),
    i_stream_get_name(stream));
}

ssize_t i_stream_read_memarea(struct istream *stream)
{
  struct istream_private *_stream = stream->real_stream;
  size_t old_size;
  ssize_t ret;

  if (unlikely(stream->closed || stream->stream_errno != 0)) {
    stream->eof = TRUE;
    errno = stream->stream_errno;
    return -1;
  }

  i_stream_verify_reader(stream);
  stream->eof = FALSE;

  if (_stream->parent != NULL)
    i_stream_seek(_stream->parent, _stream->parent_expected_offset);

  old_size = _stream->pos - _stream->skip;
  if (_stream->pos < _stream->high_pos) {
    /* we're here because we seeked back within the read buffer. */
    ret = _stream->high_pos - _stream->pos;
    _stream->pos = _stream->high_pos;
    _stream->high_pos = 0;
  } else {
    struct istream_reader reader = {
      .prev = istream_cur_reader,
      .stream = stream,
      .ioloop = current_ioloop,
    };

    _stream->high_pos = 0;
    _stream->io_pending_until_read = FALSE;
    istream_cur_reader = &reader;
    ret = _stream->read(_stream);
    istream_cur_reader = reader.prev;
  }
  i_assert(_stream->skip <= _stream->pos);
  i_assert(old_size <= _stream->pos - _stream->skip);
  switch (ret) {
  case -2:
    i_assert(_stream->skip != _stream->pos);
    break;
  case -1:
    if (stream->stream_errno != 0) {
      /* error handling should be easier if we now just
         assume the stream is now at EOF */
      stream->eof = TRUE;
      errno = stream->stream_errno;
    } else {
      i_assert(stream->eof);
      i_assert(old_size == _stream->pos - _stream->skip);
    }
    break;
  case 0:
    i_assert(!stream->blocking);
    break;
  default:
    i_assert(ret > 0);
    i_assert(_stream->skip < _stream->pos);
    i_assert((size_t)ret+old_size == _stream->pos - _stream->skip);
    _stream->last_read_timeval = ioloop_timeval;
    break;
  }

  if (stream->stream_errno != 0) {
    /* error handling should be easier if we now just
       assume the stream is now at EOF. Note that we could get here
       even if read() didn't return -1, although that's a little
       bit sloppy istream implementation. */
    stream->eof = TRUE;
  }

  i_stream_update(_stream);
  /* verify that parents' access_counters are valid. the parent's
     i_stream_read() should guarantee this. */
  i_assert(!i_stream_is_buffer_invalid(_stream));
  return ret;
}

int i_stream_read_more_memarea(struct istream *stream,
             const unsigned char **data_r, size_t *size_r)
{
  *data_r = i_stream_get_data(stream, size_r);
  if (*size_r > 0)
    return 1;

  int ret = i_stream_read_memarea(stream);
  *data_r = i_stream_get_data(stream, size_r);
  return ret;
}

void i_stream_get_last_read_time(struct istream *stream, struct timeval *tv_r)
{
  *tv_r = stream->real_stream->last_read_timeval;
}

ssize_t i_stream_read_copy_from_parent(struct istream *istream)
{
  struct istream_private *stream = istream->real_stream;
  size_t pos;
  ssize_t ret;

  stream->pos -= stream->skip;
  stream->skip = 0;

  stream->buffer = i_stream_get_data(stream->parent, &pos);
  if (pos > stream->pos)
    ret = 0;
  else do {
    ret = i_stream_read_memarea(stream->parent);
    stream->istream.stream_errno = stream->parent->stream_errno;
    stream->istream.eof = stream->parent->eof;
    stream->buffer = i_stream_get_data(stream->parent, &pos);
    /* check again, in case the parent stream had been seeked
       backwards and the previous read() didn't get us far
       enough. */
  } while (pos <= stream->pos && ret > 0);
  if (ret == -2) {
    i_stream_update(stream);
    return -2;
  }

  ret = pos > stream->pos ? (ssize_t)(pos - stream->pos) :
    (ret == 0 ? 0 : -1);
  stream->pos = pos;
  i_assert(ret != -1 || stream->istream.eof ||
     stream->istream.stream_errno != 0);
  i_stream_update(stream);
  return ret;
}

void i_stream_free_buffer(struct istream_private *stream)
{
  if (stream->memarea != NULL) {
    memarea_unref(&stream->memarea);
    stream->w_buffer = NULL;
  } else if (stream->w_buffer != NULL) {
    i_free_and_null(stream->w_buffer);
  } else {
    /* don't know how to free it */
    return;
  }
  stream->buffer_size = 0;
}

void i_stream_skip(struct istream *stream, uoff_t count)
{
  struct istream_private *_stream = stream->real_stream;
  size_t data_size;

  data_size = _stream->pos - _stream->skip;
  if (count <= data_size) {
    /* within buffer */
    stream->v_offset += count;
    _stream->skip += count;
    if (_stream->nonpersistent_buffers &&
        _stream->skip == _stream->pos) {
      _stream->skip = _stream->pos = 0;
      i_stream_free_buffer(_stream);
    }
    return;
  }

  /* have to seek forward */
  count -= data_size;
  _stream->skip = _stream->pos;
  stream->v_offset += data_size;

  if (unlikely(stream->closed || stream->stream_errno != 0))
    return;

  _stream->seek(_stream, stream->v_offset + count, FALSE);
}

static bool i_stream_can_optimize_seek(struct istream_private *stream)
{
  if (stream->parent == NULL)
    return TRUE;

  /* use the fast route only if the parent stream hasn't been changed */
  if (stream->access_counter !=
      stream->parent->real_stream->access_counter)
    return FALSE;

  return i_stream_can_optimize_seek(stream->parent->real_stream);
}

void i_stream_seek(struct istream *stream, uoff_t v_offset)
{
  struct istream_private *_stream = stream->real_stream;

  if (v_offset >= stream->v_offset &&
      i_stream_can_optimize_seek(_stream))
    i_stream_skip(stream, v_offset - stream->v_offset);
  else {
    if (unlikely(stream->closed || stream->stream_errno != 0)) {
      stream->eof = TRUE;
      return;
    }
    stream->eof = FALSE;
    _stream->seek(_stream, v_offset, FALSE);
  }
  i_stream_update(_stream);
}

void i_stream_seek_mark(struct istream *stream, uoff_t v_offset)
{
  struct istream_private *_stream = stream->real_stream;

  if (unlikely(stream->closed || stream->stream_errno != 0))
    return;

  stream->eof = FALSE;
  _stream->seek(_stream, v_offset, TRUE);
  i_stream_update(_stream);
}

void i_stream_sync(struct istream *stream)
{
  struct istream_private *_stream = stream->real_stream;

  if (unlikely(stream->closed || stream->stream_errno != 0))
    return;

  if (_stream->sync != NULL) {
    _stream->sync(_stream);
    i_stream_update(_stream);
  }
}

int i_stream_stat(struct istream *stream, bool exact, const struct stat **st_r)
{
  struct istream_private *_stream = stream->real_stream;

  if (unlikely(stream->closed || stream->stream_errno != 0))
    return -1;

  if (_stream->stat(_stream, exact) < 0) {
    stream->eof = TRUE;
    return -1;
  }
  *st_r = &_stream->statbuf;
  return 0;
}

int i_stream_get_size(struct istream *stream, bool exact, uoff_t *size_r)
{
  struct istream_private *_stream = stream->real_stream;

  if (unlikely(stream->closed || stream->stream_errno != 0))
    return -1;

  int ret;
  if ((ret = _stream->get_size(_stream, exact, size_r)) < 0)
    stream->eof = TRUE;
  return ret;
}

bool i_stream_have_bytes_left(struct istream *stream)
{
  return i_stream_get_data_size(stream) > 0 || !stream->eof;
}

bool i_stream_read_eof(struct istream *stream)
{
  if (i_stream_get_data_size(stream) == 0)
    (void)i_stream_read(stream);
  return !i_stream_have_bytes_left(stream);
}

uoff_t i_stream_get_absolute_offset(struct istream *stream)
{
  uoff_t abs_offset = stream->v_offset;
  while (stream != NULL) {
    abs_offset += stream->real_stream->start_offset;
    stream = stream->real_stream->parent;
  }
  return abs_offset;
}

static char *i_stream_next_line_finish(struct istream_private *stream, size_t i)
{
  char *ret;
  size_t end;

  if (i > stream->skip && stream->buffer[i-1] == '\r') {
    end = i - 1;
    stream->line_crlf = TRUE;
  } else {
    end = i;
    stream->line_crlf = FALSE;
  }

  if (stream->buffer == stream->w_buffer &&
      end < stream->buffer_size) {
    /* modify the buffer directly */
    stream->w_buffer[end] = '\0';
    ret = (char *)stream->w_buffer + stream->skip;
  } else {
    /* use a temporary string to return it */
    if (stream->line_str == NULL)
      stream->line_str = str_new(default_pool, 256);
    str_truncate(stream->line_str, 0);
    if (stream->skip < end)
      str_append_data(stream->line_str, stream->buffer + stream->skip,
          end - stream->skip);
    ret = str_c_modifiable(stream->line_str);
  }

  if (i < stream->pos)
    i++;
  stream->istream.v_offset += i - stream->skip;
  stream->skip = i;
  return ret;
}

static char *i_stream_last_line(struct istream_private *_stream)
{
  if (_stream->istream.eof && _stream->skip != _stream->pos &&
      _stream->return_nolf_line) {
    /* the last line is missing LF and we want to return it. */
    return i_stream_next_line_finish(_stream, _stream->pos);
  }
  return NULL;
}

char *i_stream_next_line(struct istream *stream)
{
  struct istream_private *_stream = stream->real_stream;
  const unsigned char *pos;

  if (_stream->skip >= _stream->pos)
    return NULL;

  pos = memchr(_stream->buffer + _stream->skip, '\n',
         _stream->pos - _stream->skip);
  if (pos != NULL) {
    return i_stream_next_line_finish(_stream,
             pos - _stream->buffer);
  } else {
    return i_stream_last_line(_stream);
  }
}

char *i_stream_read_next_line(struct istream *stream)
{
  char *line;

  for (;;) {
    line = i_stream_next_line(stream);
    if (line != NULL)
      break;

    switch (i_stream_read(stream)) {
    case -2:
      io_stream_set_error(&stream->real_stream->iostream,
        "Line is too long (over %zu"
        " bytes at offset %"PRIuUOFF_T")",
        i_stream_get_data_size(stream), stream->v_offset);
      stream->stream_errno = errno = ENOBUFS;
      stream->eof = TRUE;
      return NULL;
    case -1:
      return i_stream_last_line(stream->real_stream);
    case 0:
      return NULL;
    }
  }
  return line;
}

bool i_stream_last_line_crlf(struct istream *stream)
{
  return stream->real_stream->line_crlf;
}

static bool i_stream_is_buffer_invalid(const struct istream_private *stream)
{
  if (stream->parent == NULL) {
    /* the buffer can't point to parent, because it doesn't exist */
    return FALSE;
  }
  if (stream->w_buffer != NULL) {
    /* we can pretty safely assume that the stream is using its
       own private buffer, so it can never become invalid. */
    return FALSE;
  }
  if (stream->access_counter !=
      stream->parent->real_stream->access_counter) {
    /* parent has been modified behind this stream, we can't trust
       that our buffer is valid */
    return TRUE;
  }
  return i_stream_is_buffer_invalid(stream->parent->real_stream);
}

const unsigned char *
i_stream_get_data(struct istream *stream, size_t *size_r)
{
  struct istream_private *_stream = stream->real_stream;

  if (_stream->skip >= _stream->pos) {
    *size_r = 0;
    return uchar_empty_ptr;
  }

  if (unlikely(i_stream_is_buffer_invalid(_stream))) {
    /* This stream may be using parent's buffer directly as
       _stream->buffer, but the parent stream has already been
       modified indirectly. This means that the buffer might no
       longer point to where we assume it points to. So we'll
       just return the stream as empty until it's read again.

       It's a bit ugly to suddenly drop data from the stream that
       was already read, but since this happens only with shared
       parent istreams the caller is hopefully aware enough that
       something like this might happen. The other solutions would
       be to a) try to automatically read the data back (but we
       can't handle errors..) or b) always copy data to stream's
       own buffer instead of pointing to parent's buffer (but this
       causes data copying that is nearly always unnecessary). */
    *size_r = 0;
    /* if we had already read until EOF, mark the stream again as
       not being at the end of file. */
    if (stream->stream_errno == 0) {
      _stream->skip = _stream->pos = 0;
      stream->eof = FALSE;
    }
    return uchar_empty_ptr;
  }

        *size_r = _stream->pos - _stream->skip;
        return _stream->buffer + _stream->skip;
}

size_t i_stream_get_data_size(struct istream *stream)
{
  size_t size;

  (void)i_stream_get_data(stream, &size);
  return size;
}

unsigned char *i_stream_get_modifiable_data(struct istream *stream,
              size_t *size_r)
{
  struct istream_private *_stream = stream->real_stream;

  if (_stream->skip >= _stream->pos || _stream->w_buffer == NULL) {
    *size_r = 0;
    return NULL;
  }

        *size_r = _stream->pos - _stream->skip;
        return _stream->w_buffer + _stream->skip;
}

int i_stream_read_data(struct istream *stream, const unsigned char **data_r,
           size_t *size_r, size_t threshold)
{
  ssize_t ret = 0;
  bool read_more = FALSE;

  do {
    *data_r = i_stream_get_data(stream, size_r);
    if (*size_r > threshold)
      return 1;

    /* we need more data */
    ret = i_stream_read(stream);
    if (ret > 0)
      read_more = TRUE;
  } while (ret > 0);

  *data_r = i_stream_get_data(stream, size_r);
  if (ret == -2)
    return -2;

  if (ret == 0) {
    /* need to read more */
    i_assert(!stream->blocking);
    return 0;
  }
  if (stream->stream_errno == 0 && read_more) {
    /* we read at least some new data */
    return 0;
  }
  return -1;
}

int i_stream_read_limited(struct istream *stream, const unsigned char **data_r,
        size_t *size_r, size_t limit)
{
  struct istream_private *_stream = stream->real_stream;
  int ret;

  *data_r = i_stream_get_data(stream, size_r);
  if (*size_r >= limit) {
    *size_r = limit;
    return 1;
  }

  _stream->data_limit = limit;
  ret = i_stream_read_more(stream, data_r, size_r);
  _stream->data_limit = 0;

  if (*size_r >= limit)
    *size_r = limit;
  return ret;
}

void i_stream_compress(struct istream_private *stream)
{
  i_assert(stream->memarea == NULL ||
     memarea_get_refcount(stream->memarea) == 1);

  if (stream->skip != stream->pos) {
    memmove(stream->w_buffer, stream->w_buffer + stream->skip,
      stream->pos - stream->skip);
  }
  stream->pos -= stream->skip;

  stream->skip = 0;
}

static void i_stream_w_buffer_free(void *buf)
{
  i_free(buf);
}

static void
i_stream_w_buffer_realloc(struct istream_private *stream, size_t old_size)
{
  void *new_buffer;

  if (stream->memarea != NULL &&
      memarea_get_refcount(stream->memarea) == 1) {
    /* Nobody else is referencing the memarea.
       We can just reallocate it. */
    memarea_free_without_callback(&stream->memarea);
    new_buffer = i_realloc(stream->w_buffer, old_size,
               stream->buffer_size);
  } else {
    new_buffer = i_malloc(stream->buffer_size);
    if (old_size > 0) {
      i_assert(stream->w_buffer != NULL);
      memcpy(new_buffer, stream->w_buffer, old_size);
    }
    if (stream->memarea != NULL)
      memarea_unref(&stream->memarea);
  }

  stream->w_buffer = new_buffer;
  stream->buffer = new_buffer;

  stream->memarea = memarea_init(stream->w_buffer, stream->buffer_size,
               i_stream_w_buffer_free, new_buffer);
}

void i_stream_grow_buffer(struct istream_private *stream, size_t bytes)
{
  size_t old_size, max_size;

  old_size = stream->buffer_size;

  stream->buffer_size = stream->pos + bytes;
  if (stream->buffer_size <= stream->init_buffer_size)
    stream->buffer_size = stream->init_buffer_size;
  else
    stream->buffer_size = nearest_power(stream->buffer_size);

  max_size = i_stream_get_max_buffer_size(&stream->istream);
  i_assert(max_size > 0);
  if (stream->buffer_size > max_size)
    stream->buffer_size = max_size;

  if (stream->buffer_size <= old_size)
    stream->buffer_size = old_size;
  else
    i_stream_w_buffer_realloc(stream, old_size);
}

bool i_stream_try_alloc(struct istream_private *stream,
      size_t wanted_size, size_t *size_r)
{
  i_assert(wanted_size > 0);
  i_assert(stream->buffer_size >= stream->pos);

  if (wanted_size > stream->buffer_size - stream->pos) {
    if (stream->skip > 0) {
      /* remove the unused bytes from beginning of buffer */
      if (stream->memarea != NULL &&
          memarea_get_refcount(stream->memarea) > 1) {
        /* The memarea is still referenced. We can't
           overwrite data until extra references are
           gone. */
        i_stream_w_buffer_realloc(stream, stream->buffer_size);
      }
      i_stream_compress(stream);
    } else if (stream->buffer_size < i_stream_get_max_buffer_size(&stream->istream)) {
      /* buffer is full - grow it */
      i_stream_grow_buffer(stream, I_STREAM_MIN_SIZE);
    }
  }

  if (stream->data_limit == 0 ||
      (stream->buffer_size - stream->skip) < stream->data_limit)
    *size_r = stream->buffer_size - stream->pos;
  else {
    size_t buffered = (stream->pos - stream->skip);

    if (buffered >= stream->data_limit)
      *size_r = 0;
    else
      *size_r = stream->data_limit - buffered;
  }
  i_assert(stream->w_buffer != NULL || *size_r == 0);
  return *size_r > 0;
}

bool ATTR_NOWARN_UNUSED_RESULT
i_stream_try_alloc_avoid_compress(struct istream_private *stream,
          size_t wanted_size, size_t *size_r)
{
  size_t old_skip = stream->skip;

  /* try first with skip=0, so no compression is done */
  stream->skip = 0;
  bool ret = i_stream_try_alloc(stream, wanted_size, size_r);
  stream->skip = old_skip;
  if (ret || old_skip == 0)
    return ret;
  /* it's full. try with compression. */
  return i_stream_try_alloc(stream, wanted_size, size_r);
}

void *i_stream_alloc(struct istream_private *stream, size_t size)
{
  size_t old_size, avail_size;

  (void)i_stream_try_alloc(stream, size, &avail_size);
  if (avail_size < size) {
    old_size = stream->buffer_size;
    stream->buffer_size = nearest_power(stream->pos + size);
    i_stream_w_buffer_realloc(stream, old_size);

    (void)i_stream_try_alloc(stream, size, &avail_size);
    i_assert(avail_size >= size);
  }
  return stream->w_buffer + stream->pos;
}

void i_stream_memarea_detach(struct istream_private *stream)
{
  if (stream->memarea != NULL) {
    /* Don't overwrite data in a snapshot. Allocate a new
       buffer instead. */
    memarea_unref(&stream->memarea);
    stream->buffer_size = 0;
    stream->buffer = NULL;
    stream->w_buffer = NULL;
  }
}

bool i_stream_add_data(struct istream *_stream, const unsigned char *data,
           size_t size)
{
  struct istream_private *stream = _stream->real_stream;
  size_t size2;

  if (size == 0)
    return TRUE;
  (void)i_stream_try_alloc(stream, size, &size2);
  if (size > size2)
    return FALSE;

  memcpy(stream->w_buffer + stream->pos, data, size);
  stream->pos += size;
  return TRUE;
}

struct istream *i_stream_get_root_io(struct istream *stream)
{
  struct istream_private *parent;

  while ((parent = i_stream_get_io_parent(stream->real_stream)) != NULL) {
    i_assert(stream->real_stream->io == NULL);
    stream = &parent->istream;
  }
  return stream;
}

static void i_stream_verify_pending_reachable(struct istream *owner)
{
  const struct istream_reader *reader;

  /* The IO may still be added to owner later on. But if an istream that
     is currently reading it already owns one, and it can't share it with
     owner, this pending is lost. */
  for (reader = istream_cur_reader; reader != NULL; reader = reader->prev) {
    struct istream *root;

    if (reader->ioloop != current_ioloop) {
      /* Nested ioloop - the outer istreams aren't reading
         this one. */
      break;
    }
    if (reader->stream->real_stream->hidden_inputs !=
        ISTREAM_HIDDEN_INPUTS_PANIC)
      continue;
    root = i_stream_get_root_io(reader->stream);
    if (root == owner || root->real_stream->io == NULL)
      continue;
    i_panic("i_stream_set_input_pending(%s) is lost: "
      "istream %s reads it and owns the ioloop IO",
      i_stream_get_name(owner), i_stream_get_name(root));
  }
}

void i_stream_set_input_pending(struct istream *stream, bool pending)
{
  if (!pending)
    return;

  stream->real_stream->io_pending_until_read = TRUE;

  stream = i_stream_get_root_io(stream);
  if (stream->real_stream->io != NULL)
    io_set_pending(stream->real_stream->io);
  else {
    i_stream_verify_pending_reachable(stream);
    stream->real_stream->io_pending = TRUE;
  }
}

void i_stream_switch_ioloop_to(struct istream *stream, struct ioloop *ioloop)
{
  io_stream_switch_ioloop_to(&stream->real_stream->iostream, ioloop);

  do {
    if (stream->real_stream->switch_ioloop_to != NULL) {
      stream->real_stream->switch_ioloop_to(
        stream->real_stream, ioloop);
    }
    if (stream->real_stream->io_parent != NULL) {
      /* Switch also the hidden input istream, which isn't
         reached by walking the istream-parents. */
      i_stream_switch_ioloop_to(
        stream->real_stream->io_parent, ioloop);
    }
    stream = stream->real_stream->parent;
  } while (stream != NULL);
}

void i_stream_switch_ioloop(struct istream *stream)
{
  i_stream_switch_ioloop_to(stream, current_ioloop);
}

bool i_stream_io_ever_added(struct istream *stream)
{
  return i_stream_get_root_io(stream)->real_stream->io_ever_added;
}

void i_stream_set_io(struct istream *stream, struct io *io)
{
  stream = i_stream_get_root_io(stream);

  i_assert(stream->real_stream->io == NULL);
  stream->real_stream->io = io;
  stream->real_stream->io_ever_added = TRUE;
  if (stream->real_stream->io_pending) {
    io_set_pending(io);
    stream->real_stream->io_pending = FALSE;
  }
}

void i_stream_unset_io(struct istream *stream, struct io *io)
{
  stream = i_stream_get_root_io(stream);

  i_assert(stream->real_stream->io == io);
  if (io_is_pending(io))
    stream->real_stream->io_pending = TRUE;
  stream->real_stream->io = NULL;
}

static void
i_stream_default_set_max_buffer_size(struct iostream_private *stream,
             size_t max_size)
{
  struct istream_private *_stream =
    container_of(stream, struct istream_private, iostream);

  _stream->max_buffer_size = max_size;
  if (_stream->parent != NULL)
    i_stream_set_max_buffer_size(_stream->parent, max_size);
}

static void i_stream_default_close(struct iostream_private *stream,
           bool close_parent)
{
  struct istream_private *_stream =
    container_of(stream, struct istream_private, iostream);

  if (close_parent)
    i_stream_close(_stream->parent);
}

static void i_stream_default_destroy(struct iostream_private *stream)
{
  struct istream_private *_stream =
    container_of(stream, struct istream_private, iostream);

  i_stream_free_buffer(_stream);
  i_stream_unref(&_stream->parent);
}

static void
i_stream_default_seek_seekable(struct istream_private *stream,
             uoff_t v_offset, bool mark ATTR_UNUSED)
{
  stream->istream.v_offset = v_offset;
  stream->skip = stream->pos = 0;
}

void i_stream_default_seek_nonseekable(struct istream_private *stream,
               uoff_t v_offset, bool mark ATTR_UNUSED)
{
  size_t available;

  if (stream->istream.v_offset > v_offset)
    i_panic("stream %s doesn't support seeking backwards",
      i_stream_get_name(&stream->istream));

  while (stream->istream.v_offset < v_offset) {
    (void)i_stream_read(&stream->istream);

    available = stream->pos - stream->skip;
    if (available == 0) {
      if (stream->istream.stream_errno != 0) {
        /* read failed */
        return;
      }
      io_stream_set_error(&stream->iostream,
        "Can't seek to offset %"PRIuUOFF_T
        ", because we have data only up to offset %"
        PRIuUOFF_T" (eof=%d)", v_offset,
        stream->istream.v_offset, stream->istream.eof ? 1 : 0);
      stream->istream.stream_errno = ESPIPE;
      return;
    }
    if (available <= v_offset - stream->istream.v_offset)
      i_stream_skip(&stream->istream, available);
    else {
      i_stream_skip(&stream->istream,
              v_offset - stream->istream.v_offset);
    }
  }
}

bool i_stream_nonseekable_try_seek(struct istream_private *stream,
           uoff_t v_offset)
{
  uoff_t start_offset = stream->istream.v_offset - stream->skip;

  if (v_offset < start_offset) {
    /* have to seek backwards */
    i_stream_seek(stream->parent, stream->parent_start_offset);
    stream->parent_expected_offset = stream->parent_start_offset;
    stream->skip = stream->pos = 0;
    stream->istream.v_offset = 0;
    stream->high_pos = 0;
    return FALSE;
  }

  if (v_offset <= start_offset + stream->pos) {
    /* seeking backwards within what's already cached */
    stream->skip = v_offset - start_offset;
    stream->istream.v_offset = v_offset;
    if (stream->high_pos == 0)
      stream->high_pos = stream->pos;
    stream->pos = stream->skip;
  } else {
    /* read forward */
    i_stream_default_seek_nonseekable(stream, v_offset, FALSE);
  }
  return TRUE;
}

static int
seekable_i_stream_get_size(struct istream_private *stream)
{
  if (stream->cached_stream_size == UOFF_T_MAX) {
    uoff_t old_offset = stream->istream.v_offset;
    ssize_t ret;

    do {
      i_stream_skip(&stream->istream,
        i_stream_get_data_size(&stream->istream));
    } while ((ret = i_stream_read(&stream->istream)) > 0);
    i_assert(ret == -1);
    if (stream->istream.stream_errno != 0)
      return -1;

    stream->cached_stream_size = stream->istream.v_offset;
    i_stream_seek(&stream->istream, old_offset);
  }
  stream->statbuf.st_size = stream->cached_stream_size;
  return 0;
}

static int
i_stream_default_stat(struct istream_private *stream, bool exact)
{
  const struct stat *st;

  if (stream->parent == NULL)
    return stream->istream.stream_errno == 0 ? 0 : -1;

  if (i_stream_stat(stream->parent, exact, &st) < 0) {
    stream->istream.stream_errno = stream->parent->stream_errno;
    return -1;
  }
  stream->statbuf = *st;
  if (exact && !stream->stream_size_passthrough) {
    /* exact size is not known, even if parent returned something */
    stream->statbuf.st_size = -1;
    if (stream->istream.seekable) {
      if (seekable_i_stream_get_size(stream) < 0)
        return -1;
    }
  } else {
    /* When exact=FALSE always return the parent stat's size, even
       if we know the exact value. This is necessary because
       otherwise e.g. mbox code can see two different values and
       think that the mbox file keeps changing. */
  }
  return 0;
}

static int
i_stream_default_get_size(struct istream_private *stream,
        bool exact, uoff_t *size_r)
{
  if (stream->stat(stream, exact) < 0)
    return -1;
  if (stream->statbuf.st_size == -1)
    return 0;

  *size_r = stream->statbuf.st_size;
  return 1;
}

void i_stream_init_parent(struct istream_private *_stream,
        struct istream *parent)
{
  _stream->access_counter = parent->real_stream->access_counter;
  _stream->parent = parent;
  _stream->parent_start_offset = parent->v_offset;
  _stream->parent_expected_offset = parent->v_offset;
  _stream->start_offset = parent->v_offset;
  /* if parent stream is an istream-error, copy the error */
  _stream->istream.stream_errno = parent->stream_errno;
  _stream->istream.eof = parent->eof;
  i_stream_ref(parent);
}

struct istream *
i_stream_create(struct istream_private *_stream, struct istream *parent, int fd,
    enum istream_hidden_inputs hidden_inputs,
    enum istream_create_flag flags)
{
  bool noop_snapshot = (flags & ISTREAM_CREATE_FLAG_NOOP_SNAPSHOT) != 0;

  /* io_parent is what ISTREAM_HIDDEN_INPUTS_DECLARED declares, so one
     without the other is a mistake in either direction. */
  i_assert((_stream->io_parent != NULL) ==
     (hidden_inputs == ISTREAM_HIDDEN_INPUTS_DECLARED));
  _stream->hidden_inputs = hidden_inputs;

  _stream->fd = fd;
  if (parent != NULL)
    i_stream_init_parent(_stream, parent);
  else if (_stream->memarea == NULL && !noop_snapshot) {
    /* The stream has no parent and no memarea yet. We'll assume
       that it wants to be using memareas for the reads. */
    _stream->memarea = memarea_init_empty();
  }
  _stream->istream.real_stream = _stream;

  if (_stream->iostream.close == NULL)
    _stream->iostream.close = i_stream_default_close;
  if (_stream->iostream.destroy == NULL)
    _stream->iostream.destroy = i_stream_default_destroy;
  if (_stream->seek == NULL) {
    _stream->seek = _stream->istream.seekable ?
      i_stream_default_seek_seekable :
      i_stream_default_seek_nonseekable;
  }
  if (_stream->stat == NULL)
    _stream->stat = i_stream_default_stat;
  if (_stream->get_size == NULL)
    _stream->get_size = i_stream_default_get_size;
  if (_stream->snapshot == NULL) {
    _stream->snapshot = noop_snapshot ?
      i_stream_noop_snapshot :
      i_stream_default_snapshot;
  }
  if (_stream->iostream.set_max_buffer_size == NULL) {
    _stream->iostream.set_max_buffer_size =
      i_stream_default_set_max_buffer_size;
  }
  if (_stream->init_buffer_size == 0)
    _stream->init_buffer_size = I_STREAM_MIN_SIZE;

  i_zero(&_stream->statbuf);
  _stream->statbuf.st_size = -1;
  _stream->statbuf.st_atime =
    _stream->statbuf.st_mtime =
    _stream->statbuf.st_ctime = ioloop_time;
  _stream->cached_stream_size = UOFF_T_MAX;

  io_stream_init(&_stream->iostream);

  if (_stream->istream.stream_errno != 0)
    _stream->istream.eof = TRUE;

  return &_stream->istream;
}

struct istream *i_stream_create_error(int stream_errno)
{
  struct istream_private *stream;

  stream = i_new(struct istream_private, 1);
  stream->istream.closed = TRUE;
  stream->istream.readable_fd = FALSE;
  stream->istream.blocking = TRUE;
  stream->istream.seekable = TRUE;
  stream->istream.eof = TRUE;
  stream->istream.stream_errno = stream_errno;
  /* Nothing can ever actually be read from this stream, but set a
     reasonable max_buffer_size anyway since some filter istreams don't
     behave properly otherwise. */
  stream->max_buffer_size = IO_BLOCK_SIZE;
  i_stream_create(stream, NULL, -1,
      ISTREAM_HIDDEN_INPUTS_NONE, 0);
  i_stream_set_name(&stream->istream, "(error)");
  return &stream->istream;
}

struct istream *
i_stream_create_error_str(int stream_errno, const char *fmt, ...)
{
  struct istream *input;
  va_list args;

  va_start(args, fmt);
  input = i_stream_create_error(stream_errno);
  io_stream_set_verror(&input->real_stream->iostream, fmt, args);
  va_end(args);
  return input;
}

Coverage Report

Created: 2026-09-14 06:49