//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
//  This source code is licensed under both the GPLv2 (found in the
//  COPYING file in the root directory) and Apache 2.0 License
//  (found in the LICENSE.Apache file in the root directory).

#include "rocksdb/utilities/sim_cache.h"
#include <atomic>
#include "monitoring/statistics.h"
#include "port/port.h"
#include "rocksdb/env.h"
#include "util/file_reader_writer.h"
#include "util/mutexlock.h"
#include "util/string_util.h"

namespace rocksdb {

namespace {

class CacheActivityLogger {
 public:
  CacheActivityLogger()
      : activity_logging_enabled_(false), max_logging_size_(0) {}

  ~CacheActivityLogger() {
    MutexLock l(&mutex_);

    StopLoggingInternal();
  }

  Status StartLogging(const std::string& activity_log_file, Env* env,
                      uint64_t max_logging_size = 0) {
    assert(activity_log_file != "");
    assert(env != nullptr);

    Status status;
    EnvOptions env_opts;
    std::unique_ptr<WritableFile> log_file;

    MutexLock l(&mutex_);

    // Stop existing logging if any
    StopLoggingInternal();

    // Open log file
    status = env->NewWritableFile(activity_log_file, &log_file, env_opts);
    if (!status.ok()) {
      return status;
    }
    file_writer_.reset(new WritableFileWriter(std::move(log_file), env_opts));

    max_logging_size_ = max_logging_size;
    activity_logging_enabled_.store(true);

    return status;
  }

  void StopLogging() {
    MutexLock l(&mutex_);

    StopLoggingInternal();
  }

  void ReportLookup(const Slice& key) {
    if (activity_logging_enabled_.load() == false) {
      return;
    }

    std::string log_line = "LOOKUP - " + key.ToString(true) + "\n";

    // line format: "LOOKUP - <KEY>"
    MutexLock l(&mutex_);
    Status s = file_writer_->Append(log_line);
    if (!s.ok() && bg_status_.ok()) {
      bg_status_ = s;
    }
    if (MaxLoggingSizeReached() || !bg_status_.ok()) {
      // Stop logging if we have reached the max file size or
      // encountered an error
      StopLoggingInternal();
    }
  }

  void ReportAdd(const Slice& key, size_t size) {
    if (activity_logging_enabled_.load() == false) {
      return;
    }

    std::string log_line = "ADD - ";
    log_line += key.ToString(true);
    log_line += " - ";
    AppendNumberTo(&log_line, size);
		log_line += "\n";

    // line format: "ADD - <KEY> - <KEY-SIZE>"
    MutexLock l(&mutex_);
    Status s = file_writer_->Append(log_line);
    if (!s.ok() && bg_status_.ok()) {
      bg_status_ = s;
    }

    if (MaxLoggingSizeReached() || !bg_status_.ok()) {
      // Stop logging if we have reached the max file size or
      // encountered an error
      StopLoggingInternal();
    }
  }

  Status& bg_status() {
    MutexLock l(&mutex_);
    return bg_status_;
  }

 private:
  bool MaxLoggingSizeReached() {
    mutex_.AssertHeld();

    return (max_logging_size_ > 0 &&
            file_writer_->GetFileSize() >= max_logging_size_);
  }

  void StopLoggingInternal() {
    mutex_.AssertHeld();

    if (!activity_logging_enabled_) {
      return;
    }

    activity_logging_enabled_.store(false);
    Status s = file_writer_->Close();
    if (!s.ok() && bg_status_.ok()) {
      bg_status_ = s;
    }
  }

  // Mutex to sync writes to file_writer, and all following
  // class data members
  port::Mutex mutex_;
  // Indicates if logging is currently enabled
  // atomic to allow reads without mutex
  std::atomic<bool> activity_logging_enabled_;
  // When reached, we will stop logging and close the file
  // Value of 0 means unlimited
  uint64_t max_logging_size_;
  std::unique_ptr<WritableFileWriter> file_writer_;
  Status bg_status_;
};

// SimCacheImpl definition
class SimCacheImpl : public SimCache {
 public:
  // capacity for real cache (ShardedLRUCache)
  // test_capacity for key only cache
  SimCacheImpl(std::shared_ptr<Cache> cache, size_t sim_capacity,
               int num_shard_bits)
      : cache_(cache),
        key_only_cache_(NewLRUCache(sim_capacity, num_shard_bits)),
        miss_times_(0),
        hit_times_(0) {}

  virtual ~SimCacheImpl() {}
  virtual void SetCapacity(size_t capacity) override {
    cache_->SetCapacity(capacity);
  }

  virtual void SetStrictCapacityLimit(bool strict_capacity_limit) override {
    cache_->SetStrictCapacityLimit(strict_capacity_limit);
  }

  virtual Status Insert(const Slice& key, void* value, size_t charge,
                        void (*deleter)(const Slice& key, void* value),
                        Handle** handle, Priority priority) override {
    // The handle and value passed in are for real cache, so we pass nullptr
    // to key_only_cache_ for both instead. Also, the deleter function pointer
    // will be called by user to perform some external operation which should
    // be applied only once. Thus key_only_cache accepts an empty function.
    // *Lambda function without capture can be assgined to a function pointer
    Handle* h = key_only_cache_->Lookup(key);
    if (h == nullptr) {
      key_only_cache_->Insert(key, nullptr, charge,
                              [](const Slice& k, void* v) {}, nullptr,
                              priority);
    } else {
      key_only_cache_->Release(h);
    }

    cache_activity_logger_.ReportAdd(key, charge);

    return cache_->Insert(key, value, charge, deleter, handle, priority);
  }

  virtual Handle* Lookup(const Slice& key, Statistics* stats) override {
    Handle* h = key_only_cache_->Lookup(key);
    if (h != nullptr) {
      key_only_cache_->Release(h);
      inc_hit_counter();
      RecordTick(stats, SIM_BLOCK_CACHE_HIT);
    } else {
      inc_miss_counter();
      RecordTick(stats, SIM_BLOCK_CACHE_MISS);
    }

    cache_activity_logger_.ReportLookup(key);

    return cache_->Lookup(key, stats);
  }

  virtual bool Ref(Handle* handle) override { return cache_->Ref(handle); }

  virtual bool Release(Handle* handle, bool force_erase = false) override {
    return cache_->Release(handle, force_erase);
  }

  virtual void Erase(const Slice& key) override {
    cache_->Erase(key);
    key_only_cache_->Erase(key);
  }

  virtual void* Value(Handle* handle) override { return cache_->Value(handle); }

  virtual uint64_t NewId() override { return cache_->NewId(); }

  virtual size_t GetCapacity() const override { return cache_->GetCapacity(); }

  virtual bool HasStrictCapacityLimit() const override {
    return cache_->HasStrictCapacityLimit();
  }

  virtual size_t GetUsage() const override { return cache_->GetUsage(); }

  virtual size_t GetUsage(Handle* handle) const override {
    return cache_->GetUsage(handle);
  }

  virtual size_t GetPinnedUsage() const override {
    return cache_->GetPinnedUsage();
  }

  virtual void DisownData() override {
    cache_->DisownData();
    key_only_cache_->DisownData();
  }

  virtual void ApplyToAllCacheEntries(void (*callback)(void*, size_t),
                                      bool thread_safe) override {
    // only apply to _cache since key_only_cache doesn't hold value
    cache_->ApplyToAllCacheEntries(callback, thread_safe);
  }

  virtual void EraseUnRefEntries() override {
    cache_->EraseUnRefEntries();
    key_only_cache_->EraseUnRefEntries();
  }

  virtual size_t GetSimCapacity() const override {
    return key_only_cache_->GetCapacity();
  }
  virtual size_t GetSimUsage() const override {
    return key_only_cache_->GetUsage();
  }
  virtual void SetSimCapacity(size_t capacity) override {
    key_only_cache_->SetCapacity(capacity);
  }

  virtual uint64_t get_miss_counter() const override {
    return miss_times_.load(std::memory_order_relaxed);
  }

  virtual uint64_t get_hit_counter() const override {
    return hit_times_.load(std::memory_order_relaxed);
  }

  virtual void reset_counter() override {
    miss_times_.store(0, std::memory_order_relaxed);
    hit_times_.store(0, std::memory_order_relaxed);
    SetTickerCount(stats_, SIM_BLOCK_CACHE_HIT, 0);
    SetTickerCount(stats_, SIM_BLOCK_CACHE_MISS, 0);
  }

  virtual std::string ToString() const override {
    std::string res;
    res.append("SimCache MISSes: " + std::to_string(get_miss_counter()) + "\n");
    res.append("SimCache HITs:    " + std::to_string(get_hit_counter()) + "\n");
    char buff[350];
    auto lookups = get_miss_counter() + get_hit_counter();
    snprintf(buff, sizeof(buff), "SimCache HITRATE: %.2f%%\n",
             (lookups == 0 ? 0 : get_hit_counter() * 100.0f / lookups));
    res.append(buff);
    return res;
  }

  virtual std::string GetPrintableOptions() const override {
    std::string ret;
    ret.reserve(20000);
    ret.append("    cache_options:\n");
    ret.append(cache_->GetPrintableOptions());
    ret.append("    sim_cache_options:\n");
    ret.append(key_only_cache_->GetPrintableOptions());
    return ret;
  }

  virtual Status StartActivityLogging(const std::string& activity_log_file,
                                      Env* env,
                                      uint64_t max_logging_size = 0) override {
    return cache_activity_logger_.StartLogging(activity_log_file, env,
                                               max_logging_size);
  }

  virtual void StopActivityLogging() override {
    cache_activity_logger_.StopLogging();
  }

  virtual Status GetActivityLoggingStatus() override {
    return cache_activity_logger_.bg_status();
  }

 private:
  std::shared_ptr<Cache> cache_;
  std::shared_ptr<Cache> key_only_cache_;
  std::atomic<uint64_t> miss_times_;
  std::atomic<uint64_t> hit_times_;
  Statistics* stats_;
  CacheActivityLogger cache_activity_logger_;

  void inc_miss_counter() {
    miss_times_.fetch_add(1, std::memory_order_relaxed);
  }
  void inc_hit_counter() { hit_times_.fetch_add(1, std::memory_order_relaxed); }
};

}  // end anonymous namespace

// For instrumentation purpose, use NewSimCache instead
std::shared_ptr<SimCache> NewSimCache(std::shared_ptr<Cache> cache,
                                      size_t sim_capacity, int num_shard_bits) {
  if (num_shard_bits >= 20) {
    return nullptr;  // the cache cannot be sharded into too many fine pieces
  }
  return std::make_shared<SimCacheImpl>(cache, sim_capacity, num_shard_bits);
}

}  // end namespace rocksdb