1 files changed, 139 insertions, 0 deletions
diff --git a/src/tools/immutable_object_cache/CacheController.cc b/src/tools/immutable_object_cache/CacheController.cc
new file mode 100644
index 000000000..ae1636839
--- /dev/null
+++ b/src/tools/immutable_object_cache/CacheController.cc
@@ -0,0 +1,139 @@
+// -*- 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 {
+
+CacheController::CacheController(CephContext *cct,
+                                 const std::vector<const char*> &args):
+  m_args(args), m_cct(cct) {
+  ldout(m_cct, 20) << dendl;
+}
+
+CacheController::~CacheController() {
+  delete m_cache_server;
+  delete m_object_cache_store;
+}
+
+int CacheController::init() {
+  ldout(m_cct, 20) << dendl;
+  m_object_cache_store = new ObjectCacheStore(m_cct);
+  // TODO(dehao): make this configurable
+  int r = m_object_cache_store->init(true);
+  if (r < 0) {
+    lderr(m_cct) << "init error\n" << dendl;
+    return r;
+  }
+
+  r = m_object_cache_store->init_cache();
+  if (r < 0) {
+    lderr(m_cct) << "init error\n" << dendl;
+  }
+
+  return r;
+}
+
+int CacheController::shutdown() {
+  ldout(m_cct, 20) << dendl;
+
+  int r;
+  if (m_cache_server != nullptr) {
+    r = m_cache_server->stop();
+    if (r < 0) {
+      lderr(m_cct) << "stop error\n" << dendl;
+      return r;
+    }
+  }
+
+  r = m_object_cache_store->shutdown();
+  if (r < 0) {
+    lderr(m_cct) << "stop error\n" << dendl;
+    return r;
+  }
+
+  return r;
+}
+
+void CacheController::handle_signal(int signum) {
+  shutdown();
+}
+
+int CacheController::run() {
+  try {
+    std::string controller_path =
+      m_cct->_conf.get_val<std::string>("immutable_object_cache_sock");
+    if (controller_path.empty()) {
+      lderr(m_cct) << "'immutable_object_cache_sock' path not set" << dendl;
+      return -EINVAL;
+    }
+
+    std::remove(controller_path.c_str());
+
+    m_cache_server = new CacheServer(m_cct, controller_path,
+      std::bind(&CacheController::handle_request, this,
+                std::placeholders::_1, std::placeholders::_2));
+
+    int ret = m_cache_server->run();
+    if (ret != 0) {
+      return ret;
+    }
+
+    return 0;
+  } catch (std::exception& e) {
+    lderr(m_cct) << "Exception: " << e.what() << dendl;
+    return -EFAULT;
+  }
+}
+
+void CacheController::handle_request(CacheSession* session,
+                                     ObjectCacheRequest* req) {
+  ldout(m_cct, 20) << dendl;
+
+  switch (req->get_request_type()) {
+    case RBDSC_REGISTER: {
+      // TODO(dehao): skip register and allow clients to lookup directly
+
+      auto req_reg_data = reinterpret_cast <ObjectCacheRegData*> (req);
+      session->set_client_version(req_reg_data->version);
+
+      ObjectCacheRequest* reply = new ObjectCacheRegReplyData(
+        RBDSC_REGISTER_REPLY, req->seq);
+      session->send(reply);
+      break;
+    }
+    case RBDSC_READ: {
+      // lookup object in local cache store
+      std::string cache_path;
+      ObjectCacheReadData* req_read_data =
+        reinterpret_cast <ObjectCacheReadData*> (req);
+      bool return_dne_path = session->client_version().empty();
+      int ret = m_object_cache_store->lookup_object(
+        req_read_data->pool_namespace, req_read_data->pool_id,
+        req_read_data->snap_id, req_read_data->object_size,
+        req_read_data->oid, return_dne_path, cache_path);
+      ObjectCacheRequest* reply = nullptr;
+      if (ret != OBJ_CACHE_PROMOTED && ret != OBJ_CACHE_DNE) {
+        reply = new ObjectCacheReadRadosData(RBDSC_READ_RADOS, req->seq);
+      } else {
+        reply = new ObjectCacheReadReplyData(RBDSC_READ_REPLY,
+                                             req->seq, cache_path);
+      }
+      session->send(reply);
+      break;
+    }
+    default:
+      ldout(m_cct, 5) << "can't recongize request" << dendl;
+      ceph_assert(0);
+  }
+}
+
+}  // namespace immutable_obj_cache
+}  // namespace ceph