From 3b59d0b0706dc54b4fdb8bcca06ceb35f497a7e3 Mon Sep 17 00:00:00 2001
From: Dai Zhiwei <daizhiwei3@huawei.com>
Date: Fri, 24 Jun 2022 19:33:54 +0800
Subject: [PATCH 1/2] add zlib uadk plugin

---
 CMakeLists.txt                          |   7 +
 cmake/modules/Finduadkzip.cmake         |  25 ++
 src/CMakeLists.txt                      |   4 +
 src/common/legacy_config_opts.h         |   1 +
 src/common/options.cc                   |   4 +
 src/compressor/CMakeLists.txt           |   9 +
 src/compressor/Compressor.h             |   9 +
 src/compressor/UadkCompressionAccel.cc  | 446 ++++++++++++++++++++++++
 src/compressor/UadkCompressionAccel.h   |  45 +++
 src/compressor/zlib/ZlibCompressor.cc   |  29 ++
 src/compressor/zlib/ZlibCompressor.h    |  12 +
 src/include/config-h.in.cmake           |   3 +
 src/test/compressor/test_compression.cc |  84 +++++
 13 files changed, 678 insertions(+)
 create mode 100644 cmake/modules/Finduadkzip.cmake
 create mode 100644 src/compressor/UadkCompressionAccel.cc
 create mode 100644 src/compressor/UadkCompressionAccel.h

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 5253399df5..19c4305784 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -279,6 +279,13 @@ if(WITH_QATZIP)
   set(HAVE_QATZIP ${QATZIP_FOUND})
 endif(WITH_QATZIP)
 
+option(WITH_UADKZIP "Enable UADKZIP" OFF)
+if(WITH_UADKZIP)
+  find_package(uadkzip REQUIRED)
+  set(HAVE_UADKZIP ${UADKZIP_FOUND})
+  message("HAVE_UADKZIP" ${HAVE_UADKZIP})
+endif(WITH_UADKZIP)
+
 # needs mds and? XXX
 option(WITH_LIBCEPHFS "libcephfs client library" ON)
 
diff --git a/cmake/modules/Finduadkzip.cmake b/cmake/modules/Finduadkzip.cmake
new file mode 100644
index 0000000000..ca3a699cde
--- /dev/null
+++ b/cmake/modules/Finduadkzip.cmake
@@ -0,0 +1,25 @@
+# - Find uadkzip
+# Find the uadkzip compression library and includes
+#
+# UADK_INCLUDE_DIR - where to find uadkzip.h, etc.
+# UADK_LIBRARIES - List of libraries when using uadkzip.
+# UADK_FOUND - True if uadkzip found.
+
+find_path(UADKZIP_INCLUDE_DIR NAMES uadk/wd_comp.h)
+message("UADKZIP_INCLUDE_DIR: " ${UADKZIP_INCLUDE_DIR})
+
+find_library(UADKZIP_LIBRARIES NAMES wd_comp)
+message("UADKZIP_LIBRARIES: " ${UADKZIP_LIBRARIES})
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(uadkzip DEFAULT_MSG UADKZIP_LIBRARIES UADKZIP_INCLUDE_DIR)
+
+if(uadkzip_FOUND AND NOT TARGET uadkzip::uadkzip)
+  add_library(uadkzip::uadkzip UNKNOWN IMPORTED)
+  set_target_properties(uadkzip::uadkzip PROPERTIES
+    INTERFACE_INCLUDE_DIRECTORIES "${UADKZIP_INCLUDE_DIR}"
+    IMPORTED_LINK_INTERFACE_LANGUAGES "C"
+    IMPORTED_LOCATION "${UADKZIP_LIBRARIES}")
+endif()
+
+mark_as_advanced(UADKZIP_LIBRARIES UADKZIP_INCLUDE_DIR)
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index cc53f9a725..f33398c8ee 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -505,6 +505,10 @@ if(HAVE_QATZIP)
   list(APPEND ceph_common_deps ${QATZIP_LIBRARIES})
 endif()
 
+if(HAVE_UADKZIP)
+  list(APPEND ceph_common_deps ${UADKZIP_LIBRARIES})
+endif()
+
 if(WITH_DPDK)
   list(APPEND ceph_common_deps common_async_dpdk)
 endif()
diff --git a/src/common/legacy_config_opts.h b/src/common/legacy_config_opts.h
index d22061f8a7..10c4d7d206 100644
--- a/src/common/legacy_config_opts.h
+++ b/src/common/legacy_config_opts.h
@@ -86,6 +86,7 @@ OPTION(compressor_zlib_winsize, OPT_INT) //regular zlib compression winsize, not
 OPTION(compressor_zstd_level, OPT_INT) //regular zstd compression level
 
 OPTION(qat_compressor_enabled, OPT_BOOL)
+OPTION(uadk_compressor_enabled, OPT_BOOL)
 
 OPTION(plugin_crypto_accelerator, OPT_STR)
 
diff --git a/src/common/options.cc b/src/common/options.cc
index 2f28123ba5..cef26003c2 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -815,6 +815,10 @@ std::vector<Option> get_global_options() {
     .set_default(false)
     .set_description("Enable Intel QAT acceleration support for compression if available"),
 
+    Option("uadk_compressor_enabled", Option::TYPE_BOOL, Option::LEVEL_ADVANCED)
+    .set_default(false)
+    .set_description("Enable UADK acceleration support for compression if available"),
+
     Option("plugin_crypto_accelerator", Option::TYPE_STR, Option::LEVEL_ADVANCED)
     .set_default("crypto_isal")
     .set_description("Crypto accelerator library to use"),
diff --git a/src/compressor/CMakeLists.txt b/src/compressor/CMakeLists.txt
index d8b9fd83cd..9f8b36d689 100644
--- a/src/compressor/CMakeLists.txt
+++ b/src/compressor/CMakeLists.txt
@@ -4,6 +4,11 @@ set(compressor_srcs
 if (HAVE_QATZIP)
   list(APPEND compressor_srcs QatAccel.cc)
 endif()
+
+if (HAVE_UADKZIP)
+  list(APPEND compressor_srcs UadkCompressionAccel.cc)
+endif()
+
 add_library(compressor_objs OBJECT ${compressor_srcs})
 
 ## compressor plugins
@@ -27,6 +32,10 @@ if(HAVE_QATZIP)
   target_link_libraries(compressor PRIVATE ${QATZIP_LIBRARIES})
 endif()
 
+if (HAVE_UADKZIP)
+  target_link_libraries(compressor PRIVATE ${UADKZIP_LIBRARIES})
+endif()
+
 set(ceph_compressor_libs
     ceph_snappy
     ceph_zlib
diff --git a/src/compressor/Compressor.h b/src/compressor/Compressor.h
index 0a45a990a8..4a7e844250 100644
--- a/src/compressor/Compressor.h
+++ b/src/compressor/Compressor.h
@@ -27,6 +27,10 @@
   #include "QatAccel.h"
 #endif
 
+#ifdef HAVE_UADKZIP
+  #include "UadkCompressionAccel.h"
+#endif
+
 namespace TOPNSPC {
 
 class Compressor;
@@ -75,6 +79,11 @@ public:
   QatAccel qat_accel;
 #endif
 
+#ifdef HAVE_UADKZIP
+  bool uadk_cmprs_enabled;
+  UadkCompressionAccel uadk_cmprs_accel;
+#endif
+
   static const char* get_comp_alg_name(int a);
   static boost::optional<CompressionAlgorithm> get_comp_alg_type(std::string_view s);
 
diff --git a/src/compressor/UadkCompressionAccel.cc b/src/compressor/UadkCompressionAccel.cc
new file mode 100644
index 0000000000..40ce082df1
--- /dev/null
+++ b/src/compressor/UadkCompressionAccel.cc
@@ -0,0 +1,446 @@
+/* 
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2022 Huawei Technologies Co., Ltd All rights reserved.
+ *
+ * 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.
+ *
+ */
+
+#include <atomic>
+#include <pthread.h>
+#include "unistd.h"
+#include "common/debug.h"
+#include "UadkCompressionAccel.h"
+
+#define dout_context g_ceph_context
+#define dout_subsys ceph_subsys_compressor
+#undef dout_prefix
+#define dout_prefix _prefix(_dout)
+
+#define NEED_MORE_OUT_BUFFER  5
+#define PROCESS_NOT_FINISH    6
+#define UADK_MIN_BUFFER       (32*1024)
+#define UADK_MAX_BUFFER       (8*1024*1024)
+#define UADK_DEFAULT_WIN_SIZE 15
+#define UADK_PF_SIZE          1024
+
+static ostream&
+_prefix(std::ostream* _dout)
+{
+  return *_dout << "Compression--UADK: ";
+}
+
+static std::atomic<bool> init_called = { false };
+static std::atomic<size_t> uadk_compressor_thread_num = 0;
+std::mutex uadk_lock;
+
+struct UadkEngine {
+  struct wd_ctx_config ctx_cfg;
+  struct wd_sched *sched;
+  int numa_id;
+} engine;
+
+
+static int lib_poll_func(__u32 pos, __u32 expect, __u32 *count)
+{
+  int ret = wd_comp_poll_ctx(pos, expect, count);
+  if (ret < 0)
+    return ret;
+  return 0;
+}
+
+static unsigned int get_uadk_ctx_num()
+{
+  char *env_set = getenv("WD_SYNC_CTX_NUM");
+  if (env_set == nullptr) {
+    return 2;
+  } else {
+    int num = stoi(string(env_set));
+    if (num > UADK_PF_SIZE) {
+      num = UADK_PF_SIZE;
+    } else {
+      num = std::max(num, 2);
+    }
+    return (unsigned int)num;
+  }
+}
+
+static int uadk_init()
+{
+  if (init_called) {
+    dout(10) << __func__ << ": UADK already init." << dendl;
+    return 0;
+  }
+
+  int ret = 0;
+  engine.sched = wd_sched_rr_alloc(SCHED_POLICY_RR, 2, 4, lib_poll_func);
+
+  if (engine.sched == nullptr) {
+    derr << __func__ << ": wd_sched_rr_alloc fail" << dendl;
+    return -ENOMEM;
+  }
+  engine.sched->name = "sched_rr";
+
+  struct uacce_dev *uadk_dev = wd_get_accel_dev("zlib");
+  if (uadk_dev == nullptr) {
+    derr << __func__ << ": cannot get uadk device " << dendl;
+    ret = -ECANCELED;
+    return ret;
+  }
+  engine.numa_id = uadk_dev->numa_id;
+  unsigned int CMPRS_CTX_NUM = get_uadk_ctx_num();
+  engine.ctx_cfg.ctx_num = CMPRS_CTX_NUM;
+  engine.ctx_cfg.ctxs = new wd_ctx[CMPRS_CTX_NUM];
+
+  unsigned int i;
+
+  /******** request ctxs (compress ctx num + decompress ctx num) ********/
+  for (i = 0; i != CMPRS_CTX_NUM; ++i) {
+    engine.ctx_cfg.ctxs[i].ctx = wd_request_ctx(uadk_dev);
+    if (!engine.ctx_cfg.ctxs[i].ctx) {
+      derr << __func__ << ": UADK ctx ERROR !" << dendl;
+      ret = -ECANCELED;
+      goto out_fill;
+    }
+  }
+
+  struct sched_params param;
+  /******** create sched instance for compress ctx ********/
+  for(unsigned int m = 0; m != CMPRS_CTX_NUM / 2; ++m) {
+    engine.ctx_cfg.ctxs[m].op_type = WD_DIR_COMPRESS;
+    engine.ctx_cfg.ctxs[m].ctx_mode = CTX_MODE_SYNC;
+  }
+  param.numa_id = engine.numa_id;
+  param.type = WD_DIR_COMPRESS;
+  param.mode = CTX_MODE_SYNC;
+  param.begin = 0;
+  param.end = CMPRS_CTX_NUM / 2 - 1;
+
+  ret = wd_sched_rr_instance((const struct wd_sched *)engine.sched, &param);
+  if (ret < 0) {
+    derr << __func__ << ": Fail to fill compress sched region."
+	 << "(" << ret << ")" << dendl;
+    goto out_fill;
+  }
+
+  /******** create sched instance for decompress ctx ********/
+  for(unsigned int m = CMPRS_CTX_NUM / 2; m != CMPRS_CTX_NUM; ++m) {
+    engine.ctx_cfg.ctxs[m].op_type = WD_DIR_DECOMPRESS;
+    engine.ctx_cfg.ctxs[m].ctx_mode = CTX_MODE_SYNC;
+  }
+  param.type = WD_DIR_DECOMPRESS;
+  param.mode = CTX_MODE_SYNC;
+  param.begin = CMPRS_CTX_NUM / 2;
+  param.end = CMPRS_CTX_NUM - 1;
+  ret = wd_sched_rr_instance((const struct wd_sched *)engine.sched, &param);
+  if (ret < 0) {
+    derr << __func__ << ": Fail to fill decompress sched region."
+	 << "(" << ret << ")" << dendl;
+    goto out_fill;
+  }
+
+  ret = wd_comp_init(&engine.ctx_cfg, engine.sched);
+  if (ret != 0) {
+    derr << __func__ << ": fail to init UADK !"
+	 << "(" << ret << ")" << dendl;
+    goto out_fill;
+  }
+
+  free(uadk_dev);
+  uadk_dev = nullptr;
+  init_called = true;
+  return 0;
+
+out_fill:
+  for (unsigned int j = 0; j != i; ++j)
+    wd_release_ctx(engine.ctx_cfg.ctxs[j].ctx);
+
+  delete[] engine.ctx_cfg.ctxs;
+  wd_sched_rr_release(engine.sched);
+  engine.sched = nullptr;
+  free(uadk_dev);
+  uadk_dev = nullptr;
+  return ret;
+}
+
+bool UadkCompressionAccel::init()
+{
+  ++uadk_compressor_thread_num;
+
+  if (init_called) {
+    dout(10) << __func__ << ": UADK already init." << dendl;
+    return true;
+  }
+
+  uadk_lock.lock();
+  int ret = uadk_init();
+  uadk_lock.unlock();
+
+  if (ret != 0) {
+    derr << __func__ << ": fail to init uadk.(ret=" << ret << ")" << dendl;
+    --uadk_compressor_thread_num;
+    return false;
+  }
+
+  return true;
+}
+
+handle_t UadkCompressionAccel::create_comp_session()
+{
+  struct wd_comp_sess_setup setup;
+  struct sched_params ss_param = {0};
+
+  setup.op_type = WD_DIR_COMPRESS;
+  setup.alg_type = WD_ZLIB;
+  setup.comp_lv = WD_COMP_L1;
+  setup.win_sz = WD_COMP_WS_8K;
+
+  ss_param.type = setup.op_type;
+  ss_param.numa_id = engine.numa_id;
+  setup.sched_param = &ss_param;
+  handle_t h_comp_sess = wd_comp_alloc_sess(&setup);
+  return h_comp_sess;
+}
+
+void UadkCompressionAccel::free_comp_session(handle_t h_comp_sess)
+{
+  if (h_comp_sess) {
+    wd_comp_free_sess(h_comp_sess);
+    h_comp_sess = 0;
+  }
+}
+
+handle_t UadkCompressionAccel::create_decomp_session()
+{
+  struct wd_comp_sess_setup de_setup;
+  struct sched_params ss_de_param = {0};
+
+  de_setup.op_type = WD_DIR_DECOMPRESS;
+  de_setup.alg_type = WD_ZLIB;
+  de_setup.comp_lv = WD_COMP_L1;
+  de_setup.win_sz = WD_COMP_WS_32K;
+
+  ss_de_param.type = de_setup.op_type;
+  ss_de_param.numa_id = engine.numa_id;
+  de_setup.sched_param = &ss_de_param;
+  handle_t h_decomp_sess = wd_comp_alloc_sess(&de_setup);
+  return h_decomp_sess;
+}
+
+void UadkCompressionAccel::free_decomp_session(handle_t h_decomp_sess)
+{
+  if (h_decomp_sess) {
+    wd_comp_free_sess(h_decomp_sess);
+    h_decomp_sess = 0;
+  }
+}
+
+int UadkCompressionAccel::uadk_do_compress(handle_t h_sess, const unsigned char* in, unsigned int &inlen,
+		                           unsigned char *out, unsigned int &outlen, bool last_packet)
+{
+  struct wd_comp_req req;
+
+  req.op_type = WD_DIR_COMPRESS;
+  req.src = const_cast<unsigned char*>(in);
+  req.src_len = inlen;
+  req.dst = out;
+  req.dst_len = outlen;
+  req.data_fmt = WD_FLAT_BUF;
+  req.cb = nullptr;
+  if (last_packet)
+    req.last = 1;
+  else
+    req.last = 0;
+
+  int ret = wd_do_comp_strm(h_sess, &req);
+  if (ret == 0) {
+    if (inlen > req.src_len) {
+      inlen = req.src_len;
+      outlen = req.dst_len;
+      return NEED_MORE_OUT_BUFFER;
+    } else {
+      outlen = req.dst_len;
+      return ret;
+    }
+  }
+
+  return ret;
+}
+
+int UadkCompressionAccel::compress(const bufferlist &in, bufferlist &out, boost::optional<int32_t> &compressor_message)
+{
+  handle_t h_comp_sess = create_comp_session();
+  unsigned int begin = 1;
+  unsigned int out_len = 0;
+  compressor_message = UADK_DEFAULT_WIN_SIZE;
+  for (ceph::bufferlist::buffers_t::const_iterator i = in.buffers().begin(); i != in.buffers().end();) {
+    const unsigned char* c_in = (unsigned char*) (*i).c_str();
+    unsigned int len = (*i).length();
+    unsigned int in_len = len;
+    int ret = 0;
+    ++i;
+
+    bool last_ptr = (i == in.buffers().end()) ? true : false;
+
+    do {
+      if (len * 2 < UADK_MIN_BUFFER) {
+        out_len = UADK_MIN_BUFFER;
+      } else {
+        out_len = std::min<size_t>(UADK_MAX_BUFFER, len * 2);
+      }
+      bufferptr ptr = buffer::create_small_page_aligned(out_len);
+      unsigned char* c_out = (unsigned char*)ptr.c_str() + begin;
+      in_len = std::min<size_t>(UADK_MAX_BUFFER, in_len);
+      if (begin) {
+        ptr.c_str()[0] = 0;
+	out_len -= begin;
+      }
+
+      bool last_packet = last_ptr && (in_len == len);
+      memset(c_out, 0, out_len);
+      ret = uadk_do_compress(h_comp_sess, c_in, in_len, c_out, out_len, last_packet);
+      if (ret < 0) {
+        derr << __func__ << ": UADK deflation failed."
+	     << "(" << ret << ")" << dendl;
+	free_comp_session(h_comp_sess);
+	return ret;
+      }
+
+      c_in += in_len;
+      in_len = len - in_len;
+      len = in_len;
+
+      out.append(ptr, 0, out_len + begin);
+      begin = 0;
+    } while (ret == NEED_MORE_OUT_BUFFER || len > 0);
+  }
+
+  free_comp_session(h_comp_sess);
+  return 0;
+}
+
+int UadkCompressionAccel::uadk_do_decompress(handle_t h_sess, const unsigned char *in, unsigned int &inlen,
+		                             unsigned char *out, unsigned int &outlen)
+{
+  struct wd_comp_req req;
+
+  req.op_type = WD_DIR_DECOMPRESS;
+  req.data_fmt = WD_FLAT_BUF;
+  req.cb = nullptr;
+
+  req.src = const_cast<unsigned char*>(in);
+  req.src_len = inlen;
+  req.dst = out;
+  req.dst_len = outlen;
+
+  int ret = wd_do_comp_strm(h_sess, &req);
+
+  if (ret == 0) {
+    if (inlen > req.src_len) {
+      inlen = req.src_len;
+      outlen = req.dst_len;
+      return NEED_MORE_OUT_BUFFER;
+    } else if (req.status != WD_STREAM_END) {
+      inlen = req.src_len;
+      outlen = req.dst_len;
+      return PROCESS_NOT_FINISH;
+    } else {
+      outlen = req.dst_len;
+      return ret;
+    }
+  }
+
+  return ret;
+}
+
+unsigned int cal_approx_ratio(unsigned int n, unsigned m)
+{
+  unsigned int x = 0;
+  m /= n;
+  while (m != 0) {
+    m >>= 1;
+    ++x;
+  }
+  return x + 1;
+}
+
+int UadkCompressionAccel::decompress(bufferlist::const_iterator &p, size_t compressed_len, bufferlist &dst,
+		                     boost::optional<int32_t> compressor_message)
+{
+  handle_t h_decomp_sess = create_decomp_session();
+  unsigned int begin = 1;
+  unsigned int out_len = 0;
+  unsigned int probe_ratio = 2;
+  bufferptr ptr;
+  size_t remaining = std::min<size_t>(p.get_remaining(), compressed_len);
+
+  while (remaining) {
+    const char *c_in;
+    unsigned int len = p.get_ptr_and_advance(remaining, &c_in) - begin;
+    unsigned int in_len = len;
+    unsigned char *in = (unsigned char *)c_in + begin;
+    int ret = 0;
+
+    remaining -= (in_len + begin);
+    begin = 0;
+
+    do {
+      if ((len << probe_ratio) < UADK_MIN_BUFFER) {
+        out_len = UADK_MIN_BUFFER;
+      } else {
+        out_len = std::min<size_t>(UADK_MAX_BUFFER, (len << probe_ratio));
+      }
+      ptr = buffer::create_small_page_aligned(out_len);
+      unsigned char* out = (unsigned char*)ptr.c_str();
+      in_len = std::min<size_t>(UADK_MAX_BUFFER, in_len);
+      memset(out, 0, out_len);
+      ret = uadk_do_decompress(h_decomp_sess, in, in_len, out, out_len);
+      if (ret < 0) {
+        derr << __func__ << ": UADK inflation failed.(ret=" << ret << ")" << dendl;
+	free_decomp_session(h_decomp_sess);
+	return ret;
+      }
+
+     probe_ratio = cal_approx_ratio(in_len, out_len);
+     in += in_len;
+     in_len = len - in_len;
+     len = in_len;
+     dst.append(ptr, 0, out_len);
+    } while (ret == NEED_MORE_OUT_BUFFER || (ret == PROCESS_NOT_FINISH && remaining ==0) || len > 0);
+  }
+
+  free_decomp_session(h_decomp_sess);
+  return 0;
+}
+
+int UadkCompressionAccel::decompress(const bufferlist &in, bufferlist &out, boost::optional<int32_t> compressor_message)
+{
+  auto i = in.begin();
+  return decompress(i, in.length(), out, compressor_message);
+}
+
+void UadkCompressionAccel::destory()
+{
+  if (!init_called) {
+    return;
+  }
+
+  if (--uadk_compressor_thread_num != 0) {
+    dout(10) << __func__ << ": " << uadk_compressor_thread_num << " threads need uadk zip" << dendl;
+    return;
+  }
+
+  wd_comp_uninit();
+
+  for (unsigned int i = 0; i < engine.ctx_cfg.ctx_num; i++) {
+    wd_release_ctx(engine.ctx_cfg.ctxs[i].ctx);
+  }
+  delete[] engine.ctx_cfg.ctxs;
+  wd_sched_rr_release(engine.sched);
+  engine.sched = nullptr;
+  init_called = false;
+}
diff --git a/src/compressor/UadkCompressionAccel.h b/src/compressor/UadkCompressionAccel.h
new file mode 100644
index 0000000000..b2c7bb4cc4
--- /dev/null
+++ b/src/compressor/UadkCompressionAccel.h
@@ -0,0 +1,45 @@
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (c) 2022 Huawei Technologies Co., Ltd All rights reserved.
+ *
+ * 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_UADKCOMPRESSIONACCEL_H
+#define CEPH_UADKCOMPRESSIONACCEL_H
+
+#include <boost/optional.hpp>
+#include "include/buffer.h"
+
+extern "C" {
+#include <uadk/wd_comp.h>
+#include <uadk/wd.h>
+#include <uadk/wd_sched.h>
+}
+
+class UadkCompressionAccel {
+  public:
+      UadkCompressionAccel() {  }
+      ~UadkCompressionAccel() { destory(); }
+
+      bool init();
+      void destory();
+
+      int compress(const bufferlist &in, bufferlist &out, boost::optional<int32_t> &compressor_message);
+      int decompress(const bufferlist &in, bufferlist &out, boost::optional<int32_t> compressor_message);
+      int decompress(bufferlist::const_iterator &p, size_t compressed_len, bufferlist &dst, boost::optional<int32_t> compressor_message);
+  private:
+      int uadk_do_compress(handle_t h_sess, const unsigned char *in, unsigned int &inlen, unsigned char *out, unsigned int &outlen, bool last_packet);
+      int uadk_do_decompress(handle_t h_sess, const unsigned char *in, unsigned int &inlen, unsigned char *out, unsigned int &outlen);
+      handle_t create_comp_session();
+      handle_t create_decomp_session();
+      void free_comp_session(handle_t h_sess);
+      void free_decomp_session(handle_t h_sess);
+};
+
+#endif
diff --git a/src/compressor/zlib/ZlibCompressor.cc b/src/compressor/zlib/ZlibCompressor.cc
index 27b43c49cb..78cb60ac9c 100644
--- a/src/compressor/zlib/ZlibCompressor.cc
+++ b/src/compressor/zlib/ZlibCompressor.cc
@@ -174,6 +174,16 @@ int ZlibCompressor::compress(const bufferlist &in, bufferlist &out, boost::optio
   if (qat_enabled)
     return qat_accel.compress(in, out, compressor_message);
 #endif
+
+#ifdef HAVE_UADKZIP
+  if (uadk_cmprs_enabled) {
+    int uadk_ret = uadk_cmprs_accel.compress(in, out, compressor_message);
+    if (uadk_ret == 0) {
+      return uadk_ret;
+    }
+  }
+#endif
+
 #if __x86_64__ && defined(HAVE_NASM_X64_AVX2)
   if (isal_enabled)
     return isal_compress(in, out, compressor_message);
@@ -191,6 +201,15 @@ int ZlibCompressor::decompress(bufferlist::const_iterator &p, size_t compressed_
     return qat_accel.decompress(p, compressed_size, out, compressor_message);
 #endif
 
+#ifdef HAVE_UADKZIP
+  if (uadk_cmprs_enabled) {
+    int uadk_ret = uadk_cmprs_accel.decompress(p, compressed_size, out, compressor_message);
+    if (uadk_ret == 0) {
+      return uadk_ret;
+    }
+  }
+#endif
+
   int ret;
   unsigned have;
   z_stream strm;
@@ -250,6 +269,16 @@ int ZlibCompressor::decompress(const bufferlist &in, bufferlist &out, boost::opt
   if (qat_enabled)
     return qat_accel.decompress(in, out, compressor_message);
 #endif
+
+#ifdef HAVE_UADKZIP
+  if (uadk_cmprs_enabled) {
+    int uadk_ret = uadk_cmprs_accel.decompress(in, out, compressor_message);
+    if (uadk_ret == 0) {
+      return uadk_ret;
+    }
+  }
+#endif
+
   auto i = std::cbegin(in);
   return decompress(i, in.length(), out, compressor_message);
 }
diff --git a/src/compressor/zlib/ZlibCompressor.h b/src/compressor/zlib/ZlibCompressor.h
index b8e984110a..3931d88c79 100644
--- a/src/compressor/zlib/ZlibCompressor.h
+++ b/src/compressor/zlib/ZlibCompressor.h
@@ -32,6 +32,18 @@ public:
     else
       qat_enabled = false;
 #endif
+
+#ifdef HAVE_UADKZIP
+    if (cct->_conf->compressor_zlib_winsize >= 9) {
+        if (cct->_conf->uadk_compressor_enabled && uadk_cmprs_accel.init())
+           uadk_cmprs_enabled = true;
+        else
+           uadk_cmprs_enabled = false;
+  }
+  else {
+    uadk_cmprs_enabled = false;
+  }
+#endif
   }
 
   int compress(const ceph::buffer::list &in, ceph::buffer::list &out, boost::optional<int32_t> &compressor_message) override;
diff --git a/src/include/config-h.in.cmake b/src/include/config-h.in.cmake
index 155f9df6d1..c01aa1d908 100644
--- a/src/include/config-h.in.cmake
+++ b/src/include/config-h.in.cmake
@@ -357,6 +357,9 @@
 /* Defined if Intel QAT compress/decompress is supported */
 #cmakedefine HAVE_QATZIP
 
+/* Defined if Hisi UADK compress/decompress is supported */
+#cmakedefine HAVE_UADKZIP
+
 /* Define if seastar is available. */
 #cmakedefine HAVE_SEASTAR
 
diff --git a/src/test/compressor/test_compression.cc b/src/test/compressor/test_compression.cc
index a8a66398cb..da385223d0 100644
--- a/src/test/compressor/test_compression.cc
+++ b/src/test/compressor/test_compression.cc
@@ -608,3 +608,87 @@ TEST(QAT, enc_noqat_dec_qat) {
 }
 
 #endif	// HAVE_QATZIP
+
+#ifdef HAVE_UADKZIP
+TEST(UADK, enc_uadk_dec_nouadk) {
+  const char* alg_collection[] = {"zlib"};
+
+  for (auto alg : alg_collection) {
+    g_conf().set_val("uadk_compressor_enabled", "true");
+    g_conf().set_val("compressor_zlib_winsize", "15");
+    g_ceph_context->_conf.apply_changes(nullptr);
+    CompressorRef hw = Compressor::create(g_ceph_context, alg);
+    if (hw == NULL) 
+      return;
+
+    g_conf().set_val("uadk_compressor_enabled", "false");
+    g_conf().set_val("compressor_zlib_winsize", "15");
+    g_ceph_context->_conf.apply_changes(nullptr);
+    CompressorRef sw = Compressor::create(g_ceph_context, alg);
+
+    //generate random buffer
+    for (int cnt = 0; cnt < 100; cnt++) {
+      srand(cnt + 1000);
+      int log2 = (rand()%18) + 1;
+      int size = (rand() % (1 << log2)) + 1;
+
+      char test[size];
+      for (int i = 0; i < size; ++i)
+	test[i] = rand()%256;
+      bufferlist in, out;
+      in.append(test, size);
+
+      boost::optional<int32_t> compressor_message;
+      int res = hw->compress(in, out, compressor_message);
+      EXPECT_EQ(res, 0);
+      bufferlist after;
+      res = sw->decompress(out, after, compressor_message);
+      EXPECT_EQ(res, 0);
+      bufferlist exp;
+      exp.append(test, size);
+      EXPECT_TRUE(exp.contents_equal(after));
+    }
+  }
+}
+
+TEST(UADK, enc_nouadk_dec_uadk) {
+  const char* alg_collection[] = {"zlib"};
+
+  for (auto alg : alg_collection) {
+    g_conf().set_val("uadk_compressor_enabled", "true");
+    g_conf().set_val("compressor_zlib_winsize", "15");
+    g_ceph_context->_conf.apply_changes(nullptr);
+    CompressorRef hw = Compressor::create(g_ceph_context, alg);
+    if (hw == NULL) 
+      return;
+    g_conf().set_val("uadk_compressor_enabled", "false");
+    g_conf().set_val("compressor_zlib_winsize", "15");
+    g_ceph_context->_conf.apply_changes(nullptr);
+    CompressorRef sw = Compressor::create(g_ceph_context, alg);
+
+    //generate random buffer
+    for (int cnt = 0; cnt < 100; cnt++) {
+      srand(cnt + 1000);
+      int log2 = (rand()%18) +1;
+      int size = (rand() % (1 << log2)) + 1;
+
+      char test[size];
+      for (int i = 0; i < size; ++i)
+        test[i] = rand()%256;
+      bufferlist in, out;
+      in.append(test, size);
+
+      boost::optional<int32_t> compressor_message;
+      int res = sw->compress(in, out, compressor_message);
+      EXPECT_EQ(res, 0);
+      bufferlist after;
+      res = hw->decompress(out, after, compressor_message);
+      EXPECT_EQ(res, 0);
+      bufferlist exp;
+      exp.append(test, size);
+      EXPECT_TRUE(exp.contents_equal(after));
+    }
+  }
+}
+
+#endif //HAVE_UADKZIP
-- 
Gitee


From e482d3470601581ce4a89625e3a994babe272a34 Mon Sep 17 00:00:00 2001
From: Dai Zhiwei <daizhiwei3@huawei.com>
Date: Tue, 5 Jul 2022 15:29:06 +0800
Subject: [PATCH 2/2] fix unittest_compression error

---
 src/test/compressor/test_compression.cc | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/src/test/compressor/test_compression.cc b/src/test/compressor/test_compression.cc
index da385223d0..dc8ca9e7a6 100644
--- a/src/test/compressor/test_compression.cc
+++ b/src/test/compressor/test_compression.cc
@@ -642,6 +642,8 @@ TEST(UADK, enc_uadk_dec_nouadk) {
       int res = hw->compress(in, out, compressor_message);
       EXPECT_EQ(res, 0);
       bufferlist after;
+      if (hw->uadk_cmprs_enabled == true)
+	hw->uadk_cmprs_enabled = false;
       res = sw->decompress(out, after, compressor_message);
       EXPECT_EQ(res, 0);
       bufferlist exp;
@@ -682,11 +684,14 @@ TEST(UADK, enc_nouadk_dec_uadk) {
       int res = sw->compress(in, out, compressor_message);
       EXPECT_EQ(res, 0);
       bufferlist after;
+      if (hw->uadk_cmprs_enabled == false)
+        hw->uadk_cmprs_enabled = true;
       res = hw->decompress(out, after, compressor_message);
       EXPECT_EQ(res, 0);
       bufferlist exp;
       exp.append(test, size);
       EXPECT_TRUE(exp.contents_equal(after));
+      hw->uadk_cmprs_enabled = false;
     }
   }
 }
-- 
Gitee