--- /dev/null
+From 9f8ff821dfc98dfc3cdb557b736ce455a3ae6162 Mon Sep 17 00:00:00 2001
+From: Yuan Zhou <yuan.zhou@intel.com>
+Date: Thu, 16 Aug 2018 17:28:46 +0800
+Subject: [PATCH 08/10] librbd: implement async cache lookup and read
+
+Signed-off-by: Yuan Zhou <yuan.zhou@intel.com>
+---
+ .../SharedPersistentObjectCacherObjectDispatch.cc | 63 ++++++++++++----------
+ .../SharedPersistentObjectCacherObjectDispatch.h | 7 +++
+ src/tools/rbd_cache/CacheController.cc | 9 ++--
+ src/tools/rbd_cache/CacheControllerSocket.hpp | 8 ++-
+ .../rbd_cache/CacheControllerSocketClient.hpp | 49 +++++++++--------
+ src/tools/rbd_cache/CacheControllerSocketCommon.h | 12 +++++
+ 6 files changed, 94 insertions(+), 54 deletions(-)
+
+diff --git a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc b/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc
+index 2aa5cad..407ce49 100644
+--- a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc
++++ b/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc
+@@ -29,13 +29,8 @@ SharedPersistentObjectCacherObjectDispatch<I>::SharedPersistentObjectCacherObjec
+
+ template <typename I>
+ SharedPersistentObjectCacherObjectDispatch<I>::~SharedPersistentObjectCacherObjectDispatch() {
+- if (m_object_store) {
+ delete m_object_store;
+- }
+-
+- if (m_cache_client) {
+ delete m_cache_client;
+- }
+ }
+
+ template <typename I>
+@@ -88,34 +83,48 @@ bool SharedPersistentObjectCacherObjectDispatch<I>::read(
+ ldout(cct, 20) << "object_no=" << object_no << " " << object_off << "~"
+ << object_len << dendl;
+
+- // ensure we aren't holding the cache lock post-read
+ 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->connected && m_object_store) {
+- bool exists;
+ m_cache_client->lookup_object(m_image_ctx->data_ctx.get_pool_name(),
+- m_image_ctx->id, oid, &exists);
+-
+- // try to read from parent image
+- ldout(cct, 20) << "SRO cache object exists:" << exists << dendl;
+- if (exists) {
+- int r = m_object_store->read_object(oid, read_data, object_off, object_len, on_dispatched);
+- if (r != 0) {
+- *dispatch_result = io::DISPATCH_RESULT_COMPLETE;
+- on_dispatched->complete(r);
+- return true;
+- }
+- }
++ m_image_ctx->id, oid, ctx);
+ }
+-
+- ldout(cct, 20) << "Continue read from RADOS" << dendl;
+- *dispatch_result = io::DISPATCH_RESULT_CONTINUE;
+- on_dispatched->complete(0);
+ 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);
++ 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;
+@@ -123,26 +132,26 @@ void SharedPersistentObjectCacherObjectDispatch<I>::client_handle_request(std::s
+ rbd::cache::rbdsc_req_type_t *io_ctx = (rbd::cache::rbdsc_req_type_t*)(msg.c_str());
+
+ switch (io_ctx->type) {
+- case RBDSC_REGISTER_REPLY: {
++ 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 RBDSC_READ_REPLY: {
++ case rbd::cache::RBDSC_READ_REPLY: {
+ ldout(cct, 20) << "SRO cache client start to read cache" << dendl;
+ //TODO(): should call read here
+
+ break;
+ }
+- case RBDSC_READ_RADOS: {
++ 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" << dendl;
++ default: ldout(cct, 20) << "nothing" << io_ctx->type <<dendl;
+ break;
+
+ }
+diff --git a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h b/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h
+index 200688f..36b868a 100644
+--- a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h
++++ b/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h
+@@ -112,6 +112,13 @@ public:
+
+ 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);
++
+ ImageCtxT* m_image_ctx;
+
+ void client_handle_request(std::string msg);
+diff --git a/src/tools/rbd_cache/CacheController.cc b/src/tools/rbd_cache/CacheController.cc
+index cefcf28..c9d674b 100644
+--- a/src/tools/rbd_cache/CacheController.cc
++++ b/src/tools/rbd_cache/CacheController.cc
+@@ -76,7 +76,7 @@ void CacheController::run() {
+ }
+ }
+
+-void CacheController::handle_request(uint64_t sesstion_id, std::string msg){
++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;
+@@ -86,7 +86,7 @@ void CacheController::handle_request(uint64_t sesstion_id, std::string msg){
+ // 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(sesstion_id, std::string((char*)io_ctx, msg.size()));
++ m_cache_server->send(session_id, std::string((char*)io_ctx, msg.size()));
+
+ break;
+ }
+@@ -98,7 +98,10 @@ void CacheController::handle_request(uint64_t sesstion_id, std::string msg){
+ } else {
+ io_ctx->type = RBDSC_READ_REPLY;
+ }
+- m_cache_server->send(sesstion_id, std::string((char*)io_ctx, msg.size()));
++ if (io_ctx->type != RBDSC_READ_REPLY) {
++ assert(0);
++ }
++ m_cache_server->send(session_id, std::string((char*)io_ctx, msg.size()));
+
+ break;
+ }
+diff --git a/src/tools/rbd_cache/CacheControllerSocket.hpp b/src/tools/rbd_cache/CacheControllerSocket.hpp
+index 967af1d..d178b58 100644
+--- a/src/tools/rbd_cache/CacheControllerSocket.hpp
++++ b/src/tools/rbd_cache/CacheControllerSocket.hpp
+@@ -43,7 +43,9 @@ public:
+ void handle_read(const boost::system::error_code& error, size_t bytes_transferred) {
+
+ if (!error) {
+-
++ if(bytes_transferred != 544){
++ assert(0);
++ }
+ process_msg(session_id, std::string(data_, bytes_transferred));
+
+ }
+@@ -51,7 +53,8 @@ public:
+
+ void handle_write(const boost::system::error_code& error) {
+ if (!error) {
+- socket_.async_read_some(boost::asio::buffer(data_),
++ boost::asio::async_read(socket_, boost::asio::buffer(data_),
++ boost::asio::transfer_exactly(544),
+ boost::bind(&session::handle_read,
+ shared_from_this(),
+ boost::asio::placeholders::error,
+@@ -63,6 +66,7 @@ public:
+
+ boost::asio::async_write(socket_,
+ boost::asio::buffer(msg.c_str(), msg.size()),
++ boost::asio::transfer_exactly(544),
+ boost::bind(&session::handle_write,
+ shared_from_this(),
+ boost::asio::placeholders::error));
+diff --git a/src/tools/rbd_cache/CacheControllerSocketClient.hpp b/src/tools/rbd_cache/CacheControllerSocketClient.hpp
+index 56b79ce..3b0ca00 100644
+--- a/src/tools/rbd_cache/CacheControllerSocketClient.hpp
++++ b/src/tools/rbd_cache/CacheControllerSocketClient.hpp
+@@ -8,6 +8,7 @@
+ #include <boost/bind.hpp>
+ #include <boost/algorithm/string.hpp>
+ #include "include/assert.h"
++#include "include/Context.h"
+ #include "CacheControllerSocketCommon.h"
+
+
+@@ -26,8 +27,12 @@ public:
+ m_client_process_msg(processmsg),
+ ep_(stream_protocol::endpoint(file))
+ {
+- std::thread thd([this](){io_service_.run(); });
+- thd.detach();
++ io_thread.reset(new std::thread([this](){io_service_.run(); }));
++ }
++
++ ~CacheClient() {
++ io_service_.stop();
++ io_thread->join();
+ }
+
+ void run(){
+@@ -53,7 +58,8 @@ public:
+ message->offset = 0;
+ message->length = 0;
+ boost::asio::async_write(socket_, boost::asio::buffer((char*)message, message->size()),
+- [this](const boost::system::error_code& err, size_t cb) {
++ [this, message](const boost::system::error_code& err, size_t cb) {
++ delete message;
+ if (!err) {
+ boost::asio::async_read(socket_, boost::asio::buffer(buffer_),
+ boost::asio::transfer_exactly(544),
+@@ -72,7 +78,7 @@ public:
+ return 0;
+ }
+
+- int lookup_object(std::string pool_name, std::string vol_name, std::string object_id, bool* result) {
++ 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());
+@@ -82,49 +88,48 @@ public:
+ message->length = 0;
+
+ boost::asio::async_write(socket_, boost::asio::buffer((char*)message, message->size()),
+- [this, result](const boost::system::error_code& err, size_t cb) {
++ [this, on_finish, message](const boost::system::error_code& err, size_t cb) {
++ delete message;
+ if (!err) {
+- get_result(result);
++ get_result(on_finish);
+ } else {
+ return -1;
+ }
+ });
+- std::unique_lock<std::mutex> lk(m);
+- //cv.wait(lk);
+- cv.wait_for(lk, std::chrono::milliseconds(100));
++
+ return 0;
+ }
+
+- void get_result(bool* result) {
++ void get_result(Context* on_finish) {
+ boost::asio::async_read(socket_, boost::asio::buffer(buffer_),
+ boost::asio::transfer_exactly(544),
+- [this, result](const boost::system::error_code& err, size_t cb) {
++ [this, on_finish](const boost::system::error_code& err, size_t cb) {
++ if (cb != 544) {
++ assert(0);
++ }
+ if (!err) {
+ rbdsc_req_type_t *io_ctx = (rbdsc_req_type_t*)(buffer_);
+ if (io_ctx->type == RBDSC_READ_REPLY) {
+- *result = true;
++ on_finish->complete(true);
++ return;
+ } else {
+- *result = false;
++ on_finish->complete(false);
++ return;
+ }
+- cv.notify_one();
+- m_client_process_msg(std::string(buffer_, cb));
+ } else {
+- return -1;
++ assert(0);
++ return on_finish->complete(false);
+ }
+ });
+ }
+
+- void handle_connect(const boost::system::error_code& error) {
+- //TODO(): open librbd snap
+- }
+-
+- void handle_write(const boost::system::error_code& error) {
+- }
+
+ private:
+ boost::asio::io_service& io_service_;
+ boost::asio::io_service::work io_service_work_;
+ stream_protocol::socket socket_;
++
++ std::shared_ptr<std::thread> io_thread;
+ ClientProcessMsg m_client_process_msg;
+ stream_protocol::endpoint ep_;
+ char buffer_[1024];
+diff --git a/src/tools/rbd_cache/CacheControllerSocketCommon.h b/src/tools/rbd_cache/CacheControllerSocketCommon.h
+index a9d73a8..e026ec8 100644
+--- a/src/tools/rbd_cache/CacheControllerSocketCommon.h
++++ b/src/tools/rbd_cache/CacheControllerSocketCommon.h
+@@ -4,6 +4,7 @@
+ #ifndef CACHE_CONTROLLER_SOCKET_COMMON_H
+ #define CACHE_CONTROLLER_SOCKET_COMMON_H
+
++/*
+ #define RBDSC_REGISTER 0X11
+ #define RBDSC_READ 0X12
+ #define RBDSC_LOOKUP 0X13
+@@ -11,10 +12,21 @@
+ #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;
+--
+2.7.4
+
Code Review / stor4nfv.git / blobdiff