Merge branch 'wip-compressor'

Reviewed-by: Sage Weil <sage@redhat.com>

ceph.spec.in

@@ -845,6 +845,8 @@ rm -rf $RPM_BUILD_ROOT
 %{_libdir}/rados-classes/libcls_journal.so*
 %dir %{_libdir}/ceph/erasure-code
 %{_libdir}/ceph/erasure-code/libec_*.so*
+%dir %{_libdir}/ceph/compressor
+%{_libdir}/ceph/compressor/libceph_*.so*
 %if 0%{?_with_lttng}
 %{_libdir}/libos_tp.so*
 %{_libdir}/libosd_tp.so*

qa/workunits/ceph-helpers.sh

@@ -310,6 +310,7 @@ function run_mon() {
         --paxos-propose-interval=0.1 \
         --osd-crush-chooseleaf-type=0 \
         --erasure-code-dir=.libs \
+        --plugin-dir=.libs \
         --debug-mon 20 \
         --debug-ms 20 \
         --debug-paxos 20 \
@@ -542,6 +543,7 @@ function activate_osd() {
     ceph_args+=" --osd-data=$osd_data"
     ceph_args+=" --chdir="
     ceph_args+=" --erasure-code-dir=.libs"
+    ceph_args+=" --plugin-dir=.libs"
     ceph_args+=" --osd-class-dir=.libs"
     ceph_args+=" --run-dir=$dir"
     ceph_args+=" --debug-osd=20"

src/Makefile-env.am

@@ -201,7 +201,7 @@ LIBPERFGLUE = libperfglue.la
 LIBAUTH = libauth.la
 LIBMSG = libmsg.la
 LIBCRUSH = libcrush.la
-LIBCOMPRESSOR = libcompressor.la -lsnappy
+LIBCOMPRESSOR = libcompressor.la
 LIBJSON_SPIRIT = libjson_spirit.la
 LIBKV = libkv.a
 LIBLOG = liblog.la

src/Makefile.am

@@ -25,6 +25,7 @@ include mds/Makefile.am
 include os/Makefile.am
 include osd/Makefile.am
 include erasure-code/Makefile.am
+include compressor/Makefile.am
 include osdc/Makefile.am
 include client/Makefile.am
 include global/Makefile.am
@@ -46,7 +47,6 @@ include rbd_replay/Makefile.am
 include test/Makefile.am
 include tools/Makefile.am
 include Makefile-rocksdb.am
-include compressor/Makefile.am
 include tracing/Makefile.am
 include pybind/Makefile.am
 

src/common/Makefile.am

@@ -157,6 +157,7 @@ noinst_HEADERS += \
 # important; libmsg before libauth!
 LIBCOMMON_DEPS += \
 	$(LIBERASURE_CODE) \
+	$(LIBCOMPRESSOR) \
 	$(LIBMSG) $(LIBAUTH) \
 	$(LIBCRUSH) $(LIBJSON_SPIRIT) $(LIBLOG) $(LIBARCH) \
 	$(BOOST_RANDOM_LIBS)

src/common/config_opts.h

@@ -39,6 +39,7 @@ OPTION(restapi_log_level, OPT_STR, "") 	// default set by Python code
 OPTION(restapi_base_url, OPT_STR, "")	// "
 OPTION(fatal_signal_handlers, OPT_BOOL, true)
 OPTION(erasure_code_dir, OPT_STR, CEPH_PKGLIBDIR"/erasure-code") // default location for erasure-code plugins
+OPTION(compression_dir, OPT_STR, CEPH_PKGLIBDIR"/compressor") // default location for compression plugins
 
 OPTION(log_file, OPT_STR, "/var/log/ceph/$cluster-$name.log") // default changed by common_preinit()
 OPTION(log_max_new, OPT_INT, 1000) // default changed by common_preinit()
@@ -575,6 +576,9 @@ OPTION(osd_pool_default_size, OPT_INT, 3)
 OPTION(osd_pool_default_min_size, OPT_INT, 0)  // 0 means no specific default; ceph will use size-size/2
 OPTION(osd_pool_default_pg_num, OPT_INT, 8) // number of PGs for new pools. Configure in global or mon section of ceph.conf
 OPTION(osd_pool_default_pgp_num, OPT_INT, 8) // number of PGs for placement purposes. Should be equal to pg_num
+OPTION(osd_compression_plugins, OPT_STR,
+       "snappy"
+       ) // list of compression plugins
 OPTION(osd_pool_default_erasure_code_profile,
        OPT_STR,
        "plugin=jerasure "

src/compressor/AsyncCompressor.cc

@@ -21,9 +21,9 @@
 #define dout_prefix *_dout << "compressor "
 
 AsyncCompressor::AsyncCompressor(CephContext *c):
-  compressor(Compressor::create(c->_conf->async_compressor_type)), cct(c),
+  compressor(Compressor::create(c, c->_conf->async_compressor_type)), cct(c),
   job_id(0),
-  compress_tp(g_ceph_context, "AsyncCompressor::compressor_tp", "tp_async_compr", cct->_conf->async_compressor_threads, "async_compressor_threads"),
+  compress_tp(cct, "AsyncCompressor::compressor_tp", "tp_async_compr", cct->_conf->async_compressor_threads, "async_compressor_threads"),
   job_lock("AsyncCompressor::job_lock"),
   compress_wq(this, c->_conf->async_compressor_thread_timeout, c->_conf->async_compressor_thread_suicide_timeout, &compress_tp) {
 }

src/compressor/AsyncCompressor.h

@@ -25,7 +25,7 @@
 
 class AsyncCompressor {
  private:
-  Compressor *compressor;
+  CompressorRef compressor;
   CephContext *cct;
   atomic_t job_id;
   vector<int> coreids;

src/compressor/CMakeLists.txt

@@ -0,0 +1,10 @@
+## compressor plugins
+
+set(compressorlibdir ${LIBRARY_OUTPUT_PATH}/compressor)
+
+add_subdirectory(snappy)
+
+add_library(compressor_objs OBJECT Compressor.cc)
+
+add_custom_target(compressor_plugins DEPENDS
+    ceph_snappy)

src/compressor/CompressionPlugin.h

@@ -0,0 +1,45 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph distributed storage 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 COMPRESSION_PLUGIN_H
+#define COMPRESSION_PLUGIN_H
+
+#include "common/Mutex.h"
+#include "common/PluginRegistry.h"
+#include "Compressor.h"
+
+namespace ceph {
+
+  class CompressionPlugin :  public Plugin {
+  public:
+    CompressorRef compressor;
+
+    CompressionPlugin(CephContext *cct) : Plugin(cct),
+                                          compressor(0) 
+    {}
+    
+    virtual ~CompressionPlugin() {}
+
+    virtual int factory(CompressorRef *cs,
+			                  ostream *ss) = 0;
+
+    virtual const char* name() {return "CompressionPlugin";}
+  };
+
+}
+
+#endif

src/compressor/Compressor.cc

@@ -13,13 +13,21 @@
  */
 
 #include "Compressor.h"
-#include "SnappyCompressor.h"
+#include "CompressionPlugin.h"
 
 
-Compressor* Compressor::create(const string &type)
+CompressorRef Compressor::create(CephContext *cct, const string &type)
 {
-  if (type == "snappy")
-    return new SnappyCompressor();
-
-  assert(0);
+  CompressorRef cs_impl = NULL;
+  stringstream ss;
+  PluginRegistry *reg = cct->get_plugin_registry();
+  CompressionPlugin *factory = dynamic_cast<CompressionPlugin*>(reg->get_with_load("compressor", type));
+  if (factory == NULL) {
+    lderr(cct) << __func__ << " cannot load compressor of type " << type << dendl;
+    return NULL;
+  }
+  int err = factory->factory(&cs_impl, &ss);
+  if (err)
+    lderr(cct) << __func__ << " factory return error " << err << dendl;
+  return cs_impl;
 }

src/compressor/Compressor.h

@@ -18,13 +18,19 @@
 #include "include/int_types.h"
 #include "include/Context.h"
 
+class Compressor;
+typedef shared_ptr<Compressor> CompressorRef;
+
+
 class Compressor {
  public:
   virtual ~Compressor() {}
   virtual int compress(bufferlist &in, bufferlist &out) = 0;
   virtual int decompress(bufferlist &in, bufferlist &out) = 0;
 
-  static Compressor *create(const string &type);
+  static CompressorRef create(CephContext *cct, const string &type);
 };
 
+
+
 #endif

src/compressor/Makefile.am

@@ -1,11 +1,20 @@
+compressorlibdir = $(pkglibdir)/compressor
+compressorlib_LTLIBRARIES =  
+
+include compressor/snappy/Makefile.am
+
 libcompressor_la_SOURCES = \
 	compressor/Compressor.cc \
 	compressor/AsyncCompressor.cc
+compressor/CompressionPlugin.cc: ./ceph_ver.h
+libcompressor_la_DEPENDENCIES = $(compressorlib_LTLIBRARIES)
+if LINUX
+libcompressor_la_LIBADD = -ldl
+endif # LINUX
 noinst_LTLIBRARIES += libcompressor.la
 
-libcompressor_la_LIBADD = $(LIBCOMMON)
-
 noinst_HEADERS += \
 	compressor/Compressor.h \
 	compressor/AsyncCompressor.h \
-	compressor/SnappyCompressor.h
+    compressor/CompressionPlugin.h
+

src/compressor/snappy/CMakeLists.txt

@@ -0,0 +1,13 @@
+# snappy
+
+set(snappy_sources
+  CompressionPluginSnappy.cc
+  $<TARGET_OBJECTS:compressor_objs>
+)
+
+add_library(ceph_snappy SHARED ${snappy_sources})
+add_dependencies(ceph_snappy ${CMAKE_SOURCE_DIR}/src/ceph_ver.h)
+target_link_libraries(ceph_snappy ${EXTRALIBS})
+set_target_properties(ceph_snappy PROPERTIES VERSION 2.14.0 SOVERSION 2)
+set(CMAKE_SHARED_LINKER_FLAGS "${CMAKE_SHARED_LINKER_FLAGS} -lsnappy")
+install(TARGETS ceph_snappy DESTINATION lib/compressor)

src/compressor/snappy/CompressionPluginSnappy.cc

@@ -0,0 +1,57 @@
+/*
+ * 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 "ceph_ver.h"
+#include "compressor/CompressionPlugin.h"
+#include "SnappyCompressor.h"
+// -----------------------------------------------------------------------------
+
+class CompressionPluginSnappy : public CompressionPlugin {
+
+public:
+
+  CompressionPluginSnappy(CephContext* cct) : CompressionPlugin(cct)
+  {}
+
+  virtual int factory(CompressorRef *cs,
+                      ostream *ss)
+  {
+    if (compressor == 0) {
+      SnappyCompressor *interface = new SnappyCompressor();
+      compressor = CompressorRef(interface);
+    }
+    *cs = compressor;
+    return 0;
+  }
+};
+
+// -----------------------------------------------------------------------------
+
+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));
+}

src/compressor/snappy/Makefile.am

@@ -0,0 +1,22 @@
+# snappy plugin
+noinst_HEADERS += \
+  compressor/snappy/SnappyCompressor.h
+
+snappy_sources = \
+  compressor/Compressor.cc \
+  compressor/snappy/CompressionPluginSnappy.cc
+
+compressor/snappy/CompressionPluginSnappy.cc: ./ceph_ver.h
+
+libceph_snappy_la_SOURCES = ${snappy_sources}
+libceph_snappy_la_CFLAGS = ${AM_CFLAGS}  \
+  -I$(srcdir)/compressor/snappy/snappy-1.2.8
+libceph_snappy_la_CXXFLAGS= ${AM_CXXFLAGS} \
+  -I$(srcdir)/compressor/snappy/snappy-1.2.8
+libceph_snappy_la_LIBADD = $(LIBCRUSH) $(PTHREAD_LIBS) $(EXTRALIBS)
+libceph_snappy_la_LDFLAGS = ${AM_LDFLAGS} -lsnappy -version-info 2:0:0
+if LINUX
+libceph_snappy_la_LDFLAGS += -export-symbols-regex '.*__compressor_.*'
+endif
+
+compressorlib_LTLIBRARIES += libceph_snappy.la

src/compressor/snappy/SnappyCompressor.h

@@ -18,7 +18,7 @@
 #include <snappy.h>
 #include <snappy-sinksource.h>
 #include "include/buffer.h"
-#include "Compressor.h"
+#include "compressor/Compressor.h"
 
 class BufferlistSource : public snappy::Source {
   list<bufferptr>::const_iterator pb;
@@ -52,6 +52,7 @@ class BufferlistSource : public snappy::Source {
 class SnappyCompressor : public Compressor {
  public:
   virtual ~SnappyCompressor() {}
+  virtual const char* get_method_name() { return "snappy"; }
   virtual int compress(bufferlist &src, bufferlist &dst) {
     BufferlistSource source(src);
     bufferptr ptr(snappy::MaxCompressedLength(src.length()));

src/test/Makefile.am

@@ -1,5 +1,6 @@
 include test/erasure-code/Makefile.am
 include test/messenger/Makefile.am
+include test/compressor/Makefile.am
 
 if ENABLE_CLIENT
 include test/Makefile-client.am
@@ -425,7 +426,7 @@ check_PROGRAMS += unittest_subprocess
 
 unittest_async_compressor_SOURCES = test/common/test_async_compressor.cc
 unittest_async_compressor_CXXFLAGS = $(UNITTEST_CXXFLAGS)
-unittest_async_compressor_LDADD = $(UNITTEST_LDADD) $(CEPH_GLOBAL) $(LIBCOMPRESSOR)
+unittest_async_compressor_LDADD = $(UNITTEST_LDADD) $(CEPH_GLOBAL) $(LIBCOMPRESSOR) $(LIBCOMMON)
 check_PROGRAMS += unittest_async_compressor
 
 check_SCRIPTS += test/pybind/test_ceph_argparse.py

src/test/ceph-disk.sh

@@ -49,6 +49,7 @@ CEPH_ARGS+=" --log-file=$DIR/\$name.log"
 CEPH_ARGS+=" --pid-file=$DIR/\$name.pidfile"
 if test -d .libs ; then
     CEPH_ARGS+=" --erasure-code-dir=.libs"
+    CEPH_ARGS+=" --compression-dir=.libs"
 fi
 CEPH_ARGS+=" --auth-supported=none"
 CEPH_ARGS+=" --osd-journal-size=100"

src/test/compressor/Makefile.am

@@ -0,0 +1,49 @@
+if ENABLE_SERVER
+if WITH_OSD
+
+
+libceph_example_la_SOURCES = \
+	compressor/Compressor.cc \
+	test/compressor/compressor_plugin_example.cc
+noinst_HEADERS += test/compressor/compressor_example.h
+test/compressor/compressor_plugin_example.cc: ./ceph_ver.h
+libceph_example_la_CFLAGS = ${AM_CFLAGS}
+libceph_example_la_CXXFLAGS= ${AM_CXXFLAGS}
+libceph_example_la_LIBADD = $(LIBCRUSH) $(PTHREAD_LIBS) $(EXTRALIBS)
+libceph_example_la_LDFLAGS = ${AM_LDFLAGS} -export-symbols-regex '.*__compression_.*'
+compressorlib_LTLIBRARIES += libceph_example.la
+
+unittest_compression_plugin_SOURCES = \
+	compressor/Compressor.cc \
+	test/compressor/test_compression_plugin.cc 
+unittest_compression_plugin_CXXFLAGS = $(UNITTEST_CXXFLAGS)
+unittest_compression_plugin_LDADD = $(LIBOSD) $(LIBCOMMON) $(UNITTEST_LDADD) $(CEPH_GLOBAL)
+if LINUX
+unittest_compression_plugin_LDADD += -ldl
+endif
+check_TESTPROGRAMS += unittest_compression_plugin
+
+unittest_compression_snappy_SOURCES = \
+	test/compressor/test_compression_snappy.cc \
+	${snappy_sources}
+unittest_compression_snappy_CXXFLAGS = $(UNITTEST_CXXFLAGS)
+unittest_compression_snappy_LDADD = $(LIBOSD) $(LIBCOMMON) $(UNITTEST_LDADD) $(CEPH_GLOBAL)
+unittest_compression_snappy_LDFLAGS = -lsnappy
+if LINUX
+unittest_compression_snappy_LDADD += -ldl
+endif
+check_TESTPROGRAMS += unittest_compression_snappy
+
+unittest_compression_plugin_snappy_SOURCES = \
+	test/compressor/test_compression_plugin_snappy.cc \
+	${snappy_sources}
+unittest_compression_plugin_snappy_CXXFLAGS = ${AM_CXXFLAGS} ${UNITTEST_CXXFLAGS}
+unittest_compression_plugin_snappy_LDADD = $(LIBOSD) $(LIBCOMMON) $(UNITTEST_LDADD) $(CEPH_GLOBAL) $(LIBCOMPRESSOR)
+unittest_compression_plugin_snappy_LDFLAGS = -lsnappy
+if LINUX
+unittest_compression_plugin_snappy_LDADD += -ldl
+endif
+check_TESTPROGRAMS += unittest_compression_plugin_snappy
+
+endif # WITH_OSD
+endif # ENABLE_SERVER

src/test/compressor/compressor_example.h

@@ -0,0 +1,53 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph distributed storage 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_COMPRESSOR_EXAMPLE_H
+#define CEPH_COMPRESSOR_EXAMPLE_H
+
+#include <unistd.h>
+#include <errno.h>
+#include <algorithm>
+#include <sstream>
+
+#include "crush/CrushWrapper.h"
+#include "osd/osd_types.h"
+#include "compressor/Compressor.h"
+
+class CompressorExample : public Compressor {
+public:
+  virtual ~CompressorExample() {}
+
+  virtual int compress(bufferlist &in, bufferlist &out)
+  {
+    out = in;
+    return 0;
+  }
+
+  virtual int decompress(bufferlist &in, bufferlist &out)
+  {
+    out = in;
+    return 0;
+  }
+
+  virtual const char* get_method_name()
+  {
+    return "example";
+  }
+
+};
+
+#endif

src/test/compressor/compressor_plugin_example.cc

@@ -0,0 +1,57 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph distributed storage 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 <unistd.h>
+
+#include "ceph_ver.h"
+#include "compressor/CompressionPlugin.h"
+#include "compressor_example.h"
+
+class CompressorPluginExample : public CompressionPlugin {
+public:
+
+  CompressorPluginExample(CephContext* cct) : CompressionPlugin(cct)
+  {}
+
+  virtual int factory(CompressorRef *cs,
+		      ostream *ss)
+  {
+    if (compressor == 0) {
+      CompressorExample *interface = new CompressorExample();
+      compressor = CompressorRef(interface);
+    }
+    *cs = compressor;
+    return 0;
+  }
+};
+
+// -----------------------------------------------------------------------------
+
+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 CompressorPluginExample(cct));
+}

src/test/compressor/test_compression_plugin.cc

@@ -0,0 +1,71 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph distributed storage 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 <errno.h>
+#include <signal.h>
+#include <stdlib.h>
+#include <gtest/gtest.h>
+#include "global/global_init.h"
+#include "compressor/CompressionPlugin.h"
+#include "common/ceph_argparse.h"
+#include "global/global_context.h"
+#include "common/config.h"
+
+TEST(CompressionPlugin, all)
+{
+  string directory(".libs");
+  CompressorRef compressor;
+  PluginRegistry *reg = g_ceph_context->get_plugin_registry();
+  EXPECT_TRUE(reg);
+  CompressionPlugin *factory = dynamic_cast<CompressionPlugin*>(reg->get_with_load("compressor", "invalid"));
+  EXPECT_FALSE(factory);
+  factory = dynamic_cast<CompressionPlugin*>(reg->get_with_load("compressor", "example"));
+  EXPECT_TRUE(factory);
+  stringstream ss;
+  EXPECT_EQ(0, factory->factory(&compressor, &ss));
+  EXPECT_TRUE(compressor.get());
+  {
+    Mutex::Locker l(reg->lock);
+    EXPECT_EQ(-ENOENT, reg->remove("compressor", "does not exist"));
+    EXPECT_EQ(0, reg->remove("compressor", "example"));
+    EXPECT_EQ(0, reg->load("compressor", "example"));
+  }
+}
+
+int main(int argc, char **argv) {
+  vector<const char*> args;
+  argv_to_vec(argc, (const char **)argv, args);
+
+  global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT, CODE_ENVIRONMENT_UTILITY, 0);
+  common_init_finish(g_ceph_context);
+
+  system("mkdir -p .libs/compressor");
+  system("cp .libs/libceph_example.so* .libs/compressor/");
+  g_conf->set_val("plugin_dir", ".libs", false, false);
+
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+
+/*
+ * Local Variables:
+ * compile-command: "cd ../.. ; make -j4 && 
+ *   make unittest_compression_plugin && 
+ *   valgrind --tool=memcheck \
+ *      ./unittest_compression_plugin \
+ *      --gtest_filter=*.* --log-to-stderr=true --debug-osd=20"
+ * End:
+ */

src/test/compressor/test_compression_plugin_snappy.cc

@@ -0,0 +1,58 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph distributed storage 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 <errno.h>
+#include <signal.h>
+#include <stdlib.h>
+#include <gtest/gtest.h>
+#include "global/global_init.h"
+#include "compressor/Compressor.h"
+#include "common/ceph_argparse.h"
+#include "global/global_context.h"
+#include "common/config.h"
+
+TEST(CompressionPluginSnappy, factory)
+{
+  CompressorRef compressor = Compressor::create(g_ceph_context, "snappy");
+  cout << compressor;
+  EXPECT_TRUE(compressor.get());
+}
+
+int main(int argc, char **argv) {
+  vector<const char*> args;
+  argv_to_vec(argc, (const char **)argv, args);
+
+  global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT, CODE_ENVIRONMENT_UTILITY, 0);
+  common_init_finish(g_ceph_context);
+
+  system("mkdir -p .libs/compressor");
+  system("cp .libs/libceph_snappy.so* .libs/compressor/");
+
+  g_conf->set_val("plugin_dir", ".libs", false, false);
+
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+
+/*
+ * Local Variables:
+ * compile-command: "cd ../.. ; make -j4 && 
+ *   make unittest_compression_plugin_snappy && 
+ *   valgrind --tool=memcheck \
+ *      ./unittest_compression_plugin_snappy \
+ *      --gtest_filter=*.* --log-to-stderr=true --debug-osd=20"
+ * End:
+ */

src/test/compressor/test_compression_snappy.cc

@@ -0,0 +1,62 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph distributed storage 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 <errno.h>
+#include <string.h>
+#include <gtest/gtest.h>
+#include "global/global_init.h"
+#include "compressor/snappy/SnappyCompressor.h"
+#include "common/ceph_argparse.h"
+#include "global/global_context.h"
+#include "common/config.h"
+
+TEST(SnappyCompressor, compress_decompress)
+{
+  SnappyCompressor sp;
+  EXPECT_EQ(sp.get_method_name(), "snappy");
+  char* test = "This is test text";
+  int len = strlen(test);
+  bufferlist in, out;
+  in.append(test, len);
+  int res = sp.compress(in, out);
+  EXPECT_EQ(res, 0);
+  bufferlist after;
+  res = sp.decompress(out, after);
+  EXPECT_EQ(res, 0);
+}
+
+int main(int argc, char **argv) {
+  vector<const char*> args;
+  argv_to_vec(argc, (const char **)argv, args);
+
+  global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT, CODE_ENVIRONMENT_UTILITY, 0);
+  common_init_finish(g_ceph_context);
+
+  g_conf->set_val("compression_dir", ".libs", false, false);
+
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+
+/*
+ * Local Variables:
+ * compile-command: "cd ../.. ; make -j4 && 
+ *   make unittest_compression_snappy && 
+ *   valgrind --tool=memcheck \
+ *      ./unittest_compression_snappy \
+ *      --gtest_filter=*.* --log-to-stderr=true --debug-osd=20"
+ * End:
+ */

src/test/mon/mkfs.sh

@@ -43,6 +43,7 @@ function mon_mkfs() {
         --id $MON_ID \
         --fsid $fsid \
         --erasure-code-dir=.libs \
+        --compression-dir=.libs \
         --mkfs \
         --mon-data=$MON_DIR \
         --mon-initial-members=$MON_ID \
@@ -57,6 +58,7 @@ function mon_run() {
         --mon-osd-full-ratio=.99 \
         --mon-data-avail-crit=1 \
         --erasure-code-dir=.libs \
+        --compression-dir=.libs \
         --mon-data=$MON_DIR \
         --log-file=$MON_DIR/log \
         --mon-cluster-log-file=$MON_DIR/log \
@@ -84,6 +86,7 @@ function auth_none() {
         --mon-osd-full-ratio=.99 \
         --mon-data-avail-crit=1 \
         --erasure-code-dir=.libs \
+        --compression-dir=.libs \
         --mon-data=$MON_DIR \
         --extract-monmap $MON_DIR/monmap
 
@@ -151,6 +154,7 @@ function makedir() {
         --mon-osd-full-ratio=.99 \
         --mon-data-avail-crit=1 \
         --erasure-code-dir=.libs \
+        --compression-dir=.libs \
         --mkfs \
         --mon-data=$toodeep 2>&1 | tee $DIR/makedir.log
     grep 'toodeep.*No such file' $DIR/makedir.log > /dev/null

src/vstart.sh

@@ -33,17 +33,31 @@ if [ -e CMakeCache.txt ]; then
     ln -sf ../${file} ec_plugins/`basename $file`
   done
   [ -z "$EC_PATH" ] && EC_PATH=./ec_plugins
+  # check for compression plugins
+  mkdir -p .libs/compressor
+  for file in ./src/compressor/*/libcs_*.so*;
+  do
+    ln -sf ../${file} .libs/compressor/`basename $file`
+  done
+else
+    mkdir -p .libs/compressor
+    for f in `ls -d compressor/*/`; 
+    do 
+        cp .libs/libceph_`basename $f`.so* .libs/compressor/;
+    done
 fi
 
 if [ -z "$CEPH_BUILD_ROOT" ]; then
         [ -z "$CEPH_BIN" ] && CEPH_BIN=.
         [ -z "$CEPH_LIB" ] && CEPH_LIB=.libs
         [ -z $EC_PATH ] && EC_PATH=$CEPH_LIB
+        [ -z $CS_PATH ] && CS_PATH=$CEPH_LIB
         [ -z $OBJCLASS_PATH ] && OBJCLASS_PATH=$CEPH_LIB
 else
         [ -z $CEPH_BIN ] && CEPH_BIN=$CEPH_BUILD_ROOT/bin
         [ -z $CEPH_LIB ] && CEPH_LIB=$CEPH_BUILD_ROOT/lib
         [ -z $EC_PATH ] && EC_PATH=$CEPH_LIB/erasure-code
+        [ -z $CS_PATH ] && CS_PATH=$CEPH_LIB/compressor
         [ -z $OBJCLASS_PATH ] && OBJCLASS_PATH=$CEPH_LIB/rados-classes
 fi
 
@@ -433,6 +447,7 @@ if [ "$start_mon" -eq 1 ]; then
         mon data avail warn = 10
         mon data avail crit = 1
         erasure code dir = $EC_PATH
+        plugin dir = $CS_PATH
         osd pool default erasure code profile = plugin=jerasure technique=reed_sol_van k=2 m=1 ruleset-failure-domain=osd
         rgw frontends = fastcgi, civetweb port=$CEPH_RGW_PORT
         rgw dns name = localhost