4 files changed, 208 insertions, 0 deletions
diff --git a/src/compressor/snappy/CMakeLists.txt b/src/compressor/snappy/CMakeLists.txt
new file mode 100644
index 00000000..0bb233d4
--- /dev/null
+++ b/src/compressor/snappy/CMakeLists.txt
@@ -0,0 +1,13 @@
+# snappy
+
+set(snappy_sources
+  CompressionPluginSnappy.cc
+)
+
+add_library(ceph_snappy SHARED ${snappy_sources})
+target_link_libraries(ceph_snappy PRIVATE snappy::snappy)
+set_target_properties(ceph_snappy PROPERTIES
+  VERSION 2.0.0
+  SOVERSION 2
+  INSTALL_RPATH "")
+install(TARGETS ceph_snappy DESTINATION ${compressor_plugin_dir})
diff --git a/src/compressor/snappy/CompressionPluginSnappy.cc b/src/compressor/snappy/CompressionPluginSnappy.cc
new file mode 100644
index 00000000..96abf84f
--- /dev/null
+++ b/src/compressor/snappy/CompressionPluginSnappy.cc
@@ -0,0 +1,38 @@
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2015 Mirantis, Inc.
+ *
+ * Author: Alyona Kiseleva <akiselyova@mirantis.com>
+ *
+ *  This library is free software; you can redistribute it and/or
+ *  modify it under the terms of the GNU Lesser General Public
+ *  License as published by the Free Software Foundation; either
+ *  version 2.1 of the License, or (at your option) any later version.
+ *
+ */
+
+
+// -----------------------------------------------------------------------------
+#include "acconfig.h"
+#include "ceph_ver.h"
+#include "common/ceph_context.h"
+#include "CompressionPluginSnappy.h"
+
+// -----------------------------------------------------------------------------
+
+const char *__ceph_plugin_version()
+{
+  return CEPH_GIT_NICE_VER;
+}
+
+// -----------------------------------------------------------------------------
+
+int __ceph_plugin_init(CephContext *cct,
+                       const std::string& type,
+                       const std::string& name)
+{
+  PluginRegistry *instance = cct->get_plugin_registry();
+
+  return instance->add(type, name, new CompressionPluginSnappy(cct));
+}
diff --git a/src/compressor/snappy/CompressionPluginSnappy.h b/src/compressor/snappy/CompressionPluginSnappy.h
new file mode 100644
index 00000000..19b69765
--- /dev/null
+++ b/src/compressor/snappy/CompressionPluginSnappy.h
@@ -0,0 +1,42 @@
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2015 Mirantis, Inc.
+ *
+ * Author: Alyona Kiseleva <akiselyova@mirantis.com>
+ *
+ *  This library is free software; you can redistribute it and/or
+ *  modify it under the terms of the GNU Lesser General Public
+ *  License as published by the Free Software Foundation; either
+ *  version 2.1 of the License, or (at your option) any later version.
+ *
+ */
+
+#ifndef CEPH_COMPRESSION_PLUGIN_SNAPPY_H
+#define CEPH_COMPRESSION_PLUGIN_SNAPPY_H
+
+// -----------------------------------------------------------------------------
+#include "compressor/CompressionPlugin.h"
+#include "SnappyCompressor.h"
+// -----------------------------------------------------------------------------
+
+class CompressionPluginSnappy : public CompressionPlugin {
+
+public:
+
+  explicit CompressionPluginSnappy(CephContext* cct) : CompressionPlugin(cct)
+  {}
+
+  int factory(CompressorRef *cs,
+                      std::ostream *ss) override
+  {
+    if (compressor == 0) {
+      SnappyCompressor *interface = new SnappyCompressor(cct);
+      compressor = CompressorRef(interface);
+    }
+    *cs = compressor;
+    return 0;
+  }
+};
+
+#endif
diff --git a/src/compressor/snappy/SnappyCompressor.h b/src/compressor/snappy/SnappyCompressor.h
new file mode 100644
index 00000000..0291a923
--- /dev/null
+++ b/src/compressor/snappy/SnappyCompressor.h
@@ -0,0 +1,115 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2015 Haomai Wang <haomaiwang@gmail.com>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#ifndef CEPH_SNAPPYCOMPRESSOR_H
+#define CEPH_SNAPPYCOMPRESSOR_H
+
+#include <snappy.h>
+#include <snappy-sinksource.h>
+#include "common/config.h"
+#include "compressor/Compressor.h"
+#include "include/buffer.h"
+
+class CEPH_BUFFER_API BufferlistSource : public snappy::Source {
+  bufferlist::const_iterator pb;
+  size_t remaining;
+
+ public:
+  explicit BufferlistSource(bufferlist::const_iterator _pb, size_t _input_len)
+    : pb(_pb),
+      remaining(_input_len) {
+    remaining = std::min(remaining, (size_t)pb.get_remaining());
+  }
+  size_t Available() const override {
+    return remaining;
+  }
+  const char *Peek(size_t *len) override {
+    const char *data = NULL;
+    *len = 0;
+    size_t avail = Available();
+    if (avail) {
+      auto ptmp = pb;
+      *len = ptmp.get_ptr_and_advance(avail, &data);
+    }
+    return data;
+  }
+  void Skip(size_t n) override {
+    ceph_assert(n <= remaining);
+    pb.advance(n);
+    remaining -= n;
+  }
+
+  bufferlist::const_iterator get_pos() const {
+    return pb;
+  }
+};
+
+class SnappyCompressor : public Compressor {
+ public:
+  SnappyCompressor(CephContext* cct) : Compressor(COMP_ALG_SNAPPY, "snappy") {
+#ifdef HAVE_QATZIP
+    if (cct->_conf->qat_compressor_enabled && qat_accel.init("snappy"))
+      qat_enabled = true;
+    else
+      qat_enabled = false;
+#endif
+  }
+
+  int compress(const bufferlist &src, bufferlist &dst) override {
+#ifdef HAVE_QATZIP
+    if (qat_enabled)
+      return qat_accel.compress(src, dst);
+#endif
+    BufferlistSource source(const_cast<bufferlist&>(src).begin(), src.length());
+    bufferptr ptr = buffer::create_small_page_aligned(
+      snappy::MaxCompressedLength(src.length()));
+    snappy::UncheckedByteArraySink sink(ptr.c_str());
+    snappy::Compress(&source, &sink);
+    dst.append(ptr, 0, sink.CurrentDestination() - ptr.c_str());
+    return 0;
+  }
+
+  int decompress(const bufferlist &src, bufferlist &dst) override {
+#ifdef HAVE_QATZIP
+    if (qat_enabled)
+      return qat_accel.decompress(src, dst);
+#endif
+    auto i = src.begin();
+    return decompress(i, src.length(), dst);
+  }
+
+  int decompress(bufferlist::const_iterator &p,
+		 size_t compressed_len,
+		 bufferlist &dst) override {
+#ifdef HAVE_QATZIP
+    if (qat_enabled)
+      return qat_accel.decompress(p, compressed_len, dst);
+#endif
+    snappy::uint32 res_len = 0;
+    BufferlistSource source_1(p, compressed_len);
+    if (!snappy::GetUncompressedLength(&source_1, &res_len)) {
+      return -1;
+    }
+    BufferlistSource source_2(p, compressed_len);
+    bufferptr ptr(res_len);
+    if (snappy::RawUncompress(&source_2, ptr.c_str())) {
+      p = source_2.get_pos();
+      dst.append(ptr);
+      return 0;
+    }
+    return -2;
+  }
+};
+
+#endif