LRUPersistentCache.h

Go to the documentation of this file.

/*
 * Copyright 2017-present Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#pragma once

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <future>
#include <map>
#include <thread>

#include <boost/noncopyable.hpp>
#include <folly/Executor.h>
#include <folly/dynamic.h>
#include <wangle/client/persistence/LRUInMemoryCache.h>
#include <wangle/client/persistence/PersistentCache.h>
#include <wangle/client/persistence/PersistentCacheCommon.h>

namespace wangle {

template<typename K, typename V>
class CachePersistence {
 public:
  CachePersistence() : persistedVersion_(0) {}

  virtual ~CachePersistence() = default;

  bool persistVersionedData(
      const folly::dynamic& kvPairs, const CacheDataVersion& version) {
    auto result = persist(kvPairs);
    if (result) {
      persistedVersion_ = version;
    }
    return result;
  }

  virtual CacheDataVersion getLastPersistedVersion() const {
    return persistedVersion_;
  }

  virtual void setPersistedVersion(CacheDataVersion version) noexcept {
    persistedVersion_ = version;
  }

  virtual bool persist(const folly::dynamic& kvPairs) noexcept = 0;

  virtual folly::Optional<folly::dynamic> load() noexcept = 0;

  virtual void clear() = 0;

 private:
  CacheDataVersion persistedVersion_;
};

namespace client {
namespace persistence {
constexpr std::chrono::milliseconds DEFAULT_CACHE_SYNC_INTERVAL =
    std::chrono::milliseconds(5000);
constexpr int DEFAULT_CACHE_SYNC_RETRIES = 3;
} // namespace persistence
} // namespace client

template <typename K, typename V, typename MutexT = std::mutex>
class LRUPersistentCache
    : public PersistentCache<K, V>,
      public std::enable_shared_from_this<LRUPersistentCache<K, V, MutexT>>,
      private boost::noncopyable {
 public:
  using Ptr = std::shared_ptr<LRUPersistentCache<K, V, MutexT>>;

  explicit LRUPersistentCache(
      std::size_t cacheCapacity,
      std::chrono::milliseconds syncInterval =
          client::persistence::DEFAULT_CACHE_SYNC_INTERVAL,
      int nSyncRetries = client::persistence::DEFAULT_CACHE_SYNC_RETRIES,
      std::unique_ptr<CachePersistence<K, V>> persistence = nullptr);

  LRUPersistentCache(
    std::shared_ptr<folly::Executor> executor,
    std::size_t cacheCapacity,
    std::chrono::milliseconds syncInterval,
    int nSyncRetries,
    std::unique_ptr<CachePersistence<K, V>> persistence);

  ~LRUPersistentCache() override;

  bool hasPendingUpdates();

  folly::Optional<V> get(const K& key) override {
    return cache_.get(key);
  }

  void put(const K& key, const V& val) override;

  bool remove(const K& key) override {
    return cache_.remove(key);
  }

  void clear(bool clearPersistence = false) override {
    cache_.clear();
    if (clearPersistence) {
      auto persistence = getPersistence();
      if (persistence) {
        persistence->clear();
      }
    }
  }

  size_t size() override {
    return cache_.size();
  }

  void setSyncOnDestroy(bool syncOnDestroy) {
    syncOnDestroy_ = syncOnDestroy;
  }

  void setPersistence(std::unique_ptr<CachePersistence<K, V>> persistence);

 private:
  void setPersistenceHelper(
    std::unique_ptr<CachePersistence<K, V>> persistence,
    bool syncVersion) noexcept;

  CacheDataVersion load(CachePersistence<K, V>& persistence) noexcept;

  void sync();
  void oneShotSync();
  static void* syncThreadMain(void* arg);

  bool syncNow(CachePersistence<K, V>& persistence);

  std::shared_ptr<CachePersistence<K, V>> getPersistence();

 private:
  // Our threadsafe in memory cache
  LRUInMemoryCache<K, V, MutexT> cache_;

  // used to signal syncer thread
  bool stopSyncer_{false};
  // mutex used to synchronize syncer_ on destruction, tied to stopSyncerCV_
  std::mutex stopSyncerMutex_;
  // condvar used to wakeup syncer on exit
  std::condition_variable stopSyncerCV_;

  // We do not schedule the same task more than one to the executor
  std::atomic_flag executorScheduled_ = ATOMIC_FLAG_INIT;

  // sync interval in milliseconds
  const std::chrono::milliseconds syncInterval_{
      client::persistence::DEFAULT_CACHE_SYNC_INTERVAL};
  // limit on no. of sync attempts
  const int nSyncRetries_{client::persistence::DEFAULT_CACHE_SYNC_RETRIES};
  // Sync try count across executor tasks
  int nSyncTries_{0};
  std::chrono::steady_clock::time_point lastExecutorScheduleTime_;

  // Whether or not to do 1 sync before destruction.
  bool syncOnDestroy_{false};

  // persistence layer
  // we use a shared pointer since the syncer thread might be operating on
  // it when the client decides to set a new one
  std::shared_ptr<CachePersistence<K, V>> persistence_;
  // for locking access to persistence set/get
  MutexT persistenceLock_;

  // thread for periodic sync
  std::thread syncer_;

  // executor for periodic sync.
  std::shared_ptr<folly::Executor> executor_;
};

}

#include <wangle/client/persistence/LRUPersistentCache-inl.h>