summaryrefslogtreecommitdiff
path: root/extra/mariabackup/xbstream_write.c
diff options
context:
space:
mode:
authorMarko Mäkelä <marko.makela@mariadb.com>2017-05-06 14:36:46 +0300
committerMarko Mäkelä <marko.makela@mariadb.com>2017-05-06 14:36:46 +0300
commit14c6f00a9f25430f995fb42c768e19a9d2a980e8 (patch)
treeb0ff411e6ca13668124a66ded3b5bb5a7721e595 /extra/mariabackup/xbstream_write.c
parentbaad0f3484ec3079a09a206576290091cc823428 (diff)
parentb82c602db588cfa688278ef772050c004590c124 (diff)
downloadmariadb-git-14c6f00a9f25430f995fb42c768e19a9d2a980e8.tar.gz
Merge 10.1 into 10.2
Also, include fixes by Vladislav Vaintroub to the aws_key_management plugin. The AWS C++ SDK specifically depends on OPENSSL_LIBRARIES, not generic SSL_LIBRARIES (such as YaSSL).
Diffstat (limited to 'extra/mariabackup/xbstream_write.c')
-rw-r--r--extra/mariabackup/xbstream_write.c294
1 files changed, 294 insertions, 0 deletions
diff --git a/extra/mariabackup/xbstream_write.c b/extra/mariabackup/xbstream_write.c
new file mode 100644
index 00000000000..978be71e7dd
--- /dev/null
+++ b/extra/mariabackup/xbstream_write.c
@@ -0,0 +1,294 @@
+/******************************************************
+Copyright (c) 2011-2017 Percona LLC and/or its affiliates.
+
+The xbstream format writer implementation.
+
+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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
+
+*******************************************************/
+
+#include <mysql_version.h>
+#include <my_base.h>
+#include <zlib.h>
+#include "common.h"
+#include "xbstream.h"
+#include "crc_glue.h"
+
+/* Group writes smaller than this into a single chunk */
+#define XB_STREAM_MIN_CHUNK_SIZE (10 * 1024 * 1024)
+
+struct xb_wstream_struct {
+ pthread_mutex_t mutex;
+};
+
+struct xb_wstream_file_struct {
+ xb_wstream_t *stream;
+ char *path;
+ size_t path_len;
+ char chunk[XB_STREAM_MIN_CHUNK_SIZE];
+ char *chunk_ptr;
+ size_t chunk_free;
+ my_off_t offset;
+ void *userdata;
+ xb_stream_write_callback *write;
+};
+
+static int xb_stream_flush(xb_wstream_file_t *file);
+static int xb_stream_write_chunk(xb_wstream_file_t *file,
+ const void *buf, size_t len);
+static int xb_stream_write_eof(xb_wstream_file_t *file);
+
+static
+ssize_t
+xb_stream_default_write_callback(xb_wstream_file_t *file __attribute__((unused)),
+ void *userdata __attribute__((unused)),
+ const void *buf, size_t len)
+{
+ if (my_write(my_fileno(stdout), buf, len, MYF(MY_WME | MY_NABP)))
+ return -1;
+ return len;
+}
+
+xb_wstream_t *
+xb_stream_write_new(void)
+{
+ xb_wstream_t *stream;
+
+ stream = (xb_wstream_t *) my_malloc(sizeof(xb_wstream_t), MYF(MY_FAE));
+ pthread_mutex_init(&stream->mutex, NULL);
+
+ return stream;;
+}
+
+xb_wstream_file_t *
+xb_stream_write_open(xb_wstream_t *stream, const char *path,
+ MY_STAT *mystat __attribute__((unused)),
+ void *userdata,
+ xb_stream_write_callback *onwrite)
+{
+ xb_wstream_file_t *file;
+ size_t path_len;
+
+ path_len = strlen(path);
+
+ if (path_len > FN_REFLEN) {
+ msg("xb_stream_write_open(): file path is too long.\n");
+ return NULL;
+ }
+
+ file = (xb_wstream_file_t *) my_malloc(sizeof(xb_wstream_file_t) +
+ path_len + 1, MYF(MY_FAE));
+
+ file->path = (char *) (file + 1);
+#ifdef _WIN32
+ /* Normalize path on Windows, so we can restore elsewhere.*/
+ {
+ int i;
+ for (i = 0; ; i++) {
+ file->path[i] = (path[i] == '\\') ? '/' : path[i];
+ if (!path[i])
+ break;
+ }
+ }
+#else
+ memcpy(file->path, path, path_len + 1);
+#endif
+ file->path_len = path_len;
+
+ file->stream = stream;
+ file->offset = 0;
+ file->chunk_ptr = file->chunk;
+ file->chunk_free = XB_STREAM_MIN_CHUNK_SIZE;
+ if (onwrite) {
+#ifdef __WIN__
+ setmode(fileno(stdout), _O_BINARY);
+#endif
+ file->userdata = userdata;
+ file->write = onwrite;
+ } else {
+ file->userdata = NULL;
+ file->write = xb_stream_default_write_callback;
+ }
+
+ return file;
+}
+
+int
+xb_stream_write_data(xb_wstream_file_t *file, const void *buf, size_t len)
+{
+ if (len < file->chunk_free) {
+ memcpy(file->chunk_ptr, buf, len);
+ file->chunk_ptr += len;
+ file->chunk_free -= len;
+
+ return 0;
+ }
+
+ if (xb_stream_flush(file))
+ return 1;
+
+ return xb_stream_write_chunk(file, buf, len);
+}
+
+int
+xb_stream_write_close(xb_wstream_file_t *file)
+{
+ if (xb_stream_flush(file) ||
+ xb_stream_write_eof(file)) {
+ my_free(file);
+ return 1;
+ }
+
+ my_free(file);
+
+ return 0;
+}
+
+int
+xb_stream_write_done(xb_wstream_t *stream)
+{
+ pthread_mutex_destroy(&stream->mutex);
+
+ my_free(stream);
+
+ return 0;
+}
+
+static
+int
+xb_stream_flush(xb_wstream_file_t *file)
+{
+ if (file->chunk_ptr == file->chunk) {
+ return 0;
+ }
+
+ if (xb_stream_write_chunk(file, file->chunk,
+ file->chunk_ptr - file->chunk)) {
+ return 1;
+ }
+
+ file->chunk_ptr = file->chunk;
+ file->chunk_free = XB_STREAM_MIN_CHUNK_SIZE;
+
+ return 0;
+}
+
+static
+int
+xb_stream_write_chunk(xb_wstream_file_t *file, const void *buf, size_t len)
+{
+ /* Chunk magic + flags + chunk type + path_len + path + len + offset +
+ checksum */
+ uchar tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 + 4 +
+ FN_REFLEN + 8 + 8 + 4];
+ uchar *ptr;
+ xb_wstream_t *stream = file->stream;
+ ulong checksum;
+
+ /* Write xbstream header */
+ ptr = tmpbuf;
+
+ /* Chunk magic */
+ memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
+ ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;
+
+ *ptr++ = 0; /* Chunk flags */
+
+ *ptr++ = (uchar) XB_CHUNK_TYPE_PAYLOAD; /* Chunk type */
+
+ int4store(ptr, file->path_len); /* Path length */
+ ptr += 4;
+
+ memcpy(ptr, file->path, file->path_len); /* Path */
+ ptr += file->path_len;
+
+ int8store(ptr, len); /* Payload length */
+ ptr += 8;
+
+ checksum = crc32_iso3309(0, buf, (uint)len); /* checksum */
+
+ pthread_mutex_lock(&stream->mutex);
+
+ int8store(ptr, file->offset); /* Payload offset */
+ ptr += 8;
+
+ int4store(ptr, checksum);
+ ptr += 4;
+
+ xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));
+
+ if (file->write(file, file->userdata, tmpbuf, ptr-tmpbuf) == -1)
+ goto err;
+
+
+ if (file->write(file, file->userdata, buf, len) == -1) /* Payload */
+ goto err;
+
+ file->offset+= len;
+
+ pthread_mutex_unlock(&stream->mutex);
+
+ return 0;
+
+err:
+
+ pthread_mutex_unlock(&stream->mutex);
+
+ return 1;
+}
+
+static
+int
+xb_stream_write_eof(xb_wstream_file_t *file)
+{
+ /* Chunk magic + flags + chunk type + path_len + path */
+ uchar tmpbuf[sizeof(XB_STREAM_CHUNK_MAGIC) - 1 + 1 + 1 + 4 +
+ FN_REFLEN];
+ uchar *ptr;
+ xb_wstream_t *stream = file->stream;
+
+ pthread_mutex_lock(&stream->mutex);
+
+ /* Write xbstream header */
+ ptr = tmpbuf;
+
+ /* Chunk magic */
+ memcpy(ptr, XB_STREAM_CHUNK_MAGIC, sizeof(XB_STREAM_CHUNK_MAGIC) - 1);
+ ptr += sizeof(XB_STREAM_CHUNK_MAGIC) - 1;
+
+ *ptr++ = 0; /* Chunk flags */
+
+ *ptr++ = (uchar) XB_CHUNK_TYPE_EOF; /* Chunk type */
+
+ int4store(ptr, file->path_len); /* Path length */
+ ptr += 4;
+
+ memcpy(ptr, file->path, file->path_len); /* Path */
+ ptr += file->path_len;
+
+ xb_ad(ptr <= tmpbuf + sizeof(tmpbuf));
+
+ if (file->write(file, file->userdata, tmpbuf,
+ (ulonglong) (ptr - tmpbuf)) == -1)
+ goto err;
+
+ pthread_mutex_unlock(&stream->mutex);
+
+ return 0;
+err:
+
+ pthread_mutex_unlock(&stream->mutex);
+
+ return 1;
+}