ceph/src/rocksdb/utilities/simulator_cache/sim_cache.cc

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5
   6 #include "rocksdb/utilities/sim_cache.h"
   7
   8 #include <atomic>
   9 #include <iomanip>
  10
  11 #include "file/writable_file_writer.h"
  12 #include "monitoring/statistics.h"
  13 #include "port/port.h"
  14 #include "rocksdb/env.h"
  15 #include "rocksdb/file_system.h"
  16 #include "util/mutexlock.h"
  17
  18 namespace ROCKSDB_NAMESPACE {
  19
  20 namespace {
  21
  22 class CacheActivityLogger {
  23  public:
  24   CacheActivityLogger()
  25       : activity_logging_enabled_(false), max_logging_size_(0) {}
  26
  27   ~CacheActivityLogger() {
  28     MutexLock l(&mutex_);
  29
  30     StopLoggingInternal();
  31     bg_status_.PermitUncheckedError();
  32   }
  33
  34   Status StartLogging(const std::string& activity_log_file, Env* env,
  35                       uint64_t max_logging_size = 0) {
  36     assert(activity_log_file != "");
  37     assert(env != nullptr);
  38
  39     Status status;
  40     FileOptions file_opts;
  41
  42     MutexLock l(&mutex_);
  43
  44     // Stop existing logging if any
  45     StopLoggingInternal();
  46
  47     // Open log file
  48     status = WritableFileWriter::Create(env->GetFileSystem(), activity_log_file,
  49                                         file_opts, &file_writer_, nullptr);
  50     if (!status.ok()) {
  51       return status;
  52     }
  53
  54     max_logging_size_ = max_logging_size;
  55     activity_logging_enabled_.store(true);
  56
  57     return status;
  58   }
  59
  60   void StopLogging() {
  61     MutexLock l(&mutex_);
  62
  63     StopLoggingInternal();
  64   }
  65
  66   void ReportLookup(const Slice& key) {
  67     if (activity_logging_enabled_.load() == false) {
  68       return;
  69     }
  70
  71     std::ostringstream oss;
  72     // line format: "LOOKUP - <KEY>"
  73     oss << "LOOKUP - " << key.ToString(true) << std::endl;
  74
  75     MutexLock l(&mutex_);
  76     Status s = file_writer_->Append(oss.str());
  77     if (!s.ok() && bg_status_.ok()) {
  78       bg_status_ = s;
  79     }
  80     if (MaxLoggingSizeReached() || !bg_status_.ok()) {
  81       // Stop logging if we have reached the max file size or
  82       // encountered an error
  83       StopLoggingInternal();
  84     }
  85   }
  86
  87   void ReportAdd(const Slice& key, size_t size) {
  88     if (activity_logging_enabled_.load() == false) {
  89       return;
  90     }
  91
  92     std::ostringstream oss;
  93     // line format: "ADD - <KEY> - <KEY-SIZE>"
  94     oss << "ADD - " << key.ToString(true) << " - " << size << std::endl;
  95     MutexLock l(&mutex_);
  96     Status s = file_writer_->Append(oss.str());
  97     if (!s.ok() && bg_status_.ok()) {
  98       bg_status_ = s;
  99     }
 100
 101     if (MaxLoggingSizeReached() || !bg_status_.ok()) {
 102       // Stop logging if we have reached the max file size or
 103       // encountered an error
 104       StopLoggingInternal();
 105     }
 106   }
 107
 108   Status& bg_status() {
 109     MutexLock l(&mutex_);
 110     return bg_status_;
 111   }
 112
 113  private:
 114   bool MaxLoggingSizeReached() {
 115     mutex_.AssertHeld();
 116
 117     return (max_logging_size_ > 0 &&
 118             file_writer_->GetFileSize() >= max_logging_size_);
 119   }
 120
 121   void StopLoggingInternal() {
 122     mutex_.AssertHeld();
 123
 124     if (!activity_logging_enabled_) {
 125       return;
 126     }
 127
 128     activity_logging_enabled_.store(false);
 129     Status s = file_writer_->Close();
 130     if (!s.ok() && bg_status_.ok()) {
 131       bg_status_ = s;
 132     }
 133   }
 134
 135   // Mutex to sync writes to file_writer, and all following
 136   // class data members
 137   port::Mutex mutex_;
 138   // Indicates if logging is currently enabled
 139   // atomic to allow reads without mutex
 140   std::atomic<bool> activity_logging_enabled_;
 141   // When reached, we will stop logging and close the file
 142   // Value of 0 means unlimited
 143   uint64_t max_logging_size_;
 144   std::unique_ptr<WritableFileWriter> file_writer_;
 145   Status bg_status_;
 146 };
 147
 148 // SimCacheImpl definition
 149 class SimCacheImpl : public SimCache {
 150  public:
 151   // capacity for real cache (ShardedLRUCache)
 152   // test_capacity for key only cache
 153   SimCacheImpl(std::shared_ptr<Cache> sim_cache, std::shared_ptr<Cache> cache)
 154       : cache_(cache),
 155         key_only_cache_(sim_cache),
 156         miss_times_(0),
 157         hit_times_(0),
 158         stats_(nullptr) {}
 159
 160   ~SimCacheImpl() override {}
 161   void SetCapacity(size_t capacity) override { cache_->SetCapacity(capacity); }
 162
 163   void SetStrictCapacityLimit(bool strict_capacity_limit) override {
 164     cache_->SetStrictCapacityLimit(strict_capacity_limit);
 165   }
 166
 167   using Cache::Insert;
 168   Status Insert(const Slice& key, void* value, size_t charge,
 169                 void (*deleter)(const Slice& key, void* value), Handle** handle,
 170                 Priority priority) override {
 171     // The handle and value passed in are for real cache, so we pass nullptr
 172     // to key_only_cache_ for both instead. Also, the deleter function pointer
 173     // will be called by user to perform some external operation which should
 174     // be applied only once. Thus key_only_cache accepts an empty function.
 175     // *Lambda function without capture can be assgined to a function pointer
 176     Handle* h = key_only_cache_->Lookup(key);
 177     if (h == nullptr) {
 178       // TODO: Check for error here?
 179       auto s = key_only_cache_->Insert(
 180           key, nullptr, charge, [](const Slice& /*k*/, void* /*v*/) {}, nullptr,
 181           priority);
 182       s.PermitUncheckedError();
 183     } else {
 184       key_only_cache_->Release(h);
 185     }
 186
 187     cache_activity_logger_.ReportAdd(key, charge);
 188     if (!cache_) {
 189       return Status::OK();
 190     }
 191     return cache_->Insert(key, value, charge, deleter, handle, priority);
 192   }
 193
 194   using Cache::Lookup;
 195   Handle* Lookup(const Slice& key, Statistics* stats) override {
 196     Handle* h = key_only_cache_->Lookup(key);
 197     if (h != nullptr) {
 198       key_only_cache_->Release(h);
 199       inc_hit_counter();
 200       RecordTick(stats, SIM_BLOCK_CACHE_HIT);
 201     } else {
 202       inc_miss_counter();
 203       RecordTick(stats, SIM_BLOCK_CACHE_MISS);
 204     }
 205
 206     cache_activity_logger_.ReportLookup(key);
 207     if (!cache_) {
 208       return nullptr;
 209     }
 210     return cache_->Lookup(key, stats);
 211   }
 212
 213   bool Ref(Handle* handle) override { return cache_->Ref(handle); }
 214
 215   using Cache::Release;
 216   bool Release(Handle* handle, bool erase_if_last_ref = false) override {
 217     return cache_->Release(handle, erase_if_last_ref);
 218   }
 219
 220   void Erase(const Slice& key) override {
 221     cache_->Erase(key);
 222     key_only_cache_->Erase(key);
 223   }
 224
 225   void* Value(Handle* handle) override { return cache_->Value(handle); }
 226
 227   uint64_t NewId() override { return cache_->NewId(); }
 228
 229   size_t GetCapacity() const override { return cache_->GetCapacity(); }
 230
 231   bool HasStrictCapacityLimit() const override {
 232     return cache_->HasStrictCapacityLimit();
 233   }
 234
 235   size_t GetUsage() const override { return cache_->GetUsage(); }
 236
 237   size_t GetUsage(Handle* handle) const override {
 238     return cache_->GetUsage(handle);
 239   }
 240
 241   size_t GetCharge(Handle* handle) const override {
 242     return cache_->GetCharge(handle);
 243   }
 244
 245   DeleterFn GetDeleter(Handle* handle) const override {
 246     return cache_->GetDeleter(handle);
 247   }
 248
 249   size_t GetPinnedUsage() const override { return cache_->GetPinnedUsage(); }
 250
 251   void DisownData() override {
 252     cache_->DisownData();
 253     key_only_cache_->DisownData();
 254   }
 255
 256   void ApplyToAllCacheEntries(void (*callback)(void*, size_t),
 257                               bool thread_safe) override {
 258     // only apply to _cache since key_only_cache doesn't hold value
 259     cache_->ApplyToAllCacheEntries(callback, thread_safe);
 260   }
 261
 262   void ApplyToAllEntries(
 263       const std::function<void(const Slice& key, void* value, size_t charge,
 264                                DeleterFn deleter)>& callback,
 265       const ApplyToAllEntriesOptions& opts) override {
 266     cache_->ApplyToAllEntries(callback, opts);
 267   }
 268
 269   void EraseUnRefEntries() override {
 270     cache_->EraseUnRefEntries();
 271     key_only_cache_->EraseUnRefEntries();
 272   }
 273
 274   size_t GetSimCapacity() const override {
 275     return key_only_cache_->GetCapacity();
 276   }
 277   size_t GetSimUsage() const override { return key_only_cache_->GetUsage(); }
 278   void SetSimCapacity(size_t capacity) override {
 279     key_only_cache_->SetCapacity(capacity);
 280   }
 281
 282   uint64_t get_miss_counter() const override {
 283     return miss_times_.load(std::memory_order_relaxed);
 284   }
 285
 286   uint64_t get_hit_counter() const override {
 287     return hit_times_.load(std::memory_order_relaxed);
 288   }
 289
 290   void reset_counter() override {
 291     miss_times_.store(0, std::memory_order_relaxed);
 292     hit_times_.store(0, std::memory_order_relaxed);
 293     SetTickerCount(stats_, SIM_BLOCK_CACHE_HIT, 0);
 294     SetTickerCount(stats_, SIM_BLOCK_CACHE_MISS, 0);
 295   }
 296
 297   std::string ToString() const override {
 298     std::ostringstream oss;
 299     oss << "SimCache MISSes:  " << get_miss_counter() << std::endl;
 300     oss << "SimCache HITs:    " << get_hit_counter() << std::endl;
 301     auto lookups = get_miss_counter() + get_hit_counter();
 302     oss << "SimCache HITRATE: " << std::fixed << std::setprecision(2)
 303         << (lookups == 0 ? 0 : get_hit_counter() * 100.0f / lookups)
 304         << std::endl;
 305     return oss.str();
 306   }
 307
 308   std::string GetPrintableOptions() const override {
 309     std::ostringstream oss;
 310     oss << "    cache_options:" << std::endl;
 311     oss << cache_->GetPrintableOptions();
 312     oss << "    sim_cache_options:" << std::endl;
 313     oss << key_only_cache_->GetPrintableOptions();
 314     return oss.str();
 315   }
 316
 317   Status StartActivityLogging(const std::string& activity_log_file, Env* env,
 318                               uint64_t max_logging_size = 0) override {
 319     return cache_activity_logger_.StartLogging(activity_log_file, env,
 320                                                max_logging_size);
 321   }
 322
 323   void StopActivityLogging() override { cache_activity_logger_.StopLogging(); }
 324
 325   Status GetActivityLoggingStatus() override {
 326     return cache_activity_logger_.bg_status();
 327   }
 328
 329  private:
 330   std::shared_ptr<Cache> cache_;
 331   std::shared_ptr<Cache> key_only_cache_;
 332   std::atomic<uint64_t> miss_times_;
 333   std::atomic<uint64_t> hit_times_;
 334   Statistics* stats_;
 335   CacheActivityLogger cache_activity_logger_;
 336
 337   void inc_miss_counter() {
 338     miss_times_.fetch_add(1, std::memory_order_relaxed);
 339   }
 340   void inc_hit_counter() { hit_times_.fetch_add(1, std::memory_order_relaxed); }
 341 };
 342
 343 }  // end anonymous namespace
 344
 345 // For instrumentation purpose, use NewSimCache instead
 346 std::shared_ptr<SimCache> NewSimCache(std::shared_ptr<Cache> cache,
 347                                       size_t sim_capacity, int num_shard_bits) {
 348   LRUCacheOptions co;
 349   co.capacity = sim_capacity;
 350   co.num_shard_bits = num_shard_bits;
 351   co.metadata_charge_policy = kDontChargeCacheMetadata;
 352   return NewSimCache(NewLRUCache(co), cache, num_shard_bits);
 353 }
 354
 355 std::shared_ptr<SimCache> NewSimCache(std::shared_ptr<Cache> sim_cache,
 356                                       std::shared_ptr<Cache> cache,
 357                                       int num_shard_bits) {
 358   if (num_shard_bits >= 20) {
 359     return nullptr;  // the cache cannot be sharded into too many fine pieces
 360   }
 361   return std::make_shared<SimCacheImpl>(sim_cache, cache);
 362 }
 363
 364 }  // namespace ROCKSDB_NAMESPACE