diff options
Diffstat (limited to 'src/ceph/0010-librbd-new-namespace-ceph-immutable-obj-cache.patch')
| -rw-r--r-- | src/ceph/0010-librbd-new-namespace-ceph-immutable-obj-cache.patch | 3510 |
1 files changed, 3510 insertions, 0 deletions
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 new file mode 100644 index 0000000..46040e6 --- /dev/null +++ b/src/ceph/0010-librbd-new-namespace-ceph-immutable-obj-cache.patch @@ -0,0 +1,3510 @@ +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 + |