objcache2

Signed-off-by: Coldwings <coldwings@me.com>

common/objectcachev2.h

@@ -0,0 +1,255 @@
+#pragma once
+
+#include <photon/common/alog.h>
+#include <photon/common/lockfree_queue.h>
+#include <photon/photon.h>
+#include <photon/thread/list.h>
+#include <photon/thread/thread11.h>
+#include <photon/thread/timer.h>
+#include <photon/thread/workerpool.h>
+
+#include <memory>
+#include <unordered_set>
+
+/**
+
+A simplify object cache implementation.
+
+`ObjectCacheV2` shows better performance compare to `ObjectCache`, with several
+improvements:
+1. Less lock in both read and write.
+2. Object destruction always goes in background photon thread called reclaimer.
+3. Self adjustable timeout for reclaimer. No needs to set cycler timer.
+4. New `update` API, able to immediatly substitude objects.
+5. `ObjectCacheV2` no longer support acquire/release API.
+
+It should work as `ObjectCache` in code do not depends on acquire/release API.
+
+**/
+
+template <typename K, typename VPtr>
+class ObjectCacheV2 {
+protected:
+ using V = std::remove_pointer_t<VPtr>;
+
+ struct Box : public intrusive_list_node<Box> {
+ const K key;
+ std::shared_ptr<V> ref;
+ photon::spinlock boxlock;
+ photon::spinlock reflock;
+ uint64_t lastcreate = 0;
+ uint64_t timestamp = 0;
+
+ struct Updater {
+ Box* box = nullptr;
+
+ ~Updater() {}
+
+ // Returned RefPtr is the old one
+ // other reader will get new one after updated
+ std::shared_ptr<V> update(V* val, uint64_t ts = 0) {
+ while (box->reflock.try_lock() != 0) {
+ photon::thread_yield();
+ }
+ DEFER(box->reflock.unlock());
+ auto r = std::shared_ptr<V>(val);
+ if (ts) box->lastcreate = ts;
+ std::swap(r, box->ref);
+ return r;
+ }
+
+ explicit Updater(Box* b) : box(b) {}
+
+ Updater(const Updater&) = delete;
+ Updater(Updater&& rhs) : box(nullptr) { *this = std::move(rhs); }
+ Updater& operator=(const Updater&) = delete;
+ Updater& operator=(Updater&& rhs) {
+ std::swap(box, rhs.box);
+ return *this;
+ }
+ operator bool() const { return box != nullptr; }
+ };
+
+ Box() = default;
+ template <typename KeyType>
+ explicit Box(KeyType&& key)
+ : key(std::forward<KeyType>(key)), ref(nullptr) {}
+
+ ~Box() {}
+
+ Updater writer() { return Updater(this); }
+ std::shared_ptr<V> reader() {
+ SCOPED_LOCK(reflock);
+ return ref;
+ }
+ };
+ struct ItemHash {
+ size_t operator()(const Box* x) const { return std::hash<K>()(x->key); }
+ };
+ struct ItemEqual {
+ bool operator()(const Box* a, const Box* b) const {
+ return a->key == b->key;
+ }
+ };
+
+ photon::thread* reclaimer = nullptr;
+ photon::common::RingChannel<
+ LockfreeMPMCRingQueue<std::shared_ptr<V>*, 4096>>
+ reclaim_queue;
+ photon::spinlock maplock;
+ std::unordered_set<Box*, ItemHash, ItemEqual> map;
+ intrusive_list<Box> cycle_list;
+ uint64_t lifespan;
+ photon::Timer _timer;
+
+ void __update(Box* item, V* val) {
+ reclaim_queue.template send<PhotonPause>(
+ new std::shared_ptr<V>(item->writer().update(val, photon::now)));
+ }
+
+ template <typename KeyType>
+ Box* __find_or_create_item(const KeyType& key) {
+ Box keyitem(key);
+ auto pkey = &keyitem;
+ Box* item = nullptr;
+ SCOPED_LOCK(maplock);
+ auto it = map.find(pkey);
+ if (it == map.end()) {
+ item = new Box(key);
+ map.emplace(item);
+ } else
+ item = *it;
+ assert(item);
+ item->timestamp = photon::now;
+ cycle_list.pop(item);
+ cycle_list.push_back(item);
+ return item;
+ }
+
+ uint64_t __expire() {
+ intrusive_list<Box> delete_list;
+ uint64_t now = photon::now;
+ {
+ SCOPED_LOCK(maplock);
+ if (cycle_list.empty()) return 0;
+ if (photon::sat_add(cycle_list.front()->timestamp, lifespan) >
+ now) {
+ return photon::sat_add(cycle_list.front()->timestamp,
+ lifespan) -
+ now;
+ }
+ auto x = cycle_list.front();
+ while (x && (photon::sat_add(x->timestamp, lifespan) < now)) {
+ cycle_list.pop(x);
+ __update(x, nullptr);
+ map.erase(x);
+ delete_list.push_back(x);
+ x = cycle_list.front();
+ }
+ }
+ delete_list.delete_all();
+ return 0;
+ }
+
+public:
+ struct Borrow {
+ ObjectCacheV2* _oc = nullptr;
+ Box* _item = nullptr;
+ std::shared_ptr<V> _reader;
+ bool _recycle = false;
+
+ Borrow() : _reader(nullptr) {}
+
+ Borrow(ObjectCacheV2* oc, Box* item, std::shared_ptr<V>&& reader)
+ : _oc(oc),
+ _item(item),
+ _reader(std::move(reader)),
+ _recycle(false) {}
+
+ Borrow(Borrow&& rhs) : _reader(nullptr) { *this = std::move(rhs); }
+
+ Borrow& operator=(Borrow&& rhs) {
+ std::swap(_oc, rhs._oc);
+ std::swap(_item, rhs._item);
+ std::swap(_reader, rhs._reader);
+ std::swap(_recycle, rhs._recycle);
+ return *this;
+ }
+
+ ~Borrow() {
+ if (_recycle) {
+ _oc->__update(_item, nullptr);
+ }
+ }
+
+ bool recycle() { return _recycle; }
+
+ bool recycle(bool x) { return _recycle = x; }
+
+ V& operator*() const { return *_reader; }
+ V* operator->() const { return &*_reader; }
+ operator bool() const { return (bool)_reader; }
+ };
+
+ template <typename KeyType, typename Ctor>
+ Borrow borrow(KeyType&& key, Ctor&& ctor, uint64_t cooldown = 0UL) {
+ auto item = __find_or_create_item(std::forward<KeyType>(key));
+ auto r = item->reader();
+ while (!r) {
+ if (item->boxlock.try_lock() == 0) {
+ DEFER(item->boxlock.unlock());
+ auto lr = item->reader();
+ if (!lr) {
+ if (photon::sat_add(item->lastcreate, cooldown) <=
+ photon::now) {
+ __update(item, ctor());
+ }
+ return Borrow(this, item, item->reader());
+ }
+ }
+ photon::thread_yield();
+ r = item->reader();
+ }
+ return Borrow(this, item, item->reader());
+ }
+
+ template <typename KeyType>
+ Borrow borrow(KeyType&& key) {
+ return borrow(&key, [&]() { return new V(); });
+ }
+
+ template <typename KeyType, typename Ctor>
+ Borrow update(KeyType&& key, Ctor&& ctor) {
+ auto item = __find_or_create_item(std::forward<KeyType>(key));
+ __update(item, ctor());
+ return Borrow(this, item, item->reader());
+ }
+
+ ObjectCacheV2(uint64_t lifespan)
+ : lifespan(lifespan),
+ _timer(1UL * 1000 * 1000, {this, &ObjectCacheV2::__expire}, true,
+ photon::DEFAULT_STACK_SIZE) {
+ reclaimer = photon::thread_create11([this] {
+ while (true) {
+ auto r = reclaim_queue.recv();
+ if (!r) {
+ break;
+ } else {
+ delete r;
+ }
+ }
+ });
+ photon::thread_enable_join(reclaimer);
+ }
+
+ ~ObjectCacheV2() {
+ _timer.stop();
+ if (reclaimer) {
+ reclaim_queue.template send<PhotonPause>(nullptr);
+ photon::thread_join((photon::join_handle*)reclaimer);
+ reclaimer = nullptr;
+ }
+ SCOPED_LOCK(maplock);
+ cycle_list.delete_all();
+ }
+};

common/test/CMakeLists.txt

@@ -5,6 +5,10 @@ add_executable(test-objcache test_objcache.cpp)
 target_link_libraries(test-objcache PRIVATE photon_shared)
 add_test(NAME test-objcache COMMAND $<TARGET_FILE:test-objcache>)
 
+add_executable(perf-objcache perf_objcache.cpp)
+target_link_libraries(perf-objcache PRIVATE photon_shared)
+add_test(NAME perf-objcache COMMAND $<TARGET_FILE:test-objcache>)
+
 add_executable(test-common test.cpp)
 target_link_libraries(test-common PRIVATE photon_shared)
 add_test(NAME test-common COMMAND $<TARGET_FILE:test-common>)

common/test/perf_objcache.cpp

@@ -2,16 +2,15 @@
 #include <photon/photon.h>
 #include <photon/thread/thread.h>
 
+#include <algorithm>
 #include <array>
 #include <chrono>
 #include <random>
 #include <thread>
-#include <unordered_map>
 #include <vector>
-#include <algorithm>
-#include "../../test/ci-tools.h"
 
 #include "../expirecontainer.h"
+#include "../objectcachev2.h"
 
 constexpr size_t count = 10UL * 1024;
 
@@ -29,11 +28,13 @@ void ready() {
  LOG_INFO("Random key generated");
 }
 
+template <template <class, class> class OC>
 void *task(void *arg) {
- auto oc = (ObjectCache<std::string, std::string *> *)arg;
+ auto oc = (OC<std::string, std::string *> *)arg;
  std::array<uint64_t, count> k = keys;
  std::random_shuffle(k.begin(), k.end());
 
+ auto start = std::chrono::steady_clock::now();
  for (const auto &x : k) {
  auto strx = std::to_string(x);
  auto b = oc->borrow(strx, [&strx] {
@@ -42,38 +43,47 @@ void *task(void *arg) {
  return new std::string(strx);
  });
  }
+ auto done = std::chrono::steady_clock::now();
+ LOG_INFO("spent ` ms",
+ std::chrono::duration_cast<std::chrono::milliseconds>(done - start)
+ .count());
  return nullptr;
 }
 
-void test_objcache(ObjectCache<std::string, std::string *> &oc) {
+template <template <class, class> class OC>
+void test_objcache(OC<std::string, std::string *> &oc, const char *name) {
  std::vector<photon::join_handle *> jhs;
- LOG_INFO("Query to ObjectCache");
- auto start = std::chrono::steady_clock::now();
+ LOG_INFO("Query to ", name);
  for (int i = 0; i < 4; i++)
  jhs.emplace_back(
- photon::thread_enable_join(photon::thread_create(task, &oc)));
+ photon::thread_enable_join(photon::thread_create(task<OC>, &oc)));
  for (auto &x : jhs) photon::thread_join(x);
- auto done = std::chrono::steady_clock::now();
- LOG_INFO("spent ` ms",
- std::chrono::duration_cast<std::chrono::milliseconds>(done - start)
- .count());
 }
 
-int main() {
- photon::init(0, 0);
- DEFER(photon::fini());
- ObjectCache<std::string, std::string *> oc(-1UL);
- ready();
+template <template <class, class> class OC>
+void test(const char *name) {
+ OC<std::string, std::string *> oc(-1UL);
  std::vector<std::thread> ths;
+ photon::semaphore sem(0);
  for (int i = 0; i < 10; i++) {
  ths.emplace_back([&] {
- photon::init(0, 0);
- DEFER(photon::fini());
- test_objcache(oc);
+ photon::vcpu_init();
+ DEFER(photon::vcpu_fini());
+ test_objcache<OC>(oc, name);
+ sem.signal(1);
  });
  }
+ sem.wait(10);
  for (auto &x : ths) {
  x.join();
  }
+}
+
+int main() {
+ photon::vcpu_init();
+ DEFER(photon::vcpu_fini());
+ ready();
+ test<ObjectCache>("ObjectCache");
+ test<ObjectCacheV2>("ObjectCacheV2");
  return 0;
 }