src/ceph/src/compressor/lz4/LZ4Compressor.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2017 Haomai Wang <haomaiwang@gmail.com>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #ifndef CEPH_LZ4COMPRESSOR_H
  16 #define CEPH_LZ4COMPRESSOR_H
  17
  18 #include <lz4.h>
  19
  20 #include "compressor/Compressor.h"
  21 #include "include/buffer.h"
  22 #include "include/encoding.h"
  23 #include "common/Tub.h"
  24
  25
  26 class LZ4Compressor : public Compressor {
  27  public:
  28   LZ4Compressor() : Compressor(COMP_ALG_LZ4, "lz4") {}
  29
  30   int compress(const bufferlist &src, bufferlist &dst) override {
  31     bufferptr outptr = buffer::create_page_aligned(
  32       LZ4_compressBound(src.length()));
  33     LZ4_stream_t lz4_stream;
  34     LZ4_resetStream(&lz4_stream);
  35
  36     auto p = src.begin();
  37     size_t left = src.length();
  38     int pos = 0;
  39     const char *data;
  40     unsigned num = src.get_num_buffers();
  41     uint32_t origin_len;
  42     uint32_t compressed_len;
  43     ::encode((uint32_t)num, dst);
  44     while (left) {
  45       origin_len = p.get_ptr_and_advance(left, &data);
  46       compressed_len = LZ4_compress_fast_continue(
  47         &lz4_stream, data, outptr.c_str()+pos, origin_len,
  48         outptr.length()-pos, 1);
  49       if (compressed_len <= 0)
  50         return -1;
  51       pos += compressed_len;
  52       left -= origin_len;
  53       ::encode(origin_len, dst);
  54       ::encode(compressed_len, dst);
  55     }
  56     assert(p.end());
  57
  58     dst.append(outptr, 0, pos);
  59     return 0;
  60   }
  61
  62   int decompress(const bufferlist &src, bufferlist &dst) override {
  63     bufferlist::iterator i = const_cast<bufferlist&>(src).begin();
  64     return decompress(i, src.length(), dst);
  65   }
  66
  67   int decompress(bufferlist::iterator &p,
  68                  size_t compressed_len,
  69                  bufferlist &dst) override {
  70     uint32_t count;
  71     std::vector<std::pair<uint32_t, uint32_t> > compressed_pairs;
  72     ::decode(count, p);
  73     compressed_pairs.resize(count);
  74     uint32_t total_origin = 0;
  75     for (unsigned i = 0; i < count; ++i) {
  76       ::decode(compressed_pairs[i].first, p);
  77       ::decode(compressed_pairs[i].second, p);
  78       total_origin += compressed_pairs[i].first;
  79     }
  80     compressed_len -= (sizeof(uint32_t) + sizeof(uint32_t) * count * 2);
  81
  82     bufferptr dstptr(total_origin);
  83     LZ4_streamDecode_t lz4_stream_decode;
  84     LZ4_setStreamDecode(&lz4_stream_decode, nullptr, 0);
  85
  86     bufferptr cur_ptr = p.get_current_ptr();
  87     bufferptr *ptr = &cur_ptr;
  88     Tub<bufferptr> data_holder;
  89     if (compressed_len != cur_ptr.length()) {
  90       data_holder.construct(compressed_len);
  91       p.copy_deep(compressed_len, *data_holder);
  92       ptr = data_holder.get();
  93     }
  94
  95     char *c_in = ptr->c_str();
  96     char *c_out = dstptr.c_str();
  97     for (unsigned i = 0; i < count; ++i) {
  98       int r = LZ4_decompress_safe_continue(
  99           &lz4_stream_decode, c_in, c_out, compressed_pairs[i].second, compressed_pairs[i].first);
 100       if (r == (int)compressed_pairs[i].first) {
 101         c_in += compressed_pairs[i].second;
 102         c_out += compressed_pairs[i].first;
 103       } else if (r < 0) {
 104         return -1;
 105       } else if (r != (int)compressed_pairs[i].first) {
 106         return -2;
 107       }
 108     }
 109     dst.push_back(std::move(dstptr));
 110     return 0;
 111   }
 112 };
 113
 114 #endif