diff options
Diffstat (limited to 'xlators/features/compress/src')
| -rw-r--r-- | xlators/features/compress/src/Makefile.am | 19 | ||||
| -rw-r--r-- | xlators/features/compress/src/cdc-helper.c | 527 | ||||
| -rw-r--r-- | xlators/features/compress/src/cdc-mem-types.h | 23 | ||||
| -rw-r--r-- | xlators/features/compress/src/cdc.c | 348 | ||||
| -rw-r--r-- | xlators/features/compress/src/cdc.h | 99 |
5 files changed, 1016 insertions, 0 deletions
diff --git a/xlators/features/compress/src/Makefile.am b/xlators/features/compress/src/Makefile.am new file mode 100644 index 00000000000..98271a9f3fc --- /dev/null +++ b/xlators/features/compress/src/Makefile.am @@ -0,0 +1,19 @@ +xlator_LTLIBRARIES = cdc.la + +xlatordir = $(libdir)/glusterfs/$(PACKAGE_VERSION)/xlator/features + +noinst_HEADERS = cdc.h cdc-mem-types.h + +cdc_la_LDFLAGS = -module $(GF_XLATOR_DEFAULT_LDFLAGS) + +cdc_la_SOURCES = cdc.c cdc-helper.c +cdc_la_LIBADD = $(top_builddir)/libglusterfs/src/libglusterfs.la $(ZLIB_LIBS) + +AM_CPPFLAGS = $(GF_CPPFLAGS) -I$(top_srcdir)/libglusterfs/src \ + -I$(top_srcdir)/rpc/xdr/src -I$(top_builddir)/rpc/xdr/src \ + -fPIC -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -D$(GF_HOST_OS) \ + $(LIBZ_CFLAGS) + +AM_CFLAGS = -Wall $(GF_CFLAGS) + +CLEANFILES = diff --git a/xlators/features/compress/src/cdc-helper.c b/xlators/features/compress/src/cdc-helper.c new file mode 100644 index 00000000000..f973ff56cf5 --- /dev/null +++ b/xlators/features/compress/src/cdc-helper.c @@ -0,0 +1,527 @@ +/* + Copyright (c) 2013 Red Hat, Inc. <http://www.redhat.com> + This file is part of GlusterFS. + + This file is licensed to you under your choice of the GNU Lesser + General Public License, version 3 or any later version (LGPLv3 or + later), or the GNU General Public License, version 2 (GPLv2), in all + cases as published by the Free Software Foundation. +*/ + +#include <glusterfs/glusterfs.h> +#include <glusterfs/logging.h> +#include <glusterfs/syscall.h> + +#include "cdc.h" +#include "cdc-mem-types.h" + +#ifdef HAVE_LIB_Z +#include "zlib.h" +#endif + +#ifdef HAVE_LIB_Z +/* gzip header looks something like this + * (RFC 1950) + * + * +---+---+---+---+---+---+---+---+---+---+ + * |ID1|ID2|CM |FLG| MTIME |XFL|OS | + * +---+---+---+---+---+---+---+---+---+---+ + * + * Data is usually sent without this header i.e + * Data sent = <compressed-data> + trailer(8) + * The trailer contains the checksum. + * + * gzip_header is added only during debugging. + * Refer to the function cdc_dump_iovec_to_disk + */ +static const char gzip_header[10] = {'\037', '\213', Z_DEFLATED, 0, 0, 0, 0, + 0, 0, GF_CDC_OS_ID}; + +static int32_t +cdc_next_iovec(xlator_t *this, cdc_info_t *ci) +{ + int ret = -1; + + ci->ncount++; + /* check for iovec overflow -- should not happen */ + if (ci->ncount == MAX_IOVEC) { + gf_log(this->name, GF_LOG_ERROR, + "Zlib output buffer overflow" + " ->ncount (%d) | ->MAX_IOVEC (%d)", + ci->ncount, MAX_IOVEC); + goto out; + } + + ret = 0; + +out: + return ret; +} + +static void +cdc_put_long(unsigned char *string, unsigned long x) +{ + string[0] = (unsigned char)(x & 0xff); + string[1] = (unsigned char)((x & 0xff00) >> 8); + string[2] = (unsigned char)((x & 0xff0000) >> 16); + string[3] = (unsigned char)((x & 0xff000000) >> 24); +} + +static unsigned long +cdc_get_long(unsigned char *buf) +{ + return ((unsigned long)buf[0]) | (((unsigned long)buf[1]) << 8) | + (((unsigned long)buf[2]) << 16) | (((unsigned long)buf[3]) << 24); +} + +static int32_t +cdc_init_gzip_trailer(xlator_t *this, cdc_priv_t *priv, cdc_info_t *ci) +{ + int ret = -1; + char *buf = NULL; + + ret = cdc_next_iovec(this, ci); + if (ret) + goto out; + + buf = CURR_VEC(ci).iov_base = (char *)GF_CALLOC(1, GF_CDC_VALIDATION_SIZE, + gf_cdc_mt_gzip_trailer_t); + + if (!CURR_VEC(ci).iov_base) + goto out; + + CURR_VEC(ci).iov_len = GF_CDC_VALIDATION_SIZE; + + cdc_put_long((unsigned char *)&buf[0], ci->crc); + cdc_put_long((unsigned char *)&buf[4], ci->stream.total_in); + + ret = 0; + +out: + return ret; +} + +static int32_t +cdc_alloc_iobuf_and_init_vec(xlator_t *this, cdc_priv_t *priv, cdc_info_t *ci, + int size) +{ + int ret = -1; + int alloc_len = 0; + struct iobuf *iobuf = NULL; + + ret = cdc_next_iovec(this, ci); + if (ret) + goto out; + + alloc_len = size ? size : ci->buffer_size; + + iobuf = iobuf_get2(this->ctx->iobuf_pool, alloc_len); + if (!iobuf) + goto out; + + ret = iobref_add(ci->iobref, iobuf); + if (ret) + goto out; + + /* Initialize this iovec */ + CURR_VEC(ci).iov_base = iobuf->ptr; + CURR_VEC(ci).iov_len = alloc_len; + + ret = 0; + +out: + return ret; +} + +static void +cdc_init_zlib_output_stream(cdc_priv_t *priv, cdc_info_t *ci, int size) +{ + ci->stream.next_out = (unsigned char *)CURR_VEC(ci).iov_base; + ci->stream.avail_out = size ? size : ci->buffer_size; +} + +/* This routine is for testing and debugging only. + * Data written = header(10) + <compressed-data> + trailer(8) + * So each gzip dump file is at least 18 bytes in size. + */ +void +cdc_dump_iovec_to_disk(xlator_t *this, cdc_info_t *ci, const char *file) +{ + int i = 0; + int fd = 0; + size_t written = 0; + size_t total_written = 0; + + fd = open(file, O_WRONLY | O_CREAT | O_TRUNC, 0777); + if (fd < 0) { + gf_log(this->name, GF_LOG_ERROR, "Cannot open file: %s", file); + return; + } + + written = sys_write(fd, (char *)gzip_header, 10); + total_written += written; + for (i = 0; i < ci->ncount; i++) { + written = sys_write(fd, (char *)ci->vec[i].iov_base, + ci->vec[i].iov_len); + total_written += written; + } + + gf_log(this->name, GF_LOG_DEBUG, "dump'd %zu bytes to %s", total_written, + GF_CDC_DEBUG_DUMP_FILE); + + sys_close(fd); +} + +static int32_t +cdc_flush_libz_buffer(cdc_priv_t *priv, xlator_t *this, cdc_info_t *ci, + int (*libz_func)(z_streamp, int), int flush) +{ + int32_t ret = Z_OK; + int done = 0; + unsigned int deflate_len = 0; + + for (;;) { + deflate_len = ci->buffer_size - ci->stream.avail_out; + + if (deflate_len != 0) { + CURR_VEC(ci).iov_len = deflate_len; + + ret = cdc_alloc_iobuf_and_init_vec(this, priv, ci, 0); + if (ret) { + ret = Z_MEM_ERROR; + break; + } + + /* Re-position Zlib output buffer */ + cdc_init_zlib_output_stream(priv, ci, 0); + } + + if (done) { + ci->ncount--; + break; + } + + ret = libz_func(&ci->stream, flush); + + if (ret == Z_BUF_ERROR) { + ret = Z_OK; + ci->ncount--; + break; + } + + done = (ci->stream.avail_out != 0 || ret == Z_STREAM_END); + + if (ret != Z_OK && ret != Z_STREAM_END) + break; + } + + return ret; +} + +static int32_t +do_cdc_compress(struct iovec *vec, xlator_t *this, cdc_priv_t *priv, + cdc_info_t *ci) +{ + int ret = -1; + + /* Initialize defalte */ + ret = deflateInit2(&ci->stream, priv->cdc_level, Z_DEFLATED, + priv->window_size, priv->mem_level, Z_DEFAULT_STRATEGY); + + if (ret) { + gf_log(this->name, GF_LOG_ERROR, "unable to init Zlib (retval: %d)", + ret); + goto out; + } + + ret = cdc_alloc_iobuf_and_init_vec(this, priv, ci, 0); + if (ret) + goto out; + + /* setup output buffer */ + cdc_init_zlib_output_stream(priv, ci, 0); + + /* setup input buffer */ + ci->stream.next_in = (unsigned char *)vec->iov_base; + ci->stream.avail_in = vec->iov_len; + + ci->crc = crc32(ci->crc, (const Bytef *)vec->iov_base, vec->iov_len); + + gf_log(this->name, GF_LOG_DEBUG, "crc=%lu len=%d buffer_size=%d", ci->crc, + ci->stream.avail_in, ci->buffer_size); + + /* compress !! */ + while (ci->stream.avail_in != 0) { + if (ci->stream.avail_out == 0) { + CURR_VEC(ci).iov_len = ci->buffer_size; + + ret = cdc_alloc_iobuf_and_init_vec(this, priv, ci, 0); + if (ret) + break; + + /* Re-position Zlib output buffer */ + cdc_init_zlib_output_stream(priv, ci, 0); + } + + ret = deflate(&ci->stream, Z_NO_FLUSH); + if (ret != Z_OK) + break; + } + +out: + return ret; +} + +int32_t +cdc_compress(xlator_t *this, cdc_priv_t *priv, cdc_info_t *ci, dict_t **xdata) +{ + int ret = -1; + int i = 0; + + ci->iobref = iobref_new(); + if (!ci->iobref) + goto out; + + if (!*xdata) { + *xdata = dict_new(); + if (!*xdata) { + gf_log(this->name, GF_LOG_ERROR, + "Cannot allocate xdata" + " dict"); + goto out; + } + } + + /* data */ + for (i = 0; i < ci->count; i++) { + ret = do_cdc_compress(&ci->vector[i], this, priv, ci); + if (ret != Z_OK) + goto deflate_cleanup_out; + } + + /* flush zlib buffer */ + ret = cdc_flush_libz_buffer(priv, this, ci, deflate, Z_FINISH); + if (!(ret == Z_OK || ret == Z_STREAM_END)) { + gf_log(this->name, GF_LOG_ERROR, "Compression Error: ret (%d)", ret); + ret = -1; + goto deflate_cleanup_out; + } + + /* trailer */ + ret = cdc_init_gzip_trailer(this, priv, ci); + if (ret) + goto deflate_cleanup_out; + + gf_log(this->name, GF_LOG_DEBUG, "Compressed %ld to %ld bytes", + ci->stream.total_in, ci->stream.total_out); + + ci->nbytes = ci->stream.total_out + GF_CDC_VALIDATION_SIZE; + + /* set deflated canary value for identification */ + ret = dict_set_int32(*xdata, GF_CDC_DEFLATE_CANARY_VAL, 1); + if (ret) { + /* Send uncompressed data if we can't _tell_ the client + * that deflated data is on it's way. So, we just log + * the failure and continue as usual. + */ + gf_log(this->name, GF_LOG_ERROR, + "Data deflated, but could not set canary" + " value in dict for identification"); + } + + /* This is to be used in testing */ + if (priv->debug) { + cdc_dump_iovec_to_disk(this, ci, GF_CDC_DEBUG_DUMP_FILE); + } + +deflate_cleanup_out: + (void)deflateEnd(&ci->stream); + +out: + return ret; +} + +/* deflate content is checked by the presence of a canary + * value in the dict as the key + */ +static int32_t +cdc_check_content_for_deflate(dict_t *xdata) +{ + return dict_get(xdata, GF_CDC_DEFLATE_CANARY_VAL) ? -1 : 0; +} + +static unsigned long +cdc_extract_crc(char *trailer) +{ + return cdc_get_long((unsigned char *)&trailer[0]); +} + +static unsigned long +cdc_extract_size(char *trailer) +{ + return cdc_get_long((unsigned char *)&trailer[4]); +} + +static int32_t +cdc_validate_inflate(cdc_info_t *ci, unsigned long crc, unsigned long len) +{ + return !((crc == ci->crc) + /* inflated length is hidden inside + * Zlib stream struct */ + && (len == ci->stream.total_out)); +} + +static int32_t +do_cdc_decompress(xlator_t *this, cdc_priv_t *priv, cdc_info_t *ci) +{ + int ret = -1; + int i = 0; + int len = 0; + char *inflte = NULL; + char *trailer = NULL; + struct iovec vec = { + 0, + }; + unsigned long computed_crc = 0; + unsigned long computed_len = 0; + + ret = inflateInit2(&ci->stream, priv->window_size); + if (ret) { + gf_log(this->name, GF_LOG_ERROR, "Zlib: Unable to initialize inflate"); + goto out; + } + + vec = THIS_VEC(ci, 0); + + trailer = (char *)(((char *)vec.iov_base) + vec.iov_len - + GF_CDC_VALIDATION_SIZE); + + /* CRC of uncompressed data */ + computed_crc = cdc_extract_crc(trailer); + + /* size of uncomrpessed data */ + computed_len = cdc_extract_size(trailer); + + gf_log(this->name, GF_LOG_DEBUG, "crc=%lu len=%lu buffer_size=%d", + computed_crc, computed_len, ci->buffer_size); + + inflte = vec.iov_base; + len = vec.iov_len - GF_CDC_VALIDATION_SIZE; + + /* allocate buffer of the original length of the data */ + ret = cdc_alloc_iobuf_and_init_vec(this, priv, ci, 0); + if (ret) + goto out; + + /* setup output buffer */ + cdc_init_zlib_output_stream(priv, ci, 0); + + /* setup input buffer */ + ci->stream.next_in = (unsigned char *)inflte; + ci->stream.avail_in = len; + + while (ci->stream.avail_in != 0) { + if (ci->stream.avail_out == 0) { + CURR_VEC(ci).iov_len = ci->buffer_size; + + ret = cdc_alloc_iobuf_and_init_vec(this, priv, ci, 0); + if (ret) + break; + + /* Re-position Zlib output buffer */ + cdc_init_zlib_output_stream(priv, ci, 0); + } + + ret = inflate(&ci->stream, Z_NO_FLUSH); + if (ret == Z_STREAM_ERROR) + break; + } + + /* flush zlib buffer */ + ret = cdc_flush_libz_buffer(priv, this, ci, inflate, Z_SYNC_FLUSH); + if (!(ret == Z_OK || ret == Z_STREAM_END)) { + gf_log(this->name, GF_LOG_ERROR, "Decompression Error: ret (%d)", ret); + ret = -1; + goto out; + } + + /* compute CRC of the uncompresses data to check for + * correctness */ + + for (i = 0; i < ci->ncount; i++) { + ci->crc = crc32(ci->crc, (const Bytef *)ci->vec[i].iov_base, + ci->vec[i].iov_len); + } + + /* validate inflated data */ + ret = cdc_validate_inflate(ci, computed_crc, computed_len); + if (ret) { + gf_log(this->name, GF_LOG_ERROR, + "Checksum or length mismatched in inflated data"); + } + +out: + return ret; +} + +int32_t +cdc_decompress(xlator_t *this, cdc_priv_t *priv, cdc_info_t *ci, dict_t *xdata) +{ + int32_t ret = -1; + + /* check for deflate content */ + if (!cdc_check_content_for_deflate(xdata)) { + gf_log(this->name, GF_LOG_DEBUG, + "Content not deflated, passing through ..."); + goto passthrough_out; + } + + ci->iobref = iobref_new(); + if (!ci->iobref) + goto passthrough_out; + + /* do we need to do this? can we assume that one iovec + * will hold per request data every time? + * + * server/client protocol seems to deal with a single + * iovec even if op_ret > 1M. So, it looks ok to + * assume that a single iovec will contain all the + * data (This saves us a lot from finding the trailer + * and the data since it could have been split-up onto + * two adjacent iovec's. + * + * But, in case this translator is loaded above quick-read + * for some reason, then it's entirely possible that we get + * multiple iovec's... + * + * This case (handled below) is not tested. (by loading the + * xlator below quick-read) + */ + + /* @@ I_HOPE_THIS_IS_NEVER_HIT */ + if (ci->count > 1) { + gf_log(this->name, GF_LOG_WARNING, + "unable to handle" + " multiple iovecs (%d in number)", + ci->count); + goto inflate_cleanup_out; + /* TODO: coallate all iovecs in one */ + } + + ret = do_cdc_decompress(this, priv, ci); + if (ret) + goto inflate_cleanup_out; + + ci->nbytes = ci->stream.total_out; + + gf_log(this->name, GF_LOG_DEBUG, "Inflated %ld to %ld bytes", + ci->stream.total_in, ci->stream.total_out); + +inflate_cleanup_out: + (void)inflateEnd(&ci->stream); + +passthrough_out: + return ret; +} + +#endif diff --git a/xlators/features/compress/src/cdc-mem-types.h b/xlators/features/compress/src/cdc-mem-types.h new file mode 100644 index 00000000000..928afdd2efe --- /dev/null +++ b/xlators/features/compress/src/cdc-mem-types.h @@ -0,0 +1,23 @@ +/* + Copyright (c) 2013 Red Hat, Inc. <http://www.redhat.com> + This file is part of GlusterFS. + + This file is licensed to you under your choice of the GNU Lesser + General Public License, version 3 or any later version (LGPLv3 or + later), or the GNU General Public License, version 2 (GPLv2), in all + cases as published by the Free Software Foundation. +*/ + +#ifndef __CDC_MEM_TYPES_H +#define __CDC_MEM_TYPES_H + +#include <glusterfs/mem-types.h> + +enum gf_cdc_mem_types { + gf_cdc_mt_priv_t = gf_common_mt_end + 1, + gf_cdc_mt_vec_t = gf_common_mt_end + 2, + gf_cdc_mt_gzip_trailer_t = gf_common_mt_end + 3, + gf_cdc_mt_end = gf_common_mt_end + 4, +}; + +#endif diff --git a/xlators/features/compress/src/cdc.c b/xlators/features/compress/src/cdc.c new file mode 100644 index 00000000000..b0b51e914ed --- /dev/null +++ b/xlators/features/compress/src/cdc.c @@ -0,0 +1,348 @@ +/* + Copyright (c) 2013 Red Hat, Inc. <http://www.redhat.com> + This file is part of GlusterFS. + + This file is licensed to you under your choice of the GNU Lesser + General Public License, version 3 or any later version (LGPLv3 or + later), or the GNU General Public License, version 2 (GPLv2), in all + cases as published by the Free Software Foundation. +*/ + +#include <sys/uio.h> + +#include <glusterfs/xlator.h> +#include <glusterfs/defaults.h> +#include <glusterfs/logging.h> + +#include "cdc.h" +#include "cdc-mem-types.h" + +static void +cdc_cleanup_iobref(cdc_info_t *ci) +{ + assert(ci->iobref != NULL); + iobref_clear(ci->iobref); +} + +int32_t +cdc_readv_cbk(call_frame_t *frame, void *cookie, xlator_t *this, int32_t op_ret, + int32_t op_errno, struct iovec *vector, int32_t count, + struct iatt *stbuf, struct iobref *iobref, dict_t *xdata) +{ + int ret = -1; + cdc_priv_t *priv = NULL; + cdc_info_t ci = { + 0, + }; + + GF_VALIDATE_OR_GOTO("cdc", this, default_out); + GF_VALIDATE_OR_GOTO(this->name, frame, default_out); + + priv = this->private; + + if (op_ret <= 0) + goto default_out; + + if ((priv->min_file_size != 0) && (op_ret < priv->min_file_size)) + goto default_out; + + ci.count = count; + ci.ibytes = op_ret; + ci.vector = vector; + ci.buf = NULL; + ci.iobref = NULL; + ci.ncount = 0; + ci.crc = 0; + ci.buffer_size = GF_CDC_DEF_BUFFERSIZE; + + /* A readv compresses on the server side and decompresses on the client side + */ + if (priv->op_mode == GF_CDC_MODE_SERVER) { + ret = cdc_compress(this, priv, &ci, &xdata); + } else if (priv->op_mode == GF_CDC_MODE_CLIENT) { + ret = cdc_decompress(this, priv, &ci, xdata); + } else { + gf_log(this->name, GF_LOG_ERROR, "Invalid operation mode (%d)", + priv->op_mode); + } + + if (ret) + goto default_out; + + STACK_UNWIND_STRICT(readv, frame, ci.nbytes, op_errno, ci.vec, ci.ncount, + stbuf, iobref, xdata); + cdc_cleanup_iobref(&ci); + return 0; + +default_out: + STACK_UNWIND_STRICT(readv, frame, op_ret, op_errno, vector, count, stbuf, + iobref, xdata); + return 0; +} + +int32_t +cdc_readv(call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size, + off_t offset, uint32_t flags, dict_t *xdata) +{ + fop_readv_cbk_t cbk = NULL; + +#ifdef HAVE_LIB_Z + cbk = cdc_readv_cbk; +#else + cbk = default_readv_cbk; +#endif + STACK_WIND(frame, cbk, FIRST_CHILD(this), FIRST_CHILD(this)->fops->readv, + fd, size, offset, flags, xdata); + return 0; +} + +int32_t +cdc_writev_cbk(call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, struct iatt *prebuf, + struct iatt *postbuf, dict_t *xdata) +{ + STACK_UNWIND_STRICT(writev, frame, op_ret, op_errno, prebuf, postbuf, + xdata); + return 0; +} + +int32_t +cdc_writev(call_frame_t *frame, xlator_t *this, fd_t *fd, struct iovec *vector, + int32_t count, off_t offset, uint32_t flags, struct iobref *iobref, + dict_t *xdata) +{ + int ret = -1; + cdc_priv_t *priv = NULL; + cdc_info_t ci = { + 0, + }; + size_t isize = 0; + + GF_VALIDATE_OR_GOTO("cdc", this, err); + GF_VALIDATE_OR_GOTO(this->name, frame, err); + + priv = this->private; + + isize = iov_length(vector, count); + + if (isize <= 0) + goto default_out; + + if ((priv->min_file_size != 0) && (isize < priv->min_file_size)) + goto default_out; + + ci.count = count; + ci.ibytes = isize; + ci.vector = vector; + ci.buf = NULL; + ci.iobref = NULL; + ci.ncount = 0; + ci.crc = 0; + ci.buffer_size = GF_CDC_DEF_BUFFERSIZE; + + /* A writev compresses on the client side and decompresses on the server + * side + */ + if (priv->op_mode == GF_CDC_MODE_CLIENT) { + ret = cdc_compress(this, priv, &ci, &xdata); + } else if (priv->op_mode == GF_CDC_MODE_SERVER) { + ret = cdc_decompress(this, priv, &ci, xdata); + } else { + gf_log(this->name, GF_LOG_ERROR, "Invalid operation mode (%d) ", + priv->op_mode); + } + + if (ret) + goto default_out; + + STACK_WIND(frame, cdc_writev_cbk, FIRST_CHILD(this), + FIRST_CHILD(this)->fops->writev, fd, ci.vec, ci.ncount, offset, + flags, iobref, xdata); + + cdc_cleanup_iobref(&ci); + return 0; + +default_out: + STACK_WIND(frame, cdc_writev_cbk, FIRST_CHILD(this), + FIRST_CHILD(this)->fops->writev, fd, vector, count, offset, + flags, iobref, xdata); + return 0; +err: + STACK_UNWIND_STRICT(writev, frame, -1, EINVAL, NULL, NULL, NULL); + return 0; +} + +int32_t +mem_acct_init(xlator_t *this) +{ + int ret = -1; + + if (!this) + return ret; + + ret = xlator_mem_acct_init(this, gf_cdc_mt_end); + + if (ret != 0) { + gf_log(this->name, GF_LOG_ERROR, + "Memory accounting init" + "failed"); + return ret; + } + + return ret; +} + +int32_t +init(xlator_t *this) +{ + int ret = -1; + char *temp_str = NULL; + cdc_priv_t *priv = NULL; + + GF_VALIDATE_OR_GOTO("cdc", this, err); + + if (!this->children || this->children->next) { + gf_log(this->name, GF_LOG_ERROR, "Need subvolume == 1"); + goto err; + } + + if (!this->parents) { + gf_log(this->name, GF_LOG_WARNING, "Dangling volume. Check volfile"); + } + + priv = GF_CALLOC(1, sizeof(*priv), gf_cdc_mt_priv_t); + if (!priv) { + goto err; + } + + /* Check if debug mode is turned on */ + GF_OPTION_INIT("debug", priv->debug, bool, err); + if (priv->debug) { + gf_log(this->name, GF_LOG_DEBUG, "CDC debug option turned on"); + } + + /* Set Gzip Window Size */ + GF_OPTION_INIT("window-size", priv->window_size, int32, err); + if ((priv->window_size > GF_CDC_MAX_WINDOWSIZE) || + (priv->window_size < GF_CDC_DEF_WINDOWSIZE)) { + gf_log(this->name, GF_LOG_WARNING, + "Invalid gzip window size (%d), using default", + priv->window_size); + priv->window_size = GF_CDC_DEF_WINDOWSIZE; + } + + /* Set Gzip (De)Compression Level */ + GF_OPTION_INIT("compression-level", priv->cdc_level, int32, err); + if (((priv->cdc_level < 1) || (priv->cdc_level > 9)) && + (priv->cdc_level != GF_CDC_DEF_COMPRESSION)) { + gf_log(this->name, GF_LOG_WARNING, + "Invalid gzip (de)compression level (%d)," + " using default", + priv->cdc_level); + priv->cdc_level = GF_CDC_DEF_COMPRESSION; + } + + /* Set Gzip Memory Level */ + GF_OPTION_INIT("mem-level", priv->mem_level, int32, err); + if ((priv->mem_level < 1) || (priv->mem_level > 9)) { + gf_log(this->name, GF_LOG_WARNING, + "Invalid gzip memory level, using the default"); + priv->mem_level = GF_CDC_DEF_MEMLEVEL; + } + + /* Set min file size to enable compression */ + GF_OPTION_INIT("min-size", priv->min_file_size, int32, err); + + /* Mode of operation - Server/Client */ + ret = dict_get_str(this->options, "mode", &temp_str); + if (ret) { + gf_log(this->name, GF_LOG_CRITICAL, "Operation mode not specified !!"); + goto err; + } + + if (GF_CDC_MODE_IS_CLIENT(temp_str)) { + priv->op_mode = GF_CDC_MODE_CLIENT; + } else if (GF_CDC_MODE_IS_SERVER(temp_str)) { + priv->op_mode = GF_CDC_MODE_SERVER; + } else { + gf_log(this->name, GF_LOG_CRITICAL, + "Bogus operation mode (%s) specified", temp_str); + goto err; + } + + this->private = priv; + gf_log(this->name, GF_LOG_DEBUG, "CDC xlator loaded in (%s) mode", + temp_str); + return 0; + +err: + if (priv) + GF_FREE(priv); + + return -1; +} + +void +fini(xlator_t *this) +{ + cdc_priv_t *priv = this->private; + + if (priv) + GF_FREE(priv); + this->private = NULL; + return; +} + +struct xlator_fops fops = { + .readv = cdc_readv, + .writev = cdc_writev, +}; + +struct xlator_cbks cbks = {}; + +struct volume_options options[] = { + {.key = {"window-size"}, + .default_value = "-15", + .type = GF_OPTION_TYPE_INT, + .description = "Size of the zlib history buffer."}, + {.key = {"mem-level"}, + .default_value = "8", + .type = GF_OPTION_TYPE_INT, + .description = "Memory allocated for internal compression state. " + "1 uses minimum memory but is slow and reduces " + "compression ratio; memLevel=9 uses maximum memory " + "for optimal speed. The default value is 8."}, + {.key = {"compression-level"}, + .default_value = "-1", + .type = GF_OPTION_TYPE_INT, + .description = "Compression levels \n" + "0 : no compression, 1 : best speed, \n" + "9 : best compression, -1 : default compression "}, + {.key = {"min-size"}, + .default_value = "0", + .type = GF_OPTION_TYPE_INT, + .description = "Data is compressed only when its size exceeds this."}, + {.key = {"mode"}, + .value = {"server", "client"}, + .type = GF_OPTION_TYPE_STR, + .description = "Set on the basis of where the xlator is loaded. " + "This option should NOT be configured by user."}, + {.key = {"debug"}, + .default_value = "false", + .type = GF_OPTION_TYPE_BOOL, + .description = "This is used in testing. Will dump compressed data " + "to disk as a gzip file."}, + {.key = {NULL}}, +}; + +xlator_api_t xlator_api = { + .init = init, + .fini = fini, + .mem_acct_init = mem_acct_init, + .op_version = {GD_OP_VERSION_3_9_0}, + .fops = &fops, + .cbks = &cbks, + .options = options, + .identifier = "cdc", + .category = GF_TECH_PREVIEW, +}; diff --git a/xlators/features/compress/src/cdc.h b/xlators/features/compress/src/cdc.h new file mode 100644 index 00000000000..cb87b06a989 --- /dev/null +++ b/xlators/features/compress/src/cdc.h @@ -0,0 +1,99 @@ +/* + Copyright (c) 2013 Red Hat, Inc. <http://www.redhat.com> + This file is part of GlusterFS. + + This file is licensed to you under your choice of the GNU Lesser + General Public License, version 3 or any later version (LGPLv3 or + later), or the GNU General Public License, version 2 (GPLv2), in all + cases as published by the Free Software Foundation. +*/ + +#ifndef __CDC_H +#define __CDC_H + +#ifdef HAVE_LIB_Z +#include "zlib.h" +#endif + +#include <glusterfs/xlator.h> + +#ifndef MAX_IOVEC +#define MAX_IOVEC 16 +#endif + +typedef struct cdc_priv { + int window_size; + int mem_level; + int cdc_level; + int min_file_size; + int op_mode; + gf_boolean_t debug; + gf_lock_t lock; +} cdc_priv_t; + +typedef struct cdc_info { + /* input bits */ + int count; + int32_t ibytes; + struct iovec *vector; + struct iatt *buf; + + /* output bits */ + int ncount; + int nbytes; + int buffer_size; + struct iovec vec[MAX_IOVEC]; + struct iobref *iobref; + + /* zlib bits */ +#ifdef HAVE_LIB_Z + z_stream stream; +#endif + unsigned long crc; +} cdc_info_t; + +#define NVEC(ci) (ci->ncount - 1) +#define CURR_VEC(ci) ci->vec[ci->ncount - 1] +#define THIS_VEC(ci, i) ci->vector[i] + +/* Gzip defaults */ +#define GF_CDC_DEF_WINDOWSIZE -15 /* default value */ +#define GF_CDC_MAX_WINDOWSIZE -8 /* max value */ + +#ifdef HAVE_LIB_Z +#define GF_CDC_DEF_COMPRESSION Z_DEFAULT_COMPRESSION +#else +#define GF_CDC_DEF_COMPRESSION -1 +#endif + +#define GF_CDC_DEF_MEMLEVEL 8 +#define GF_CDC_DEF_BUFFERSIZE 262144 // 256K - default compression buffer size + +/* Operation mode + * If xlator is loaded on client, readv decompresses and writev compresses + * If xlator is loaded on server, readv compresses and writev decompresses + */ +#define GF_CDC_MODE_CLIENT 0 +#define GF_CDC_MODE_SERVER 1 + +/* min size of data to do cmpression + * 0 == compress even 1byte + */ +#define GF_CDC_MIN_CHUNK_SIZE 0 + +#define GF_CDC_VALIDATION_SIZE 8 + +#define GF_CDC_OS_ID 0xFF +#define GF_CDC_DEFLATE_CANARY_VAL "deflate" +#define GF_CDC_DEBUG_DUMP_FILE "/tmp/cdcdump.gz" + +#define GF_CDC_MODE_IS_CLIENT(m) (strcmp(m, "client") == 0) + +#define GF_CDC_MODE_IS_SERVER(m) (strcmp(m, "server") == 0) + +int32_t +cdc_compress(xlator_t *this, cdc_priv_t *priv, cdc_info_t *ci, dict_t **xdata); +int32_t +cdc_decompress(xlator_t *this, cdc_priv_t *priv, cdc_info_t *ci, dict_t *xdata); + +#endif |