~akopytov/percona-xtrabackup/bug1210266-2.1

/******************************************************

Copyright (c) 2012 Percona Ireland Ltd.

buffer datasink for XtraBackup.

This program is free software; you can redistribute it and/or modify

it under the terms of the GNU General Public License as published by

the Free Software Foundation; version 2 of the License.

This program is distributed in the hope that it will be useful,

but WITHOUT ANY WARRANTY; without even the implied warranty of

MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

GNU General Public License for more details.

You should have received a copy of the GNU General Public License

along with this program; if not, write to the Free Software

Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

*******************************************************/

/* Does buffered output to a destination datasink set with ds_set_pipe().

Writes to the destination datasink are guaranteed to not be smaller than a

specified buffer size (DS_DEFAULT_BUFFER_SIZE by default), with the only

exception for the last write for a file. */

#include <mysql_version.h>

#include <my_base.h>

#include "ds_buffer.h"

#include "common.h"

#include "datasink.h"

#define DS_DEFAULT_BUFFER_SIZE (64 * 1024)

typedef struct {

	ds_file_t	*dst_file;

	char		*buf;

	size_t		 pos;

	size_t		size;

} ds_buffer_file_t;

typedef struct {

	size_t	buffer_size;

} ds_buffer_ctxt_t;

static ds_ctxt_t *buffer_init(const char *root);

static ds_file_t *buffer_open(ds_ctxt_t *ctxt, const char *path,

			      MY_STAT *mystat);

static int buffer_write(ds_file_t *file, const void *buf, size_t len);

static int buffer_close(ds_file_t *file);

static void buffer_deinit(ds_ctxt_t *ctxt);

datasink_t datasink_buffer = {

	&buffer_init,

	&buffer_open,

	&buffer_write,

	&buffer_close,

	&buffer_deinit

};

/* Change the default buffer size */

void ds_buffer_set_size(ds_ctxt_t *ctxt, size_t size)

{

	ds_buffer_ctxt_t *buffer_ctxt = (ds_buffer_ctxt_t *) ctxt->ptr;

	buffer_ctxt->buffer_size = size;

}

static ds_ctxt_t *

buffer_init(const char *root)

{

	ds_ctxt_t		*ctxt;

	ds_buffer_ctxt_t	*buffer_ctxt;

	ctxt = my_malloc(sizeof(ds_ctxt_t) + sizeof(ds_buffer_ctxt_t),

			 MYF(MY_FAE));

	buffer_ctxt = (ds_buffer_ctxt_t *) (ctxt + 1);

	buffer_ctxt->buffer_size = DS_DEFAULT_BUFFER_SIZE;

	ctxt->ptr = buffer_ctxt;

	ctxt->root = my_strdup(root, MYF(MY_FAE));

	return ctxt;

}

static ds_file_t *

buffer_open(ds_ctxt_t *ctxt, const char *path, MY_STAT *mystat)

{

	ds_buffer_ctxt_t	*buffer_ctxt;

	ds_ctxt_t		*pipe_ctxt;

	ds_file_t		*dst_file;

	ds_file_t		*file;

	ds_buffer_file_t	*buffer_file;

	pipe_ctxt = ctxt->pipe_ctxt;

	xb_a(pipe_ctxt != NULL);

	dst_file = ds_open(pipe_ctxt, path, mystat);

	if (dst_file == NULL) {

		exit(EXIT_FAILURE);

	}

	buffer_ctxt = (ds_buffer_ctxt_t *) ctxt->ptr;

	file = (ds_file_t *) my_malloc(sizeof(ds_file_t) +

				       sizeof(ds_buffer_file_t) +

				       buffer_ctxt->buffer_size,

				       MYF(MY_FAE));

	buffer_file = (ds_buffer_file_t *) (file + 1);

	buffer_file->dst_file = dst_file;

	buffer_file->buf = (char *) (buffer_file + 1);

	buffer_file->size = buffer_ctxt->buffer_size;

	buffer_file->pos = 0;

	file->path = dst_file->path;

	file->ptr = buffer_file;

	return file;

}

static int

buffer_write(ds_file_t *file, const void *buf, size_t len)

{

	ds_buffer_file_t	*buffer_file;

	buffer_file = (ds_buffer_file_t *) file->ptr;

	while (len > 0) {

		if (buffer_file->pos + len > buffer_file->size) {

			if (buffer_file->pos > 0) {

				size_t bytes;

				bytes = buffer_file->size - buffer_file->pos;

				memcpy(buffer_file->buf + buffer_file->pos, buf,

				       bytes);

				if (ds_write(buffer_file->dst_file,

					     buffer_file->buf,

					     buffer_file->size)) {

					return 1;

				}

				buffer_file->pos = 0;

				buf = (const char *) buf + bytes;

				len -= bytes;

			} else {

				/* We don't have any buffered bytes, just write

				the entire source buffer */

				if (ds_write(buffer_file->dst_file, buf, len)) {

					return 1;

				}

				break;

			}

		} else {

			memcpy(buffer_file->buf + buffer_file->pos, buf, len);

			buffer_file->pos += len;

			break;

		}

	}

	return 0;

}

static int

buffer_close(ds_file_t *file)

{

	ds_buffer_file_t	*buffer_file;

	buffer_file = (ds_buffer_file_t *) file->ptr;

	if (buffer_file->pos > 0) {

		ds_write(buffer_file->dst_file, buffer_file->buf,

			 buffer_file->pos);

	}

	ds_close(buffer_file->dst_file);

	MY_FREE(file);

	return 0;

}

static void

buffer_deinit(ds_ctxt_t *ctxt)

{

	MY_FREE(ctxt->root);

	MY_FREE(ctxt);

}