From dd4804fb05ad8aca51516b0112975cc91ef85a6b Mon Sep 17 00:00:00 2001
From: Yuan Zhou <yuan.zhou@intel.com>
Date: Wed, 8 Aug 2018 15:31:47 +0800
Subject: [PATCH 07/10] librbd: cleanup policy based promotion/eviction

Signed-off-by: Yuan Zhou <yuan.zhou@intel.com>
---
 src/common/options.cc                              |   4 +
 .../rbd_cache/CacheControllerSocketClient.hpp      |   3 +-
 src/tools/rbd_cache/ObjectCacheStore.cc            |  63 +++----
 src/tools/rbd_cache/ObjectCacheStore.h             |  10 +-
 src/tools/rbd_cache/Policy.hpp                     |  18 +-
 src/tools/rbd_cache/SimplePolicy.hpp               | 188 +++++++++------------
 6 files changed, 141 insertions(+), 145 deletions(-)

diff --git a/src/common/options.cc b/src/common/options.cc
index 7839a31..b334c1e 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_entries", Option::TYPE_INT, Option::LEVEL_ADVANCED)
+    .set_default(4096)
+    .set_description("shared ssd caching data entries"),
+
     Option("rbd_non_blocking_aio", Option::TYPE_BOOL, Option::LEVEL_ADVANCED)
     .set_default(true)
     .set_description("process AIO ops from a dispatch thread to prevent blocking"),
diff --git a/src/tools/rbd_cache/CacheControllerSocketClient.hpp b/src/tools/rbd_cache/CacheControllerSocketClient.hpp
index 4e1f36c..56b79ce 100644
--- a/src/tools/rbd_cache/CacheControllerSocketClient.hpp
+++ b/src/tools/rbd_cache/CacheControllerSocketClient.hpp
@@ -90,7 +90,8 @@ public:
         }
     });
     std::unique_lock<std::mutex> lk(m);
-    cv.wait(lk);
+    //cv.wait(lk);
+    cv.wait_for(lk, std::chrono::milliseconds(100));
     return 0;
   }
 
diff --git a/src/tools/rbd_cache/ObjectCacheStore.cc b/src/tools/rbd_cache/ObjectCacheStore.cc
index b39fe66..99f90d6 100644
--- a/src/tools/rbd_cache/ObjectCacheStore.cc
+++ b/src/tools/rbd_cache/ObjectCacheStore.cc
@@ -15,7 +15,12 @@ namespace cache {
 ObjectCacheStore::ObjectCacheStore(CephContext *cct, ContextWQ* work_queue)
       : m_cct(cct), m_work_queue(work_queue),
         m_rados(new librados::Rados()) {
-  m_policy = new SimplePolicy(4096, 0.9); // TODO
+
+  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() {
@@ -35,7 +40,16 @@ int ObjectCacheStore::init(bool reset) {
     lderr(m_cct) << "fail to conect to cluster" << dendl;
     return ret;
   }
-  //TODO(): check existing cache objects
+
+  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;
 }
 
@@ -58,10 +72,6 @@ int ObjectCacheStore::do_promote(std::string pool_name, std::string object_name)
   
   librados::IoCtx* ioctx = m_ioctxs[pool_name]; 
 
-  //promoting: update metadata 
-  m_policy->update_status(cache_file_name, PROMOTING);
-  assert(PROMOTING == m_policy->get_status(cache_file_name));
-
   librados::bufferlist* read_buf = new librados::bufferlist();
   int object_size = 4096*1024; //TODO(): read config from image metadata
 
@@ -83,9 +93,9 @@ int ObjectCacheStore::do_promote(std::string pool_name, std::string object_name)
   ret = cache_file.write_object_to_file(*read_buf, object_size);
   
   // update metadata
-  assert(PROMOTING == m_policy->get_status(cache_file_name));
-  m_policy->update_status(cache_file_name, PROMOTED);
-  assert(PROMOTED == m_policy->get_status(cache_file_name));
+  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;
 
@@ -97,18 +107,15 @@ int ObjectCacheStore::lookup_object(std::string pool_name, std::string object_na
 
   std::string cache_file_name =  pool_name + object_name;
 
-  // TODO lookup and return status;
-
   CACHESTATUS ret;
   ret = m_policy->lookup_object(cache_file_name);
 
   switch(ret) {
-    case NONE:
+    case OBJ_CACHE_NONE:
       return do_promote(pool_name, object_name);
-    case PROMOTING:
-      return -1;
-    case PROMOTED:
+    case OBJ_CACHE_PROMOTED:
       return 0;
+    case OBJ_CACHE_PROMOTING:
     default:
       return -1;
   }
@@ -117,26 +124,14 @@ int ObjectCacheStore::lookup_object(std::string pool_name, std::string object_na
 void ObjectCacheStore::evict_thread_body() {
   int ret;
   while(m_evict_go) {
-    std::string temp_cache_file;
-
-    ret = m_policy->evict_object(temp_cache_file);
-    if(ret == 0) {
-      continue;
-    }
-
-    // TODO
-    // delete temp_cache_file file.
-
-    assert(EVICTING == m_policy->get_status(temp_cache_file));
-
-    m_policy->update_status(temp_cache_file, EVICTED);
-
-    assert(NONE == m_policy->get_status(temp_cache_file));
+    ret = evict_objects();
   }
 }
 
 
 int ObjectCacheStore::shutdown() {
+  m_evict_go = false;
+  evict_thd->join();
   m_rados->shutdown();
   return 0;
 }
@@ -165,5 +160,13 @@ int ObjectCacheStore::promote_object(librados::IoCtx* ioctx, std::string object_
   
 }
 
+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
index 5118a73..ba0e1f1 100644
--- a/src/tools/rbd_cache/ObjectCacheStore.h
+++ b/src/tools/rbd_cache/ObjectCacheStore.h
@@ -15,6 +15,7 @@
 #include "librbd/cache/SharedPersistentObjectCacherFile.h"
 #include "SimplePolicy.hpp"
 
+
 using librados::Rados;
 using librados::IoCtx;
 
@@ -40,10 +41,9 @@ class ObjectCacheStore
 
     int lock_cache(std::string vol_name);
 
-    void evict_thread_body();
-
   private:
-    int _evict_object();
+    void evict_thread_body();
+    int evict_objects();
 
     int do_promote(std::string pool_name, std::string object_name);
 
@@ -61,8 +61,8 @@ class ObjectCacheStore
     librbd::cache::SyncFile *m_cache_file;
 
     Policy* m_policy;
-
-    bool m_evict_go;
+    std::thread* evict_thd;
+    bool m_evict_go = false;
 };
 
 } // namespace rbd
diff --git a/src/tools/rbd_cache/Policy.hpp b/src/tools/rbd_cache/Policy.hpp
index 575c294..711e3bd 100644
--- a/src/tools/rbd_cache/Policy.hpp
+++ b/src/tools/rbd_cache/Policy.hpp
@@ -1,12 +1,16 @@
 #ifndef RBD_CACHE_POLICY_HPP
 #define RBD_CACHE_POLICY_HPP
 
+#include <list>
+#include <string>
+
+namespace rbd {
+namespace cache {
+
 enum CACHESTATUS {
-  NONE = 0,
-  PROMOTING,
-  PROMOTED,
-  EVICTING,
-  EVICTED,
+  OBJ_CACHE_NONE = 0,
+  OBJ_CACHE_PROMOTING,
+  OBJ_CACHE_PROMOTED,
 };
 
 
@@ -18,5 +22,9 @@ public:
   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
index a0d8de7..e785de1 100644
--- a/src/tools/rbd_cache/SimplePolicy.hpp
+++ b/src/tools/rbd_cache/SimplePolicy.hpp
@@ -3,138 +3,93 @@
 
 #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 level)
-    : m_level(level),
-      m_lock("SimplePolicy"),
-      m_entry_count(block_num)
+  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")
   {
 
-    Entry m_entries[m_entry_count];
-
-    for(auto &entry : m_entries) {
-      m_free_lru.lru_insert_bot(&entry);
+    for(uint64_t i = 0; i < m_entry_count; i++) {
+      m_free_list.push_back(new Entry());
     }
+
   }
 
-  ~SimplePolicy() {}
+  ~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) {
-    Mutex::Locker locker(m_lock);
 
-    auto entry_it = m_oid_to_entry.find(cache_file_name);
-    if(entry_it == m_oid_to_entry.end()) {
-      return NONE;
+    //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;
 
-    LRU* lru;
-    if(entry->status == PROMOTED) {
-      lru = &m_promoted_lru;
-    } else {
-      lru = &m_handing_lru;
+    if(entry->status == OBJ_CACHE_PROMOTED) {
+      // touch it
+      m_promoted_lru.lru_touch(entry);
     }
 
-    // touch it
-    lru->lru_remove(entry);
-    lru->lru_insert_top(entry);
-
     return entry->status;
   }
 
   int evict_object(std::string& out_cache_file_name) {
-    Mutex::Locker locker(m_lock);
-
-    // still have enough free space, don't need to evict lru.
-    uint64_t temp_current_size = m_oid_to_entry.size();
-    float temp_current_evict_level = temp_current_size / m_entry_count;
-    if(temp_current_evict_level < m_level) {
-      return 0;
-    }
-
-    // when all entries are USING, PROMOTING or EVICTING, just busy waiting.
-    if(m_promoted_lru.lru_get_size() == 0) {
-      return 0;
-    }
-
-    assert(m_promoted_lru.lru_get_size() != 0);
-
-    // evict one item from promoted lru
-    Entry *entry = reinterpret_cast<Entry*>(m_promoted_lru.lru_get_next_expire());
-    assert(entry != nullptr);
-
-    assert(entry->status == PROMOTED);
-
-    out_cache_file_name = entry->cache_file_name;
-    entry->status = EVICTING;
-
-    m_promoted_lru.lru_remove(entry);
-    m_handing_lru.lru_insert_top(entry);
+    RWLock::WLocker locker(m_cache_map_lock);
 
     return 1;
   }
 
   // TODO(): simplify the logic
-  void update_status(std::string _file_name, CACHESTATUS _status) {
-    Mutex::Locker locker(m_lock);
+  void update_status(std::string file_name, CACHESTATUS new_status) {
+    RWLock::WLocker locker(m_cache_map_lock);
 
     Entry* entry;
-    auto entry_it = m_oid_to_entry.find(_file_name);
+    auto entry_it = m_cache_map.find(file_name);
 
     // just check.
-    if(_status == PROMOTING) {
-      assert(m_oid_to_entry.find(_file_name) == m_oid_to_entry.end());
+    if(new_status == OBJ_CACHE_PROMOTING) {
+      assert(entry_it == m_cache_map.end());
     }
 
-    // miss this object.
-    if(entry_it == m_oid_to_entry.end() && _status == PROMOTING) {
-      entry = reinterpret_cast<Entry*>(m_free_lru.lru_get_next_expire());
-      if(entry == nullptr) {
-        assert(0); // namely evict thread have some problems.
-      }
-
-      entry->status = PROMOTING;
-
-      m_oid_to_entry[_file_name] = entry;
-      m_free_lru.lru_remove(entry);
-      m_handing_lru.lru_insert_top(entry);
-
-      return;
-    }
-
-    assert(entry_it != m_oid_to_entry.end());
+    assert(entry_it != m_cache_map.end());
 
     entry = entry_it->second;
 
-    // promoting action have been finished, so update it.
-    if(entry->status == PROMOTING && _status== PROMOTED) {
-      m_handing_lru.lru_remove(entry);
+    // 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 = PROMOTED;
-      return;
-    }
-
-    // will delete this cache file
-    if(entry->status == PROMOTED && _status == EVICTING) {
-      m_promoted_lru.lru_remove(entry);
-      m_handing_lru.lru_insert_top(entry);
-      entry->status = EVICTING;
-      return;
-    }
-
-
-    if(_status == EVICTED) {
-      m_oid_to_entry.erase(entry_it);
-      m_handing_lru.lru_remove(entry);
-      m_free_lru.lru_insert_bot(entry);
+      entry->status = new_status;
       return;
     }
 
@@ -142,39 +97,64 @@ public:
   }
 
   // get entry status
-  CACHESTATUS get_status(std::string _file_name) {
-    Mutex::Locker locker(m_lock);
-    auto entry_it = m_oid_to_entry.find(_file_name);
-    if(entry_it == m_oid_to_entry.end()) {
-      return NONE;
+  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(NONE){}
+      Entry() : status(OBJ_CACHE_NONE){}
       std::string cache_file_name;
       void encode(bufferlist &bl){}
       void decode(bufferlist::iterator &it){}
   };
 
-  std::unordered_map<std::string, Entry*> m_oid_to_entry;
+  float m_watermark;
+  uint64_t m_entry_count;
 
-  LRU m_free_lru;
-  LRU m_handing_lru; // include promoting status or evicting status
-  LRU m_promoted_lru; // include promoted, using status.
+  std::unordered_map<std::string, Entry*> m_cache_map;
+  RWLock m_cache_map_lock;
 
-  mutable Mutex m_lock;
+  std::deque<Entry*> m_free_list;
+  Mutex m_free_list_lock;
 
-  float m_level;
-  uint64_t m_entry_count;
+  LRU m_promoted_lru; // include promoted, using status.
 
 };
 
+} // namespace cache
+} // namespace rbd
 #endif
-- 
2.7.4