src/ceph/src/compressor/zlib/ZlibCompressor.cc

   1 /*
   2  * Ceph - scalable distributed file system
   3  *
   4  * Copyright (C) 2015 Mirantis, Inc.
   5  *
   6  * Author: Alyona Kiseleva <akiselyova@mirantis.com>
   7  *
   8  *  This library is free software; you can redistribute it and/or
   9  *  modify it under the terms of the GNU Lesser General Public
  10  *  License as published by the Free Software Foundation; either
  11  *  version 2.1 of the License, or (at your option) any later version.
  12  *
  13  */
  14
  15 // -----------------------------------------------------------------------------
  16 #include "common/debug.h"
  17 #include "ZlibCompressor.h"
  18 #include "osd/osd_types.h"
  19 #include "isa-l/include/igzip_lib.h"
  20 // -----------------------------------------------------------------------------
  21
  22 #include <zlib.h>
  23
  24 // -----------------------------------------------------------------------------
  25 #define dout_context cct
  26 #define dout_subsys ceph_subsys_compressor
  27 #undef dout_prefix
  28 #define dout_prefix _prefix(_dout)
  29 // -----------------------------------------------------------------------------
  30
  31 // -----------------------------------------------------------------------------
  32
  33 static ostream&
  34 _prefix(std::ostream* _dout)
  35 {
  36   return *_dout << "ZlibCompressor: ";
  37 }
  38 // -----------------------------------------------------------------------------
  39
  40 #define MAX_LEN (CEPH_PAGE_SIZE)
  41
  42 // default window size for Zlib 1.2.8, negated for raw deflate
  43 #define ZLIB_DEFAULT_WIN_SIZE -15
  44
  45 // desired memory usage level. increasing to 9 doesn't speed things up
  46 // significantly (helps only on >=16K blocks) and sometimes degrades
  47 // compression ratio.
  48 #define ZLIB_MEMORY_LEVEL 8
  49
  50 int ZlibCompressor::zlib_compress(const bufferlist &in, bufferlist &out)
  51 {
  52   int ret;
  53   unsigned have;
  54   z_stream strm;
  55   unsigned char* c_in;
  56   int begin = 1;
  57
  58   /* allocate deflate state */
  59   strm.zalloc = Z_NULL;
  60   strm.zfree = Z_NULL;
  61   strm.opaque = Z_NULL;
  62   ret = deflateInit2(&strm, cct->_conf->compressor_zlib_level, Z_DEFLATED, ZLIB_DEFAULT_WIN_SIZE, ZLIB_MEMORY_LEVEL, Z_DEFAULT_STRATEGY);
  63   if (ret != Z_OK) {
  64     dout(1) << "Compression init error: init return "
  65          << ret << " instead of Z_OK" << dendl;
  66     return -1;
  67   }
  68
  69   for (std::list<buffer::ptr>::const_iterator i = in.buffers().begin();
  70       i != in.buffers().end();) {
  71
  72     c_in = (unsigned char*) (*i).c_str();
  73     long unsigned int len = (*i).length();
  74     ++i;
  75
  76     strm.avail_in = len;
  77     int flush = i != in.buffers().end() ? Z_NO_FLUSH : Z_FINISH;
  78
  79     strm.next_in = c_in;
  80     do {
  81       bufferptr ptr = buffer::create_page_aligned(MAX_LEN);
  82       strm.next_out = (unsigned char*)ptr.c_str() + begin;
  83       strm.avail_out = MAX_LEN - begin;
  84       if (begin) {
  85         // put a compressor variation mark in front of compressed stream, not used at the moment
  86         ptr.c_str()[0] = 0;
  87         begin = 0;
  88       }
  89       ret = deflate(&strm, flush);    /* no bad return value */
  90       if (ret == Z_STREAM_ERROR) {
  91          dout(1) << "Compression error: compress return Z_STREAM_ERROR("
  92               << ret << ")" << dendl;
  93          deflateEnd(&strm);
  94          return -1;
  95       }
  96       have = MAX_LEN - strm.avail_out;
  97       out.append(ptr, 0, have);
  98     } while (strm.avail_out == 0);
  99     if (strm.avail_in != 0) {
 100       dout(10) << "Compression error: unused input" << dendl;
 101       deflateEnd(&strm);
 102       return -1;
 103     }
 104   }
 105
 106   deflateEnd(&strm);
 107   return 0;
 108 }
 109
 110 #if __x86_64__ && defined(HAVE_BETTER_YASM_ELF64)
 111 int ZlibCompressor::isal_compress(const bufferlist &in, bufferlist &out)
 112 {
 113   int ret;
 114   unsigned have;
 115   isal_zstream strm;
 116   unsigned char* c_in;
 117   int begin = 1;
 118
 119   /* allocate deflate state */
 120   isal_deflate_init(&strm);
 121   strm.end_of_stream = 0;
 122
 123   for (std::list<buffer::ptr>::const_iterator i = in.buffers().begin();
 124       i != in.buffers().end();) {
 125
 126     c_in = (unsigned char*) (*i).c_str();
 127     long unsigned int len = (*i).length();
 128     ++i;
 129
 130     strm.avail_in = len;
 131     strm.end_of_stream = (i == in.buffers().end());
 132     strm.flush = FINISH_FLUSH;
 133
 134     strm.next_in = c_in;
 135
 136     do {
 137       bufferptr ptr = buffer::create_page_aligned(MAX_LEN);
 138       strm.next_out = (unsigned char*)ptr.c_str() + begin;
 139       strm.avail_out = MAX_LEN - begin;
 140       if (begin) {
 141         // put a compressor variation mark in front of compressed stream, not used at the moment
 142         ptr.c_str()[0] = 1;
 143         begin = 0;
 144       }
 145       ret = isal_deflate(&strm);
 146       if (ret != COMP_OK) {
 147          dout(1) << "Compression error: isal_deflate return error ("
 148               << ret << ")" << dendl;
 149          return -1;
 150       }
 151       have = MAX_LEN - strm.avail_out;
 152       out.append(ptr, 0, have);
 153     } while (strm.avail_out == 0);
 154     if (strm.avail_in != 0) {
 155       dout(10) << "Compression error: unused input" << dendl;
 156       return -1;
 157     }
 158   }
 159
 160   return 0;
 161 }
 162 #endif
 163
 164 int ZlibCompressor::compress(const bufferlist &in, bufferlist &out)
 165 {
 166 #if __x86_64__ && defined(HAVE_BETTER_YASM_ELF64)
 167   if (isal_enabled)
 168     return isal_compress(in, out);
 169   else
 170     return zlib_compress(in, out);
 171 #else
 172   return zlib_compress(in, out);
 173 #endif
 174 }
 175
 176 int ZlibCompressor::decompress(bufferlist::iterator &p, size_t compressed_size, bufferlist &out)
 177 {
 178   int ret;
 179   unsigned have;
 180   z_stream strm;
 181   const char* c_in;
 182   int begin = 1;
 183
 184   /* allocate inflate state */
 185   strm.zalloc = Z_NULL;
 186   strm.zfree = Z_NULL;
 187   strm.opaque = Z_NULL;
 188   strm.avail_in = 0;
 189   strm.next_in = Z_NULL;
 190
 191   // choose the variation of compressor
 192   ret = inflateInit2(&strm, ZLIB_DEFAULT_WIN_SIZE);
 193   if (ret != Z_OK) {
 194     dout(1) << "Decompression init error: init return "
 195          << ret << " instead of Z_OK" << dendl;
 196     return -1;
 197   }
 198
 199   size_t remaining = MIN(p.get_remaining(), compressed_size);
 200
 201   while(remaining) {
 202     long unsigned int len = p.get_ptr_and_advance(remaining, &c_in);
 203     remaining -= len;
 204     strm.avail_in = len - begin;
 205     strm.next_in = (unsigned char*)c_in + begin;
 206     begin = 0;
 207
 208     do {
 209       strm.avail_out = MAX_LEN;
 210       bufferptr ptr = buffer::create_page_aligned(MAX_LEN);
 211       strm.next_out = (unsigned char*)ptr.c_str();
 212       ret = inflate(&strm, Z_NO_FLUSH);
 213       if (ret != Z_OK && ret != Z_STREAM_END && ret != Z_BUF_ERROR) {
 214        dout(1) << "Decompression error: decompress return "
 215             << ret << dendl;
 216        inflateEnd(&strm);
 217        return -1;
 218       }
 219       have = MAX_LEN - strm.avail_out;
 220       out.append(ptr, 0, have);
 221     } while (strm.avail_out == 0);
 222   }
 223
 224   /* clean up and return */
 225   (void)inflateEnd(&strm);
 226   return 0;
 227 }
 228
 229 int ZlibCompressor::decompress(const bufferlist &in, bufferlist &out)
 230 {
 231   bufferlist::iterator i = const_cast<bufferlist&>(in).begin();
 232   return decompress(i, in.length(), out);
 233 }