+++ /dev/null
-From 55b29a71c238ac465d05035a51808a3b616a8f46 Mon Sep 17 00:00:00 2001
-From: Yuan Zhou <yuan.zhou@intel.com>
-Date: Wed, 5 Sep 2018 14:40:54 +0800
-Subject: [PATCH 09/10] librbd: clean up on rbd shared cache
-
-Signed-off-by: Dehao Shang <dehao.shang@intel.com>
-Signed-off-by: Yuan Zhou <yuan.zhou@intel.com>
----
- src/common/options.cc | 4 +
- .../SharedPersistentObjectCacherObjectDispatch.cc | 28 ++-
- .../SharedPersistentObjectCacherObjectDispatch.h | 3 +-
- src/tools/rbd_cache/CacheController.cc | 11 +-
- src/tools/rbd_cache/CacheController.h | 1 -
- src/tools/rbd_cache/CacheControllerSocket.hpp | 213 +++++++++++++------
- .../rbd_cache/CacheControllerSocketClient.hpp | 226 ++++++++++++++-------
- src/tools/rbd_cache/CacheControllerSocketCommon.h | 2 +
- 8 files changed, 340 insertions(+), 148 deletions(-)
-
-diff --git a/src/common/options.cc b/src/common/options.cc
-index b334c1e..3172744 100644
---- a/src/common/options.cc
-+++ b/src/common/options.cc
-@@ -6365,6 +6365,10 @@ static std::vector<Option> get_rbd_options() {
- .set_default("/tmp")
- .set_description("shared ssd caching data dir"),
-
-+ Option("rbd_shared_cache_sock", Option::TYPE_STR, Option::LEVEL_ADVANCED)
-+ .set_default("/tmp/rbd_shared_ro_cache_sock")
-+ .set_description("shared ssd caching domain socket"),
-+
- Option("rbd_shared_cache_entries", Option::TYPE_INT, Option::LEVEL_ADVANCED)
- .set_default(4096)
- .set_description("shared ssd caching data entries"),
-diff --git a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc b/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc
-index 407ce49..7cbc019 100644
---- a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc
-+++ b/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.cc
-@@ -33,6 +33,7 @@ SharedPersistentObjectCacherObjectDispatch<I>::~SharedPersistentObjectCacherObje
- 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;
-@@ -44,11 +45,11 @@ void SharedPersistentObjectCacherObjectDispatch<I>::init() {
- return;
- }
-
-- ldout(cct, 20) << "parent image: setup SRO cache client = " << dendl;
-+ ldout(cct, 5) << "parent image: setup SRO cache client = " << dendl;
-
-- std::string controller_path = "/tmp/rbd_shared_readonly_cache_demo";
-- m_cache_client = new rbd::cache::CacheClient(io_service, controller_path.c_str(),
-- ([&](std::string s){client_handle_request(s);}));
-+ 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) {
-@@ -78,18 +79,29 @@ bool SharedPersistentObjectCacherObjectDispatch<I>::read(
- 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->connected && m_object_store) {
-+ 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);
- }
-@@ -109,6 +121,7 @@ int SharedPersistentObjectCacherObjectDispatch<I>::handle_read_cache(
- // 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
-@@ -123,7 +136,6 @@ int SharedPersistentObjectCacherObjectDispatch<I>::handle_read_cache(
- return false;
- }
- }
--
- template <typename I>
- void SharedPersistentObjectCacherObjectDispatch<I>::client_handle_request(std::string msg) {
- auto cct = m_image_ctx->cct;
-@@ -133,7 +145,7 @@ void SharedPersistentObjectCacherObjectDispatch<I>::client_handle_request(std::s
-
- switch (io_ctx->type) {
- case rbd::cache::RBDSC_REGISTER_REPLY: {
-- // open cache handler for volume
-+ // 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);
-
-@@ -153,7 +165,7 @@ void SharedPersistentObjectCacherObjectDispatch<I>::client_handle_request(std::s
- }
- 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 36b868a..5685244 100644
---- a/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h
-+++ b/src/librbd/cache/SharedPersistentObjectCacherObjectDispatch.h
-@@ -118,12 +118,11 @@ private:
- 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;
-
-- void client_handle_request(std::string msg);
- rbd::cache::CacheClient *m_cache_client = nullptr;
-- boost::asio::io_service io_service;
- };
-
- } // namespace cache
-diff --git a/src/tools/rbd_cache/CacheController.cc b/src/tools/rbd_cache/CacheController.cc
-index c9d674b..620192c 100644
---- a/src/tools/rbd_cache/CacheController.cc
-+++ b/src/tools/rbd_cache/CacheController.cc
-@@ -65,12 +65,12 @@ void CacheController::handle_signal(int signum){}
- void CacheController::run() {
- try {
- //TODO(): use new socket path
-- std::string controller_path = "/tmp/rbd_shared_readonly_cache_demo";
-+ 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(io_service, controller_path,
-- ([&](uint64_t p, std::string s){handle_request(p, s);}));
-- io_service.run();
-+ 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";
- }
-@@ -105,7 +105,8 @@ void CacheController::handle_request(uint64_t session_id, std::string msg){
-
- break;
- }
--
-+ std::cout<<"can't recongize request"<<std::endl;
-+ assert(0); // TODO replace it.
- }
- }
-
-diff --git a/src/tools/rbd_cache/CacheController.h b/src/tools/rbd_cache/CacheController.h
-index 0e3abc1..0e23484 100644
---- a/src/tools/rbd_cache/CacheController.h
-+++ b/src/tools/rbd_cache/CacheController.h
-@@ -41,7 +41,6 @@ class CacheController {
- void handle_request(uint64_t sesstion_id, std::string msg);
-
- private:
-- boost::asio::io_service io_service;
- CacheServer *m_cache_server;
- std::vector<const char*> m_args;
- CephContext *m_cct;
-diff --git a/src/tools/rbd_cache/CacheControllerSocket.hpp b/src/tools/rbd_cache/CacheControllerSocket.hpp
-index d178b58..2ff7477 100644
---- a/src/tools/rbd_cache/CacheControllerSocket.hpp
-+++ b/src/tools/rbd_cache/CacheControllerSocket.hpp
-@@ -11,6 +11,7 @@
- #include <string>
- #include <boost/bind.hpp>
- #include <boost/asio.hpp>
-+#include <boost/asio/error.hpp>
- #include <boost/algorithm/string.hpp>
- #include "CacheControllerSocketCommon.h"
-
-@@ -23,110 +24,202 @@ 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)
-- : session_id(session_id), socket_(io_service), process_msg(processmsg) {}
-+ : m_session_id(session_id), m_dm_socket(io_service), process_msg(processmsg) {}
-
- stream_protocol::socket& socket() {
-- return socket_;
-+ return m_dm_socket;
- }
-
- void start() {
--
-- boost::asio::async_read(socket_, boost::asio::buffer(data_),
-- boost::asio::transfer_exactly(544),
-+ 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));
-+ 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) {
-- if(bytes_transferred != 544){
-- assert(0);
-- }
-- process_msg(session_id, std::string(data_, bytes_transferred));
-+ 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) {
-- if (!error) {
-- 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,
-- boost::asio::placeholders::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(socket_,
-+ boost::asio::async_write(m_dm_socket,
- boost::asio::buffer(msg.c_str(), msg.size()),
-- boost::asio::transfer_exactly(544),
-+ boost::asio::transfer_exactly(RBDSC_MSG_LEN),
- boost::bind(&session::handle_write,
-- shared_from_this(),
-- boost::asio::placeholders::error));
-+ shared_from_this(),
-+ boost::asio::placeholders::error,
-+ boost::asio::placeholders::bytes_transferred));
-
- }
-
- private:
-- uint64_t session_id;
-- stream_protocol::socket socket_;
-+ 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 data_[1024];
-+ char m_buffer[1024];
- };
-
- typedef std::shared_ptr<session> session_ptr;
-
- class CacheServer {
- public:
-- CacheServer(boost::asio::io_service& io_service,
-- const std::string& file, ProcessMsg processmsg)
-- : io_service_(io_service),
-- server_process_msg(processmsg),
-- acceptor_(io_service, stream_protocol::endpoint(file))
-- {
-- session_ptr new_session(new session(session_id, io_service_, server_process_msg));
-- 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
-- if (!error) {
-- new_session->start();
-- session_map.emplace(session_id, new_session);
-- session_id++;
-- new_session.reset(new session(session_id, io_service_, server_process_msg));
-- acceptor_.async_accept(new_session->socket(),
-- boost::bind(&CacheServer::handle_accept, this, new_session,
-- boost::asio::placeholders::error));
-+ 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 = session_map.find(session_id);
-- if (it != session_map.end()) {
-+ 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:
-- boost::asio::io_service& io_service_;
-- ProcessMsg server_process_msg;
-- stream_protocol::acceptor acceptor_;
-- uint64_t session_id = 1;
-- std::map<uint64_t, session_ptr> session_map;
-+ 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
-diff --git a/src/tools/rbd_cache/CacheControllerSocketClient.hpp b/src/tools/rbd_cache/CacheControllerSocketClient.hpp
-index 3b0ca00..964f888 100644
---- a/src/tools/rbd_cache/CacheControllerSocketClient.hpp
-+++ b/src/tools/rbd_cache/CacheControllerSocketClient.hpp
-@@ -4,9 +4,12 @@
- #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"
-@@ -19,32 +22,64 @@ namespace cache {
-
- class CacheClient {
- public:
-- CacheClient(boost::asio::io_service& io_service,
-- const std::string& file, ClientProcessMsg processmsg)
-- : io_service_(io_service),
-- io_service_work_(io_service),
-- socket_(io_service),
-+ 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),
-- ep_(stream_protocol::endpoint(file))
-+ m_ep(stream_protocol::endpoint(file)),
-+ m_session_work(false),
-+ cct(ceph_ctx)
- {
-- io_thread.reset(new std::thread([this](){io_service_.run(); }));
-+ // TODO wrapper io_service
-+ std::thread thd([this](){
-+ m_io_service.run();});
-+ thd.detach();
- }
-
-- ~CacheClient() {
-- io_service_.stop();
-- io_thread->join();
-+ void run(){
- }
-
-- 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() {
-- try {
-- socket_.connect(ep_);
-- } catch (std::exception& e) {
-+ 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;
- }
-- connected = true;
-+
-+ std::cout<<"connect success"<<std::endl;
-+
- return 0;
- }
-
-@@ -57,27 +92,51 @@ public:
- message->vol_size = vol_size;
- message->offset = 0;
- message->length = 0;
-- boost::asio::async_write(socket_, boost::asio::buffer((char*)message, message->size()),
-- [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),
-- [this](const boost::system::error_code& err, size_t cb) {
-- if (!err) {
-- m_client_process_msg(std::string(buffer_, cb));
-- } else {
-- return -1;
-- }
-- });
-- } else {
-- return -1;
-- }
-- });
-+
-+ 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;
-@@ -87,59 +146,82 @@ public:
- message->offset = 0;
- message->length = 0;
-
-- boost::asio::async_write(socket_, boost::asio::buffer((char*)message, message->size()),
-+ 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) {
-+ 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);
-- } else {
-- return -1;
-- }
- });
-
- return 0;
- }
-
- void get_result(Context* on_finish) {
-- boost::asio::async_read(socket_, boost::asio::buffer(buffer_),
-- boost::asio::transfer_exactly(544),
-+ 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 (cb != 544) {
-- assert(0);
-- }
-- if (!err) {
-- rbdsc_req_type_t *io_ctx = (rbdsc_req_type_t*)(buffer_);
-- if (io_ctx->type == RBDSC_READ_REPLY) {
-- on_finish->complete(true);
-- return;
-- } else {
-- on_finish->complete(false);
-- return;
-- }
-- } else {
-- assert(0);
-- return on_finish->complete(false);
-- }
-+ 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& io_service_;
-- boost::asio::io_service::work io_service_work_;
-- stream_protocol::socket socket_;
--
-- std::shared_ptr<std::thread> io_thread;
-+ 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 ep_;
-- char buffer_[1024];
-- int block_size_ = 1024;
--
-- std::condition_variable cv;
-- std::mutex m;
--
--public:
-- bool connected = false;
-+ 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
-diff --git a/src/tools/rbd_cache/CacheControllerSocketCommon.h b/src/tools/rbd_cache/CacheControllerSocketCommon.h
-index e026ec8..e17529a 100644
---- a/src/tools/rbd_cache/CacheControllerSocketCommon.h
-+++ b/src/tools/rbd_cache/CacheControllerSocketCommon.h
-@@ -55,6 +55,8 @@ struct rbdsc_req_type_t {
- }
- };
-
-+static const int RBDSC_MSG_LEN = sizeof(rbdsc_req_type_t);
-+
- } // namespace cache
- } // namespace rbd
- #endif
---
-2.7.4
-
Code Review / stor4nfv.git / blobdiff