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