ostream-file.c revision 52b34f874a22615c67a9e1a1ee1679153559e377
c25356d5978632df6203437e1953bcb29e0c736fTimo Sirainen/* Copyright (c) 2002-2009 Dovecot authors, see the included COPYING file */
c25356d5978632df6203437e1953bcb29e0c736fTimo Sirainen
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen/* @UNSAFE: whole file */
49e358eebea107aad9919dcc4bd88cee8519ba2eTimo Sirainen
49e358eebea107aad9919dcc4bd88cee8519ba2eTimo Sirainen#include "lib.h"
49e358eebea107aad9919dcc4bd88cee8519ba2eTimo Sirainen#include "ioloop.h"
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen#include "write-full.h"
dd62b77c932d1b518f2a3e4bf80e36542becc256Timo Sirainen#include "network.h"
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen#include "sendfile-util.h"
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen#include "istream.h"
03f5c621d06d6b6d77a145196c9633a7aa64dc78Timo Sirainen#include "istream-internal.h"
c06f4017027263cf3a08becc551f5126409e2a83Timo Sirainen#include "ostream-internal.h"
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen
da2aa032ccfa8e7e4a4380ef738014549f4d2c2dTimo Sirainen#include <unistd.h>
da2aa032ccfa8e7e4a4380ef738014549f4d2c2dTimo Sirainen#include <sys/stat.h>
411d6baa37f31d90730e90c4a28c43e1974bbe58Timo Sirainen#ifdef HAVE_SYS_UIO_H
7e1f68ad71d3485f1882142837b01f7a98ca8467Timo Sirainen# include <sys/uio.h>
7e1f68ad71d3485f1882142837b01f7a98ca8467Timo Sirainen#endif
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen
252db51b6c0a605163326b3ea5d09e9936ca3b29Timo Sirainen/* try to keep the buffer size within 4k..128k. ReiserFS may actually return
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen 128k as optimal size. */
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen#define DEFAULT_OPTIMAL_BLOCK_SIZE 4096
5ac0b0bf32898c63da086ae169674ecac151a31eTimo Sirainen#define MAX_OPTIMAL_BLOCK_SIZE (128*1024)
5ac0b0bf32898c63da086ae169674ecac151a31eTimo Sirainen
2526d52441ef368215ab6bf04fd0356d3b09d235Timo Sirainen#define IS_STREAM_EMPTY(fstream) \
2526d52441ef368215ab6bf04fd0356d3b09d235Timo Sirainen ((fstream)->head == (fstream)->tail && !(fstream)->full)
09801f106cd531a28b4e03ec665e44c421264560Timo Sirainen
09801f106cd531a28b4e03ec665e44c421264560Timo Sirainen#define MAX_SSIZE_T(size) \
09801f106cd531a28b4e03ec665e44c421264560Timo Sirainen ((size) < SSIZE_T_MAX ? (size_t)(size) : SSIZE_T_MAX)
fe363b433b8038a69b55169da9dca27892ad7d18Timo Sirainen
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainenstruct file_ostream {
6ef7e31619edfaa17ed044b45861d106a86191efTimo Sirainen struct ostream_private ostream;
fe363b433b8038a69b55169da9dca27892ad7d18Timo Sirainen
10c96a244935de4add8213ba0b894178dfb889a5Timo Sirainen int fd;
bdcb00145ad87765e3fd22d4ebc4d2c029a326b9Timo Sirainen struct io *io;
bdcb00145ad87765e3fd22d4ebc4d2c029a326b9Timo Sirainen uoff_t buffer_offset;
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen uoff_t real_offset;
8cb72c59d5ea4e9e5f638d7ec840bb853f5a188eTimo Sirainen
8cb72c59d5ea4e9e5f638d7ec840bb853f5a188eTimo Sirainen unsigned char *buffer; /* ring-buffer */
8cb72c59d5ea4e9e5f638d7ec840bb853f5a188eTimo Sirainen size_t buffer_size, max_buffer_size, optimal_block_size;
8cb72c59d5ea4e9e5f638d7ec840bb853f5a188eTimo Sirainen size_t head, tail; /* first unsent/unused byte */
8cb72c59d5ea4e9e5f638d7ec840bb853f5a188eTimo Sirainen
8cb72c59d5ea4e9e5f638d7ec840bb853f5a188eTimo Sirainen unsigned int full:1; /* if head == tail, is buffer empty or full? */
8cb72c59d5ea4e9e5f638d7ec840bb853f5a188eTimo Sirainen unsigned int file:1;
cd56a23e21f1df3f79648cf07e2f4385e2fadebbTimo Sirainen unsigned int corked:1;
cd56a23e21f1df3f79648cf07e2f4385e2fadebbTimo Sirainen unsigned int flush_pending:1;
cd56a23e21f1df3f79648cf07e2f4385e2fadebbTimo Sirainen unsigned int socket_cork_set:1;
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen unsigned int no_socket_cork:1;
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen unsigned int no_sendfile:1;
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen unsigned int autoclose_fd:1;
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen};
d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen
5ac0b0bf32898c63da086ae169674ecac151a31eTimo Sirainenstatic void stream_send_io(struct file_ostream *fstream);
5ac0b0bf32898c63da086ae169674ecac151a31eTimo Sirainen
a10ed8c47534b4c6b6bf2711ccfe577e720a47b4Timo Sirainenstatic void stream_closed(struct file_ostream *fstream)
a10ed8c47534b4c6b6bf2711ccfe577e720a47b4Timo Sirainen{
5ac0b0bf32898c63da086ae169674ecac151a31eTimo Sirainen if (fstream->io != NULL)
27a44fcfd8d19bffe0f267f20a2b5d3fe7600fddTimo Sirainen io_remove(&fstream->io);
27a44fcfd8d19bffe0f267f20a2b5d3fe7600fddTimo Sirainen
27a44fcfd8d19bffe0f267f20a2b5d3fe7600fddTimo Sirainen if (fstream->autoclose_fd && fstream->fd != -1) {
c28f6aa0b70af4811c9ace9114fe827c2f503455Timo Sirainen if (close(fstream->fd) < 0)
c28f6aa0b70af4811c9ace9114fe827c2f503455Timo Sirainen i_error("file_ostream.close() failed: %m");
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen }
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen fstream->fd = -1;
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen fstream->ostream.ostream.closed = TRUE;
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen}
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainenstatic void o_stream_file_close(struct iostream_private *stream)
07e4875d250e7a7157cd99132aafc773cf3cdf83Timo Sirainen{
07e4875d250e7a7157cd99132aafc773cf3cdf83Timo Sirainen struct file_ostream *fstream = (struct file_ostream *)stream;
07e4875d250e7a7157cd99132aafc773cf3cdf83Timo Sirainen
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen /* flush output before really closing it */
7662010b03ffe5f2a6ecf4b4eb220d1c65efea76Timo Sirainen o_stream_flush(&fstream->ostream.ostream);
7662010b03ffe5f2a6ecf4b4eb220d1c65efea76Timo Sirainen
7662010b03ffe5f2a6ecf4b4eb220d1c65efea76Timo Sirainen stream_closed(fstream);
7662010b03ffe5f2a6ecf4b4eb220d1c65efea76Timo Sirainen}
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainenstatic void o_stream_file_destroy(struct iostream_private *stream)
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen{
0a49b316fc729e5d57268ffa63c7122ac73f994cTimo Sirainen struct file_ostream *fstream = (struct file_ostream *)stream;
0a49b316fc729e5d57268ffa63c7122ac73f994cTimo Sirainen
51e1a1c280ccb461a15827f7987d09cb9708b6e3Timo Sirainen i_free(fstream->buffer);
51e1a1c280ccb461a15827f7987d09cb9708b6e3Timo Sirainen}
51e1a1c280ccb461a15827f7987d09cb9708b6e3Timo Sirainen
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainenstatic void
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Siraineno_stream_file_set_max_buffer_size(struct iostream_private *stream,
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen size_t max_size)
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen{
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen struct file_ostream *fstream = (struct file_ostream *)stream;
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen fstream->max_buffer_size = max_size;
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen}
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainenstatic void update_buffer(struct file_ostream *fstream, size_t size)
602a0434db30d8e3292d1c161a803d96a879a74fTimo Sirainen{
602a0434db30d8e3292d1c161a803d96a879a74fTimo Sirainen size_t used;
602a0434db30d8e3292d1c161a803d96a879a74fTimo Sirainen
602a0434db30d8e3292d1c161a803d96a879a74fTimo Sirainen if (IS_STREAM_EMPTY(fstream) || size == 0)
602a0434db30d8e3292d1c161a803d96a879a74fTimo Sirainen return;
01f4ee4a0243f3fe9af763e1a540cd5cff0d63f5Timo Sirainen
07e4875d250e7a7157cd99132aafc773cf3cdf83Timo Sirainen if (fstream->head < fstream->tail) {
7d207b1e77a7b5e3fda640e353acfc86d261fedfTimo Sirainen /* ...HXXXT... */
7d207b1e77a7b5e3fda640e353acfc86d261fedfTimo Sirainen used = fstream->tail - fstream->head;
7d207b1e77a7b5e3fda640e353acfc86d261fedfTimo Sirainen i_assert(size <= used);
7d207b1e77a7b5e3fda640e353acfc86d261fedfTimo Sirainen fstream->head += size;
7d207b1e77a7b5e3fda640e353acfc86d261fedfTimo Sirainen } else {
01f4ee4a0243f3fe9af763e1a540cd5cff0d63f5Timo Sirainen /* XXXT...HXXX */
4b9f99761df5014c659cd87fddaf6854af428cfcTimo Sirainen used = fstream->buffer_size - fstream->head;
4b9f99761df5014c659cd87fddaf6854af428cfcTimo Sirainen if (size > used) {
4b9f99761df5014c659cd87fddaf6854af428cfcTimo Sirainen size -= used;
7e1f68ad71d3485f1882142837b01f7a98ca8467Timo Sirainen i_assert(size <= fstream->tail);
68a4946b12583b88fa802e52ebee45cd96056772Timo Sirainen fstream->head = size;
a3c197999dfe2b0c8ea38cb77cfa5e95026005c0Timo Sirainen } else {
a3c197999dfe2b0c8ea38cb77cfa5e95026005c0Timo Sirainen fstream->head += size;
923115fd382904fa13bb09bf307bf2835b52df60Timo Sirainen }
923115fd382904fa13bb09bf307bf2835b52df60Timo Sirainen
923115fd382904fa13bb09bf307bf2835b52df60Timo Sirainen fstream->full = FALSE;
7e1f68ad71d3485f1882142837b01f7a98ca8467Timo Sirainen }
89e195dfb5c4b0efd9b9f459771a4467674e5b1fTimo Sirainen
51e1a1c280ccb461a15827f7987d09cb9708b6e3Timo Sirainen if (fstream->head == fstream->tail)
51e1a1c280ccb461a15827f7987d09cb9708b6e3Timo Sirainen fstream->head = fstream->tail = 0;
c0435c854a0e7246373b9752d163095cc4fbe985Timo Sirainen
89e195dfb5c4b0efd9b9f459771a4467674e5b1fTimo Sirainen if (fstream->head == fstream->buffer_size)
137ea7ca34005345aa2304a940149b7f3774d727Timo Sirainen fstream->head = 0;
89e195dfb5c4b0efd9b9f459771a4467674e5b1fTimo Sirainen}
6f08b98ac63c25b747120d0c8f8e319b4e26ab0fTimo Sirainen
6f08b98ac63c25b747120d0c8f8e319b4e26ab0fTimo Sirainenstatic void o_stream_socket_cork(struct file_ostream *fstream)
6f08b98ac63c25b747120d0c8f8e319b4e26ab0fTimo Sirainen{
7e1f68ad71d3485f1882142837b01f7a98ca8467Timo Sirainen if (fstream->corked && !fstream->socket_cork_set) {
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen if (!fstream->no_socket_cork) {
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen if (net_set_cork(fstream->fd, TRUE) < 0)
68a4946b12583b88fa802e52ebee45cd96056772Timo Sirainen fstream->no_socket_cork = TRUE;
68a4946b12583b88fa802e52ebee45cd96056772Timo Sirainen else
3785910c303507db5f629684e6dde2cc7f83668eTimo Sirainen fstream->socket_cork_set = TRUE;
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen }
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen }
68a4946b12583b88fa802e52ebee45cd96056772Timo Sirainen}
c06f4017027263cf3a08becc551f5126409e2a83Timo Sirainen
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainenstatic int o_stream_lseek(struct file_ostream *fstream)
699fdc186f982f70d990820796eaa0f12133e27cTimo Sirainen{
699fdc186f982f70d990820796eaa0f12133e27cTimo Sirainen off_t ret;
699fdc186f982f70d990820796eaa0f12133e27cTimo Sirainen
c06f4017027263cf3a08becc551f5126409e2a83Timo Sirainen if (fstream->real_offset == fstream->buffer_offset)
c06f4017027263cf3a08becc551f5126409e2a83Timo Sirainen return 0;
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen
282a436a74d8835edb45cc019b1c916013013fd3Timo Sirainen ret = lseek(fstream->fd, (off_t)fstream->buffer_offset, SEEK_SET);
282a436a74d8835edb45cc019b1c916013013fd3Timo Sirainen if (ret < 0) {
282a436a74d8835edb45cc019b1c916013013fd3Timo Sirainen fstream->ostream.ostream.stream_errno = errno;
282a436a74d8835edb45cc019b1c916013013fd3Timo Sirainen return -1;
282a436a74d8835edb45cc019b1c916013013fd3Timo Sirainen }
ecc81625167ed96c04c02aa190a1ea5baa65b474Timo Sirainen
if (ret != (off_t)fstream->buffer_offset) {
fstream->ostream.ostream.stream_errno = EINVAL;
return -1;
}
fstream->real_offset = fstream->buffer_offset;
return 0;
}
static ssize_t o_stream_writev(struct file_ostream *fstream,
const struct const_iovec *iov, int iov_size)
{
ssize_t ret, ret2;
size_t size, sent;
bool partial;
int i;
o_stream_socket_cork(fstream);
if (iov_size == 1) {
if (!fstream->file ||
fstream->real_offset == fstream->buffer_offset) {
ret = write(fstream->fd, iov->iov_base, iov->iov_len);
if (ret > 0)
fstream->real_offset += ret;
} else {
ret = pwrite(fstream->fd, iov->iov_base, iov->iov_len,
fstream->buffer_offset);
}
partial = ret != (ssize_t)iov->iov_len;
} else {
if (o_stream_lseek(fstream) < 0)
return -1;
sent = 0; partial = FALSE;
while (iov_size > IOV_MAX) {
size = 0;
for (i = 0; i < IOV_MAX; i++)
size += iov[i].iov_len;
ret = writev(fstream->fd, (const struct iovec *)iov,
IOV_MAX);
if (ret != (ssize_t)size) {
partial = TRUE;
break;
}
fstream->real_offset += ret;
fstream->buffer_offset += ret;
sent += ret;
iov += IOV_MAX;
iov_size -= IOV_MAX;
}
if (iov_size <= IOV_MAX) {
size = 0;
for (i = 0; i < iov_size; i++)
size += iov[i].iov_len;
ret = writev(fstream->fd, (const struct iovec *)iov,
iov_size);
partial = ret != (ssize_t)size;
}
if (ret > 0) {
fstream->real_offset += ret;
ret += sent;
} else if (!fstream->file && sent > 0) {
/* return what we managed to get sent */
ret = sent;
}
}
if (ret < 0) {
if (errno == EAGAIN || errno == EINTR)
return 0;
fstream->ostream.ostream.stream_errno = errno;
stream_closed(fstream);
return -1;
}
if (unlikely(ret == 0 && fstream->file)) {
/* assume out of disk space */
fstream->ostream.ostream.stream_errno = ENOSPC;
stream_closed(fstream);
return -1;
}
fstream->buffer_offset += ret;
if (partial && fstream->file) {
/* we failed to write everything to a file. either we ran out
of disk space or we're writing to NFS. try to write the
rest to resolve this. */
size = ret;
while (iov_size > 0 && size >= iov->iov_len) {
size -= iov->iov_len;
iov++;
iov_size--;
}
i_assert(iov_size > 0);
if (size == 0)
ret2 = o_stream_writev(fstream, iov, iov_size);
else {
/* write the first iov separately */
struct const_iovec new_iov;
new_iov.iov_base =
CONST_PTR_OFFSET(iov->iov_base, size);
new_iov.iov_len = iov->iov_len - size;
ret2 = o_stream_writev(fstream, &new_iov, 1);
if (ret2 > 0) {
i_assert((size_t)ret2 == new_iov.iov_len);
/* write the rest */
if (iov_size > 1) {
ret += ret2;
ret2 = o_stream_writev(fstream, iov + 1,
iov_size - 1);
}
}
}
if (ret2 <= 0)
return ret2;
ret += ret2;
}
return ret;
}
/* returns how much of vector was used */
static int o_stream_fill_iovec(struct file_ostream *fstream,
struct const_iovec iov[2])
{
if (IS_STREAM_EMPTY(fstream))
return 0;
if (fstream->head < fstream->tail) {
iov[0].iov_base = fstream->buffer + fstream->head;
iov[0].iov_len = fstream->tail - fstream->head;
return 1;
} else {
iov[0].iov_base = fstream->buffer + fstream->head;
iov[0].iov_len = fstream->buffer_size - fstream->head;
if (fstream->tail == 0)
return 1;
else {
iov[1].iov_base = fstream->buffer;
iov[1].iov_len = fstream->tail;
return 2;
}
}
}
static int buffer_flush(struct file_ostream *fstream)
{
struct const_iovec iov[2];
int iov_len;
ssize_t ret;
iov_len = o_stream_fill_iovec(fstream, iov);
if (iov_len > 0) {
ret = o_stream_writev(fstream, iov, iov_len);
if (ret < 0)
return -1;
update_buffer(fstream, ret);
}
return IS_STREAM_EMPTY(fstream) ? 1 : 0;
}
static void o_stream_file_cork(struct ostream_private *stream, bool set)
{
struct file_ostream *fstream = (struct file_ostream *)stream;
int ret;
if (fstream->corked != set && !stream->ostream.closed) {
if (set && fstream->io != NULL)
io_remove(&fstream->io);
else if (!set) {
/* buffer flushing might close the stream */
ret = buffer_flush(fstream);
if (fstream->io == NULL &&
(ret == 0 || fstream->flush_pending) &&
!stream->ostream.closed) {
fstream->io = io_add(fstream->fd, IO_WRITE,
stream_send_io, fstream);
}
}
if (fstream->socket_cork_set) {
i_assert(!set);
if (net_set_cork(fstream->fd, FALSE) < 0)
fstream->no_socket_cork = TRUE;
fstream->socket_cork_set = FALSE;
}
fstream->corked = set;
}
}
static int o_stream_file_flush(struct ostream_private *stream)
{
struct file_ostream *fstream = (struct file_ostream *) stream;
return buffer_flush(fstream);
}
static void
o_stream_file_flush_pending(struct ostream_private *stream, bool set)
{
struct file_ostream *fstream = (struct file_ostream *) stream;
fstream->flush_pending = set;
if (set && !fstream->corked && fstream->io == NULL) {
fstream->io = io_add(fstream->fd, IO_WRITE,
stream_send_io, fstream);
}
}
static size_t get_unused_space(const struct file_ostream *fstream)
{
if (fstream->head > fstream->tail) {
/* XXXT...HXXX */
return fstream->head - fstream->tail;
} else if (fstream->head < fstream->tail) {
/* ...HXXXT... */
return (fstream->buffer_size - fstream->tail) + fstream->head;
} else {
/* either fully unused or fully used */
return fstream->full ? 0 : fstream->buffer_size;
}
}
static size_t o_stream_file_get_used_size(const struct ostream_private *stream)
{
const struct file_ostream *fstream =
(const struct file_ostream *)stream;
return fstream->buffer_size - get_unused_space(fstream);
}
static int o_stream_file_seek(struct ostream_private *stream, uoff_t offset)
{
struct file_ostream *fstream = (struct file_ostream *)stream;
if (offset > OFF_T_MAX || !fstream->file) {
stream->ostream.stream_errno = EINVAL;
return -1;
}
if (buffer_flush(fstream) < 0)
return -1;
stream->ostream.offset = offset;
fstream->buffer_offset = offset;
return 1;
}
static void o_stream_grow_buffer(struct file_ostream *fstream, size_t bytes)
{
size_t size, new_size, end_size;
size = nearest_power(fstream->buffer_size + bytes);
if (size > fstream->max_buffer_size) {
/* limit the size */
size = fstream->max_buffer_size;
} else if (fstream->corked) {
/* try to use optimal buffer size with corking */
new_size = I_MIN(fstream->optimal_block_size,
fstream->max_buffer_size);
if (new_size > size)
size = new_size;
}
if (size <= fstream->buffer_size)
return;
fstream->buffer = i_realloc(fstream->buffer,
fstream->buffer_size, size);
if (fstream->tail <= fstream->head && !IS_STREAM_EMPTY(fstream)) {
/* move head forward to end of buffer */
end_size = fstream->buffer_size - fstream->head;
memmove(fstream->buffer + size - end_size,
fstream->buffer + fstream->head, end_size);
fstream->head = size - end_size;
}
fstream->full = FALSE;
fstream->buffer_size = size;
}
static void stream_send_io(struct file_ostream *fstream)
{
struct ostream *ostream = &fstream->ostream.ostream;
int ret;
/* Set flush_pending = FALSE first before calling the flush callback,
and change it to TRUE only if callback returns 0. That way the
callback can call o_stream_set_flush_pending() again and we don't
forget it even if flush callback returns 1. */
fstream->flush_pending = FALSE;
o_stream_ref(ostream);
if (fstream->ostream.callback != NULL)
ret = fstream->ostream.callback(fstream->ostream.context);
else
ret = o_stream_file_flush(&fstream->ostream);
if (ret == 0)
fstream->flush_pending = TRUE;
if (!fstream->flush_pending && IS_STREAM_EMPTY(fstream)) {
if (fstream->io != NULL) {
/* all sent */
io_remove(&fstream->io);
}
} else if (!fstream->ostream.ostream.closed) {
/* Add the IO handler if it's not there already. Callback
might have just returned 0 without there being any data
to be sent. */
if (fstream->io == NULL) {
fstream->io = io_add(fstream->fd, IO_WRITE,
stream_send_io, fstream);
}
}
o_stream_unref(&ostream);
}
static size_t o_stream_add(struct file_ostream *fstream,
const void *data, size_t size)
{
size_t unused, sent;
int i;
unused = get_unused_space(fstream);
if (unused < size)
o_stream_grow_buffer(fstream, size-unused);
sent = 0;
for (i = 0; i < 2 && sent < size && !fstream->full; i++) {
unused = fstream->tail >= fstream->head ?
fstream->buffer_size - fstream->tail :
fstream->head - fstream->tail;
if (unused > size-sent)
unused = size-sent;
memcpy(fstream->buffer + fstream->tail,
CONST_PTR_OFFSET(data, sent), unused);
sent += unused;
fstream->tail += unused;
if (fstream->tail == fstream->buffer_size)
fstream->tail = 0;
if (fstream->head == fstream->tail)
fstream->full = TRUE;
}
if (sent != 0 && fstream->io == NULL &&
!fstream->corked && !fstream->file) {
fstream->io = io_add(fstream->fd, IO_WRITE, stream_send_io,
fstream);
}
return sent;
}
static ssize_t o_stream_file_sendv(struct ostream_private *stream,
const struct const_iovec *iov,
unsigned int iov_count)
{
struct file_ostream *fstream = (struct file_ostream *)stream;
size_t size, total_size, added, optimal_size;
unsigned int i;
ssize_t ret = 0;
for (i = 0, size = 0; i < iov_count; i++)
size += iov[i].iov_len;
total_size = size;
if (size > get_unused_space(fstream) && !IS_STREAM_EMPTY(fstream)) {
if (o_stream_file_flush(stream) < 0)
return -1;
}
optimal_size = I_MIN(fstream->optimal_block_size,
fstream->max_buffer_size);
if (IS_STREAM_EMPTY(fstream) &&
(!fstream->corked || size >= optimal_size)) {
/* send immediately */
ret = o_stream_writev(fstream, iov, iov_count);
if (ret < 0)
return -1;
size = ret;
while (size > 0 && iov_count > 0 && size >= iov[0].iov_len) {
size -= iov[0].iov_len;
iov++;
iov_count--;
}
if (iov_count == 0)
i_assert(size == 0);
else {
added = o_stream_add(fstream,
CONST_PTR_OFFSET(iov[0].iov_base, size),
iov[0].iov_len - size);
ret += added;
if (added != iov[0].iov_len - size) {
/* buffer full */
stream->ostream.offset += ret;
return ret;
}
iov++;
iov_count--;
}
}
/* buffer it, at least partly */
for (i = 0; i < iov_count; i++) {
added = o_stream_add(fstream, iov[i].iov_base, iov[i].iov_len);
ret += added;
if (added != iov[i].iov_len)
break;
}
stream->ostream.offset += ret;
i_assert((size_t)ret <= total_size);
i_assert((size_t)ret == total_size || !fstream->file);
return ret;
}
static off_t io_stream_sendfile(struct ostream_private *outstream,
struct istream *instream, int in_fd)
{
struct file_ostream *foutstream = (struct file_ostream *)outstream;
const struct stat *st;
uoff_t start_offset;
uoff_t in_size, offset, send_size, v_offset;
ssize_t ret;
st = i_stream_stat(instream, TRUE);
if (st == NULL) {
outstream->ostream.stream_errno = instream->stream_errno;
return -1;
}
in_size = st->st_size;
o_stream_socket_cork(foutstream);
/* flush out any data in buffer */
if ((ret = buffer_flush(foutstream)) <= 0)
return ret;
if (o_stream_lseek(foutstream) < 0)
return -1;
start_offset = v_offset = instream->v_offset;
do {
offset = instream->real_stream->abs_start_offset + v_offset;
send_size = in_size - v_offset;
ret = safe_sendfile(foutstream->fd, in_fd, &offset,
MAX_SSIZE_T(send_size));
if (ret <= 0) {
if (ret == 0 || errno == EINTR || errno == EAGAIN) {
ret = 0;
break;
}
outstream->ostream.stream_errno = errno;
if (errno != EINVAL) {
/* close only if error wasn't because
sendfile() isn't supported */
stream_closed(foutstream);
}
break;
}
v_offset += ret;
foutstream->real_offset += ret;
foutstream->buffer_offset += ret;
outstream->ostream.offset += ret;
} while ((uoff_t)ret != send_size);
i_stream_seek(instream, v_offset);
return ret < 0 ? -1 : (off_t)(instream->v_offset - start_offset);
}
static off_t io_stream_copy(struct ostream_private *outstream,
struct istream *instream)
{
struct file_ostream *foutstream = (struct file_ostream *)outstream;
uoff_t start_offset;
struct const_iovec iov[3];
int iov_len;
const unsigned char *data;
size_t size, skip_size;
ssize_t ret;
int pos;
iov_len = o_stream_fill_iovec(foutstream, iov);
skip_size = 0;
for (pos = 0; pos < iov_len; pos++)
skip_size += iov[pos].iov_len;
start_offset = instream->v_offset;
for (;;) {
(void)i_stream_read_data(instream, &data, &size,
foutstream->optimal_block_size-1);
if (size == 0) {
/* all sent */
break;
}
pos = iov_len++;
iov[pos].iov_base = (void *) data;
iov[pos].iov_len = size;
ret = o_stream_writev(foutstream, iov, iov_len);
if (ret < 0)
return -1;
if (skip_size > 0) {
if ((size_t)ret < skip_size) {
update_buffer(foutstream, ret);
skip_size -= ret;
ret = 0;
} else {
update_buffer(foutstream, skip_size);
ret -= skip_size;
skip_size = 0;
}
}
outstream->ostream.offset += ret;
i_stream_skip(instream, ret);
if ((size_t)ret != iov[pos].iov_len)
break;
i_assert(skip_size == 0);
iov_len = 0;
}
return (off_t) (instream->v_offset - start_offset);
}
static off_t io_stream_copy_backwards(struct ostream_private *outstream,
struct istream *instream, uoff_t in_size)
{
struct file_ostream *foutstream = (struct file_ostream *)outstream;
uoff_t in_start_offset, in_offset, in_limit, out_offset;
const unsigned char *data;
size_t buffer_size, size, read_size;
ssize_t ret;
i_assert(IS_STREAM_EMPTY(foutstream));
/* figure out optimal buffer size */
buffer_size = instream->real_stream->buffer_size;
if (buffer_size == 0 || buffer_size > foutstream->buffer_size) {
if (foutstream->optimal_block_size > foutstream->buffer_size) {
o_stream_grow_buffer(foutstream,
foutstream->optimal_block_size -
foutstream->buffer_size);
}
buffer_size = foutstream->buffer_size;
}
in_start_offset = instream->v_offset;
in_offset = in_limit = in_size;
out_offset = outstream->ostream.offset + (in_offset - in_start_offset);
while (in_offset > in_start_offset) {
if (in_offset - in_start_offset <= buffer_size)
read_size = in_offset - in_start_offset;
else
read_size = buffer_size;
in_offset -= read_size;
out_offset -= read_size;
for (;;) {
i_assert(in_offset <= in_limit);
i_stream_seek(instream, in_offset);
read_size = in_limit - in_offset;
(void)i_stream_read_data(instream, &data, &size,
read_size-1);
if (size >= read_size) {
size = read_size;
if (instream->mmaped) {
/* we'll have to write it through
buffer or the file gets corrupted */
i_assert(size <=
foutstream->buffer_size);
memcpy(foutstream->buffer, data, size);
data = foutstream->buffer;
}
break;
}
/* buffer too large probably, try with smaller */
read_size -= size;
in_offset += read_size;
out_offset += read_size;
buffer_size -= read_size;
}
in_limit -= size;
ret = pwrite_full(foutstream->fd, data, size, out_offset);
if (ret < 0) {
/* error */
outstream->ostream.stream_errno = errno;
return -1;
}
i_stream_skip(instream, size);
}
outstream->ostream.offset += in_size - in_start_offset;
return (off_t) (in_size - in_start_offset);
}
static off_t o_stream_file_send_istream(struct ostream_private *outstream,
struct istream *instream)
{
struct file_ostream *foutstream = (struct file_ostream *)outstream;
const struct stat *st;
off_t in_abs_offset, ret;
int in_fd;
in_fd = i_stream_get_fd(instream);
if (in_fd == foutstream->fd) {
/* copying data within same fd. we'll have to be careful with
seeks and overlapping writes. */
st = i_stream_stat(instream, TRUE);
if (st == NULL) {
outstream->ostream.stream_errno = instream->stream_errno;
return -1;
}
i_assert(instream->v_offset <= (uoff_t)st->st_size);
in_abs_offset = instream->real_stream->abs_start_offset +
instream->v_offset;
ret = (off_t)outstream->ostream.offset - in_abs_offset;
if (ret == 0) {
/* copying data over itself. we don't really
need to do that, just fake it. */
return st->st_size - instream->v_offset;
}
if (ret > 0 && st->st_size > ret) {
/* overlapping */
i_assert(instream->seekable);
return io_stream_copy_backwards(outstream, instream,
st->st_size);
}
}
if (!foutstream->no_sendfile && in_fd != -1 && instream->seekable) {
ret = io_stream_sendfile(outstream, instream, in_fd);
if (ret >= 0 || outstream->ostream.stream_errno != EINVAL)
return ret;
/* sendfile() not supported (with this fd), fallback to
regular sending. */
outstream->ostream.stream_errno = 0;
foutstream->no_sendfile = TRUE;
}
return io_stream_copy(outstream, instream);
}
static struct file_ostream *
o_stream_create_fd_common(int fd, bool autoclose_fd)
{
struct file_ostream *fstream;
fstream = i_new(struct file_ostream, 1);
fstream->fd = fd;
fstream->autoclose_fd = autoclose_fd;
fstream->optimal_block_size = DEFAULT_OPTIMAL_BLOCK_SIZE;
fstream->ostream.iostream.close = o_stream_file_close;
fstream->ostream.iostream.destroy = o_stream_file_destroy;
fstream->ostream.iostream.set_max_buffer_size =
o_stream_file_set_max_buffer_size;
fstream->ostream.cork = o_stream_file_cork;
fstream->ostream.flush = o_stream_file_flush;
fstream->ostream.flush_pending = o_stream_file_flush_pending;
fstream->ostream.get_used_size = o_stream_file_get_used_size;
fstream->ostream.seek = o_stream_file_seek;
fstream->ostream.sendv = o_stream_file_sendv;
fstream->ostream.send_istream = o_stream_file_send_istream;
return fstream;
}
static void fstream_init_file(struct file_ostream *fstream)
{
struct stat st;
fstream->no_sendfile = TRUE;
if (fstat(fstream->fd, &st) < 0)
return;
if ((uoff_t)st.st_blksize > fstream->optimal_block_size) {
/* use the optimal block size, but with a reasonable limit */
fstream->optimal_block_size =
I_MIN(st.st_blksize, MAX_OPTIMAL_BLOCK_SIZE);
}
if (S_ISREG(st.st_mode)) {
fstream->no_socket_cork = TRUE;
fstream->file = TRUE;
}
}
struct ostream *
o_stream_create_fd(int fd, size_t max_buffer_size, bool autoclose_fd)
{
struct file_ostream *fstream;
struct ostream *ostream;
off_t offset;
fstream = o_stream_create_fd_common(fd, autoclose_fd);
fstream->max_buffer_size = max_buffer_size;
ostream = o_stream_create(&fstream->ostream);
offset = lseek(fd, 0, SEEK_CUR);
if (offset >= 0) {
ostream->offset = offset;
fstream->real_offset = offset;
fstream->buffer_offset = offset;
fstream_init_file(fstream);
} else {
if (net_getsockname(fd, NULL, NULL) < 0) {
fstream->no_sendfile = TRUE;
fstream->no_socket_cork = TRUE;
}
}
if (max_buffer_size == 0)
fstream->max_buffer_size = fstream->optimal_block_size;
return ostream;
}
struct ostream *
o_stream_create_fd_file(int fd, uoff_t offset, bool autoclose_fd)
{
struct file_ostream *fstream;
struct ostream *ostream;
if (offset == (uoff_t)-1)
offset = lseek(fd, 0, SEEK_CUR);
fstream = o_stream_create_fd_common(fd, autoclose_fd);
fstream_init_file(fstream);
fstream->max_buffer_size = fstream->optimal_block_size;
fstream->real_offset = offset;
fstream->buffer_offset = offset;
ostream = o_stream_create(&fstream->ostream);
ostream->offset = offset;
return ostream;
}