Merge lp:~sergei.glushchenko/percona-xtrabackup/20-decompress into lp:percona-xtrabackup/2.0

[Sergei Glushchenko]

BP https://blueprints.launchpad.net/percona-xtrabackup/+spec/xbstream-decompress
  Support for decompression in xbstream. New option -z been added to
  automatically decompress .qp files inside the stream. decompress datasink
  been implemented for qpress10 single file archives extraction.

#24660

[Sergei Glushchenko (sergei.glushchenko)]

http://jenkins.percona.com/view/XtraBackup/job/percona-xtrabackup-2.0-param/243/

[Stewart Smith (stewart)]

I have three thoughts:
- Should we add this to 2.0 at all? It is adding a feature and a 2.1 release probably isn't far off.
- I think that certainly in 2.1, this should be default behaviour, as it's almost certainly what the user wanted.
- parallel decompression?

[Sergei Glushchenko (sergei.glushchenko)]

> I have three thoughts:
> - Should we add this to 2.0 at all? It is adding a feature and a 2.1 release
> probably isn't far off.

We have customer using 2.0, who want this feature.

> - I think that certainly in 2.1, this should be default behaviour, as it's
> almost certainly what the user wanted.

Yes, probably you're right.

> - parallel decompression?

I've experimented with parallel decompression, implementing it similar to parallel compression and it didn't gave me any performance gain. It was even slower than thread-per-file implementation. Do you have any specific proposals how to implement parallel decompression?

[Stewart Smith (stewart)]

> > - parallel decompression?
>
> I've experimented with parallel decompression, implementing it similar to
> parallel compression and it didn't gave me any performance gain. It was even
> slower than thread-per-file implementation. Do you have any specific proposals
> how to implement parallel decompression?

Thread per file is fine.

[Alexey Kopytov (akopytov)]

As discussed on IRC, the feature will be implemented by adding a --decompress-program option to xbstream so that an external utility such as qpress can be used.

[Alexey Kopytov (akopytov)]

After further analysis and discussions it was decided that this feature does not really have real applications, while complicating the xbstream architecture significantly.

Users willing to utilize compression only to speed up transferring the entire archive to another host do not really need built-in compression in XtraBackup. That feature was implemented for different use cases. Instead, users can use normal uncompressed stream in XtraBackup (either tar, or xbstream in case of parallel or incremental backups) and then use external compression utilities (either single- or multi-threaded) to compress the stream and decompress it on remote host.

review: Disapprove

Unmerged revisions

462. By Sergei Glushchenko on 2012-08-19

BP https://blueprints.launchpad.net/percona-xtrabackup/+spec/xbstream-decompress
Support for decompression in xbstream. New option -z been added to
automatically decompress .qp files inside the stream. decompress datasink
been implemented for qpress10 one-file archives extraction.

=== modified file 'src/Makefile'
--- src/Makefile	2012-02-10 20:05:56 +0000
+++ src/Makefile	2012-08-19 12:20:47 +0000
@@ -24,7 +24,7 @@
 
 COMMON_INC = -I. -I libarchive/libarchive -I quicklz
 XTRABACKUPOBJS = xtrabackup.o stream.o local.o compress.o xbstream_write.o \
-	quicklz/quicklz.o
+	quicklz/quicklz.o decompress.o
 XBSTREAMOBJS = xbstream.o xbstream_write.o xbstream_read.o
 
 LIBARCHIVE_A = libarchive/libarchive/libarchive.a
@@ -176,7 +176,7 @@
 xbstream.o xbstream_read.o: %.o: %.c
 	$(CC) $(CFLAGS) $(INC) $(DEFS) -c $< -o $@
 
-xbstream: $(XBSTREAMOBJS) $(MYSQLOBJS) local.o
+xbstream: $(XBSTREAMOBJS) $(MYSQLOBJS) local.o decompress.o quicklz/quicklz.o
 	$(CC) $(CFLAGS) $^ $(INC) $(MYSQLOBJS) $(LIBS) -o $@
 
 xtrabackup.o: xtrabackup.c xb_regex.h
 
=== modified file 'src/datasink.h'
--- src/datasink.h	2012-02-10 20:05:56 +0000
+++ src/datasink.h	2012-08-19 12:20:47 +0000
@@ -30,6 +30,7 @@
 	struct datasink_struct	*datasink;
 	char 			*root;
 	void			*ptr;
+	struct ds_ctxt	*pipe_ctxt;
 } ds_ctxt_t;
 
 typedef struct {
 
=== added file 'src/decompress.c'
--- src/decompress.c	1970-01-01 00:00:00 +0000
+++ src/decompress.c	2012-08-19 12:20:47 +0000
@@ -0,0 +1,453 @@
+/******************************************************
+Copyright (c) 2011 Percona Inc.
+
+Compressing datasink implementation 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
+
+*******************************************************/
+
+
+#include <my_base.h>
+#include <quicklz.h>
+#include <univ.i>
+#include <zlib.h>
+#include "common.h"
+#include "datasink.h"
+
+#define COMPRESS_CHUNK_SIZE (64 * 1024UL)
+#define MY_QLZ_COMPRESS_OVERHEAD 400
+
+#define DS_DECOMPRESS_BUF_SIZE (COMPRESS_CHUNK_SIZE * 10UL)
+
+typedef struct {
+	pthread_t		id;
+	pthread_mutex_t		data_mutex;
+	pthread_cond_t		not_full_cond;
+	pthread_cond_t		not_empty_cond;
+	char			data[DS_DECOMPRESS_BUF_SIZE];
+	char*			rptr;
+	char*			wptr;
+	size_t			bytes_avail;
+	size_t			data_size;
+	my_bool			eof;
+	my_bool			error;
+	ds_ctxt_t		*ds;
+} ds_decomp_ctxt_t;
+
+static
+void *
+deqpress_worker_thread_func(void *arg);
+
+static ds_ctxt_t *decomp_init(const char *root);
+static ds_file_t *decomp_open(ds_ctxt_t *ctxt, const char *path,
+				MY_STAT *mystat);
+static int decomp_write(ds_file_t *file, const void *buf, size_t len);
+static int decomp_close(ds_file_t *file);
+static void decomp_deinit(ds_ctxt_t *ctxt);
+
+datasink_t datasink_decompress = {
+	&decomp_init,
+	&decomp_open,
+	&decomp_write,
+	&decomp_close,
+	&decomp_deinit
+};
+
+static
+ds_ctxt_t *
+decomp_init(const char *root)
+{
+	ds_ctxt_t		*ctxt;
+
+	ctxt = (ds_ctxt_t *) my_malloc(sizeof(ds_ctxt_t),
+				       MYF(MY_FAE));
+
+	ctxt->ptr = NULL;
+	ctxt->root = my_strdup(root, MYF(MY_FAE));
+	ctxt->datasink = &datasink_decompress;
+
+	return ctxt;
+}
+
+static
+void
+decomp_deinit(ds_ctxt_t *ctxt)
+{
+	MY_FREE(ctxt->root);
+	MY_FREE(ctxt);
+}
+
+static
+ds_file_t *
+decomp_open(ds_ctxt_t *ctxt, const char *path,
+	    MY_STAT *mystat __attribute__((__unused__)))
+{
+	ds_decomp_ctxt_t	*decomp_ctxt;
+	ds_file_t		*file;
+
+	xb_a(ctxt->pipe_ctxt != NULL);
+
+	file = (ds_file_t *) my_malloc(sizeof(ds_file_t) +
+				       sizeof(ds_decomp_ctxt_t),
+				       MYF(MY_FAE));
+
+	decomp_ctxt = (ds_decomp_ctxt_t *) (file + 1);
+	file->ptr = decomp_ctxt;
+	file->path = my_strdup(path, MYF(MY_FAE));
+
+	decomp_ctxt->rptr =
+			decomp_ctxt->wptr = decomp_ctxt->data;
+	decomp_ctxt->eof = decomp_ctxt->error = FALSE;
+	decomp_ctxt->bytes_avail = 0;
+	decomp_ctxt->data_size = DS_DECOMPRESS_BUF_SIZE;
+	decomp_ctxt->ds = (ds_ctxt_t *) ctxt->pipe_ctxt;
+
+	pthread_mutex_init(&decomp_ctxt->data_mutex, NULL);
+	pthread_cond_init(&decomp_ctxt->not_full_cond, NULL);
+	pthread_cond_init(&decomp_ctxt->not_empty_cond, NULL);
+
+	if (pthread_create(&decomp_ctxt->id, NULL,
+			   deqpress_worker_thread_func, file)) {
+		msg("compress: pthread_create() failed: "
+		    "errno = %d\n", errno);
+		return NULL;
+	}
+
+	return file;
+}
+
+static
+int
+decomp_write(ds_file_t *file, const void *buf, size_t len)
+{
+	ds_decomp_ctxt_t 	*decomp_ctxt;
+	const char		*ptr = (const char*) buf;
+
+	decomp_ctxt = (ds_decomp_ctxt_t *) file->ptr;
+
+	while (len > 0) {
+
+		size_t n;
+
+		pthread_mutex_lock(&decomp_ctxt->data_mutex);
+
+		while ((decomp_ctxt->bytes_avail
+			== decomp_ctxt->data_size) &&
+		       !decomp_ctxt->error) {
+			pthread_cond_wait(
+				&decomp_ctxt->not_full_cond,
+				&decomp_ctxt->data_mutex);
+		}
+
+		if (decomp_ctxt->error) {
+			pthread_mutex_unlock(&decomp_ctxt->data_mutex);
+			return 1;
+		}
+
+		n = len < (decomp_ctxt->data_size - decomp_ctxt->bytes_avail) ?
+		    len : (decomp_ctxt->data_size - decomp_ctxt->bytes_avail);
+
+		if (decomp_ctxt->wptr + n >
+		    decomp_ctxt->data + decomp_ctxt->data_size) {
+			n = decomp_ctxt->data + decomp_ctxt->data_size -
+			    decomp_ctxt->wptr;
+		}
+
+		memcpy(decomp_ctxt->wptr, ptr, n);
+
+		len -= n;
+		ptr += n;
+		decomp_ctxt->wptr += n;
+		decomp_ctxt->bytes_avail += n;
+
+		if (decomp_ctxt->wptr ==
+		    decomp_ctxt->data + decomp_ctxt->data_size) {
+			decomp_ctxt->wptr = decomp_ctxt->data;
+		}
+
+		if (decomp_ctxt->bytes_avail > decomp_ctxt->data_size / 5)
+			pthread_cond_signal(&decomp_ctxt->not_empty_cond);
+
+		pthread_mutex_unlock(&decomp_ctxt->data_mutex);
+
+	}
+
+	return 0;
+}
+
+static
+my_bool
+decomp_read(ds_file_t *file, const void *buf, size_t len)
+{
+	ds_decomp_ctxt_t 	*decomp_ctxt;
+	char			*ptr = (char*) buf;
+
+	decomp_ctxt = (ds_decomp_ctxt_t *) file->ptr;
+
+	while (len > 0) {
+
+		size_t n;
+
+		pthread_mutex_lock(&decomp_ctxt->data_mutex);
+
+		while (decomp_ctxt->bytes_avail == 0 &&
+			!decomp_ctxt->eof) {
+			pthread_cond_wait(
+				&decomp_ctxt->not_empty_cond,
+				&decomp_ctxt->data_mutex);
+		}
+
+		if (decomp_ctxt->bytes_avail == 0 &&
+			decomp_ctxt->eof) {
+			pthread_mutex_unlock(&decomp_ctxt->data_mutex);
+			return FALSE;
+		}
+
+		n = len < decomp_ctxt->bytes_avail ?
+			 len : decomp_ctxt->bytes_avail;
+
+
+		if (decomp_ctxt->rptr + n >
+		    decomp_ctxt->data + decomp_ctxt->data_size) {
+			n = decomp_ctxt->data + decomp_ctxt->data_size -
+			    decomp_ctxt->rptr;
+		}
+
+		if (ptr) {
+			memcpy(ptr, decomp_ctxt->rptr, n);
+			ptr += n;
+		}
+
+		len -= n;
+		decomp_ctxt->rptr += n;
+		decomp_ctxt->bytes_avail -= n;
+
+		if (decomp_ctxt->rptr ==
+		    decomp_ctxt->data + decomp_ctxt->data_size) {
+			decomp_ctxt->rptr = decomp_ctxt->data;
+		}
+
+		pthread_cond_signal(&decomp_ctxt->not_full_cond);
+
+		pthread_mutex_unlock(&decomp_ctxt->data_mutex);
+
+	}
+
+	return TRUE;
+}
+
+static
+my_bool
+decomp_is_eof(ds_file_t *file)
+{
+	return ((ds_decomp_ctxt_t *) file->ptr)->eof;
+}
+
+static
+int
+decomp_close(ds_file_t *file)
+{
+	ds_decomp_ctxt_t 	*decomp_ctxt;
+
+	decomp_ctxt = (ds_decomp_ctxt_t *) file->ptr;
+
+	pthread_mutex_lock(&decomp_ctxt->data_mutex);
+
+	decomp_ctxt->eof = TRUE;
+
+	pthread_cond_broadcast(&decomp_ctxt->not_empty_cond);
+
+	pthread_mutex_unlock(&decomp_ctxt->data_mutex);
+
+	pthread_join(decomp_ctxt->id, NULL);
+
+	pthread_mutex_destroy(&decomp_ctxt->data_mutex);
+	pthread_cond_destroy(&decomp_ctxt->not_full_cond);
+	pthread_cond_destroy(&decomp_ctxt->not_empty_cond);
+
+	MY_FREE(file->path);
+	MY_FREE(file);
+
+	return 0;
+}
+
+static
+void
+decomp_set_error(ds_file_t *file)
+{
+	ds_decomp_ctxt_t 	*decomp_ctxt;
+
+	decomp_ctxt = (ds_decomp_ctxt_t *) file->ptr;
+
+	pthread_mutex_lock(&decomp_ctxt->data_mutex);
+
+	decomp_ctxt->error = TRUE;
+
+	pthread_cond_broadcast(&decomp_ctxt->not_full_cond);
+
+	pthread_mutex_unlock(&decomp_ctxt->data_mutex);
+}
+
+#define DECOMP_READ(file, buf, len) \
+	if (!decomp_read(file, buf, len)) { \
+		msg("decompress: error read %s\n", file->path); \
+		goto error; \
+	}
+
+
+static
+void *
+deqpress_worker_thread_func(void *arg)
+{
+	ds_file_t *file = (ds_file_t *) arg;
+	ds_ctxt_t *ds = ((ds_decomp_ctxt_t *) file->ptr)->ds;
+	char *comp_chunk = NULL, *decomp_chunk = NULL;
+	char tmp[10];
+	ds_file_t *arcfile = NULL;
+	ulong filename_len;
+	char filename[FN_REFLEN];
+	char dirname[FN_REFLEN];
+	size_t		dirpath_len;
+
+	my_thread_init();
+
+	DECOMP_READ(file, tmp, 8);
+
+	if (strncmp(tmp, "qpress10", 8) != 0) {
+		msg("decompress: source was not compressed with qpress %s\n",
+		    file->path);
+		goto error;
+	}
+
+	DECOMP_READ(file, tmp, 8);
+
+	ulonglong compress_chunk_size = uint8korr(tmp);
+	if (compress_chunk_size > 512*1024*1024) {
+		msg("decompress: source is corrupted: %s\n", file->path);
+		goto error;
+	}
+
+	DECOMP_READ(file, tmp, 1);
+	if (tmp[0] != 'F') {
+		msg("decompress: %s: xbtream can decompress only single-file "
+		    "qpress archives for now\n", file->path);
+		goto error;
+	}
+
+	DECOMP_READ(file, tmp, 4);
+
+	filename_len = uint4korr(tmp);
+
+	dirname_part(dirname, file->path, &dirpath_len);
+	char *filename_short = my_malloc(filename_len + 1, MYF(MY_FAE));
+	DECOMP_READ(file, filename_short, filename_len + 1);
+	snprintf(filename, FN_REFLEN,
+		 "%s/%s/%s", ds->root, dirname, filename_short);
+	MY_FREE(filename_short);
+
+	MY_STAT mystat;
+	arcfile = ds->datasink->open(ds, filename, &mystat);
+	if (arcfile == NULL) {
+		msg("decompress: failed to create file %s.\n", filename);
+		goto error;
+	}
+
+	comp_chunk = my_malloc((compress_chunk_size +
+				MY_QLZ_COMPRESS_OVERHEAD) * 2, MYF(MY_FAE));
+	decomp_chunk = comp_chunk + compress_chunk_size + MY_QLZ_COMPRESS_OVERHEAD;
+	qlz_state_decompress state;
+
+	while (1) {
+		DECOMP_READ(file, tmp, 8);
+
+		if (strncmp(tmp, "NEWBNEWB", 8) == 0) {
+			DECOMP_READ(file, tmp, 8);
+			DECOMP_READ(file, tmp, 4);
+
+			ulong adler = uint4korr(tmp);
+
+			DECOMP_READ(file, comp_chunk, 9);
+
+			size_t size_compressed =
+				qlz_size_compressed(comp_chunk);
+			if (size_compressed > compress_chunk_size +
+			    MY_QLZ_COMPRESS_OVERHEAD) {
+				msg("decompress: chunk is too large %s\n",
+				     file->path);
+				goto error;
+			}
+
+			DECOMP_READ(file, comp_chunk + 9, size_compressed - 9);
+
+			ulint adler_read = adler32(0x00000001, comp_chunk,
+							size_compressed);
+			size_t size_decompressed = qlz_decompress(comp_chunk,
+				decomp_chunk, &state);
+
+			if (size_decompressed > COMPRESS_CHUNK_SIZE
+				+ MY_QLZ_COMPRESS_OVERHEAD) {
+				msg("decompress: buffer overflow\n");
+				goto error;
+			}
+
+			if (adler != adler_read) {
+				msg("decompress: corrupted %s, "
+					"checksum mismatch\n",  
+					file->path);
+				goto error;
+			}
+
+			ds->datasink->write(arcfile, decomp_chunk,
+				size_decompressed);
+
+			continue;
+		}
+
+		if (strncmp(tmp, "ENDSENDS", 8) == 0) {
+			DECOMP_READ(file, tmp, 8);
+			break;
+		}
+
+		msg("decompress: corrupted %s\n", file->path);
+		goto error;
+	}
+
+	if (decomp_read(file, tmp, 1)) {
+		msg("decompress: error, EOF expected\n");
+		goto error;
+	}
+
+	if (!decomp_is_eof(file)) {
+		msg("decompress: error, EOF expected\n");
+		goto error;
+	}
+
+	if (arcfile) {
+		ds->datasink->close(arcfile);
+	}
+	MY_FREE(comp_chunk);
+	my_thread_end();
+	return NULL;
+
+error:
+	if (arcfile) {
+		ds->datasink->close(arcfile);
+	}
+	decomp_set_error(file);
+	MY_FREE(comp_chunk);
+	my_thread_end();
+	return NULL;
+
+}
 
=== added file 'src/decompress.h'
--- src/decompress.h	1970-01-01 00:00:00 +0000
+++ src/decompress.h	2012-08-19 12:20:47 +0000
@@ -0,0 +1,28 @@
+/******************************************************
+Copyright (c) 2011 Percona Inc.
+
+Compression interface 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_DEDSCOMPRESS_H
+#define DS_DEDSCOMPRESS_H
+
+#include "datasink.h"
+
+extern datasink_t datasink_decompress;
+
+#endif
 
=== modified file 'src/xbstream.c'
--- src/xbstream.c	2012-02-10 20:05:56 +0000
+++ src/xbstream.c	2012-08-19 12:20:47 +0000
@@ -24,6 +24,7 @@
 #include "common.h"
 #include "xbstream.h"
 #include "local.h"
+#include "decompress.h"
 
 #define XBSTREAM_VERSION "1.0"
 #define XBSTREAM_BUFFER_SIZE (1024 * 1024UL)
@@ -39,6 +40,7 @@
 static run_mode_t 	opt_mode;
 static char *		opt_directory = NULL;
 static my_bool		opt_verbose = 0;
+static my_bool		opt_decompress = 0;
 
 static struct my_option my_long_options[] =
 {
@@ -54,6 +56,9 @@
 	 GET_STR_ALLOC, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
 	{"verbose", 'v', "Print verbose output.", &opt_verbose, &opt_verbose,
 	 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+	{"decompress", 'z', "Decompress compressed files.",
+	 &opt_decompress, &opt_decompress,
+	 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
 
 	{0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
@@ -362,7 +367,9 @@
 	HASH			filehash;
 	file_entry_t		*entry;
 	datasink_t		*ds;
+	datasink_t		*ds_decomp;
 	ds_ctxt_t		*ds_ctxt;
+	ds_ctxt_t		*ds_ctxt_decomp;
 
 	stream = xb_stream_read_new();
 	if (stream == NULL) {
@@ -372,7 +379,10 @@
 
 	/* If --directory is specified, it is already set as CWD by now. */
 	ds = &datasink_local;
+	ds_decomp = &datasink_decompress;
 	ds_ctxt = ds->init(".");
+	ds_ctxt_decomp = ds_decomp->init(".");
+	ds_ctxt_decomp->pipe_ctxt = ds_ctxt;
 
 	if (my_hash_init(&filehash, &my_charset_bin, START_FILE_HASH_SIZE,
 			  0, 0, (my_hash_get_key) get_file_entry_key,
@@ -395,8 +405,18 @@
 							chunk.pathlen);
 
 		if (entry == NULL) {
-			entry = file_entry_new(ds_ctxt, chunk.path,
-					       chunk.pathlen);
+			if (opt_decompress &&
+			    strlen(chunk.path) > 3 &&
+			    strcmp(chunk.path + strlen(chunk.path) - 3,
+			    	   ".qp") == 0 &&
+			    memcmp(chunk.data, "qpress10", 8) == 0) {
+				entry = file_entry_new(ds_ctxt_decomp, chunk.path,
+						       chunk.pathlen);
+			} else {
+				entry = file_entry_new(ds_ctxt, chunk.path,
+						       chunk.pathlen);
+
+			}
 			if (entry == NULL) {
 				goto err;
 			}
@@ -420,7 +440,7 @@
 			goto err;
 		}
 
-		if (ds->write(entry->file, chunk.data, chunk.length)) {
+		if (entry->ds_ctxt->datasink->write(entry->file, chunk.data, chunk.length)) {
 			msg("%s: my_write() failed.\n", my_progname);
 			goto err;
 		}
@@ -434,12 +454,14 @@
 
 	my_hash_free(&filehash);
 	ds->deinit(ds_ctxt);
+	ds_decomp->deinit(ds_ctxt_decomp);
 	xb_stream_read_done(stream);
 
 	return 0;
 err:
 	my_hash_free(&filehash);
 	ds->deinit(ds_ctxt);
+	ds_decomp->deinit(ds_ctxt_decomp);
 	xb_stream_read_done(stream);
 
 	return 1;
 
=== added file 'test/t/ib_stream_decompress.sh'
--- test/t/ib_stream_decompress.sh	1970-01-01 00:00:00 +0000
+++ test/t/ib_stream_decompress.sh	2012-08-19 12:20:47 +0000
@@ -0,0 +1,9 @@
+############################################################################
+# Test streaming + compression (decompression with xbstream)
+############################################################################
+
+stream_format=xbstream
+stream_extract_cmd="xbstream -xvz <"
+innobackupex_options="--compress --compress-threads=4 --parallel=4"
+
+. inc/ib_stream_common.sh