Merge lp:~akopytov/percona-xtrabackup/bug1095249-2.0 into lp:percona-xtrabackup/2.0

[Alexey Kopytov]

[George Ormond Lorch III (gl-az)]

I see now why you re-wrote the ds chaining in 2.1. I don't like that the compression sink needs to manage it's connection to the next sink (the buffer in this case) rather than having it done at the higher level in xtrabackup.c like it is now done in 2.1.

Anyway, looks like a reasonable fix to the issue.

review: Approve (g2)

[Alexey Kopytov (akopytov)]

http://jenkins.percona.com/view/XtraBackup/job/percona-xtrabackup-2.0-param/319/ (though compression tests are not currently executed in Jenkins anyway).

[Laurynas Biveinis (laurynas-biveinis)]

The commit message includes the title of a wrong bug.
OK with it fixed.

review: Approve

[Alexey Kopytov (akopytov)]

Hi Laurynas,

On Sun, 06 Jan 2013 11:38:21 -0000, Laurynas Biveinis wrote:
> Review: Approve
>
> The commit message includes the title of a wrong bug.
> OK with it fixed.
>

Fixed.

=== modified file 'src/Makefile'
--- src/Makefile	2012-02-10 20:05:56 +0000
+++ src/Makefile	2013-01-07 06:22:20 +0000
@@ -23,7 +23,8 @@
 BIN_DIR=$(PREFIX)/bin
 
 COMMON_INC = -I. -I libarchive/libarchive -I quicklz
-XTRABACKUPOBJS = xtrabackup.o stream.o local.o compress.o xbstream_write.o \
+XTRABACKUPOBJS = xtrabackup.o stream.o local.o compress.o buffer.o \
+	xbstream_write.o \
 	quicklz/quicklz.o
 XBSTREAMOBJS = xbstream.o xbstream_write.o xbstream_read.o
 
 
=== added file 'src/buffer.c'
--- src/buffer.c	1970-01-01 00:00:00 +0000
+++ src/buffer.c	2013-01-07 06:22:20 +0000
@@ -0,0 +1,197 @@
+/******************************************************
+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 <my_base.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));
+	ctxt->datasink = &datasink_buffer;
+
+	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 = pipe_ctxt->datasink->open(pipe_ctxt, path, mystat);
+	if (dst_file == NULL) {
+		exit(EXIT_FAILURE);
+	}
+
+	dst_file->datasink = pipe_ctxt->datasink;
+
+	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;
+	datasink_t		*dest_ds;
+
+	buffer_file = (ds_buffer_file_t *) file->ptr;
+
+	dest_ds = buffer_file->dst_file->datasink;
+
+	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 (dest_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 (dest_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;
+	datasink_t		*dest_ds;
+
+	buffer_file = (ds_buffer_file_t *) file->ptr;
+
+	dest_ds = buffer_file->dst_file->datasink;
+
+	if (buffer_file->pos > 0) {
+		dest_ds->write(buffer_file->dst_file, buffer_file->buf,
+			       buffer_file->pos);
+	}
+
+	dest_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);
+}
 
=== added file 'src/buffer.h'
--- src/buffer.h	1970-01-01 00:00:00 +0000
+++ src/buffer.h	2013-01-07 06:22:20 +0000
@@ -0,0 +1,31 @@
+/******************************************************
+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
+
+*******************************************************/
+
+#ifndef DS_BUFFER_H
+#define DS_BUFFER_H
+
+#include "datasink.h"
+
+extern datasink_t datasink_buffer;
+
+/* Change the default buffer size */
+void ds_buffer_set_size(ds_ctxt_t *ctxt, size_t size);
+
+#endif
 
=== modified file 'src/compress.c'
--- src/compress.c	2012-06-14 10:50:28 +0000
+++ src/compress.c	2013-01-07 06:22:20 +0000
@@ -27,6 +27,7 @@
 #include "datasink.h"
 #include "stream.h"
 #include "local.h"
+#include "buffer.h"
 
 #define COMPRESS_CHUNK_SIZE (64 * 1024UL)
 #define MY_QLZ_COMPRESS_OVERHEAD 400
@@ -97,15 +98,26 @@
 	ds_ctxt_t		*ctxt;
 	ds_compress_ctxt_t	*compress_ctxt;
 	datasink_t		*dest_ds;
+	datasink_t		*pipe_ds;
 	ds_ctxt_t		*dest_ctxt;
 	comp_thread_ctxt_t	*threads;
 
+	dest_ds = &datasink_buffer;
+
+	dest_ctxt = dest_ds->init(root);
+	if (dest_ctxt == NULL) {
+		msg("compress: failed to initialize the buffer datasink.\n");
+		return NULL;
+	}
+
+	/* Use a 1 MB buffer for compressed output stream */
+	ds_buffer_set_size(dest_ctxt, 1024 * 1024);
+
 	/* Decide whether the compressed data will be stored in local files or
 	streamed to an archive */
-	dest_ds = xtrabackup_stream ? &datasink_stream : &datasink_local;
-
-	dest_ctxt = dest_ds->init(root);
-	if (dest_ctxt == NULL) {
+	pipe_ds = xtrabackup_stream ? &datasink_stream : &datasink_local;
+	dest_ctxt->pipe_ctxt = pipe_ds->init(root);
+	if (dest_ctxt->pipe_ctxt == NULL) {
 		msg("compress: failed to initialize the target datasink.\n");
 		return NULL;
 	}
@@ -323,7 +335,9 @@
 {
 	ds_compress_ctxt_t 	*comp_ctxt;
 	ds_ctxt_t		*dest_ctxt;
+	ds_ctxt_t		*pipe_ctxt;
 	datasink_t		*dest_ds;
+	datasink_t		*pipe_ds;
 
 	comp_ctxt = (ds_compress_ctxt_t *) ctxt->ptr;;
 
@@ -332,7 +346,11 @@
 	dest_ctxt = comp_ctxt->dest_ctxt;
 	dest_ds = dest_ctxt->datasink;
 
+	pipe_ctxt = dest_ctxt->pipe_ctxt;
+	pipe_ds = pipe_ctxt->datasink;
+
 	dest_ds->deinit(dest_ctxt);
+	pipe_ds->deinit(pipe_ctxt);
 
 	MY_FREE(ctxt);
 }
 
=== modified file 'src/datasink.h'
--- src/datasink.h	2012-02-10 20:05:56 +0000
+++ src/datasink.h	2013-01-07 06:22:20 +0000
@@ -26,15 +26,17 @@
 
 struct datasink_struct;
 
-typedef struct {
+typedef struct ds_ctxt {
 	struct datasink_struct	*datasink;
 	char 			*root;
 	void			*ptr;
+	struct ds_ctxt		*pipe_ctxt;
 } ds_ctxt_t;
 
 typedef struct {
-	void *ptr;
-	char *path;
+	void 			*ptr;
+	char 			*path;
+	struct datasink_struct	*datasink;
 } ds_file_t;
 
 typedef struct datasink_struct {
 
=== added file 'test/t/ib_compress_basic.sh'
--- test/t/ib_compress_basic.sh	1970-01-01 00:00:00 +0000
+++ test/t/ib_compress_basic.sh	2013-01-07 06:22:20 +0000
@@ -0,0 +1,32 @@
+########################################################################
+# Basic test for local compressed backups
+########################################################################
+
+. inc/common.sh
+
+if ! which qpress > /dev/null 2>&1 ; then
+  echo "Requires qpress to be installed" > $SKIPPED_REASON
+  exit $SKIPPED_EXIT_CODE
+fi
+
+start_server --innodb_file_per_table
+load_sakila
+
+innobackupex --compress --no-timestamp $topdir/backup
+
+stop_server
+rm -rf ${MYSQLD_DATADIR}/*
+
+cd $topdir/backup
+
+for i in *.qp;  do qpress -d $i ./; done; \
+for i in sakila/*.qp; do qpress -d $i sakila/; done
+
+innobackupex --apply-log $topdir/backup
+
+innobackupex --copy-back $topdir/backup
+
+start_server
+
+# TODO: proper validation
+run_cmd ${MYSQL} ${MYSQL_ARGS} -e "SELECT count(*) from actor" sakila