X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?p=stor4nfv.git;a=blobdiff_plain;f=src%2Fceph%2F0010-librbd-new-namespace-ceph-immutable-obj-cache.patch;fp=src%2Fceph%2F0010-librbd-new-namespace-ceph-immutable-obj-cache.patch;h=0000000000000000000000000000000000000000;hp=46040e6942addd78d0a8977a45dedd13961ba099;hb=fe427fc3f69ecc5e104a6fdb2c9f5d82f0254fdb;hpb=79b3874aa314884557bc59983cbdb99f62e77699

diff --git a/src/ceph/0010-librbd-new-namespace-ceph-immutable-obj-cache.patch b/src/ceph/0010-librbd-new-namespace-ceph-immutable-obj-cache.patch
deleted file mode 100644
index 46040e6..0000000
--- a/src/ceph/0010-librbd-new-namespace-ceph-immutable-obj-cache.patch
+++ /dev/null
@@ -1,3510 +0,0 @@
-From 9a097084d06e7186206b43d9c81d1f648791d7a4 Mon Sep 17 00:00:00 2001
-From: Yuan Zhou <yuan.zhou@intel.com>
-Date: Fri, 7 Sep 2018 08:29:51 +0800
-Subject: [PATCH 10/10] librbd: new namespace ceph immutable obj cache
-
-clean up class/func names to use the new namespace
-
-Signed-off-by: Yuan Zhou <yuan.zhou@intel.com>
----
- src/common/options.cc                              |   2 +-
- src/common/subsys.h                                |   3 +-
- src/librbd/CMakeLists.txt                          |   3 +-
- .../SharedPersistentObjectCacherObjectDispatch.cc  | 175 ----------------
- .../SharedPersistentObjectCacherObjectDispatch.h   | 133 ------------
- src/librbd/cache/SharedReadOnlyObjectDispatch.cc   | 170 +++++++++++++++
- src/librbd/cache/SharedReadOnlyObjectDispatch.h    | 126 ++++++++++++
- src/librbd/image/OpenRequest.cc                    |   4 +-
- src/tools/CMakeLists.txt                           |   2 +-
- .../ceph_immutable_object_cache/CMakeLists.txt     |  11 +
- .../ceph_immutable_object_cache/CacheClient.cc     | 205 ++++++++++++++++++
- .../ceph_immutable_object_cache/CacheClient.h      |  53 +++++
- .../ceph_immutable_object_cache/CacheController.cc | 117 +++++++++++
- .../ceph_immutable_object_cache/CacheController.h  |  53 +++++
- .../ceph_immutable_object_cache/CacheServer.cc     |  99 +++++++++
- .../ceph_immutable_object_cache/CacheServer.h      |  54 +++++
- .../ceph_immutable_object_cache/CacheSession.cc    | 115 +++++++++++
- .../ceph_immutable_object_cache/CacheSession.h     |  58 ++++++
- .../ObjectCacheStore.cc                            | 172 ++++++++++++++++
- .../ceph_immutable_object_cache/ObjectCacheStore.h |  70 +++++++
- src/tools/ceph_immutable_object_cache/Policy.hpp   |  33 +++
- .../ceph_immutable_object_cache/SimplePolicy.hpp   | 163 +++++++++++++++
- .../ceph_immutable_object_cache/SocketCommon.h     |  54 +++++
- src/tools/ceph_immutable_object_cache/main.cc      |  85 ++++++++
- src/tools/rbd_cache/CMakeLists.txt                 |   9 -
- src/tools/rbd_cache/CacheController.cc             | 116 -----------
- src/tools/rbd_cache/CacheController.h              |  54 -----
- src/tools/rbd_cache/CacheControllerSocket.hpp      | 228 --------------------
- .../rbd_cache/CacheControllerSocketClient.hpp      | 229 ---------------------
- src/tools/rbd_cache/CacheControllerSocketCommon.h  |  62 ------
- src/tools/rbd_cache/ObjectCacheStore.cc            | 172 ----------------
- src/tools/rbd_cache/ObjectCacheStore.h             |  70 -------
- src/tools/rbd_cache/Policy.hpp                     |  30 ---
- src/tools/rbd_cache/SimplePolicy.hpp               | 160 --------------
- src/tools/rbd_cache/main.cc                        |  85 --------
- 35 files changed, 1646 insertions(+), 1529 deletions(-)
- delete mode 100644 src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc
- delete mode 100644 src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h
- create mode 100644 src/librbd/cache/SharedReadOnlyObjectDispatch.cc
- create mode 100644 src/librbd/cache/SharedReadOnlyObjectDispatch.h
- create mode 100644 src/tools/ceph_immutable_object_cache/CMakeLists.txt
- create mode 100644 src/tools/ceph_immutable_object_cache/CacheClient.cc
- create mode 100644 src/tools/ceph_immutable_object_cache/CacheClient.h
- create mode 100644 src/tools/ceph_immutable_object_cache/CacheController.cc
- create mode 100644 src/tools/ceph_immutable_object_cache/CacheController.h
- create mode 100644 src/tools/ceph_immutable_object_cache/CacheServer.cc
- create mode 100644 src/tools/ceph_immutable_object_cache/CacheServer.h
- create mode 100644 src/tools/ceph_immutable_object_cache/CacheSession.cc
- create mode 100644 src/tools/ceph_immutable_object_cache/CacheSession.h
- create mode 100644 src/tools/ceph_immutable_object_cache/ObjectCacheStore.cc
- create mode 100644 src/tools/ceph_immutable_object_cache/ObjectCacheStore.h
- create mode 100644 src/tools/ceph_immutable_object_cache/Policy.hpp
- create mode 100644 src/tools/ceph_immutable_object_cache/SimplePolicy.hpp
- create mode 100644 src/tools/ceph_immutable_object_cache/SocketCommon.h
- create mode 100644 src/tools/ceph_immutable_object_cache/main.cc
- delete mode 100644 src/tools/rbd_cache/CMakeLists.txt
- delete mode 100644 src/tools/rbd_cache/CacheController.cc
- delete mode 100644 src/tools/rbd_cache/CacheController.h
- delete mode 100644 src/tools/rbd_cache/CacheControllerSocket.hpp
- delete mode 100644 src/tools/rbd_cache/CacheControllerSocketClient.hpp
- delete mode 100644 src/tools/rbd_cache/CacheControllerSocketCommon.h
- delete mode 100644 src/tools/rbd_cache/ObjectCacheStore.cc
- delete mode 100644 src/tools/rbd_cache/ObjectCacheStore.h
- delete mode 100644 src/tools/rbd_cache/Policy.hpp
- delete mode 100644 src/tools/rbd_cache/SimplePolicy.hpp
- delete mode 100644 src/tools/rbd_cache/main.cc
-
-diff --git a/src/common/options.cc b/src/common/options.cc
-index 3172744..bf00aab 100644
---- a/src/common/options.cc
-+++ b/src/common/options.cc
-@@ -6358,7 +6358,7 @@ static std::vector<Option> get_rbd_options() {
-     .set_description("time in seconds for detecting a hung thread"),
- 
-     Option("rbd_shared_cache_enabled", Option::TYPE_BOOL, Option::LEVEL_ADVANCED)
--    .set_default(true)
-+    .set_default(false)
-     .set_description("whether to enable shared ssd caching"),
- 
-     Option("rbd_shared_cache_path", Option::TYPE_STR, Option::LEVEL_ADVANCED)
-diff --git a/src/common/subsys.h b/src/common/subsys.h
-index bdd2d0e..5b532c1 100644
---- a/src/common/subsys.h
-+++ b/src/common/subsys.h
-@@ -36,9 +36,10 @@ SUBSYS(objecter, 0, 1)
- SUBSYS(rados, 0, 5)
- SUBSYS(rbd, 0, 5)
- SUBSYS(rbd_mirror, 0, 5)
--SUBSYS(rbd_replay, 0, 5)
- SUBSYS(journaler, 0, 5)
- SUBSYS(objectcacher, 0, 5)
-+SUBSYS(immutable_obj_cache, 0, 5)
-+SUBSYS(rbd_replay, 0, 5)
- SUBSYS(client, 0, 5)
- SUBSYS(osd, 1, 5)
- SUBSYS(optracker, 0, 5)
-diff --git a/src/librbd/CMakeLists.txt b/src/librbd/CMakeLists.txt
-index 540ee78..c9bfb6f 100644
---- a/src/librbd/CMakeLists.txt
-+++ b/src/librbd/CMakeLists.txt
-@@ -32,7 +32,7 @@ set(librbd_internal_srcs
-   api/Snapshot.cc
-   cache/ImageWriteback.cc
-   cache/ObjectCacherObjectDispatch.cc
--  cache/SharedPersistentObjectCacherObjectDispatch.cc
-+  cache/SharedReadOnlyObjectDispatch.cc
-   cache/SharedPersistentObjectCacher.cc
-   cache/SharedPersistentObjectCacherFile.cc
-   deep_copy/ImageCopyRequest.cc
-@@ -125,6 +125,7 @@ set(librbd_internal_srcs
-   trash/MoveRequest.cc
-   watcher/Notifier.cc
-   watcher/RewatchRequest.cc
-+  ${CMAKE_SOURCE_DIR}/src/tools/ceph_immutable_object_cache/CacheClient.cc
-   ${CMAKE_SOURCE_DIR}/src/common/ContextCompletion.cc)
- 
- add_library(rbd_api STATIC librbd.cc)
-diff --git a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc b/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc
-deleted file mode 100644
-index 7cbc019..0000000
---- a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc
-+++ /dev/null
-@@ -1,175 +0,0 @@
--// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
--// vim: ts=8 sw=2 smarttab
--
--#include "librbd/cache/SharedPersistentObjectCacherObjectDispatch.h"
--#include "common/WorkQueue.h"
--#include "librbd/ImageCtx.h"
--#include "librbd/Journal.h"
--#include "librbd/Utils.h"
--#include "librbd/LibrbdWriteback.h"
--#include "librbd/io/ObjectDispatchSpec.h"
--#include "librbd/io/ObjectDispatcher.h"
--#include "librbd/io/Utils.h"
--#include "osd/osd_types.h"
--#include "osdc/WritebackHandler.h"
--#include <vector>
--
--#define dout_subsys ceph_subsys_rbd
--#undef dout_prefix
--#define dout_prefix *_dout << "librbd::cache::SharedPersistentObjectCacherObjectDispatch: " \
--                           << this << " " << __func__ << ": "
--
--namespace librbd {
--namespace cache {
--
--template <typename I>
--SharedPersistentObjectCacherObjectDispatch<I>::SharedPersistentObjectCacherObjectDispatch(
--    I* image_ctx) : m_image_ctx(image_ctx) {
--}
--
--template <typename I>
--SharedPersistentObjectCacherObjectDispatch<I>::~SharedPersistentObjectCacherObjectDispatch() {
--    delete m_object_store;
--    delete m_cache_client;
--}
--
--// TODO if connect fails, init will return error to high layer.
--template <typename I>
--void SharedPersistentObjectCacherObjectDispatch<I>::init() {
--  auto cct = m_image_ctx->cct;
--  ldout(cct, 5) << dendl;
--
--  if (m_image_ctx->parent != nullptr) {
--    //TODO(): should we cover multi-leveled clone?
--    ldout(cct, 5) << "child image: skipping SRO cache client" << dendl;
--    return;
--  }
--
--  ldout(cct, 5) << "parent image: setup SRO cache client = " << dendl;
--
--  std::string controller_path = ((CephContext*)cct)->_conf.get_val<std::string>("rbd_shared_cache_sock");
--  m_cache_client = new rbd::cache::CacheClient(controller_path.c_str(),
--    ([&](std::string s){client_handle_request(s);}), m_image_ctx->cct);
--
--  int ret = m_cache_client->connect();
--  if (ret < 0) {
--    ldout(cct, 5) << "SRO cache client fail to connect with local controller: "
--                  << "please start rbd-cache daemon"
--		  << dendl;
--  } else {
--    ldout(cct, 5) << "SRO cache client to register volume on rbd-cache daemon: "
--                   << "name = " << m_image_ctx->id 
--                   << dendl;
--
--    ret = m_cache_client->register_volume(m_image_ctx->data_ctx.get_pool_name(),
--                                    m_image_ctx->id, m_image_ctx->size);
--
--    if (ret >= 0) {
--      // add ourself to the IO object dispatcher chain
--      m_image_ctx->io_object_dispatcher->register_object_dispatch(this);
--    }
--  }
--}
--
--template <typename I>
--bool SharedPersistentObjectCacherObjectDispatch<I>::read(
--    const std::string &oid, uint64_t object_no, uint64_t object_off,
--    uint64_t object_len, librados::snap_t snap_id, int op_flags,
--    const ZTracer::Trace &parent_trace, ceph::bufferlist* read_data,
--    io::ExtentMap* extent_map, int* object_dispatch_flags,
--    io::DispatchResult* dispatch_result, Context** on_finish,
--    Context* on_dispatched) {
--
--  // IO chained in reverse order
--
--  // Now, policy is : when session have any error, later all read will dispatched to  rados layer.
--  if(!m_cache_client->is_session_work()) {
--    *dispatch_result = io::DISPATCH_RESULT_CONTINUE;
--    on_dispatched->complete(0);
--    return true;
--    // TODO : domain socket have error, all read operation will dispatched to rados layer.
--  }
--
--  auto cct = m_image_ctx->cct;
--  ldout(cct, 20) << "object_no=" << object_no << " " << object_off << "~"
--                 << object_len << dendl;
--
--
--  on_dispatched = util::create_async_context_callback(*m_image_ctx,
--                                                      on_dispatched);
--  auto ctx = new FunctionContext([this, oid, object_off, object_len, read_data, dispatch_result, on_dispatched](bool cache) {
--    handle_read_cache(cache, oid, object_off, object_len, read_data, dispatch_result, on_dispatched);
--  });
--
--  if (m_cache_client && m_cache_client->is_session_work() && m_object_store) {
--    m_cache_client->lookup_object(m_image_ctx->data_ctx.get_pool_name(),
--      m_image_ctx->id, oid, ctx);
--  }
--  return true;
--}
--
--template <typename I>
--int SharedPersistentObjectCacherObjectDispatch<I>::handle_read_cache(
--    bool cache,
--    const std::string &oid, uint64_t object_off, uint64_t object_len,
--    ceph::bufferlist* read_data, io::DispatchResult* dispatch_result,
--    Context* on_dispatched) {
--  // IO chained in reverse order
--  auto cct = m_image_ctx->cct;
--  ldout(cct, 20) << dendl;
--
--  // try to read from parent image
--  if (cache) {
--    int r = m_object_store->read_object(oid, read_data, object_off, object_len, on_dispatched);
--    //int r = object_len;
--    if (r != 0) {
--      *dispatch_result = io::DISPATCH_RESULT_COMPLETE;
--      //TODO(): complete in syncfile
--      on_dispatched->complete(r);
--      ldout(cct, 20) << "AAAAcomplete=" << *dispatch_result <<dendl;
--      return true;
--    }
--  } else {
--    *dispatch_result = io::DISPATCH_RESULT_CONTINUE;
--    on_dispatched->complete(0);
--    ldout(cct, 20) << "BBB no cache" << *dispatch_result <<dendl;
--    return false;
--  }
--}
--template <typename I>
--void SharedPersistentObjectCacherObjectDispatch<I>::client_handle_request(std::string msg) {
--  auto cct = m_image_ctx->cct;
--  ldout(cct, 20) << dendl;
--
--  rbd::cache::rbdsc_req_type_t *io_ctx = (rbd::cache::rbdsc_req_type_t*)(msg.c_str());
--
--  switch (io_ctx->type) {
--    case rbd::cache::RBDSC_REGISTER_REPLY: {
--      // open cache handler for volume
--      ldout(cct, 20) << "SRO cache client open cache handler" << dendl;
--      m_object_store = new SharedPersistentObjectCacher<I>(m_image_ctx, m_image_ctx->shared_cache_path);
--
--      break;
--    }
--    case rbd::cache::RBDSC_READ_REPLY: {
--      ldout(cct, 20) << "SRO cache client start to read cache" << dendl;
--      //TODO(): should call read here
--
--      break;
--    }
--    case rbd::cache::RBDSC_READ_RADOS: {
--      ldout(cct, 20) << "SRO cache client start to read rados" << dendl;
--      //TODO(): should call read here
--
--      break;
--    }
--    default: ldout(cct, 20) << "nothing" << io_ctx->type <<dendl;
--      break;
--
--  }
--}
--
--} // namespace cache
--} // namespace librbd
--
--template class librbd::cache::SharedPersistentObjectCacherObjectDispatch<librbd::ImageCtx>;
-diff --git a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h b/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h
-deleted file mode 100644
-index 5685244..0000000
---- a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h
-+++ /dev/null
-@@ -1,133 +0,0 @@
--// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
--// vim: ts=8 sw=2 smarttab
--
--#ifndef CEPH_LIBRBD_CACHE_SHARED_PERSISTENT_OBJECT_CACHER_OBJECT_DISPATCH_H
--#define CEPH_LIBRBD_CACHE_SHARED_PERSISTENT_OBJECT_CACHER_OBJECT_DISPATCH_H
--
--#include "librbd/io/ObjectDispatchInterface.h"
--#include "common/Mutex.h"
--#include "osdc/ObjectCacher.h"
--#include "tools/rbd_cache/CacheControllerSocketClient.hpp"
--#include "SharedPersistentObjectCacher.h"
--
--struct WritebackHandler;
--
--namespace librbd {
--
--class ImageCtx;
--
--namespace cache {
--
--/**
-- * Facade around the OSDC object cacher to make it align with
-- * the object dispatcher interface
-- */
--template <typename ImageCtxT = ImageCtx>
--class SharedPersistentObjectCacherObjectDispatch : public io::ObjectDispatchInterface {
--public:
--  static SharedPersistentObjectCacherObjectDispatch* create(ImageCtxT* image_ctx) {
--    return new SharedPersistentObjectCacherObjectDispatch(image_ctx);
--  }
--
--  SharedPersistentObjectCacherObjectDispatch(ImageCtxT* image_ctx);
--  ~SharedPersistentObjectCacherObjectDispatch() override;
--
--  io::ObjectDispatchLayer get_object_dispatch_layer() const override {
--    return io::OBJECT_DISPATCH_LAYER_SHARED_PERSISTENT_CACHE;
--  }
--
--  void init();
--  void shut_down(Context* on_finish) {
--    m_image_ctx->op_work_queue->queue(on_finish, 0);
--  }
--
--  bool read(
--      const std::string &oid, uint64_t object_no, uint64_t object_off,
--      uint64_t object_len, librados::snap_t snap_id, int op_flags,
--      const ZTracer::Trace &parent_trace, ceph::bufferlist* read_data,
--      io::ExtentMap* extent_map, int* object_dispatch_flags,
--      io::DispatchResult* dispatch_result, Context** on_finish,
--      Context* on_dispatched) override;
--
--  bool discard(
--      const std::string &oid, uint64_t object_no, uint64_t object_off,
--      uint64_t object_len, const ::SnapContext &snapc, int discard_flags,
--      const ZTracer::Trace &parent_trace, int* object_dispatch_flags,
--      uint64_t* journal_tid, io::DispatchResult* dispatch_result,
--      Context** on_finish, Context* on_dispatched) {
--    return false;
--  }
--
--  bool write(
--      const std::string &oid, uint64_t object_no, uint64_t object_off,
--      ceph::bufferlist&& data, const ::SnapContext &snapc, int op_flags,
--      const ZTracer::Trace &parent_trace, int* object_dispatch_flags,
--      uint64_t* journal_tid, io::DispatchResult* dispatch_result,
--      Context** on_finish, Context* on_dispatched) {
--    return false;
--  }
--
--  bool write_same(
--      const std::string &oid, uint64_t object_no, uint64_t object_off,
--      uint64_t object_len, io::Extents&& buffer_extents,
--      ceph::bufferlist&& data, const ::SnapContext &snapc, int op_flags,
--      const ZTracer::Trace &parent_trace, int* object_dispatch_flags,
--      uint64_t* journal_tid, io::DispatchResult* dispatch_result,
--      Context** on_finish, Context* on_dispatched) {
--    return false;
--  }
--
--  bool compare_and_write(
--      const std::string &oid, uint64_t object_no, uint64_t object_off,
--      ceph::bufferlist&& cmp_data, ceph::bufferlist&& write_data,
--      const ::SnapContext &snapc, int op_flags,
--      const ZTracer::Trace &parent_trace, uint64_t* mismatch_offset,
--      int* object_dispatch_flags, uint64_t* journal_tid,
--      io::DispatchResult* dispatch_result, Context** on_finish,
--      Context* on_dispatched) {
--    return false;
--  }
--
--  bool flush(
--      io::FlushSource flush_source, const ZTracer::Trace &parent_trace,
--      io::DispatchResult* dispatch_result, Context** on_finish,
--      Context* on_dispatched) {
--    return false;
--  }
--
--  bool invalidate_cache(Context* on_finish) {
--    return false;
--  }
--
--  bool reset_existence_cache(Context* on_finish) {
--    return false;
--  }
--
--  void extent_overwritten(
--      uint64_t object_no, uint64_t object_off, uint64_t object_len,
--      uint64_t journal_tid, uint64_t new_journal_tid) {
--  }
--
--  SharedPersistentObjectCacher<ImageCtxT> *m_object_store = nullptr;
--
--private:
--
--  int handle_read_cache(
--      bool cache,
--      const std::string &oid, uint64_t object_off,
--      uint64_t object_len, ceph::bufferlist* read_data,
--      io::DispatchResult* dispatch_result,
--      Context* on_dispatched);
--  void client_handle_request(std::string msg);
--
--  ImageCtxT* m_image_ctx;
--
--  rbd::cache::CacheClient *m_cache_client = nullptr;
--};
--
--} // namespace cache
--} // namespace librbd
--
--extern template class librbd::cache::SharedPersistentObjectCacherObjectDispatch<librbd::ImageCtx>;
--
--#endif // CEPH_LIBRBD_CACHE_OBJECT_CACHER_OBJECT_DISPATCH_H
-diff --git a/src/librbd/cache/SharedReadOnlyObjectDispatch.cc b/src/librbd/cache/SharedReadOnlyObjectDispatch.cc
-new file mode 100644
-index 0000000..23c7dbe
---- /dev/null
-+++ b/src/librbd/cache/SharedReadOnlyObjectDispatch.cc
-@@ -0,0 +1,170 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#include "common/WorkQueue.h"
-+#include "librbd/ImageCtx.h"
-+#include "librbd/Journal.h"
-+#include "librbd/Utils.h"
-+#include "librbd/LibrbdWriteback.h"
-+#include "librbd/io/ObjectDispatchSpec.h"
-+#include "librbd/io/ObjectDispatcher.h"
-+#include "librbd/io/Utils.h"
-+#include "librbd/cache/SharedReadOnlyObjectDispatch.h"
-+#include "osd/osd_types.h"
-+#include "osdc/WritebackHandler.h"
-+
-+#include <vector>
-+
-+#define dout_subsys ceph_subsys_rbd
-+#undef dout_prefix
-+#define dout_prefix *_dout << "librbd::cache::SharedReadOnlyObjectDispatch: " \
-+                           << this << " " << __func__ << ": "
-+
-+namespace librbd {
-+namespace cache {
-+
-+template <typename I>
-+SharedReadOnlyObjectDispatch<I>::SharedReadOnlyObjectDispatch(
-+    I* image_ctx) : m_image_ctx(image_ctx) {
-+}
-+
-+template <typename I>
-+SharedReadOnlyObjectDispatch<I>::~SharedReadOnlyObjectDispatch() {
-+    delete m_object_store;
-+    delete m_cache_client;
-+}
-+
-+// TODO if connect fails, init will return error to high layer.
-+template <typename I>
-+void SharedReadOnlyObjectDispatch<I>::init() {
-+  auto cct = m_image_ctx->cct;
-+  ldout(cct, 5) << dendl;
-+
-+  if (m_image_ctx->parent != nullptr) {
-+    //TODO(): should we cover multi-leveled clone?
-+    ldout(cct, 5) << "child image: skipping SRO cache client" << dendl;
-+    return;
-+  }
-+
-+  ldout(cct, 5) << "parent image: setup SRO cache client = " << dendl;
-+
-+  std::string controller_path = ((CephContext*)cct)->_conf.get_val<std::string>("rbd_shared_cache_sock");
-+  m_cache_client = new ceph::immutable_obj_cache::CacheClient(controller_path.c_str(),
-+    ([&](std::string s){client_handle_request(s);}), m_image_ctx->cct);
-+
-+  int ret = m_cache_client->connect();
-+  if (ret < 0) {
-+    ldout(cct, 5) << "SRO cache client fail to connect with local controller: "
-+                  << "please start rbd-cache daemon"
-+		  << dendl;
-+  } else {
-+    ldout(cct, 5) << "SRO cache client to register volume on rbd-cache daemon: "
-+                   << "name = " << m_image_ctx->id 
-+                   << dendl;
-+
-+    ret = m_cache_client->register_volume(m_image_ctx->data_ctx.get_pool_name(),
-+                                          m_image_ctx->id, m_image_ctx->size);
-+
-+    if (ret >= 0) {
-+      // add ourself to the IO object dispatcher chain
-+      m_image_ctx->io_object_dispatcher->register_object_dispatch(this);
-+    }
-+  }
-+}
-+
-+template <typename I>
-+bool SharedReadOnlyObjectDispatch<I>::read(
-+    const std::string &oid, uint64_t object_no, uint64_t object_off,
-+    uint64_t object_len, librados::snap_t snap_id, int op_flags,
-+    const ZTracer::Trace &parent_trace, ceph::bufferlist* read_data,
-+    io::ExtentMap* extent_map, int* object_dispatch_flags,
-+    io::DispatchResult* dispatch_result, Context** on_finish,
-+    Context* on_dispatched) {
-+  auto cct = m_image_ctx->cct;
-+  ldout(cct, 20) << "object_no=" << object_no << " " << object_off << "~"
-+                 << object_len << dendl;
-+
-+  // if any session fails, later reads will go to rados
-+  if(!m_cache_client->is_session_work()) {
-+    *dispatch_result = io::DISPATCH_RESULT_CONTINUE;
-+    on_dispatched->complete(0);
-+    return true;
-+    // TODO(): fix domain socket error
-+  }
-+
-+  auto ctx = new FunctionContext([this, oid, object_off, object_len,
-+    read_data, dispatch_result, on_dispatched](bool cache) {
-+    handle_read_cache(cache, oid, object_off, object_len,
-+                      read_data, dispatch_result, on_dispatched);
-+  });
-+
-+  if (m_cache_client && m_cache_client->is_session_work() && m_object_store) {
-+    m_cache_client->lookup_object(m_image_ctx->data_ctx.get_pool_name(),
-+      m_image_ctx->id, oid, ctx);
-+  }
-+  return true;
-+}
-+
-+template <typename I>
-+int SharedReadOnlyObjectDispatch<I>::handle_read_cache(
-+    bool cache, const std::string &oid, uint64_t object_off,
-+    uint64_t object_len, ceph::bufferlist* read_data,
-+    io::DispatchResult* dispatch_result, Context* on_dispatched) {
-+  auto cct = m_image_ctx->cct;
-+  ldout(cct, 20) << dendl;
-+
-+  // try to read from parent image
-+  if (cache) {
-+    int r = m_object_store->read_object(oid, read_data, object_off, object_len, on_dispatched);
-+    //int r = object_len;
-+    if (r != 0) {
-+      *dispatch_result = io::DISPATCH_RESULT_COMPLETE;
-+      //TODO(): complete in syncfile
-+      on_dispatched->complete(r);
-+      ldout(cct, 20) << "read cache: " << *dispatch_result <<dendl;
-+      return true;
-+    }
-+  } else {
-+    *dispatch_result = io::DISPATCH_RESULT_CONTINUE;
-+    on_dispatched->complete(0);
-+    ldout(cct, 20) << "read rados: " << *dispatch_result <<dendl;
-+    return false;
-+  }
-+}
-+template <typename I>
-+void SharedReadOnlyObjectDispatch<I>::client_handle_request(std::string msg) {
-+  auto cct = m_image_ctx->cct;
-+  ldout(cct, 20) << dendl;
-+
-+  ceph::immutable_obj_cache::rbdsc_req_type_t *io_ctx = (ceph::immutable_obj_cache::rbdsc_req_type_t*)(msg.c_str());
-+
-+  switch (io_ctx->type) {
-+    case ceph::immutable_obj_cache::RBDSC_REGISTER_REPLY: {
-+      // open cache handler for volume
-+      ldout(cct, 20) << "SRO cache client open cache handler" << dendl;
-+      m_object_store = new SharedPersistentObjectCacher<I>(m_image_ctx, m_image_ctx->shared_cache_path);
-+
-+      break;
-+    }
-+    case ceph::immutable_obj_cache::RBDSC_READ_REPLY: {
-+      ldout(cct, 20) << "SRO cache client start to read cache" << dendl;
-+      //TODO(): should call read here
-+
-+      break;
-+    }
-+    case ceph::immutable_obj_cache::RBDSC_READ_RADOS: {
-+      ldout(cct, 20) << "SRO cache client start to read rados" << dendl;
-+      //TODO(): should call read here
-+
-+      break;
-+    }
-+    default: ldout(cct, 20) << "nothing" << io_ctx->type <<dendl;
-+      break;
-+
-+  }
-+}
-+
-+} // namespace cache
-+} // namespace librbd
-+
-+template class librbd::cache::SharedReadOnlyObjectDispatch<librbd::ImageCtx>;
-diff --git a/src/librbd/cache/SharedReadOnlyObjectDispatch.h b/src/librbd/cache/SharedReadOnlyObjectDispatch.h
-new file mode 100644
-index 0000000..9b56da9
---- /dev/null
-+++ b/src/librbd/cache/SharedReadOnlyObjectDispatch.h
-@@ -0,0 +1,126 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#ifndef CEPH_LIBRBD_CACHE_SHARED_PERSISTENT_OBJECT_CACHER_OBJECT_DISPATCH_H
-+#define CEPH_LIBRBD_CACHE_SHARED_PERSISTENT_OBJECT_CACHER_OBJECT_DISPATCH_H
-+
-+#include "common/Mutex.h"
-+#include "SharedPersistentObjectCacher.h"
-+#include "librbd/io/ObjectDispatchInterface.h"
-+#include "tools/ceph_immutable_object_cache/CacheClient.h"
-+
-+
-+namespace librbd {
-+
-+class ImageCtx;
-+
-+namespace cache {
-+
-+template <typename ImageCtxT = ImageCtx>
-+class SharedReadOnlyObjectDispatch : public io::ObjectDispatchInterface {
-+public:
-+  static SharedReadOnlyObjectDispatch* create(ImageCtxT* image_ctx) {
-+    return new SharedReadOnlyObjectDispatch(image_ctx);
-+  }
-+
-+  SharedReadOnlyObjectDispatch(ImageCtxT* image_ctx);
-+  ~SharedReadOnlyObjectDispatch() override;
-+
-+  io::ObjectDispatchLayer get_object_dispatch_layer() const override {
-+    return io::OBJECT_DISPATCH_LAYER_SHARED_PERSISTENT_CACHE;
-+  }
-+
-+  void init();
-+  void shut_down(Context* on_finish) {
-+    m_image_ctx->op_work_queue->queue(on_finish, 0);
-+  }
-+
-+  bool read(
-+      const std::string &oid, uint64_t object_no, uint64_t object_off,
-+      uint64_t object_len, librados::snap_t snap_id, int op_flags,
-+      const ZTracer::Trace &parent_trace, ceph::bufferlist* read_data,
-+      io::ExtentMap* extent_map, int* object_dispatch_flags,
-+      io::DispatchResult* dispatch_result, Context** on_finish,
-+      Context* on_dispatched) override;
-+
-+  bool discard(
-+      const std::string &oid, uint64_t object_no, uint64_t object_off,
-+      uint64_t object_len, const ::SnapContext &snapc, int discard_flags,
-+      const ZTracer::Trace &parent_trace, int* object_dispatch_flags,
-+      uint64_t* journal_tid, io::DispatchResult* dispatch_result,
-+      Context** on_finish, Context* on_dispatched) {
-+    return false;
-+  }
-+
-+  bool write(
-+      const std::string &oid, uint64_t object_no, uint64_t object_off,
-+      ceph::bufferlist&& data, const ::SnapContext &snapc, int op_flags,
-+      const ZTracer::Trace &parent_trace, int* object_dispatch_flags,
-+      uint64_t* journal_tid, io::DispatchResult* dispatch_result,
-+      Context** on_finish, Context* on_dispatched) {
-+    return false;
-+  }
-+
-+  bool write_same(
-+      const std::string &oid, uint64_t object_no, uint64_t object_off,
-+      uint64_t object_len, io::Extents&& buffer_extents,
-+      ceph::bufferlist&& data, const ::SnapContext &snapc, int op_flags,
-+      const ZTracer::Trace &parent_trace, int* object_dispatch_flags,
-+      uint64_t* journal_tid, io::DispatchResult* dispatch_result,
-+      Context** on_finish, Context* on_dispatched) {
-+    return false;
-+  }
-+
-+  bool compare_and_write(
-+      const std::string &oid, uint64_t object_no, uint64_t object_off,
-+      ceph::bufferlist&& cmp_data, ceph::bufferlist&& write_data,
-+      const ::SnapContext &snapc, int op_flags,
-+      const ZTracer::Trace &parent_trace, uint64_t* mismatch_offset,
-+      int* object_dispatch_flags, uint64_t* journal_tid,
-+      io::DispatchResult* dispatch_result, Context** on_finish,
-+      Context* on_dispatched) {
-+    return false;
-+  }
-+
-+  bool flush(
-+      io::FlushSource flush_source, const ZTracer::Trace &parent_trace,
-+      io::DispatchResult* dispatch_result, Context** on_finish,
-+      Context* on_dispatched) {
-+    return false;
-+  }
-+
-+  bool invalidate_cache(Context* on_finish) {
-+    return false;
-+  }
-+
-+  bool reset_existence_cache(Context* on_finish) {
-+    return false;
-+  }
-+
-+  void extent_overwritten(
-+      uint64_t object_no, uint64_t object_off, uint64_t object_len,
-+      uint64_t journal_tid, uint64_t new_journal_tid) {
-+  }
-+
-+private:
-+
-+  int handle_read_cache(
-+      bool cache,
-+      const std::string &oid, uint64_t object_off,
-+      uint64_t object_len, ceph::bufferlist* read_data,
-+      io::DispatchResult* dispatch_result,
-+      Context* on_dispatched);
-+  void client_handle_request(std::string msg);
-+
-+  ImageCtxT* m_image_ctx;
-+
-+  ceph::immutable_obj_cache::CacheClient *m_cache_client = nullptr;
-+  SharedPersistentObjectCacher<ImageCtxT> *m_object_store = nullptr;
-+};
-+
-+} // namespace cache
-+} // namespace librbd
-+
-+extern template class librbd::cache::SharedReadOnlyObjectDispatch<librbd::ImageCtx>;
-+
-+#endif // CEPH_LIBRBD_CACHE_OBJECT_CACHER_OBJECT_DISPATCH_H
-diff --git a/src/librbd/image/OpenRequest.cc b/src/librbd/image/OpenRequest.cc
-index 30a7b66..57ce92f 100644
---- a/src/librbd/image/OpenRequest.cc
-+++ b/src/librbd/image/OpenRequest.cc
-@@ -8,7 +8,7 @@
- #include "librbd/ImageCtx.h"
- #include "librbd/Utils.h"
- #include "librbd/cache/ObjectCacherObjectDispatch.h"
--#include "librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc"
-+#include "librbd/cache/SharedReadOnlyObjectDispatch.cc"
- #include "librbd/image/CloseRequest.h"
- #include "librbd/image/RefreshRequest.h"
- #include "librbd/image/SetSnapRequest.h"
-@@ -457,7 +457,7 @@ Context *OpenRequest<I>::send_init_cache(int *result) {
-     // enable Shared Read-only cache for parent image
-     if (m_image_ctx->child != nullptr && m_image_ctx->shared_cache_enabled ) {
-       ldout(cct, 10) << this << " " << "setting up parent cache"<< dendl;
--      auto sro_cache = cache::SharedPersistentObjectCacherObjectDispatch<I>::create(m_image_ctx);
-+      auto sro_cache = cache::SharedReadOnlyObjectDispatch<I>::create(m_image_ctx);
-       sro_cache->init();
-     }
- 
-diff --git a/src/tools/CMakeLists.txt b/src/tools/CMakeLists.txt
-index 72ab342..f7c5872 100644
---- a/src/tools/CMakeLists.txt
-+++ b/src/tools/CMakeLists.txt
-@@ -99,7 +99,6 @@ endif(WITH_CEPHFS)
- if(WITH_RBD)
-   add_subdirectory(rbd)
-   add_subdirectory(rbd_mirror)
--  add_subdirectory(rbd_cache)
-   if(LINUX)
-     add_subdirectory(rbd_nbd)
-   endif()
-@@ -108,4 +107,5 @@ if(WITH_RBD)
-   endif()
- endif(WITH_RBD)
- 
-+add_subdirectory(ceph_immutable_object_cache)
- add_subdirectory(ceph-dencoder)
-diff --git a/src/tools/ceph_immutable_object_cache/CMakeLists.txt b/src/tools/ceph_immutable_object_cache/CMakeLists.txt
-new file mode 100644
-index 0000000..c7c7af3
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/CMakeLists.txt
-@@ -0,0 +1,11 @@
-+add_executable(ceph-immutable-object-cache
-+  ${CMAKE_SOURCE_DIR}/src/librbd/cache/SharedPersistentObjectCacherFile.cc
-+  ObjectCacheStore.cc
-+  CacheController.cc
-+  CacheServer.cc
-+  CacheSession.cc
-+  main.cc)
-+target_link_libraries(ceph-immutable-object-cache
-+  librados
-+  global)
-+install(TARGETS ceph-immutable-object-cache DESTINATION bin)
-diff --git a/src/tools/ceph_immutable_object_cache/CacheClient.cc b/src/tools/ceph_immutable_object_cache/CacheClient.cc
-new file mode 100644
-index 0000000..a7116bf
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/CacheClient.cc
-@@ -0,0 +1,205 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#include "CacheClient.h"
-+
-+#define dout_context g_ceph_context
-+#define dout_subsys ceph_subsys_immutable_obj_cache
-+#undef dout_prefix
-+#define dout_prefix *_dout << "ceph::cache::CacheControllerSocketClient: " << this << " " \
-+                           << __func__ << ": "
-+
-+
-+using boost::asio::local::stream_protocol;
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+ CacheClient::CacheClient(const std::string& file, ClientProcessMsg processmsg, CephContext* ceph_ctx)
-+    : m_io_service_work(m_io_service),
-+      m_dm_socket(m_io_service),
-+      m_client_process_msg(processmsg),
-+      m_ep(stream_protocol::endpoint(file)),
-+      m_session_work(false),
-+      cct(ceph_ctx)
-+  {
-+     // TODO wrapper io_service
-+     std::thread thd([this](){m_io_service.run();});
-+     thd.detach();
-+  }
-+
-+  void CacheClient::run(){
-+  }
-+
-+  bool CacheClient::is_session_work() {
-+    return m_session_work.load() == true;
-+  }
-+
-+  // just when error occur, call this method.
-+  void CacheClient::close() {
-+    m_session_work.store(false);
-+    boost::system::error_code close_ec;
-+    m_dm_socket.close(close_ec);
-+    if(close_ec) {
-+       ldout(cct, 20) << "close: " << close_ec.message() << dendl;
-+    }
-+    ldout(cct, 20) << "session don't work, later all request will be dispatched to rados layer" << dendl;
-+  }
-+
-+  int CacheClient::connect() {
-+    boost::system::error_code ec;
-+    m_dm_socket.connect(m_ep, ec);
-+    if(ec) {
-+      if(ec == boost::asio::error::connection_refused) {
-+        ldout(cct, 20) << ec.message() << " : maybe rbd-cache Controller don't startup. "
-+                  << "Now data will be read from ceph cluster " << dendl;
-+      } else {
-+        ldout(cct, 20) << "connect: " << ec.message() << dendl;
-+      }
-+
-+      if(m_dm_socket.is_open()) {
-+        // Set to indicate what error occurred, if any.
-+        // Note that, even if the function indicates an error,
-+        // the underlying descriptor is closed.
-+        boost::system::error_code close_ec;
-+        m_dm_socket.close(close_ec);
-+        if(close_ec) {
-+          ldout(cct, 20) << "close: " << close_ec.message() << dendl;
-+        }
-+      }
-+      return -1;
-+    }
-+
-+    ldout(cct, 20) <<"connect success"<< dendl;
-+
-+    return 0;
-+  }
-+
-+  int CacheClient::register_volume(std::string pool_name, std::string vol_name, uint64_t vol_size) {
-+    // cache controller will init layout
-+    rbdsc_req_type_t *message = new rbdsc_req_type_t();
-+    message->type = RBDSC_REGISTER;
-+    memcpy(message->pool_name, pool_name.c_str(), pool_name.size());
-+    memcpy(message->vol_name, vol_name.c_str(), vol_name.size());
-+    message->vol_size = vol_size;
-+    message->offset = 0;
-+    message->length = 0;
-+
-+    uint64_t ret;
-+    boost::system::error_code ec;
-+
-+    ret = boost::asio::write(m_dm_socket, boost::asio::buffer((char*)message, message->size()), ec);
-+    if(ec) {
-+      ldout(cct, 20) << "write fails : " << ec.message() << dendl;
-+      return -1;
-+    }
-+
-+    if(ret != message->size()) {
-+      ldout(cct, 20) << "write fails : ret != send_bytes " << dendl;
-+      return -1;
-+    }
-+
-+    // hard code TODO
-+    ret = boost::asio::read(m_dm_socket, boost::asio::buffer(m_recv_buffer, RBDSC_MSG_LEN), ec);
-+    if(ec == boost::asio::error::eof) {
-+      ldout(cct, 20) << "recv eof" << dendl;
-+      return -1;
-+    }
-+
-+    if(ec) {
-+      ldout(cct, 20) << "write fails : " << ec.message() << dendl;
-+      return -1;
-+    }
-+
-+    if(ret != RBDSC_MSG_LEN) {
-+      ldout(cct, 20) << "write fails : ret != receive bytes " << dendl;
-+      return -1;
-+    }
-+
-+    m_client_process_msg(std::string(m_recv_buffer, ret));
-+
-+    delete message;
-+
-+    ldout(cct, 20) << "register volume success" << dendl;
-+
-+    // TODO
-+    m_session_work.store(true);
-+
-+    return 0;
-+  }
-+
-+  // if occur any error, we just return false. Then read from rados.
-+  int CacheClient::lookup_object(std::string pool_name, std::string vol_name, std::string object_id, Context* on_finish) {
-+    rbdsc_req_type_t *message = new rbdsc_req_type_t();
-+    message->type = RBDSC_READ;
-+    memcpy(message->pool_name, pool_name.c_str(), pool_name.size());
-+    memcpy(message->vol_name, object_id.c_str(), object_id.size());
-+    message->vol_size = 0;
-+    message->offset = 0;
-+    message->length = 0;
-+
-+    boost::asio::async_write(m_dm_socket,
-+                             boost::asio::buffer((char*)message, message->size()),
-+                             boost::asio::transfer_exactly(RBDSC_MSG_LEN),
-+        [this, on_finish, message](const boost::system::error_code& err, size_t cb) {
-+          delete message;
-+          if(err) {
-+            ldout(cct, 20) << "lookup_object: async_write fails." << err.message() << dendl;
-+            close();
-+            on_finish->complete(false);
-+            return;
-+          }
-+          if(cb != RBDSC_MSG_LEN) {
-+            ldout(cct, 20) << "lookup_object: async_write fails. in-complete request" << dendl;
-+            close();
-+            on_finish->complete(false);
-+            return;
-+          }
-+          get_result(on_finish);
-+    });
-+
-+    return 0;
-+  }
-+
-+  void CacheClient::get_result(Context* on_finish) {
-+    boost::asio::async_read(m_dm_socket, boost::asio::buffer(m_recv_buffer, RBDSC_MSG_LEN),
-+                            boost::asio::transfer_exactly(RBDSC_MSG_LEN),
-+        [this, on_finish](const boost::system::error_code& err, size_t cb) {
-+          if(err == boost::asio::error::eof) {
-+            ldout(cct, 20) << "get_result: ack is EOF." << dendl;
-+            close();
-+            on_finish->complete(false);
-+            return;
-+          }
-+          if(err) {
-+            ldout(cct, 20) << "get_result: async_read fails:" << err.message() << dendl;
-+            close();
-+            on_finish->complete(false); // TODO replace this assert with some metohds.
-+            return;
-+          }
-+          if (cb != RBDSC_MSG_LEN) {
-+            close();
-+            ldout(cct, 20) << "get_result: in-complete ack." << dendl;
-+	    on_finish->complete(false); // TODO: replace this assert with some methods.
-+          }
-+
-+	  rbdsc_req_type_t *io_ctx = (rbdsc_req_type_t*)(m_recv_buffer);
-+
-+          // TODO: re-occur yuan's bug
-+          if(io_ctx->type == RBDSC_READ) {
-+            ldout(cct, 20) << "get rbdsc_read... " << dendl;
-+            assert(0);
-+          }
-+
-+          if (io_ctx->type == RBDSC_READ_REPLY) {
-+	    on_finish->complete(true);
-+            return;
-+          } else {
-+	    on_finish->complete(false);
-+            return;
-+          }
-+    });
-+  }
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-diff --git a/src/tools/ceph_immutable_object_cache/CacheClient.h b/src/tools/ceph_immutable_object_cache/CacheClient.h
-new file mode 100644
-index 0000000..d82ab8f
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/CacheClient.h
-@@ -0,0 +1,53 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#ifndef CEPH_CACHE_CLIENT_H
-+#define CEPH_CACHE_CLIENT_H
-+
-+#include <atomic>
-+#include <boost/asio.hpp>
-+#include <boost/bind.hpp>
-+#include <boost/asio/error.hpp>
-+#include <boost/algorithm/string.hpp>
-+#include "librbd/ImageCtx.h"
-+#include "include/assert.h"
-+#include "include/Context.h"
-+#include "SocketCommon.h"
-+
-+
-+using boost::asio::local::stream_protocol;
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+class CacheClient {
-+public:
-+  CacheClient(const std::string& file, ClientProcessMsg processmsg, CephContext* ceph_ctx);
-+  void run();
-+  bool is_session_work();
-+
-+  void close();
-+  int connect();
-+
-+  int register_volume(std::string pool_name, std::string vol_name, uint64_t vol_size);
-+  int lookup_object(std::string pool_name, std::string vol_name, std::string object_id, Context* on_finish);
-+  void get_result(Context* on_finish);
-+
-+private:
-+  boost::asio::io_service m_io_service;
-+  boost::asio::io_service::work m_io_service_work;
-+  stream_protocol::socket m_dm_socket;
-+  ClientProcessMsg m_client_process_msg;
-+  stream_protocol::endpoint m_ep;
-+  char m_recv_buffer[1024];
-+
-+  // atomic modfiy for this variable.
-+  // thread 1 : asio callback thread modify it.
-+  // thread 2 : librbd read it.
-+  std::atomic<bool> m_session_work;
-+  CephContext* cct;
-+};
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+#endif
-diff --git a/src/tools/ceph_immutable_object_cache/CacheController.cc b/src/tools/ceph_immutable_object_cache/CacheController.cc
-new file mode 100644
-index 0000000..cb636d2
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/CacheController.cc
-@@ -0,0 +1,117 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#include "CacheController.h"
-+
-+#define dout_context g_ceph_context
-+#define dout_subsys ceph_subsys_immutable_obj_cache
-+#undef dout_prefix
-+#define dout_prefix *_dout << "ceph::cache::CacheController: " << this << " " \
-+                           << __func__ << ": "
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+class ThreadPoolSingleton : public ThreadPool {
-+public:
-+  ContextWQ *op_work_queue;
-+
-+  explicit ThreadPoolSingleton(CephContext *cct)
-+    : ThreadPool(cct, "ceph::cache::thread_pool", "tp_librbd_cache", 32,
-+                 "pcache_threads"),
-+      op_work_queue(new ContextWQ("ceph::pcache_op_work_queue",
-+                    cct->_conf.get_val<int64_t>("rbd_op_thread_timeout"),
-+                    this)) {
-+    start();
-+  }
-+  ~ThreadPoolSingleton() override {
-+    op_work_queue->drain();
-+    delete op_work_queue;
-+
-+    stop();
-+  }
-+};
-+
-+
-+CacheController::CacheController(CephContext *cct, const std::vector<const char*> &args):
-+  m_args(args), m_cct(cct) {
-+
-+}
-+
-+CacheController::~CacheController() {
-+
-+}
-+
-+int CacheController::init() {
-+  ThreadPoolSingleton* thread_pool_singleton = &m_cct->lookup_or_create_singleton_object<ThreadPoolSingleton>(
-+    "ceph::cache::thread_pool", false, m_cct);
-+  pcache_op_work_queue = thread_pool_singleton->op_work_queue;
-+
-+  m_object_cache_store = new ObjectCacheStore(m_cct, pcache_op_work_queue);
-+  //TODO(): make this configurable
-+  int r = m_object_cache_store->init(true);
-+  if (r < 0) {
-+    lderr(m_cct) << "init error\n" << dendl;
-+  }
-+  return r;
-+}
-+
-+int CacheController::shutdown() {
-+  int r = m_object_cache_store->shutdown();
-+  return r;
-+}
-+
-+void CacheController::handle_signal(int signum){}
-+
-+void CacheController::run() {
-+  try {
-+    //TODO(): use new socket path
-+    std::string controller_path = m_cct->_conf.get_val<std::string>("rbd_shared_cache_sock");
-+    std::remove(controller_path.c_str()); 
-+    
-+    m_cache_server = new CacheServer(controller_path,
-+      ([&](uint64_t p, std::string s){handle_request(p, s);}), m_cct);
-+    m_cache_server->run();
-+  } catch (std::exception& e) {
-+    lderr(m_cct) << "Exception: " << e.what() << dendl;
-+  }
-+}
-+
-+void CacheController::handle_request(uint64_t session_id, std::string msg){
-+  rbdsc_req_type_t *io_ctx = (rbdsc_req_type_t*)(msg.c_str());
-+
-+  int ret = 0;
-+  
-+  switch (io_ctx->type) {
-+    case RBDSC_REGISTER: {
-+      // init cache layout for volume        
-+      m_object_cache_store->init_cache(io_ctx->vol_name, io_ctx->vol_size);
-+      io_ctx->type = RBDSC_REGISTER_REPLY;
-+      m_cache_server->send(session_id, std::string((char*)io_ctx, msg.size()));
-+
-+      break;
-+    }
-+    case RBDSC_READ: {
-+      // lookup object in local cache store
-+      ret = m_object_cache_store->lookup_object(io_ctx->pool_name, io_ctx->vol_name);
-+      if (ret < 0) {
-+        io_ctx->type = RBDSC_READ_RADOS;
-+      } else {
-+        io_ctx->type = RBDSC_READ_REPLY;
-+      }
-+      if (io_ctx->type != RBDSC_READ_REPLY) {
-+        assert(0);
-+      }
-+      m_cache_server->send(session_id, std::string((char*)io_ctx, msg.size()));
-+
-+      break;
-+    }
-+    ldout(m_cct, 5) << "can't recongize request" << dendl;
-+    assert(0); // TODO replace it.
-+  }
-+}
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+
-+
-diff --git a/src/tools/ceph_immutable_object_cache/CacheController.h b/src/tools/ceph_immutable_object_cache/CacheController.h
-new file mode 100644
-index 0000000..837fe36
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/CacheController.h
-@@ -0,0 +1,53 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#ifndef CEPH_CACHE_CONTROLLER_H
-+#define CEPH_CACHE_CONTROLLER_H
-+
-+#include "common/Formatter.h"
-+#include "common/admin_socket.h"
-+#include "common/debug.h"
-+#include "common/errno.h"
-+#include "common/ceph_context.h"
-+#include "common/Mutex.h"
-+#include "common/WorkQueue.h"
-+#include "include/rados/librados.hpp"
-+#include "include/rbd/librbd.h"
-+#include "include/assert.h"
-+#include "librbd/ImageCtx.h"
-+#include "librbd/ImageState.h"
-+#include "CacheServer.h"
-+#include "ObjectCacheStore.h"
-+
-+#include <thread>
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+class CacheController {
-+ public:
-+  CacheController(CephContext *cct, const std::vector<const char*> &args);
-+  ~CacheController();
-+
-+  int init();
-+
-+  int shutdown();
-+
-+  void handle_signal(int sinnum);
-+
-+  void run();
-+
-+  void handle_request(uint64_t sesstion_id, std::string msg);
-+
-+ private:
-+  CacheServer *m_cache_server;
-+  std::vector<const char*> m_args;
-+  CephContext *m_cct;
-+  ObjectCacheStore *m_object_cache_store;
-+  ContextWQ* pcache_op_work_queue;
-+};
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+
-+#endif
-diff --git a/src/tools/ceph_immutable_object_cache/CacheServer.cc b/src/tools/ceph_immutable_object_cache/CacheServer.cc
-new file mode 100644
-index 0000000..dd2d47e
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/CacheServer.cc
-@@ -0,0 +1,99 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#include "common/debug.h"
-+#include "common/ceph_context.h"
-+#include "CacheServer.h"
-+
-+#define dout_context g_ceph_context
-+#define dout_subsys ceph_subsys_immutable_obj_cache
-+#undef dout_prefix
-+#define dout_prefix *_dout << "ceph::cache::CacheControllerSocket: " << this << " " \
-+                           << __func__ << ": "
-+
-+
-+using boost::asio::local::stream_protocol;
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+CacheServer::CacheServer(const std::string& file, ProcessMsg processmsg, CephContext* cct)
-+  : cct(cct), m_server_process_msg(processmsg),
-+    m_local_path(file), m_acceptor(m_io_service) {}
-+
-+CacheServer::~CacheServer(){}
-+
-+void CacheServer::run() {
-+  bool ret;
-+  ret = start_accept();
-+  if(!ret) {
-+    return;
-+  }
-+  m_io_service.run();
-+}
-+
-+// TODO : use callback to replace this function.
-+void CacheServer::send(uint64_t session_id, std::string msg) {
-+  auto it = m_session_map.find(session_id);
-+  if (it != m_session_map.end()) {
-+    it->second->send(msg);
-+  } else {
-+    // TODO : why don't find existing session id ?
-+    ldout(cct, 20) << "don't find session id..." << dendl;
-+    assert(0);
-+  }
-+}
-+
-+// when creating one acceptor, can control every step in this way.
-+bool CacheServer::start_accept() {
-+  boost::system::error_code ec;
-+  m_acceptor.open(m_local_path.protocol(), ec);
-+  if(ec) {
-+    ldout(cct, 20) << "m_acceptor open fails: " << ec.message() << dendl;
-+    return false;
-+  }
-+
-+  // TODO control acceptor attribute.
-+
-+  m_acceptor.bind(m_local_path, ec);
-+  if(ec) {
-+    ldout(cct, 20) << "m_acceptor bind fails: " << ec.message() << dendl;
-+    return false;
-+  }
-+
-+  m_acceptor.listen(boost::asio::socket_base::max_connections, ec);
-+  if(ec) {
-+    ldout(cct, 20) << "m_acceptor listen fails: " << ec.message() << dendl;
-+    return false;
-+  }
-+
-+  accept();
-+  return true;
-+}
-+
-+void CacheServer::accept() {
-+  CacheSessionPtr new_session(new CacheSession(m_session_id, m_io_service, m_server_process_msg, cct));
-+  m_acceptor.async_accept(new_session->socket(),
-+      boost::bind(&CacheServer::handle_accept, this, new_session,
-+        boost::asio::placeholders::error));
-+}
-+
-+void CacheServer::handle_accept(CacheSessionPtr new_session, const boost::system::error_code& error) {
-+
-+  if(error) {
-+    lderr(cct) << "async accept fails : " << error.message() << dendl;
-+    assert(0); // TODO
-+  }
-+
-+  m_session_map.emplace(m_session_id, new_session);
-+  // TODO : session setting
-+  new_session->start();
-+  m_session_id++;
-+
-+  // lanuch next accept
-+  accept();
-+}
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+
-diff --git a/src/tools/ceph_immutable_object_cache/CacheServer.h b/src/tools/ceph_immutable_object_cache/CacheServer.h
-new file mode 100644
-index 0000000..6c5c133
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/CacheServer.h
-@@ -0,0 +1,54 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#ifndef CEPH_CACHE_SERVER_H
-+#define CEPH_CACHE_SERVER_H
-+
-+#include <cstdio>
-+#include <iostream>
-+#include <array>
-+#include <memory>
-+#include <string>
-+#include <boost/bind.hpp>
-+#include <boost/asio.hpp>
-+#include <boost/asio/error.hpp>
-+#include <boost/algorithm/string.hpp>
-+
-+#include "include/assert.h"
-+#include "SocketCommon.h"
-+#include "CacheSession.h"
-+
-+
-+using boost::asio::local::stream_protocol;
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+class CacheServer {
-+
-+ public:
-+  CacheServer(const std::string& file, ProcessMsg processmsg, CephContext* cct);
-+  ~CacheServer();
-+
-+  void run();
-+  void send(uint64_t session_id, std::string msg);
-+
-+ private:
-+  bool start_accept();
-+  void accept();
-+  void handle_accept(CacheSessionPtr new_session, const boost::system::error_code& error);
-+
-+ private:
-+  CephContext* cct;
-+  boost::asio::io_service m_io_service; // TODO wrapper it.
-+  ProcessMsg m_server_process_msg;
-+  stream_protocol::endpoint m_local_path;
-+  stream_protocol::acceptor m_acceptor;
-+  uint64_t m_session_id = 1;
-+  std::map<uint64_t, CacheSessionPtr> m_session_map;
-+};
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+
-+#endif
-diff --git a/src/tools/ceph_immutable_object_cache/CacheSession.cc b/src/tools/ceph_immutable_object_cache/CacheSession.cc
-new file mode 100644
-index 0000000..6cffb41
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/CacheSession.cc
-@@ -0,0 +1,115 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#include "common/debug.h"
-+#include "common/ceph_context.h"
-+#include "CacheSession.h"
-+
-+#define dout_context g_ceph_context
-+#define dout_subsys ceph_subsys_immutable_obj_cache
-+#undef dout_prefix
-+#define dout_prefix *_dout << "ceph::cache::CacheSession: " << this << " " \
-+                           << __func__ << ": "
-+
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+CacheSession::CacheSession(uint64_t session_id, boost::asio::io_service& io_service, ProcessMsg processmsg, CephContext* cct)
-+    : m_session_id(session_id), m_dm_socket(io_service), process_msg(processmsg), cct(cct)
-+    {}
-+
-+CacheSession::~CacheSession(){}
-+
-+stream_protocol::socket& CacheSession::socket() {
-+  return m_dm_socket;
-+}
-+
-+void CacheSession::start() {
-+  if(true) {
-+    serial_handing_request();
-+  } else {
-+    parallel_handing_request();
-+  }
-+}
-+// flow:
-+//
-+// recv request --> process request --> reply ack
-+//   |                                      |
-+//   --------------<-------------------------
-+void CacheSession::serial_handing_request() {
-+  boost::asio::async_read(m_dm_socket, boost::asio::buffer(m_buffer, RBDSC_MSG_LEN),
-+                          boost::asio::transfer_exactly(RBDSC_MSG_LEN),
-+                          boost::bind(&CacheSession::handle_read,
-+                                      shared_from_this(),
-+                                      boost::asio::placeholders::error,
-+                                      boost::asio::placeholders::bytes_transferred));
-+}
-+
-+// flow :
-+//
-+//              --> thread 1: process request
-+// recv request --> thread 2: process request --> reply ack
-+//              --> thread n: process request
-+//
-+void CacheSession::parallel_handing_request() {
-+  // TODO
-+}
-+
-+void CacheSession::handle_read(const boost::system::error_code& error, size_t bytes_transferred) {
-+  // when recv eof, the most proble is that client side close socket.
-+  // so, server side need to end handing_request
-+  if(error == boost::asio::error::eof) {
-+    ldout(cct, 20) << "session: async_read : " << error.message() << dendl;
-+    return;
-+  }
-+
-+  if(error) {
-+    ldout(cct, 20) << "session: async_read fails: " << error.message() << dendl;
-+    assert(0);
-+  }
-+
-+  if(bytes_transferred != RBDSC_MSG_LEN) {
-+    ldout(cct, 20) << "session : request in-complete. "<<dendl;
-+    assert(0);
-+  }
-+
-+  // TODO async_process can increse coding readable.
-+  // process_msg_callback call handle async_send
-+  process_msg(m_session_id, std::string(m_buffer, bytes_transferred));
-+}
-+
-+void CacheSession::handle_write(const boost::system::error_code& error, size_t bytes_transferred) {
-+  if (error) {
-+    ldout(cct, 20) << "session: async_write fails: " << error.message() << dendl;
-+    assert(0);
-+  }
-+
-+  if(bytes_transferred != RBDSC_MSG_LEN) {
-+    ldout(cct, 20) << "session : reply in-complete. "<<dendl;
-+    assert(0);
-+  }
-+
-+  boost::asio::async_read(m_dm_socket, boost::asio::buffer(m_buffer),
-+                          boost::asio::transfer_exactly(RBDSC_MSG_LEN),
-+                          boost::bind(&CacheSession::handle_read,
-+                          shared_from_this(),
-+                          boost::asio::placeholders::error,
-+                          boost::asio::placeholders::bytes_transferred));
-+
-+}
-+
-+void CacheSession::send(std::string msg) {
-+    boost::asio::async_write(m_dm_socket,
-+        boost::asio::buffer(msg.c_str(), msg.size()),
-+        boost::asio::transfer_exactly(RBDSC_MSG_LEN),
-+        boost::bind(&CacheSession::handle_write,
-+                    shared_from_this(),
-+                    boost::asio::placeholders::error,
-+                    boost::asio::placeholders::bytes_transferred));
-+
-+}
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+
-diff --git a/src/tools/ceph_immutable_object_cache/CacheSession.h b/src/tools/ceph_immutable_object_cache/CacheSession.h
-new file mode 100644
-index 0000000..ce2591b
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/CacheSession.h
-@@ -0,0 +1,58 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#ifndef CEPH_CACHE_SESSION_H
-+#define CEPH_CACHE_SESSION_H
-+
-+#include <iostream>
-+#include <string>
-+#include <boost/bind.hpp>
-+#include <boost/asio.hpp>
-+#include <boost/asio/error.hpp>
-+#include <boost/algorithm/string.hpp>
-+
-+#include "include/assert.h"
-+#include "SocketCommon.h"
-+
-+
-+using boost::asio::local::stream_protocol;
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+class CacheSession : public std::enable_shared_from_this<CacheSession> {
-+public:
-+  CacheSession(uint64_t session_id, boost::asio::io_service& io_service, ProcessMsg processmsg, CephContext* cct);
-+  ~CacheSession();
-+
-+  stream_protocol::socket& socket();
-+  void start();
-+  void serial_handing_request();
-+  void parallel_handing_request();
-+
-+private:
-+
-+  void handle_read(const boost::system::error_code& error, size_t bytes_transferred); 
-+
-+  void handle_write(const boost::system::error_code& error, size_t bytes_transferred);
-+
-+public:
-+  void send(std::string msg);
-+
-+private:
-+  uint64_t m_session_id;
-+  stream_protocol::socket m_dm_socket;
-+  ProcessMsg process_msg;
-+  CephContext* cct;
-+
-+  // Buffer used to store data received from the client.
-+  //std::array<char, 1024> data_;
-+  char m_buffer[1024];
-+};
-+
-+typedef std::shared_ptr<CacheSession> CacheSessionPtr;
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+
-+#endif
-diff --git a/src/tools/ceph_immutable_object_cache/ObjectCacheStore.cc b/src/tools/ceph_immutable_object_cache/ObjectCacheStore.cc
-new file mode 100644
-index 0000000..50721ca
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/ObjectCacheStore.cc
-@@ -0,0 +1,172 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#include "ObjectCacheStore.h"
-+
-+#define dout_context g_ceph_context
-+#define dout_subsys ceph_subsys_immutable_obj_cache
-+#undef dout_prefix
-+#define dout_prefix *_dout << "ceph::cache::ObjectCacheStore: " << this << " " \
-+                           << __func__ << ": "
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+ObjectCacheStore::ObjectCacheStore(CephContext *cct, ContextWQ* work_queue)
-+      : m_cct(cct), m_work_queue(work_queue),
-+        m_rados(new librados::Rados()) {
-+
-+  uint64_t object_cache_entries =
-+    cct->_conf.get_val<int64_t>("rbd_shared_cache_entries");
-+
-+  //TODO(): allow to set level
-+  m_policy = new SimplePolicy(object_cache_entries, 0.5);
-+}
-+
-+ObjectCacheStore::~ObjectCacheStore() {
-+  delete m_policy;
-+}
-+
-+int ObjectCacheStore::init(bool reset) {
-+
-+  int ret = m_rados->init_with_context(m_cct);
-+  if(ret < 0) {
-+    lderr(m_cct) << "fail to init Ceph context" << dendl;
-+    return ret;
-+  }
-+
-+  ret = m_rados->connect();
-+  if(ret < 0 ) {
-+    lderr(m_cct) << "fail to conect to cluster" << dendl;
-+    return ret;
-+  }
-+
-+  std::string cache_path = m_cct->_conf.get_val<std::string>("rbd_shared_cache_path");
-+  //TODO(): check and reuse existing cache objects
-+  if(reset) {
-+    std::string cmd = "exec rm -rf " + cache_path + "/rbd_cache*; exec mkdir -p " + cache_path;
-+    //TODO(): to use std::filesystem
-+    int r = system(cmd.c_str());
-+  }
-+
-+  evict_thd = new std::thread([this]{this->evict_thread_body();});
-+  return ret;
-+}
-+
-+int ObjectCacheStore::do_promote(std::string pool_name, std::string object_name) {
-+  int ret = 0;
-+  std::string cache_file_name =  pool_name + object_name;
-+
-+  //TODO(): lock on ioctx map
-+  if (m_ioctxs.find(pool_name) == m_ioctxs.end()) {
-+    librados::IoCtx* io_ctx = new librados::IoCtx();
-+    ret = m_rados->ioctx_create(pool_name.c_str(), *io_ctx);
-+    if (ret < 0) {
-+      lderr(m_cct) << "fail to create ioctx" << dendl;
-+      assert(0);
-+    }
-+    m_ioctxs.emplace(pool_name, io_ctx); 
-+  }
-+
-+  assert(m_ioctxs.find(pool_name) != m_ioctxs.end());
-+  
-+  librados::IoCtx* ioctx = m_ioctxs[pool_name]; 
-+
-+  librados::bufferlist* read_buf = new librados::bufferlist();
-+  int object_size = 4096*1024; //TODO(): read config from image metadata
-+
-+  //TODO(): async promote
-+  ret = promote_object(ioctx, object_name, read_buf, object_size);
-+  if (ret == -ENOENT) {
-+    read_buf->append(std::string(object_size, '0'));
-+    ret = 0;
-+  }
-+
-+  if( ret < 0) {
-+    lderr(m_cct) << "fail to read from rados" << dendl;
-+    return ret;
-+  }
-+
-+  // persistent to cache
-+  librbd::cache::SyncFile cache_file(m_cct, cache_file_name);
-+  cache_file.open();
-+  ret = cache_file.write_object_to_file(*read_buf, object_size);
-+  
-+  // update metadata
-+  assert(OBJ_CACHE_PROMOTING == m_policy->get_status(cache_file_name));
-+  m_policy->update_status(cache_file_name, OBJ_CACHE_PROMOTED);
-+  assert(OBJ_CACHE_PROMOTED == m_policy->get_status(cache_file_name));
-+
-+  return ret;
-+
-+}
-+ 
-+// return -1, client need to read data from cluster.
-+// return 0,  client directly read data from cache.
-+int ObjectCacheStore::lookup_object(std::string pool_name, std::string object_name) {
-+
-+  std::string cache_file_name =  pool_name + object_name;
-+
-+  CACHESTATUS ret;
-+  ret = m_policy->lookup_object(cache_file_name);
-+
-+  switch(ret) {
-+    case OBJ_CACHE_NONE:
-+      return do_promote(pool_name, object_name);
-+    case OBJ_CACHE_PROMOTED:
-+      return 0;
-+    case OBJ_CACHE_PROMOTING:
-+    default:
-+      return -1;
-+  }
-+}
-+
-+void ObjectCacheStore::evict_thread_body() {
-+  int ret;
-+  while(m_evict_go) {
-+    ret = evict_objects();
-+  }
-+}
-+
-+
-+int ObjectCacheStore::shutdown() {
-+  m_evict_go = false;
-+  evict_thd->join();
-+  m_rados->shutdown();
-+  return 0;
-+}
-+
-+int ObjectCacheStore::init_cache(std::string vol_name, uint64_t vol_size) {
-+  return 0;
-+}
-+
-+int ObjectCacheStore::lock_cache(std::string vol_name) {
-+  return 0;
-+}
-+
-+int ObjectCacheStore::promote_object(librados::IoCtx* ioctx, std::string object_name, librados::bufferlist* read_buf, uint64_t read_len) {
-+  int ret;
-+
-+  librados::AioCompletion* read_completion = librados::Rados::aio_create_completion(); 
-+
-+  ret = ioctx->aio_read(object_name, read_completion, read_buf, read_len, 0);
-+  if(ret < 0) {
-+    lderr(m_cct) << "fail to read from rados" << dendl;
-+    return ret;
-+  }
-+  read_completion->wait_for_complete();
-+  ret = read_completion->get_return_value();
-+  return ret;
-+  
-+}
-+
-+int ObjectCacheStore::evict_objects() {
-+  std::list<std::string> obj_list;
-+  m_policy->get_evict_list(&obj_list);
-+  for (auto& obj: obj_list) {
-+    //do_evict(obj);
-+  }
-+}
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-diff --git a/src/tools/ceph_immutable_object_cache/ObjectCacheStore.h b/src/tools/ceph_immutable_object_cache/ObjectCacheStore.h
-new file mode 100644
-index 0000000..d044b27
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/ObjectCacheStore.h
-@@ -0,0 +1,70 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#ifndef CEPH_CACHE_OBJECT_CACHE_STORE_H
-+#define CEPH_CACHE_OBJECT_CACHE_STORE_H
-+
-+#include "common/debug.h"
-+#include "common/errno.h"
-+#include "common/ceph_context.h"
-+#include "common/Mutex.h"
-+#include "include/rados/librados.hpp"
-+#include "include/rbd/librbd.h"
-+#include "librbd/ImageCtx.h"
-+#include "librbd/ImageState.h"
-+#include "librbd/cache/SharedPersistentObjectCacherFile.h"
-+#include "SimplePolicy.hpp"
-+
-+
-+using librados::Rados;
-+using librados::IoCtx;
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+typedef shared_ptr<librados::Rados> RadosRef;
-+typedef shared_ptr<librados::IoCtx> IoCtxRef;
-+
-+class ObjectCacheStore 
-+{
-+  public:
-+    ObjectCacheStore(CephContext *cct, ContextWQ* work_queue);
-+    ~ObjectCacheStore();
-+
-+    int init(bool reset);
-+
-+    int shutdown();
-+
-+    int lookup_object(std::string pool_name, std::string object_name);
-+
-+    int init_cache(std::string vol_name, uint64_t vol_size);
-+
-+    int lock_cache(std::string vol_name);
-+
-+  private:
-+    void evict_thread_body();
-+    int evict_objects();
-+
-+    int do_promote(std::string pool_name, std::string object_name);
-+
-+    int promote_object(librados::IoCtx*, std::string object_name,
-+                       librados::bufferlist* read_buf,
-+                       uint64_t length);
-+
-+    CephContext *m_cct;
-+    ContextWQ* m_work_queue;
-+    RadosRef m_rados;
-+
-+
-+    std::map<std::string, librados::IoCtx*> m_ioctxs;
-+
-+    librbd::cache::SyncFile *m_cache_file;
-+
-+    Policy* m_policy;
-+    std::thread* evict_thd;
-+    bool m_evict_go = false;
-+};
-+
-+} // namespace ceph
-+} // namespace immutable_obj_cache
-+#endif
-diff --git a/src/tools/ceph_immutable_object_cache/Policy.hpp b/src/tools/ceph_immutable_object_cache/Policy.hpp
-new file mode 100644
-index 0000000..8090202
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/Policy.hpp
-@@ -0,0 +1,33 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#ifndef CEPH_CACHE_POLICY_HPP
-+#define CEPH_CACHE_POLICY_HPP
-+
-+#include <list>
-+#include <string>
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+enum CACHESTATUS {
-+  OBJ_CACHE_NONE = 0,
-+  OBJ_CACHE_PROMOTING,
-+  OBJ_CACHE_PROMOTED,
-+};
-+
-+
-+class Policy {
-+public:
-+  Policy(){}
-+  virtual ~Policy(){};
-+  virtual CACHESTATUS lookup_object(std::string) = 0;
-+  virtual int evict_object(std::string&) = 0;
-+  virtual void update_status(std::string, CACHESTATUS) = 0;
-+  virtual CACHESTATUS get_status(std::string) = 0;
-+  virtual void get_evict_list(std::list<std::string>* obj_list) = 0;
-+};
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+#endif
-diff --git a/src/tools/ceph_immutable_object_cache/SimplePolicy.hpp b/src/tools/ceph_immutable_object_cache/SimplePolicy.hpp
-new file mode 100644
-index 0000000..757ee6a
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/SimplePolicy.hpp
-@@ -0,0 +1,163 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#ifndef CEPH_CACHE_SIMPLE_POLICY_HPP
-+#define CEPH_CACHE_SIMPLE_POLICY_HPP
-+
-+#include "Policy.hpp"
-+#include "include/lru.h"
-+#include "common/RWLock.h"
-+#include "common/Mutex.h"
-+
-+#include <vector>
-+#include <unordered_map>
-+#include <string>
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+
-+class SimplePolicy : public Policy {
-+public:
-+  SimplePolicy(uint64_t block_num, float watermark)
-+    : m_watermark(watermark), m_entry_count(block_num),
-+      m_cache_map_lock("rbd::cache::SimplePolicy::m_cache_map_lock"),
-+      m_free_list_lock("rbd::cache::SimplePolicy::m_free_list_lock")
-+  {
-+
-+    for(uint64_t i = 0; i < m_entry_count; i++) {
-+      m_free_list.push_back(new Entry());
-+    }
-+
-+  }
-+
-+  ~SimplePolicy() {
-+    for(uint64_t i = 0; i < m_entry_count; i++) {
-+      Entry* entry = reinterpret_cast<Entry*>(m_free_list.front());
-+      delete entry;
-+      m_free_list.pop_front();
-+    }
-+  }
-+
-+  CACHESTATUS lookup_object(std::string cache_file_name) {
-+
-+    //TODO(): check race condition
-+    RWLock::WLocker wlocker(m_cache_map_lock);
-+
-+    auto entry_it = m_cache_map.find(cache_file_name);
-+    if(entry_it == m_cache_map.end()) {
-+      Mutex::Locker locker(m_free_list_lock);
-+      Entry* entry = reinterpret_cast<Entry*>(m_free_list.front());
-+      assert(entry != nullptr);
-+      m_free_list.pop_front();
-+      entry->status = OBJ_CACHE_PROMOTING;
-+
-+      m_cache_map[cache_file_name] = entry;
-+
-+      return OBJ_CACHE_NONE;
-+    }
-+
-+    Entry* entry = entry_it->second;
-+
-+    if(entry->status == OBJ_CACHE_PROMOTED) {
-+      // touch it
-+      m_promoted_lru.lru_touch(entry);
-+    }
-+
-+    return entry->status;
-+  }
-+
-+  int evict_object(std::string& out_cache_file_name) {
-+    RWLock::WLocker locker(m_cache_map_lock);
-+
-+    return 1;
-+  }
-+
-+  // TODO(): simplify the logic
-+  void update_status(std::string file_name, CACHESTATUS new_status) {
-+    RWLock::WLocker locker(m_cache_map_lock);
-+
-+    Entry* entry;
-+    auto entry_it = m_cache_map.find(file_name);
-+
-+    // just check.
-+    if(new_status == OBJ_CACHE_PROMOTING) {
-+      assert(entry_it == m_cache_map.end());
-+    }
-+
-+    assert(entry_it != m_cache_map.end());
-+
-+    entry = entry_it->second;
-+
-+    // promoting is done, so update it.
-+    if(entry->status == OBJ_CACHE_PROMOTING && new_status== OBJ_CACHE_PROMOTED) {
-+      m_promoted_lru.lru_insert_top(entry);
-+      entry->status = new_status;
-+      return;
-+    }
-+
-+    assert(0);
-+  }
-+
-+  // get entry status
-+  CACHESTATUS get_status(std::string file_name) {
-+    RWLock::RLocker locker(m_cache_map_lock);
-+    auto entry_it = m_cache_map.find(file_name);
-+    if(entry_it == m_cache_map.end()) {
-+      return OBJ_CACHE_NONE;
-+    }
-+
-+    return entry_it->second->status;
-+  }
-+
-+  void get_evict_list(std::list<std::string>* obj_list) {
-+    RWLock::WLocker locker(m_cache_map_lock);
-+    // check free ratio, pop entries from LRU
-+    if (m_free_list.size() / m_entry_count < m_watermark) {
-+      int evict_num = 10; //TODO(): make this configurable
-+      for(int i = 0; i < evict_num; i++) {
-+        Entry* entry = reinterpret_cast<Entry*>(m_promoted_lru.lru_expire());
-+        if (entry == nullptr) {
-+	  continue;
-+        }
-+        std::string file_name = entry->cache_file_name;
-+        obj_list->push_back(file_name);
-+
-+        auto entry_it = m_cache_map.find(file_name);
-+        m_cache_map.erase(entry_it);
-+
-+        //mark this entry as free
-+	entry->status = OBJ_CACHE_NONE;
-+        Mutex::Locker locker(m_free_list_lock);
-+        m_free_list.push_back(entry);
-+      }
-+   }
-+  }
-+
-+private:
-+
-+  class Entry : public LRUObject {
-+    public:
-+      CACHESTATUS status;
-+      Entry() : status(OBJ_CACHE_NONE){}
-+      std::string cache_file_name;
-+      void encode(bufferlist &bl){}
-+      void decode(bufferlist::iterator &it){}
-+  };
-+
-+  float m_watermark;
-+  uint64_t m_entry_count;
-+
-+  std::unordered_map<std::string, Entry*> m_cache_map;
-+  RWLock m_cache_map_lock;
-+
-+  std::deque<Entry*> m_free_list;
-+  Mutex m_free_list_lock;
-+
-+  LRU m_promoted_lru; // include promoted, using status.
-+
-+};
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+#endif
-diff --git a/src/tools/ceph_immutable_object_cache/SocketCommon.h b/src/tools/ceph_immutable_object_cache/SocketCommon.h
-new file mode 100644
-index 0000000..53dca54
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/SocketCommon.h
-@@ -0,0 +1,54 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#ifndef CEPH_CACHE_SOCKET_COMMON_H
-+#define CEPH_CACHE_SOCKET_COMMON_H
-+
-+namespace ceph {
-+namespace immutable_obj_cache {
-+
-+static const int RBDSC_REGISTER        =  0X11;
-+static const int RBDSC_READ            =  0X12;
-+static const int RBDSC_LOOKUP          =  0X13;
-+static const int RBDSC_REGISTER_REPLY  =  0X14;
-+static const int RBDSC_READ_REPLY      =  0X15;
-+static const int RBDSC_LOOKUP_REPLY    =  0X16;
-+static const int RBDSC_READ_RADOS      =  0X17;
-+
-+
-+
-+typedef std::function<void(uint64_t, std::string)> ProcessMsg;
-+typedef std::function<void(std::string)> ClientProcessMsg;
-+typedef uint8_t rbdsc_req_type;
-+
-+//TODO(): switch to bufferlist
-+struct rbdsc_req_type_t {
-+  rbdsc_req_type type;
-+  uint64_t vol_size;
-+  uint64_t offset;
-+  uint64_t length;
-+  char pool_name[256];
-+  char vol_name[256];
-+
-+  uint64_t size() {
-+    return sizeof(rbdsc_req_type_t);
-+  }
-+
-+  std::string to_buffer() {
-+    std::stringstream ss;
-+    ss << type;
-+    ss << vol_size;
-+    ss << offset;
-+    ss << length;
-+    ss << pool_name;
-+    ss << vol_name;
-+
-+    return ss.str();
-+  }
-+};
-+
-+static const int RBDSC_MSG_LEN = sizeof(rbdsc_req_type_t);
-+
-+} // namespace immutable_obj_cache
-+} // namespace ceph
-+#endif
-diff --git a/src/tools/ceph_immutable_object_cache/main.cc b/src/tools/ceph_immutable_object_cache/main.cc
-new file mode 100644
-index 0000000..7a9131d
---- /dev/null
-+++ b/src/tools/ceph_immutable_object_cache/main.cc
-@@ -0,0 +1,85 @@
-+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-+// vim: ts=8 sw=2 smarttab
-+
-+#include "common/ceph_argparse.h"
-+#include "common/config.h"
-+#include "common/debug.h"
-+#include "common/errno.h"
-+#include "global/global_init.h"
-+#include "global/signal_handler.h"
-+#include "CacheController.h"
-+
-+#include <vector>
-+
-+ceph::immutable_obj_cache::CacheController *cachectl = nullptr;
-+
-+void usage() {
-+  std::cout << "usage: cache controller [options...]" << std::endl;
-+  std::cout << "options:\n";
-+  std::cout << "  -m monaddress[:port]      connect to specified monitor\n";
-+  std::cout << "  --keyring=<path>          path to keyring for local cluster\n";
-+  std::cout << "  --log-file=<logfile>       file to log debug output\n";
-+  std::cout << "  --debug-rbd-cachecontroller=<log-level>/<memory-level>  set rbd-mirror debug level\n";
-+  generic_server_usage();
-+}
-+
-+static void handle_signal(int signum)
-+{
-+  if (cachectl)
-+    cachectl->handle_signal(signum);
-+}
-+
-+int main(int argc, const char **argv)
-+{
-+  std::vector<const char*> args;
-+  env_to_vec(args);
-+  argv_to_vec(argc, argv, args);
-+
-+  auto cct = global_init(nullptr, args, CEPH_ENTITY_TYPE_CLIENT,
-+			 CODE_ENVIRONMENT_DAEMON,
-+			 CINIT_FLAG_UNPRIVILEGED_DAEMON_DEFAULTS);
-+
-+  for (auto i = args.begin(); i != args.end(); ++i) {
-+    if (ceph_argparse_flag(args, i, "-h", "--help", (char*)NULL)) {
-+      usage();
-+      return EXIT_SUCCESS;
-+    }
-+  }
-+
-+  if (g_conf()->daemonize) {
-+    global_init_daemonize(g_ceph_context);
-+  }
-+  g_ceph_context->enable_perf_counter();
-+
-+  common_init_finish(g_ceph_context);
-+
-+  init_async_signal_handler();
-+  register_async_signal_handler(SIGHUP, sighup_handler);
-+  register_async_signal_handler_oneshot(SIGINT, handle_signal);
-+  register_async_signal_handler_oneshot(SIGTERM, handle_signal);
-+
-+  std::vector<const char*> cmd_args;
-+  argv_to_vec(argc, argv, cmd_args);
-+
-+  // disable unnecessary librbd cache
-+  g_ceph_context->_conf.set_val_or_die("rbd_cache", "false");
-+
-+  cachectl = new ceph::immutable_obj_cache::CacheController(g_ceph_context, cmd_args);
-+  int r = cachectl->init();
-+  if (r < 0) {
-+    std::cerr << "failed to initialize: " << cpp_strerror(r) << std::endl;
-+    goto cleanup;
-+  }
-+
-+  cachectl->run();
-+
-+ cleanup:
-+  unregister_async_signal_handler(SIGHUP, sighup_handler);
-+  unregister_async_signal_handler(SIGINT, handle_signal);
-+  unregister_async_signal_handler(SIGTERM, handle_signal);
-+  shutdown_async_signal_handler();
-+
-+  delete cachectl;
-+
-+  return r < 0 ? EXIT_SUCCESS : EXIT_FAILURE;
-+}
-diff --git a/src/tools/rbd_cache/CMakeLists.txt b/src/tools/rbd_cache/CMakeLists.txt
-deleted file mode 100644
-index 597d802..0000000
---- a/src/tools/rbd_cache/CMakeLists.txt
-+++ /dev/null
-@@ -1,9 +0,0 @@
--add_executable(rbd-cache
--  ${CMAKE_SOURCE_DIR}/src/librbd/cache/SharedPersistentObjectCacherFile.cc
--  ObjectCacheStore.cc
--  CacheController.cc
--  main.cc)
--target_link_libraries(rbd-cache
--  librados
--  global)
--install(TARGETS rbd-cache DESTINATION bin)
-diff --git a/src/tools/rbd_cache/CacheController.cc b/src/tools/rbd_cache/CacheController.cc
-deleted file mode 100644
-index 620192c..0000000
---- a/src/tools/rbd_cache/CacheController.cc
-+++ /dev/null
-@@ -1,116 +0,0 @@
--// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
--// vim: ts=8 sw=2 smarttab
--
--#include "CacheController.h"
--
--#define dout_context g_ceph_context
--#define dout_subsys ceph_subsys_rbd_cache
--#undef dout_prefix
--#define dout_prefix *_dout << "rbd::cache::CacheController: " << this << " " \
--                           << __func__ << ": "
--
--namespace rbd {
--namespace cache {
--
--class ThreadPoolSingleton : public ThreadPool {
--public:
--  ContextWQ *op_work_queue;
--
--  explicit ThreadPoolSingleton(CephContext *cct)
--    : ThreadPool(cct, "librbd::cache::thread_pool", "tp_librbd_cache", 32,
--                 "pcache_threads"),
--      op_work_queue(new ContextWQ("librbd::pcache_op_work_queue",
--                    cct->_conf.get_val<int64_t>("rbd_op_thread_timeout"),
--                    this)) {
--    start();
--  }
--  ~ThreadPoolSingleton() override {
--    op_work_queue->drain();
--    delete op_work_queue;
--
--    stop();
--  }
--};
--
--
--CacheController::CacheController(CephContext *cct, const std::vector<const char*> &args):
--  m_args(args), m_cct(cct) {
--
--}
--
--CacheController::~CacheController() {
--
--}
--
--int CacheController::init() {
--  ThreadPoolSingleton* thread_pool_singleton = &m_cct->lookup_or_create_singleton_object<ThreadPoolSingleton>(
--    "rbd::cache::thread_pool", false, m_cct);
--  pcache_op_work_queue = thread_pool_singleton->op_work_queue;
--
--  m_object_cache_store = new ObjectCacheStore(m_cct, pcache_op_work_queue);
--  int r = m_object_cache_store->init(false);
--  if (r < 0) {
--    //derr << "init error\n" << dendl;
--  }
--  return r;
--}
--
--int CacheController::shutdown() {
--  int r = m_object_cache_store->shutdown();
--  return r;
--}
--
--void CacheController::handle_signal(int signum){}
--
--void CacheController::run() {
--  try {
--    //TODO(): use new socket path
--    std::string controller_path = m_cct->_conf.get_val<std::string>("rbd_shared_cache_sock");
--    std::remove(controller_path.c_str()); 
--    
--    m_cache_server = new CacheServer(controller_path,
--      ([&](uint64_t p, std::string s){handle_request(p, s);}), m_cct);
--    m_cache_server->run();
--  } catch (std::exception& e) {
--    std::cerr << "Exception: " << e.what() << "\n";
--  }
--}
--
--void CacheController::handle_request(uint64_t session_id, std::string msg){
--  rbdsc_req_type_t *io_ctx = (rbdsc_req_type_t*)(msg.c_str());
--
--  int ret = 0;
--  
--  switch (io_ctx->type) {
--    case RBDSC_REGISTER: {
--      // init cache layout for volume        
--      m_object_cache_store->init_cache(io_ctx->vol_name, io_ctx->vol_size);
--      io_ctx->type = RBDSC_REGISTER_REPLY;
--      m_cache_server->send(session_id, std::string((char*)io_ctx, msg.size()));
--
--      break;
--    }
--    case RBDSC_READ: {
--      // lookup object in local cache store
--      ret = m_object_cache_store->lookup_object(io_ctx->pool_name, io_ctx->vol_name);
--      if (ret < 0) {
--        io_ctx->type = RBDSC_READ_RADOS;
--      } else {
--        io_ctx->type = RBDSC_READ_REPLY;
--      }
--      if (io_ctx->type != RBDSC_READ_REPLY) {
--        assert(0);
--      }
--      m_cache_server->send(session_id, std::string((char*)io_ctx, msg.size()));
--
--      break;
--    }
--    std::cout<<"can't recongize request"<<std::endl;
--    assert(0); // TODO replace it.
--  }
--}
--
--} // namespace rbd
--} // namespace cache
--
--
-diff --git a/src/tools/rbd_cache/CacheController.h b/src/tools/rbd_cache/CacheController.h
-deleted file mode 100644
-index 0e23484..0000000
---- a/src/tools/rbd_cache/CacheController.h
-+++ /dev/null
-@@ -1,54 +0,0 @@
--#ifndef CACHE_CONTROLLER_H
--#define CACHE_CONTROLLER_H
--
--#include <thread>
--
--#include "common/Formatter.h"
--#include "common/admin_socket.h"
--#include "common/debug.h"
--#include "common/errno.h"
--#include "common/ceph_context.h"
--#include "common/Mutex.h"
--#include "common/WorkQueue.h"
--#include "include/rados/librados.hpp"
--#include "include/rbd/librbd.h"
--#include "include/assert.h"
--#include "librbd/ImageCtx.h"
--#include "librbd/ImageState.h"
--
--#include "CacheControllerSocket.hpp"
--#include "ObjectCacheStore.h"
--
--
--using boost::asio::local::stream_protocol;
--
--namespace rbd {
--namespace cache {
--
--class CacheController {
-- public:
--  CacheController(CephContext *cct, const std::vector<const char*> &args);
--  ~CacheController();
--
--  int init();
--
--  int shutdown();
--
--  void handle_signal(int sinnum);
--
--  void run();
--
--  void handle_request(uint64_t sesstion_id, std::string msg);
--
-- private:
--  CacheServer *m_cache_server;
--  std::vector<const char*> m_args;
--  CephContext *m_cct;
--  ObjectCacheStore *m_object_cache_store;
--  ContextWQ* pcache_op_work_queue;
--};
--
--} // namespace rbd
--} // namespace cache
--
--#endif
-diff --git a/src/tools/rbd_cache/CacheControllerSocket.hpp b/src/tools/rbd_cache/CacheControllerSocket.hpp
-deleted file mode 100644
-index 2ff7477..0000000
---- a/src/tools/rbd_cache/CacheControllerSocket.hpp
-+++ /dev/null
-@@ -1,228 +0,0 @@
--// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
--// vim: ts=8 sw=2 smarttab
--
--#ifndef CACHE_CONTROLLER_SOCKET_H
--#define CACHE_CONTROLLER_SOCKET_H
--
--#include <cstdio>
--#include <iostream>
--#include <array>
--#include <memory>
--#include <string>
--#include <boost/bind.hpp>
--#include <boost/asio.hpp>
--#include <boost/asio/error.hpp>
--#include <boost/algorithm/string.hpp>
--#include "CacheControllerSocketCommon.h"
--
--
--using boost::asio::local::stream_protocol;
--
--namespace rbd {
--namespace cache {
--
--class session : public std::enable_shared_from_this<session> {
--public:
--  session(uint64_t session_id, boost::asio::io_service& io_service, ProcessMsg processmsg)
--    : m_session_id(session_id), m_dm_socket(io_service), process_msg(processmsg) {}
--
--  stream_protocol::socket& socket() {
--    return m_dm_socket;
--  }
--
--  void start() {
--    if(true) {
--      serial_handing_request();
--    } else {
--      parallel_handing_request();
--    }
--  }
--  // flow:
--  //
--  // recv request --> process request --> reply ack
--  //   |                                      |
--  //   --------------<-------------------------
--  void serial_handing_request() {
--    boost::asio::async_read(m_dm_socket, boost::asio::buffer(m_buffer, RBDSC_MSG_LEN),
--                            boost::asio::transfer_exactly(RBDSC_MSG_LEN),
--                            boost::bind(&session::handle_read,
--                                        shared_from_this(),
--                                        boost::asio::placeholders::error,
--                                        boost::asio::placeholders::bytes_transferred));
--  }
--
--  // flow :
--  //
--  //              --> thread 1: process request
--  // recv request --> thread 2: process request --> reply ack
--  //              --> thread n: process request
--  //
--  void parallel_handing_request() {
--    // TODO
--  }
--
--private:
--
--  void handle_read(const boost::system::error_code& error, size_t bytes_transferred) {
--    // when recv eof, the most proble is that client side close socket.
--    // so, server side need to end handing_request
--    if(error == boost::asio::error::eof) {
--      std::cout<<"session: async_read : " << error.message() << std::endl;
--      return;
--    }
--
--    if(error) {
--      std::cout<<"session: async_read fails: " << error.message() << std::endl;
--      assert(0);
--    }
--
--    if(bytes_transferred != RBDSC_MSG_LEN) {
--      std::cout<<"session : request in-complete. "<<std::endl;
--      assert(0);
--    }
--
--    // TODO async_process can increse coding readable.
--    // process_msg_callback call handle async_send
--    process_msg(m_session_id, std::string(m_buffer, bytes_transferred));
--  }
--
--  void handle_write(const boost::system::error_code& error, size_t bytes_transferred) {
--    if (error) {
--      std::cout<<"session: async_write fails: " << error.message() << std::endl;
--      assert(0);
--    }
--
--    if(bytes_transferred != RBDSC_MSG_LEN) {
--      std::cout<<"session : reply in-complete. "<<std::endl;
--      assert(0);
--    }
--
--    boost::asio::async_read(m_dm_socket, boost::asio::buffer(m_buffer),
--                            boost::asio::transfer_exactly(RBDSC_MSG_LEN),
--                            boost::bind(&session::handle_read,
--                            shared_from_this(),
--                            boost::asio::placeholders::error,
--                            boost::asio::placeholders::bytes_transferred));
--
--  }
--
--public:
--  void send(std::string msg) {
--      boost::asio::async_write(m_dm_socket,
--          boost::asio::buffer(msg.c_str(), msg.size()),
--          boost::asio::transfer_exactly(RBDSC_MSG_LEN),
--          boost::bind(&session::handle_write,
--                      shared_from_this(),
--                      boost::asio::placeholders::error,
--                      boost::asio::placeholders::bytes_transferred));
--
--  }
--
--private:
--  uint64_t m_session_id;
--  stream_protocol::socket m_dm_socket;
--  ProcessMsg process_msg;
--
--  // Buffer used to store data received from the client.
--  //std::array<char, 1024> data_;
--  char m_buffer[1024];
--};
--
--typedef std::shared_ptr<session> session_ptr;
--
--class CacheServer {
--public:
--  CacheServer(const std::string& file, ProcessMsg processmsg, CephContext* cct)
--    : m_cct(cct), m_server_process_msg(processmsg),
--      m_local_path(file),
--      m_acceptor(m_io_service)
--  {}
--
--  void run() {
--    bool ret;
--    ret = start_accept();
--    if(!ret) {
--      return;
--    }
--    m_io_service.run();
--  }
--
--  // TODO : use callback to replace this function.
--  void send(uint64_t session_id, std::string msg) {
--    auto it = m_session_map.find(session_id);
--    if (it != m_session_map.end()) {
--      it->second->send(msg);
--    } else {
--      // TODO : why don't find existing session id ?
--      std::cout<<"don't find session id..."<<std::endl;
--      assert(0);
--    }
--  }
--
--private:
--  // when creating one acceptor, can control every step in this way.
--  bool start_accept() {
--    boost::system::error_code ec;
--    m_acceptor.open(m_local_path.protocol(), ec);
--    if(ec) {
--      std::cout << "m_acceptor open fails: " << ec.message() << std::endl;
--      return false;
--    }
--
--    // TODO control acceptor attribute.
--
--    m_acceptor.bind(m_local_path, ec);
--    if(ec) {
--      std::cout << "m_acceptor bind fails: " << ec.message() << std::endl;
--      return false;
--    }
--
--    m_acceptor.listen(boost::asio::socket_base::max_connections, ec);
--    if(ec) {
--      std::cout << "m_acceptor listen fails: " << ec.message() << std::endl;
--      return false;
--    }
--
--    accept();
--    return true;
--  }
--
--  void accept() {
--    session_ptr new_session(new session(m_session_id, m_io_service, m_server_process_msg));
--    m_acceptor.async_accept(new_session->socket(),
--        boost::bind(&CacheServer::handle_accept, this, new_session,
--          boost::asio::placeholders::error));
--  }
--
-- void handle_accept(session_ptr new_session, const boost::system::error_code& error) {
--    //TODO(): open librbd snap ... yuan
--
--    if(error) {
--      std::cout << "async accept fails : " << error.message() << std::endl;
--      assert(0); // TODO
--    }
--
--      // must put session into m_session_map at the front of session.start()
--    m_session_map.emplace(m_session_id, new_session);
--    // TODO : session setting
--    new_session->start();
--    m_session_id++;
--
--    // lanuch next accept
--    accept();
--  }
--
--private:
--  CephContext* m_cct;
--  boost::asio::io_service m_io_service; // TODO wrapper it.
--  ProcessMsg m_server_process_msg;
--  stream_protocol::endpoint m_local_path;
--  stream_protocol::acceptor m_acceptor;
--  uint64_t m_session_id = 1;
--  std::map<uint64_t, session_ptr> m_session_map;
--};
--
--} // namespace cache
--} // namespace rbd
--
--#endif
-diff --git a/src/tools/rbd_cache/CacheControllerSocketClient.hpp b/src/tools/rbd_cache/CacheControllerSocketClient.hpp
-deleted file mode 100644
-index 964f888..0000000
---- a/src/tools/rbd_cache/CacheControllerSocketClient.hpp
-+++ /dev/null
-@@ -1,229 +0,0 @@
--// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
--// vim: ts=8 sw=2 smarttab
--
--#ifndef CACHE_CONTROLLER_SOCKET_CLIENT_H
--#define CACHE_CONTROLLER_SOCKET_CLIENT_H
--
--#include <atomic>
--#include <boost/asio.hpp>
--#include <boost/bind.hpp>
--#include <boost/asio/error.hpp>
--#include <boost/algorithm/string.hpp>
--#include "librbd/ImageCtx.h"
--#include "include/assert.h"
--#include "include/Context.h"
--#include "CacheControllerSocketCommon.h"
--
--
--using boost::asio::local::stream_protocol;
--
--namespace rbd {
--namespace cache {
--
--class CacheClient {
--public:
--  CacheClient(const std::string& file, ClientProcessMsg processmsg, CephContext* ceph_ctx)
--    : m_io_service_work(m_io_service),
--      m_dm_socket(m_io_service),
--      m_client_process_msg(processmsg),
--      m_ep(stream_protocol::endpoint(file)),
--      m_session_work(false),
--      cct(ceph_ctx)
--  {
--     // TODO wrapper io_service
--     std::thread thd([this](){
--                      m_io_service.run();});
--     thd.detach();
--  }
--
--  void run(){
--  }
--
--  bool is_session_work() {
--    return m_session_work.load() == true;
--  }
--
--  // just when error occur, call this method.
--  void close() {
--    m_session_work.store(false);
--    boost::system::error_code close_ec;
--    m_dm_socket.close(close_ec);
--    if(close_ec) {
--       std::cout << "close: " << close_ec.message() << std::endl;
--    }
--    std::cout << "session don't work, later all request will be dispatched to rados layer" << std::endl;
--  }
--
--  int connect() {
--    boost::system::error_code ec;
--    m_dm_socket.connect(m_ep, ec);
--    if(ec) {
--      if(ec == boost::asio::error::connection_refused) {
--        std::cout << ec.message() << " : maybe rbd-cache Controller don't startup. "
--                  << "Now data will be read from ceph cluster " << std::endl;
--      } else {
--        std::cout << "connect: " << ec.message() << std::endl;
--      }
--
--      if(m_dm_socket.is_open()) {
--        // Set to indicate what error occurred, if any.
--        // Note that, even if the function indicates an error,
--        // the underlying descriptor is closed.
--        boost::system::error_code close_ec;
--        m_dm_socket.close(close_ec);
--        if(close_ec) {
--          std::cout << "close: " << close_ec.message() << std::endl;
--        }
--      }
--      return -1;
--    }
--
--    std::cout<<"connect success"<<std::endl;
--
--    return 0;
--  }
--
--  int register_volume(std::string pool_name, std::string vol_name, uint64_t vol_size) {
--    // cache controller will init layout
--    rbdsc_req_type_t *message = new rbdsc_req_type_t();
--    message->type = RBDSC_REGISTER;
--    memcpy(message->pool_name, pool_name.c_str(), pool_name.size());
--    memcpy(message->vol_name, vol_name.c_str(), vol_name.size());
--    message->vol_size = vol_size;
--    message->offset = 0;
--    message->length = 0;
--
--    uint64_t ret;
--    boost::system::error_code ec;
--
--    ret = boost::asio::write(m_dm_socket, boost::asio::buffer((char*)message, message->size()), ec);
--    if(ec) {
--      std::cout << "write fails : " << ec.message() << std::endl;
--      return -1;
--    }
--
--    if(ret != message->size()) {
--      std::cout << "write fails : ret != send_bytes "<< std::endl;
--      return -1;
--    }
--
--    // hard code TODO
--    ret = boost::asio::read(m_dm_socket, boost::asio::buffer(m_recv_buffer, RBDSC_MSG_LEN), ec);
--    if(ec == boost::asio::error::eof) {
--      std::cout<< "recv eof"<<std::endl;
--      return -1;
--    }
--
--    if(ec) {
--      std::cout << "write fails : " << ec.message() << std::endl;
--      return -1;
--    }
--
--    if(ret != RBDSC_MSG_LEN) {
--      std::cout << "write fails : ret != receive bytes " << std::endl;
--      return -1;
--    }
--
--    m_client_process_msg(std::string(m_recv_buffer, ret));
--
--    delete message;
--
--    std::cout << "register volume success" << std::endl;
--
--    // TODO
--    m_session_work.store(true);
--
--    return 0;
--  }
--
--  // if occur any error, we just return false. Then read from rados.
--  int lookup_object(std::string pool_name, std::string vol_name, std::string object_id, Context* on_finish) {
--    rbdsc_req_type_t *message = new rbdsc_req_type_t();
--    message->type = RBDSC_READ;
--    memcpy(message->pool_name, pool_name.c_str(), pool_name.size());
--    memcpy(message->vol_name, object_id.c_str(), object_id.size());
--    message->vol_size = 0;
--    message->offset = 0;
--    message->length = 0;
--
--    boost::asio::async_write(m_dm_socket,
--                             boost::asio::buffer((char*)message, message->size()),
--                             boost::asio::transfer_exactly(RBDSC_MSG_LEN),
--        [this, on_finish, message](const boost::system::error_code& err, size_t cb) {
--          delete message;
--          if(err) {
--            std::cout<< "lookup_object: async_write fails." << err.message() << std::endl;
--            close();
--            on_finish->complete(false);
--            return;
--          }
--          if(cb != RBDSC_MSG_LEN) {
--            std::cout<< "lookup_object: async_write fails. in-complete request" <<std::endl;
--            close();
--            on_finish->complete(false);
--            return;
--          }
--          get_result(on_finish);
--    });
--
--    return 0;
--  }
--
--  void get_result(Context* on_finish) {
--    boost::asio::async_read(m_dm_socket, boost::asio::buffer(m_recv_buffer, RBDSC_MSG_LEN),
--                            boost::asio::transfer_exactly(RBDSC_MSG_LEN),
--        [this, on_finish](const boost::system::error_code& err, size_t cb) {
--          if(err == boost::asio::error::eof) {
--            std::cout<<"get_result: ack is EOF." << std::endl;
--            close();
--            on_finish->complete(false);
--            return;
--          }
--          if(err) {
--            std::cout<< "get_result: async_read fails:" << err.message() << std::endl;
--            close();
--            on_finish->complete(false); // TODO replace this assert with some metohds.
--            return;
--          }
--          if (cb != RBDSC_MSG_LEN) {
--            close();
--            std::cout << "get_result: in-complete ack." << std::endl;
--	    on_finish->complete(false); // TODO: replace this assert with some methods.
--          }
--
--	  rbdsc_req_type_t *io_ctx = (rbdsc_req_type_t*)(m_recv_buffer);
--
--          // TODO: re-occur yuan's bug
--          if(io_ctx->type == RBDSC_READ) {
--            std::cout << "get rbdsc_read... " << std::endl;
--            assert(0);
--          }
--
--          if (io_ctx->type == RBDSC_READ_REPLY) {
--	    on_finish->complete(true);
--            return;
--          } else {
--	    on_finish->complete(false);
--            return;
--          }
--    });
--  }
--
--private:
--  boost::asio::io_service m_io_service;
--  boost::asio::io_service::work m_io_service_work;
--  stream_protocol::socket m_dm_socket;
--  ClientProcessMsg m_client_process_msg;
--  stream_protocol::endpoint m_ep;
--  char m_recv_buffer[1024];
--
--  // atomic modfiy for this variable.
--  // thread 1 : asio callback thread modify it.
--  // thread 2 : librbd read it.
--  std::atomic<bool> m_session_work;
--  CephContext* cct;
--};
--
--} // namespace cache
--} // namespace rbd
--#endif
-diff --git a/src/tools/rbd_cache/CacheControllerSocketCommon.h b/src/tools/rbd_cache/CacheControllerSocketCommon.h
-deleted file mode 100644
-index e17529a..0000000
---- a/src/tools/rbd_cache/CacheControllerSocketCommon.h
-+++ /dev/null
-@@ -1,62 +0,0 @@
--// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
--// vim: ts=8 sw=2 smarttab
--
--#ifndef CACHE_CONTROLLER_SOCKET_COMMON_H
--#define CACHE_CONTROLLER_SOCKET_COMMON_H
--
--/*
--#define RBDSC_REGISTER         0X11
--#define RBDSC_READ             0X12
--#define RBDSC_LOOKUP           0X13
--#define RBDSC_REGISTER_REPLY   0X14
--#define RBDSC_READ_REPLY       0X15
--#define RBDSC_LOOKUP_REPLY     0X16
--#define RBDSC_READ_RADOS       0X17
--*/
--
--namespace rbd {
--namespace cache {
--
--static const int RBDSC_REGISTER        =  0X11;
--static const int RBDSC_READ            =  0X12;
--static const int RBDSC_LOOKUP          =  0X13;
--static const int RBDSC_REGISTER_REPLY  =  0X14;
--static const int RBDSC_READ_REPLY      =  0X15;
--static const int RBDSC_LOOKUP_REPLY    =  0X16;
--static const int RBDSC_READ_RADOS      =  0X17;
--
--
--
--typedef std::function<void(uint64_t, std::string)> ProcessMsg;
--typedef std::function<void(std::string)> ClientProcessMsg;
--typedef uint8_t rbdsc_req_type;
--struct rbdsc_req_type_t {
--  rbdsc_req_type type;
--  uint64_t vol_size;
--  uint64_t offset;
--  uint64_t length;
--  char pool_name[256];
--  char vol_name[256];
--
--  uint64_t size() {
--    return sizeof(rbdsc_req_type_t);
--  }
--
--  std::string to_buffer() {
--    std::stringstream ss;
--    ss << type;
--    ss << vol_size;
--    ss << offset;
--    ss << length;
--    ss << pool_name;
--    ss << vol_name;
--
--    return ss.str();
--  }
--};
--
--static const int RBDSC_MSG_LEN = sizeof(rbdsc_req_type_t);
--
--} // namespace cache
--} // namespace rbd
--#endif
-diff --git a/src/tools/rbd_cache/ObjectCacheStore.cc b/src/tools/rbd_cache/ObjectCacheStore.cc
-deleted file mode 100644
-index 99f90d6..0000000
---- a/src/tools/rbd_cache/ObjectCacheStore.cc
-+++ /dev/null
-@@ -1,172 +0,0 @@
--// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
--// vim: ts=8 sw=2 smarttab
--
--#include "ObjectCacheStore.h"
--
--#define dout_context g_ceph_context
--#define dout_subsys ceph_subsys_rbd_cache
--#undef dout_prefix
--#define dout_prefix *_dout << "rbd::cache::ObjectCacheStore: " << this << " " \
--                           << __func__ << ": "
--
--namespace rbd {
--namespace cache {
--
--ObjectCacheStore::ObjectCacheStore(CephContext *cct, ContextWQ* work_queue)
--      : m_cct(cct), m_work_queue(work_queue),
--        m_rados(new librados::Rados()) {
--
--  uint64_t object_cache_entries =
--    cct->_conf.get_val<int64_t>("rbd_shared_cache_entries");
--
--  //TODO(): allow to set level
--  m_policy = new SimplePolicy(object_cache_entries, 0.5);
--}
--
--ObjectCacheStore::~ObjectCacheStore() {
--  delete m_policy;
--}
--
--int ObjectCacheStore::init(bool reset) {
--
--  int ret = m_rados->init_with_context(m_cct);
--  if(ret < 0) {
--    lderr(m_cct) << "fail to init Ceph context" << dendl;
--    return ret;
--  }
--
--  ret = m_rados->connect();
--  if(ret < 0 ) {
--    lderr(m_cct) << "fail to conect to cluster" << dendl;
--    return ret;
--  }
--
--  std::string cache_path = m_cct->_conf.get_val<std::string>("rbd_shared_cache_path");
--  //TODO(): check and reuse existing cache objects
--  if(reset) {
--    std::string cmd = "exec rm -rf " + cache_path + "/rbd_cache*; exec mkdir -p " + cache_path;
--    //TODO(): to use std::filesystem
--    int r = system(cmd.c_str());
--  }
--
--  evict_thd = new std::thread([this]{this->evict_thread_body();});
--  return ret;
--}
--
--int ObjectCacheStore::do_promote(std::string pool_name, std::string object_name) {
--  int ret = 0;
--  std::string cache_file_name =  pool_name + object_name;
--
--  //TODO(): lock on ioctx map
--  if (m_ioctxs.find(pool_name) == m_ioctxs.end()) {
--    librados::IoCtx* io_ctx = new librados::IoCtx();
--    ret = m_rados->ioctx_create(pool_name.c_str(), *io_ctx);
--    if (ret < 0) {
--      lderr(m_cct) << "fail to create ioctx" << dendl;
--      assert(0);
--    }
--    m_ioctxs.emplace(pool_name, io_ctx); 
--  }
--
--  assert(m_ioctxs.find(pool_name) != m_ioctxs.end());
--  
--  librados::IoCtx* ioctx = m_ioctxs[pool_name]; 
--
--  librados::bufferlist* read_buf = new librados::bufferlist();
--  int object_size = 4096*1024; //TODO(): read config from image metadata
--
--  //TODO(): async promote
--  ret = promote_object(ioctx, object_name, read_buf, object_size);
--  if (ret == -ENOENT) {
--    read_buf->append(std::string(object_size, '0'));
--    ret = 0;
--  }
--
--  if( ret < 0) {
--    lderr(m_cct) << "fail to read from rados" << dendl;
--    return ret;
--  }
--
--  // persistent to cache
--  librbd::cache::SyncFile cache_file(m_cct, cache_file_name);
--  cache_file.open();
--  ret = cache_file.write_object_to_file(*read_buf, object_size);
--  
--  // update metadata
--  assert(OBJ_CACHE_PROMOTING == m_policy->get_status(cache_file_name));
--  m_policy->update_status(cache_file_name, OBJ_CACHE_PROMOTED);
--  assert(OBJ_CACHE_PROMOTED == m_policy->get_status(cache_file_name));
--
--  return ret;
--
--}
-- 
--// return -1, client need to read data from cluster.
--// return 0,  client directly read data from cache.
--int ObjectCacheStore::lookup_object(std::string pool_name, std::string object_name) {
--
--  std::string cache_file_name =  pool_name + object_name;
--
--  CACHESTATUS ret;
--  ret = m_policy->lookup_object(cache_file_name);
--
--  switch(ret) {
--    case OBJ_CACHE_NONE:
--      return do_promote(pool_name, object_name);
--    case OBJ_CACHE_PROMOTED:
--      return 0;
--    case OBJ_CACHE_PROMOTING:
--    default:
--      return -1;
--  }
--}
--
--void ObjectCacheStore::evict_thread_body() {
--  int ret;
--  while(m_evict_go) {
--    ret = evict_objects();
--  }
--}
--
--
--int ObjectCacheStore::shutdown() {
--  m_evict_go = false;
--  evict_thd->join();
--  m_rados->shutdown();
--  return 0;
--}
--
--int ObjectCacheStore::init_cache(std::string vol_name, uint64_t vol_size) {
--  return 0;
--}
--
--int ObjectCacheStore::lock_cache(std::string vol_name) {
--  return 0;
--}
--
--int ObjectCacheStore::promote_object(librados::IoCtx* ioctx, std::string object_name, librados::bufferlist* read_buf, uint64_t read_len) {
--  int ret;
--
--  librados::AioCompletion* read_completion = librados::Rados::aio_create_completion(); 
--
--  ret = ioctx->aio_read(object_name, read_completion, read_buf, read_len, 0);
--  if(ret < 0) {
--    lderr(m_cct) << "fail to read from rados" << dendl;
--    return ret;
--  }
--  read_completion->wait_for_complete();
--  ret = read_completion->get_return_value();
--  return ret;
--  
--}
--
--int ObjectCacheStore::evict_objects() {
--  std::list<std::string> obj_list;
--  m_policy->get_evict_list(&obj_list);
--  for (auto& obj: obj_list) {
--    //do_evict(obj);
--  }
--}
--
--} // namespace cache
--} // namespace rbd
-diff --git a/src/tools/rbd_cache/ObjectCacheStore.h b/src/tools/rbd_cache/ObjectCacheStore.h
-deleted file mode 100644
-index ba0e1f1..0000000
---- a/src/tools/rbd_cache/ObjectCacheStore.h
-+++ /dev/null
-@@ -1,70 +0,0 @@
--// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
--// vim: ts=8 sw=2 smarttab
--
--#ifndef OBJECT_CACHE_STORE_H
--#define OBJECT_CACHE_STORE_H
--
--#include "common/debug.h"
--#include "common/errno.h"
--#include "common/ceph_context.h"
--#include "common/Mutex.h"
--#include "include/rados/librados.hpp"
--#include "include/rbd/librbd.h"
--#include "librbd/ImageCtx.h"
--#include "librbd/ImageState.h"
--#include "librbd/cache/SharedPersistentObjectCacherFile.h"
--#include "SimplePolicy.hpp"
--
--
--using librados::Rados;
--using librados::IoCtx;
--
--namespace rbd {
--namespace cache {
--
--typedef shared_ptr<librados::Rados> RadosRef;
--typedef shared_ptr<librados::IoCtx> IoCtxRef;
--
--class ObjectCacheStore 
--{
--  public:
--    ObjectCacheStore(CephContext *cct, ContextWQ* work_queue);
--    ~ObjectCacheStore();
--
--    int init(bool reset);
--
--    int shutdown();
--
--    int lookup_object(std::string pool_name, std::string object_name);
--
--    int init_cache(std::string vol_name, uint64_t vol_size);
--
--    int lock_cache(std::string vol_name);
--
--  private:
--    void evict_thread_body();
--    int evict_objects();
--
--    int do_promote(std::string pool_name, std::string object_name);
--
--    int promote_object(librados::IoCtx*, std::string object_name,
--                       librados::bufferlist* read_buf,
--                       uint64_t length);
--
--    CephContext *m_cct;
--    ContextWQ* m_work_queue;
--    RadosRef m_rados;
--
--
--    std::map<std::string, librados::IoCtx*> m_ioctxs;
--
--    librbd::cache::SyncFile *m_cache_file;
--
--    Policy* m_policy;
--    std::thread* evict_thd;
--    bool m_evict_go = false;
--};
--
--} // namespace rbd
--} // namespace cache
--#endif
-diff --git a/src/tools/rbd_cache/Policy.hpp b/src/tools/rbd_cache/Policy.hpp
-deleted file mode 100644
-index 711e3bd..0000000
---- a/src/tools/rbd_cache/Policy.hpp
-+++ /dev/null
-@@ -1,30 +0,0 @@
--#ifndef RBD_CACHE_POLICY_HPP
--#define RBD_CACHE_POLICY_HPP
--
--#include <list>
--#include <string>
--
--namespace rbd {
--namespace cache {
--
--enum CACHESTATUS {
--  OBJ_CACHE_NONE = 0,
--  OBJ_CACHE_PROMOTING,
--  OBJ_CACHE_PROMOTED,
--};
--
--
--class Policy {
--public:
--  Policy(){}
--  virtual ~Policy(){};
--  virtual CACHESTATUS lookup_object(std::string) = 0;
--  virtual int evict_object(std::string&) = 0;
--  virtual void update_status(std::string, CACHESTATUS) = 0;
--  virtual CACHESTATUS get_status(std::string) = 0;
--  virtual void get_evict_list(std::list<std::string>* obj_list) = 0;
--};
--
--} // namespace cache
--} // namespace rbd
--#endif
-diff --git a/src/tools/rbd_cache/SimplePolicy.hpp b/src/tools/rbd_cache/SimplePolicy.hpp
-deleted file mode 100644
-index e785de1..0000000
---- a/src/tools/rbd_cache/SimplePolicy.hpp
-+++ /dev/null
-@@ -1,160 +0,0 @@
--#ifndef RBD_CACHE_SIMPLE_POLICY_HPP
--#define RBD_CACHE_SIMPLE_POLICY_HPP
--
--#include "Policy.hpp"
--#include "include/lru.h"
--#include "common/RWLock.h"
--#include "common/Mutex.h"
--
--#include <vector>
--#include <unordered_map>
--#include <string>
--
--namespace rbd {
--namespace cache {
--
--
--class SimplePolicy : public Policy {
--public:
--  SimplePolicy(uint64_t block_num, float watermark)
--    : m_watermark(watermark), m_entry_count(block_num),
--      m_cache_map_lock("rbd::cache::SimplePolicy::m_cache_map_lock"),
--      m_free_list_lock("rbd::cache::SimplePolicy::m_free_list_lock")
--  {
--
--    for(uint64_t i = 0; i < m_entry_count; i++) {
--      m_free_list.push_back(new Entry());
--    }
--
--  }
--
--  ~SimplePolicy() {
--    for(uint64_t i = 0; i < m_entry_count; i++) {
--      Entry* entry = reinterpret_cast<Entry*>(m_free_list.front());
--      delete entry;
--      m_free_list.pop_front();
--    }
--  }
--
--  CACHESTATUS lookup_object(std::string cache_file_name) {
--
--    //TODO(): check race condition
--    RWLock::WLocker wlocker(m_cache_map_lock);
--
--    auto entry_it = m_cache_map.find(cache_file_name);
--    if(entry_it == m_cache_map.end()) {
--      Mutex::Locker locker(m_free_list_lock);
--      Entry* entry = reinterpret_cast<Entry*>(m_free_list.front());
--      assert(entry != nullptr);
--      m_free_list.pop_front();
--      entry->status = OBJ_CACHE_PROMOTING;
--
--      m_cache_map[cache_file_name] = entry;
--
--      return OBJ_CACHE_NONE;
--    }
--
--    Entry* entry = entry_it->second;
--
--    if(entry->status == OBJ_CACHE_PROMOTED) {
--      // touch it
--      m_promoted_lru.lru_touch(entry);
--    }
--
--    return entry->status;
--  }
--
--  int evict_object(std::string& out_cache_file_name) {
--    RWLock::WLocker locker(m_cache_map_lock);
--
--    return 1;
--  }
--
--  // TODO(): simplify the logic
--  void update_status(std::string file_name, CACHESTATUS new_status) {
--    RWLock::WLocker locker(m_cache_map_lock);
--
--    Entry* entry;
--    auto entry_it = m_cache_map.find(file_name);
--
--    // just check.
--    if(new_status == OBJ_CACHE_PROMOTING) {
--      assert(entry_it == m_cache_map.end());
--    }
--
--    assert(entry_it != m_cache_map.end());
--
--    entry = entry_it->second;
--
--    // promoting is done, so update it.
--    if(entry->status == OBJ_CACHE_PROMOTING && new_status== OBJ_CACHE_PROMOTED) {
--      m_promoted_lru.lru_insert_top(entry);
--      entry->status = new_status;
--      return;
--    }
--
--    assert(0);
--  }
--
--  // get entry status
--  CACHESTATUS get_status(std::string file_name) {
--    RWLock::RLocker locker(m_cache_map_lock);
--    auto entry_it = m_cache_map.find(file_name);
--    if(entry_it == m_cache_map.end()) {
--      return OBJ_CACHE_NONE;
--    }
--
--    return entry_it->second->status;
--  }
--
--  void get_evict_list(std::list<std::string>* obj_list) {
--    RWLock::WLocker locker(m_cache_map_lock);
--    // check free ratio, pop entries from LRU
--    if (m_free_list.size() / m_entry_count < m_watermark) {
--      int evict_num = 10; //TODO(): make this configurable
--      for(int i = 0; i < evict_num; i++) {
--        Entry* entry = reinterpret_cast<Entry*>(m_promoted_lru.lru_expire());
--        if (entry == nullptr) {
--	  continue;
--        }
--        std::string file_name = entry->cache_file_name;
--        obj_list->push_back(file_name);
--
--        auto entry_it = m_cache_map.find(file_name);
--        m_cache_map.erase(entry_it);
--
--        //mark this entry as free
--	entry->status = OBJ_CACHE_NONE;
--        Mutex::Locker locker(m_free_list_lock);
--        m_free_list.push_back(entry);
--      }
--   }
--  }
--
--private:
--
--  class Entry : public LRUObject {
--    public:
--      CACHESTATUS status;
--      Entry() : status(OBJ_CACHE_NONE){}
--      std::string cache_file_name;
--      void encode(bufferlist &bl){}
--      void decode(bufferlist::iterator &it){}
--  };
--
--  float m_watermark;
--  uint64_t m_entry_count;
--
--  std::unordered_map<std::string, Entry*> m_cache_map;
--  RWLock m_cache_map_lock;
--
--  std::deque<Entry*> m_free_list;
--  Mutex m_free_list_lock;
--
--  LRU m_promoted_lru; // include promoted, using status.
--
--};
--
--} // namespace cache
--} // namespace rbd
--#endif
-diff --git a/src/tools/rbd_cache/main.cc b/src/tools/rbd_cache/main.cc
-deleted file mode 100644
-index d604760..0000000
---- a/src/tools/rbd_cache/main.cc
-+++ /dev/null
-@@ -1,85 +0,0 @@
--// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
--// vim: ts=8 sw=2 smarttab
--
--#include "common/ceph_argparse.h"
--#include "common/config.h"
--#include "common/debug.h"
--#include "common/errno.h"
--#include "global/global_init.h"
--#include "global/signal_handler.h"
--#include "CacheController.h"
--
--#include <vector>
--
--rbd::cache::CacheController *cachectl = nullptr;
--
--void usage() {
--  std::cout << "usage: cache controller [options...]" << std::endl;
--  std::cout << "options:\n";
--  std::cout << "  -m monaddress[:port]      connect to specified monitor\n";
--  std::cout << "  --keyring=<path>          path to keyring for local cluster\n";
--  std::cout << "  --log-file=<logfile>       file to log debug output\n";
--  std::cout << "  --debug-rbd-cachecontroller=<log-level>/<memory-level>  set rbd-mirror debug level\n";
--  generic_server_usage();
--}
--
--static void handle_signal(int signum)
--{
--  if (cachectl)
--    cachectl->handle_signal(signum);
--}
--
--int main(int argc, const char **argv)
--{
--  std::vector<const char*> args;
--  env_to_vec(args);
--  argv_to_vec(argc, argv, args);
--
--  auto cct = global_init(nullptr, args, CEPH_ENTITY_TYPE_CLIENT,
--			 CODE_ENVIRONMENT_DAEMON,
--			 CINIT_FLAG_UNPRIVILEGED_DAEMON_DEFAULTS);
--
--  for (auto i = args.begin(); i != args.end(); ++i) {
--    if (ceph_argparse_flag(args, i, "-h", "--help", (char*)NULL)) {
--      usage();
--      return EXIT_SUCCESS;
--    }
--  }
--
--  if (g_conf()->daemonize) {
--    global_init_daemonize(g_ceph_context);
--  }
--  g_ceph_context->enable_perf_counter();
--
--  common_init_finish(g_ceph_context);
--
--  init_async_signal_handler();
--  register_async_signal_handler(SIGHUP, sighup_handler);
--  register_async_signal_handler_oneshot(SIGINT, handle_signal);
--  register_async_signal_handler_oneshot(SIGTERM, handle_signal);
--
--  std::vector<const char*> cmd_args;
--  argv_to_vec(argc, argv, cmd_args);
--
--  // disable unnecessary librbd cache
--  g_ceph_context->_conf.set_val_or_die("rbd_cache", "false");
--
--  cachectl = new rbd::cache::CacheController(g_ceph_context, cmd_args);
--  int r = cachectl->init();
--  if (r < 0) {
--    std::cerr << "failed to initialize: " << cpp_strerror(r) << std::endl;
--    goto cleanup;
--  }
--
--  cachectl->run();
--
-- cleanup:
--  unregister_async_signal_handler(SIGHUP, sighup_handler);
--  unregister_async_signal_handler(SIGINT, handle_signal);
--  unregister_async_signal_handler(SIGTERM, handle_signal);
--  shutdown_async_signal_handler();
--
--  delete cachectl;
--
--  return r < 0 ? EXIT_SUCCESS : EXIT_FAILURE;
--}
--- 
-2.7.4
-