//  Copyright (c) Meta Platforms, Inc. and affiliates.
//  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 "cache/secondary_cache_adapter.h"

#include <atomic>

#include "cache/tiered_secondary_cache.h"
#include "monitoring/perf_context_imp.h"
#include "test_util/sync_point.h"
#include "util/cast_util.h"

namespace ROCKSDB_NAMESPACE {

namespace {
// A distinct pointer value for marking "dummy" cache entries
struct Dummy {
  char val[7] = "kDummy";
};
const Dummy kDummy{};
Cache::ObjectPtr const kDummyObj = const_cast<Dummy*>(&kDummy);
const char* kTieredCacheName = "TieredCache";
}  // namespace

// When CacheWithSecondaryAdapter is constructed with the distribute_cache_res
// parameter set to true, it manages the entire memory budget across the
// primary and secondary cache. The secondary cache is assumed to be in
// memory, such as the CompressedSecondaryCache. When a placeholder entry
// is inserted by a CacheReservationManager instance to reserve memory,
// the CacheWithSecondaryAdapter ensures that the reservation is distributed
// proportionally across the primary/secondary caches.
//
// The primary block cache is initially sized to the sum of the primary cache
// budget + teh secondary cache budget, as follows -
//   |---------    Primary Cache Configured Capacity  -----------|
//   |---Secondary Cache Budget----|----Primary Cache Budget-----|
//
// A ConcurrentCacheReservationManager member in the CacheWithSecondaryAdapter,
// pri_cache_res_,
// is used to help with tracking the distribution of memory reservations.
// Initially, it accounts for the entire secondary cache budget as a
// reservation against the primary cache. This shrinks the usable capacity of
// the primary cache to the budget that the user originally desired.
//
//   |--Reservation for Sec Cache--|-Pri Cache Usable Capacity---|
//
// When a reservation placeholder is inserted into the adapter, it is inserted
// directly into the primary cache. This means the entire charge of the
// placeholder is counted against the primary cache. To compensate and count
// a portion of it against the secondary cache, the secondary cache Deflate()
// method is called to shrink it. Since the Deflate() causes the secondary
// actual usage to shrink, it is refelcted here by releasing an equal amount
// from the pri_cache_res_ reservation. The Deflate() in the secondary cache
// can be, but is not required to be, implemented using its own cache
// reservation manager.
//
// For example, if the pri/sec ratio is 70/30, and the combined capacity is
// 100MB, the intermediate and final  state after inserting a reservation
// placeholder for 10MB would be as follows -
//
//   |-Reservation for Sec Cache-|-Pri Cache Usable Capacity-|---R---|
// 1. After inserting the placeholder in primary
//   |-------  30MB -------------|------- 60MB -------------|-10MB--|
// 2. After deflating the secondary and adjusting the reservation for
//    secondary against the primary
//   |-------  27MB -------------|------- 63MB -------------|-10MB--|
//
// Likewise, when the user inserted placeholder is released, the secondary
// cache Inflate() method is called to grow it, and the pri_cache_res_
// reservation is increased by an equal amount.
//
// Another way of implementing this would have been to simply split the user
// reservation into primary and seconary components. However, this would
// require allocating a structure to track the associated secondary cache
// reservation, which adds some complexity and overhead.
//
CacheWithSecondaryAdapter::CacheWithSecondaryAdapter(
    std::shared_ptr<Cache> target,
    std::shared_ptr<SecondaryCache> secondary_cache,
    TieredAdmissionPolicy adm_policy, bool distribute_cache_res)
    : CacheWrapper(std::move(target)),
      secondary_cache_(std::move(secondary_cache)),
      adm_policy_(adm_policy),
      distribute_cache_res_(distribute_cache_res),
      placeholder_usage_(0),
      reserved_usage_(0),
      sec_reserved_(0) {
  target_->SetEvictionCallback(
      [this](const Slice& key, Handle* handle, bool was_hit) {
        return EvictionHandler(key, handle, was_hit);
      });
  if (distribute_cache_res_) {
    size_t sec_capacity = 0;
    pri_cache_res_ = std::make_shared<ConcurrentCacheReservationManager>(
        std::make_shared<CacheReservationManagerImpl<CacheEntryRole::kMisc>>(
            target_));
    Status s = secondary_cache_->GetCapacity(sec_capacity);
    assert(s.ok());
    // Initially, the primary cache is sized to uncompressed cache budget plsu
    // compressed secondary cache budget. The secondary cache budget is then
    // taken away from the primary cache through cache reservations. Later,
    // when a placeholder entry is inserted by the caller, its inserted
    // into the primary cache and the portion that should be assigned to the
    // secondary cache is freed from the reservation.
    s = pri_cache_res_->UpdateCacheReservation(sec_capacity);
    assert(s.ok());
    sec_cache_res_ratio_ = (double)sec_capacity / target_->GetCapacity();
  }
}

CacheWithSecondaryAdapter::~CacheWithSecondaryAdapter() {
  // `*this` will be destroyed before `*target_`, so we have to prevent
  // use after free
  target_->SetEvictionCallback({});
#ifndef NDEBUG
  if (distribute_cache_res_) {
    size_t sec_capacity = 0;
    Status s = secondary_cache_->GetCapacity(sec_capacity);
    assert(s.ok());
    assert(placeholder_usage_ == 0);
    assert(reserved_usage_ == 0);
    assert(pri_cache_res_->GetTotalMemoryUsed() == sec_capacity);
  }
#endif  // NDEBUG
}

bool CacheWithSecondaryAdapter::EvictionHandler(const Slice& key,
                                                Handle* handle, bool was_hit) {
  auto helper = GetCacheItemHelper(handle);
  if (helper->IsSecondaryCacheCompatible() &&
      adm_policy_ != TieredAdmissionPolicy::kAdmPolicyThreeQueue) {
    auto obj = target_->Value(handle);
    // Ignore dummy entry
    if (obj != kDummyObj) {
      bool force = false;
      if (adm_policy_ == TieredAdmissionPolicy::kAdmPolicyAllowCacheHits) {
        force = was_hit;
      } else if (adm_policy_ == TieredAdmissionPolicy::kAdmPolicyAllowAll) {
        force = true;
      }
      // Spill into secondary cache.
      secondary_cache_->Insert(key, obj, helper, force).PermitUncheckedError();
    }
  }
  // Never takes ownership of obj
  return false;
}

bool CacheWithSecondaryAdapter::ProcessDummyResult(Cache::Handle** handle,
                                                   bool erase) {
  if (*handle && target_->Value(*handle) == kDummyObj) {
    target_->Release(*handle, erase);
    *handle = nullptr;
    return true;
  } else {
    return false;
  }
}

void CacheWithSecondaryAdapter::CleanupCacheObject(
    ObjectPtr obj, const CacheItemHelper* helper) {
  if (helper->del_cb) {
    helper->del_cb(obj, memory_allocator());
  }
}

Cache::Handle* CacheWithSecondaryAdapter::Promote(
    std::unique_ptr<SecondaryCacheResultHandle>&& secondary_handle,
    const Slice& key, const CacheItemHelper* helper, Priority priority,
    Statistics* stats, bool found_dummy_entry, bool kept_in_sec_cache) {
  assert(secondary_handle->IsReady());

  ObjectPtr obj = secondary_handle->Value();
  if (!obj) {
    // Nothing found.
    return nullptr;
  }
  // Found something.
  switch (helper->role) {
    case CacheEntryRole::kFilterBlock:
      RecordTick(stats, SECONDARY_CACHE_FILTER_HITS);
      break;
    case CacheEntryRole::kIndexBlock:
      RecordTick(stats, SECONDARY_CACHE_INDEX_HITS);
      break;
    case CacheEntryRole::kDataBlock:
      RecordTick(stats, SECONDARY_CACHE_DATA_HITS);
      break;
    default:
      break;
  }
  PERF_COUNTER_ADD(secondary_cache_hit_count, 1);
  RecordTick(stats, SECONDARY_CACHE_HITS);

  // Note: SecondaryCache::Size() is really charge (from the CreateCallback)
  size_t charge = secondary_handle->Size();
  Handle* result = nullptr;
  // Insert into primary cache, possibly as a standalone+dummy entries.
  if (secondary_cache_->SupportForceErase() && !found_dummy_entry) {
    // Create standalone and insert dummy
    // Allow standalone to be created even if cache is full, to avoid
    // reading the entry from storage.
    result =
        CreateStandalone(key, obj, helper, charge, /*allow_uncharged*/ true);
    assert(result);
    PERF_COUNTER_ADD(block_cache_standalone_handle_count, 1);

    // Insert dummy to record recent use
    // TODO: try to avoid case where inserting this dummy could overwrite a
    // regular entry
    Status s = Insert(key, kDummyObj, &kNoopCacheItemHelper, /*charge=*/0,
                      /*handle=*/nullptr, priority);
    s.PermitUncheckedError();
    // Nothing to do or clean up on dummy insertion failure
  } else {
    // Insert regular entry into primary cache.
    // Don't allow it to spill into secondary cache again if it was kept there.
    Status s = Insert(
        key, obj, kept_in_sec_cache ? helper->without_secondary_compat : helper,
        charge, &result, priority);
    if (s.ok()) {
      assert(result);
      PERF_COUNTER_ADD(block_cache_real_handle_count, 1);
    } else {
      // Create standalone result instead, even if cache is full, to avoid
      // reading the entry from storage.
      result =
          CreateStandalone(key, obj, helper, charge, /*allow_uncharged*/ true);
      assert(result);
      PERF_COUNTER_ADD(block_cache_standalone_handle_count, 1);
    }
  }
  return result;
}

Status CacheWithSecondaryAdapter::Insert(const Slice& key, ObjectPtr value,
                                         const CacheItemHelper* helper,
                                         size_t charge, Handle** handle,
                                         Priority priority,
                                         const Slice& compressed_value,
                                         CompressionType type) {
  Status s = target_->Insert(key, value, helper, charge, handle, priority);
  if (s.ok() && value == nullptr && distribute_cache_res_ && handle) {
    charge = target_->GetCharge(*handle);

    MutexLock l(&cache_res_mutex_);
    placeholder_usage_ += charge;
    // Check if total placeholder reservation is more than the overall
    // cache capacity. If it is, then we don't try to charge the
    // secondary cache because we don't want to overcharge it (beyond
    // its capacity).
    // In order to make this a bit more lightweight, we also check if
    // the difference between placeholder_usage_ and reserved_usage_ is
    // atleast kReservationChunkSize and avoid any adjustments if not.
    if ((placeholder_usage_ <= target_->GetCapacity()) &&
        ((placeholder_usage_ - reserved_usage_) >= kReservationChunkSize)) {
      reserved_usage_ = placeholder_usage_ & ~(kReservationChunkSize - 1);
      size_t new_sec_reserved =
          static_cast<size_t>(reserved_usage_ * sec_cache_res_ratio_);
      size_t sec_charge = new_sec_reserved - sec_reserved_;
      s = secondary_cache_->Deflate(sec_charge);
      assert(s.ok());
      s = pri_cache_res_->UpdateCacheReservation(sec_charge,
                                                 /*increase=*/false);
      assert(s.ok());
      sec_reserved_ += sec_charge;
    }
  }
  // Warm up the secondary cache with the compressed block. The secondary
  // cache may choose to ignore it based on the admission policy.
  if (value != nullptr && !compressed_value.empty() &&
      adm_policy_ == TieredAdmissionPolicy::kAdmPolicyThreeQueue &&
      helper->IsSecondaryCacheCompatible()) {
    Status status = secondary_cache_->InsertSaved(key, compressed_value, type);
    assert(status.ok() || status.IsNotSupported());
  }

  return s;
}

Cache::Handle* CacheWithSecondaryAdapter::Lookup(const Slice& key,
                                                 const CacheItemHelper* helper,
                                                 CreateContext* create_context,
                                                 Priority priority,
                                                 Statistics* stats) {
  // NOTE: we could just StartAsyncLookup() and Wait(), but this should be a bit
  // more efficient
  Handle* result =
      target_->Lookup(key, helper, create_context, priority, stats);
  bool secondary_compatible = helper && helper->IsSecondaryCacheCompatible();
  bool found_dummy_entry =
      ProcessDummyResult(&result, /*erase=*/secondary_compatible);
  if (!result && secondary_compatible) {
    // Try our secondary cache
    bool kept_in_sec_cache = false;
    std::unique_ptr<SecondaryCacheResultHandle> secondary_handle =
        secondary_cache_->Lookup(key, helper, create_context, /*wait*/ true,
                                 found_dummy_entry, stats,
                                 /*out*/ kept_in_sec_cache);
    if (secondary_handle) {
      result = Promote(std::move(secondary_handle), key, helper, priority,
                       stats, found_dummy_entry, kept_in_sec_cache);
    }
  }
  return result;
}

bool CacheWithSecondaryAdapter::Release(Handle* handle,
                                        bool erase_if_last_ref) {
  if (erase_if_last_ref) {
    ObjectPtr v = target_->Value(handle);
    if (v == nullptr && distribute_cache_res_) {
      size_t charge = target_->GetCharge(handle);

      MutexLock l(&cache_res_mutex_);
      placeholder_usage_ -= charge;
      // Check if total placeholder reservation is more than the overall
      // cache capacity. If it is, then we do nothing as reserved_usage_ must
      // be already maxed out
      if ((placeholder_usage_ <= target_->GetCapacity()) &&
          (placeholder_usage_ < reserved_usage_)) {
        // Adjust reserved_usage_ in chunks of kReservationChunkSize, so
        // we don't hit this slow path too often.
        reserved_usage_ = placeholder_usage_ & ~(kReservationChunkSize - 1);
        size_t new_sec_reserved =
            static_cast<size_t>(reserved_usage_ * sec_cache_res_ratio_);
        size_t sec_charge = sec_reserved_ - new_sec_reserved;
        Status s = secondary_cache_->Inflate(sec_charge);
        assert(s.ok());
        s = pri_cache_res_->UpdateCacheReservation(sec_charge,
                                                   /*increase=*/true);
        assert(s.ok());
        sec_reserved_ -= sec_charge;
      }
    }
  }
  return target_->Release(handle, erase_if_last_ref);
}

Cache::ObjectPtr CacheWithSecondaryAdapter::Value(Handle* handle) {
  ObjectPtr v = target_->Value(handle);
  // TODO with stacked secondaries: might fail in EvictionHandler
  assert(v != kDummyObj);
  return v;
}

void CacheWithSecondaryAdapter::StartAsyncLookupOnMySecondary(
    AsyncLookupHandle& async_handle) {
  assert(!async_handle.IsPending());
  assert(async_handle.result_handle == nullptr);

  std::unique_ptr<SecondaryCacheResultHandle> secondary_handle =
      secondary_cache_->Lookup(
          async_handle.key, async_handle.helper, async_handle.create_context,
          /*wait*/ false, async_handle.found_dummy_entry, async_handle.stats,
          /*out*/ async_handle.kept_in_sec_cache);
  if (secondary_handle) {
    // TODO with stacked secondaries: Check & process if already ready?
    async_handle.pending_handle = secondary_handle.release();
    async_handle.pending_cache = secondary_cache_.get();
  }
}

void CacheWithSecondaryAdapter::StartAsyncLookup(
    AsyncLookupHandle& async_handle) {
  target_->StartAsyncLookup(async_handle);
  if (!async_handle.IsPending()) {
    bool secondary_compatible =
        async_handle.helper &&
        async_handle.helper->IsSecondaryCacheCompatible();
    async_handle.found_dummy_entry |= ProcessDummyResult(
        &async_handle.result_handle, /*erase=*/secondary_compatible);

    if (async_handle.Result() == nullptr && secondary_compatible) {
      // Not found and not pending on another secondary cache
      StartAsyncLookupOnMySecondary(async_handle);
    }
  }
}

void CacheWithSecondaryAdapter::WaitAll(AsyncLookupHandle* async_handles,
                                        size_t count) {
  if (count == 0) {
    // Nothing to do
    return;
  }
  // Requests that are pending on *my* secondary cache, at the start of this
  // function
  std::vector<AsyncLookupHandle*> my_pending;
  // Requests that are pending on an "inner" secondary cache (managed somewhere
  // under target_), as of the start of this function
  std::vector<AsyncLookupHandle*> inner_pending;

  // Initial accounting of pending handles, excluding those already handled
  // by "outer" secondary caches. (See cur->pending_cache = nullptr.)
  for (size_t i = 0; i < count; ++i) {
    AsyncLookupHandle* cur = async_handles + i;
    if (cur->pending_cache) {
      assert(cur->IsPending());
      assert(cur->helper);
      assert(cur->helper->IsSecondaryCacheCompatible());
      if (cur->pending_cache == secondary_cache_.get()) {
        my_pending.push_back(cur);
        // Mark as "to be handled by this caller"
        cur->pending_cache = nullptr;
      } else {
        // Remember as potentially needing a lookup in my secondary
        inner_pending.push_back(cur);
      }
    }
  }

  // Wait on inner-most cache lookups first
  // TODO with stacked secondaries: because we are not using proper
  // async/await constructs here yet, there is a false synchronization point
  // here where all the results at one level are needed before initiating
  // any lookups at the next level. Probably not a big deal, but worth noting.
  if (!inner_pending.empty()) {
    target_->WaitAll(async_handles, count);
  }

  // For those that failed to find something, convert to lookup in my
  // secondary cache.
  for (AsyncLookupHandle* cur : inner_pending) {
    if (cur->Result() == nullptr) {
      // Not found, try my secondary
      StartAsyncLookupOnMySecondary(*cur);
      if (cur->IsPending()) {
        assert(cur->pending_cache == secondary_cache_.get());
        my_pending.push_back(cur);
        // Mark as "to be handled by this caller"
        cur->pending_cache = nullptr;
      }
    }
  }

  // Wait on all lookups on my secondary cache
  {
    std::vector<SecondaryCacheResultHandle*> my_secondary_handles;
    for (AsyncLookupHandle* cur : my_pending) {
      my_secondary_handles.push_back(cur->pending_handle);
    }
    secondary_cache_->WaitAll(std::move(my_secondary_handles));
  }

  // Process results
  for (AsyncLookupHandle* cur : my_pending) {
    std::unique_ptr<SecondaryCacheResultHandle> secondary_handle(
        cur->pending_handle);
    cur->pending_handle = nullptr;
    cur->result_handle = Promote(
        std::move(secondary_handle), cur->key, cur->helper, cur->priority,
        cur->stats, cur->found_dummy_entry, cur->kept_in_sec_cache);
    assert(cur->pending_cache == nullptr);
  }
}

std::string CacheWithSecondaryAdapter::GetPrintableOptions() const {
  std::string str = target_->GetPrintableOptions();
  str.append("  secondary_cache:\n");
  str.append(secondary_cache_->GetPrintableOptions());
  return str;
}

const char* CacheWithSecondaryAdapter::Name() const {
  if (distribute_cache_res_) {
    return kTieredCacheName;
  } else {
    // To the user, at least for now, configure the underlying cache with
    // a secondary cache. So we pretend to be that cache
    return target_->Name();
  }
}

// Update the total cache capacity. If we're distributing cache reservations
// to both primary and secondary, then update the pri_cache_res_reservation
// as well. At the moment, we don't have a good way of handling the case
// where the new capacity < total cache reservations.
void CacheWithSecondaryAdapter::SetCapacity(size_t capacity) {
  size_t sec_capacity = static_cast<size_t>(
      capacity * (distribute_cache_res_ ? sec_cache_res_ratio_ : 0.0));
  size_t old_sec_capacity = 0;

  if (distribute_cache_res_) {
    MutexLock m(&cache_res_mutex_);

    Status s = secondary_cache_->GetCapacity(old_sec_capacity);
    if (!s.ok()) {
      return;
    }
    if (old_sec_capacity > sec_capacity) {
      // We're shrinking the cache. We do things in the following order to
      // avoid a temporary spike in usage over the configured capacity -
      // 1. Lower the secondary cache capacity
      // 2. Credit an equal amount (by decreasing pri_cache_res_) to the
      //    primary cache
      // 3. Decrease the primary cache capacity to the total budget
      s = secondary_cache_->SetCapacity(sec_capacity);
      if (s.ok()) {
        if (placeholder_usage_ > capacity) {
          // Adjust reserved_usage_ down
          reserved_usage_ = capacity & ~(kReservationChunkSize - 1);
        }
        size_t new_sec_reserved =
            static_cast<size_t>(reserved_usage_ * sec_cache_res_ratio_);
        s = pri_cache_res_->UpdateCacheReservation(
            (old_sec_capacity - sec_capacity) -
                (sec_reserved_ - new_sec_reserved),
            /*increase=*/false);
        sec_reserved_ = new_sec_reserved;
        assert(s.ok());
        target_->SetCapacity(capacity);
      }
    } else {
      // We're expanding the cache. Do it in the following order to avoid
      // unnecessary evictions -
      // 1. Increase the primary cache capacity to total budget
      // 2. Reserve additional memory in primary on behalf of secondary (by
      //    increasing pri_cache_res_ reservation)
      // 3. Increase secondary cache capacity
      target_->SetCapacity(capacity);
      s = pri_cache_res_->UpdateCacheReservation(
          sec_capacity - old_sec_capacity,
          /*increase=*/true);
      assert(s.ok());
      s = secondary_cache_->SetCapacity(sec_capacity);
      assert(s.ok());
    }
  } else {
    // No cache reservation distribution. Just set the primary cache capacity.
    target_->SetCapacity(capacity);
  }
}

Status CacheWithSecondaryAdapter::GetSecondaryCacheCapacity(
    size_t& size) const {
  return secondary_cache_->GetCapacity(size);
}

Status CacheWithSecondaryAdapter::GetSecondaryCachePinnedUsage(
    size_t& size) const {
  Status s;
  if (distribute_cache_res_) {
    MutexLock m(&cache_res_mutex_);
    size_t capacity = 0;
    s = secondary_cache_->GetCapacity(capacity);
    if (s.ok()) {
      size = capacity - pri_cache_res_->GetTotalMemoryUsed();
    } else {
      size = 0;
    }
  } else {
    size = 0;
  }
  return s;
}

// Update the secondary/primary allocation ratio (remember, the primary
// capacity is the total memory budget when distribute_cache_res_ is true).
// When the ratio changes, we may accumulate some error in the calculations
// for secondary cache inflate/deflate and pri_cache_res_ reservations.
// This is due to the rounding of the reservation amount.
//
// We rely on the current pri_cache_res_ total memory used to estimate the
// new secondary cache reservation after the ratio change. For this reason,
// once the ratio is lowered to 0.0 (effectively disabling the secondary
// cache and pri_cache_res_ total mem used going down to 0), we cannot
// increase the ratio and re-enable it, We might remove this limitation
// in the future.
Status CacheWithSecondaryAdapter::UpdateCacheReservationRatio(
    double compressed_secondary_ratio) {
  if (!distribute_cache_res_) {
    return Status::NotSupported();
  }

  MutexLock m(&cache_res_mutex_);
  size_t pri_capacity = target_->GetCapacity();
  size_t sec_capacity =
      static_cast<size_t>(pri_capacity * compressed_secondary_ratio);
  size_t old_sec_capacity;
  Status s = secondary_cache_->GetCapacity(old_sec_capacity);
  if (!s.ok()) {
    return s;
  }

  // Calculate the new secondary cache reservation
  // reserved_usage_ will never be > the cache capacity, so we don't
  // have to worry about adjusting it here.
  sec_cache_res_ratio_ = compressed_secondary_ratio;
  size_t new_sec_reserved =
      static_cast<size_t>(reserved_usage_ * sec_cache_res_ratio_);
  if (sec_capacity > old_sec_capacity) {
    // We're increasing the ratio, thus ending up with a larger secondary
    // cache and a smaller usable primary cache capacity. Similar to
    // SetCapacity(), we try to avoid a temporary increase in total usage
    // beyond the configured capacity -
    // 1. A higher secondary cache ratio means it gets a higher share of
    //    cache reservations. So first account for that by deflating the
    //    secondary cache
    // 2. Increase pri_cache_res_ reservation to reflect the new secondary
    //    cache utilization (increase in capacity - increase in share of cache
    //    reservation)
    // 3. Increase secondary cache capacity
    s = secondary_cache_->Deflate(new_sec_reserved - sec_reserved_);
    assert(s.ok());
    s = pri_cache_res_->UpdateCacheReservation(
        (sec_capacity - old_sec_capacity) - (new_sec_reserved - sec_reserved_),
        /*increase=*/true);
    assert(s.ok());
    sec_reserved_ = new_sec_reserved;
    s = secondary_cache_->SetCapacity(sec_capacity);
    assert(s.ok());
  } else {
    // We're shrinking the ratio. Try to avoid unnecessary evictions -
    // 1. Lower the secondary cache capacity
    // 2. Decrease pri_cache_res_ reservation to relect lower secondary
    //    cache utilization (decrease in capacity - decrease in share of cache
    //    reservations)
    // 3. Inflate the secondary cache to give it back the reduction in its
    //    share of cache reservations
    s = secondary_cache_->SetCapacity(sec_capacity);
    if (s.ok()) {
      s = pri_cache_res_->UpdateCacheReservation(
          (old_sec_capacity - sec_capacity) -
              (sec_reserved_ - new_sec_reserved),
          /*increase=*/false);
      assert(s.ok());
      s = secondary_cache_->Inflate(sec_reserved_ - new_sec_reserved);
      assert(s.ok());
      sec_reserved_ = new_sec_reserved;
    }
  }

  return s;
}

Status CacheWithSecondaryAdapter::UpdateAdmissionPolicy(
    TieredAdmissionPolicy adm_policy) {
  adm_policy_ = adm_policy;
  return Status::OK();
}

std::shared_ptr<Cache> NewTieredCache(const TieredCacheOptions& _opts) {
  if (!_opts.cache_opts) {
    return nullptr;
  }

  TieredCacheOptions opts = _opts;
  {
    bool valid_adm_policy = true;

    switch (_opts.adm_policy) {
      case TieredAdmissionPolicy::kAdmPolicyAuto:
        // Select an appropriate default policy
        if (opts.adm_policy == TieredAdmissionPolicy::kAdmPolicyAuto) {
          if (opts.nvm_sec_cache) {
            opts.adm_policy = TieredAdmissionPolicy::kAdmPolicyThreeQueue;
          } else {
            opts.adm_policy = TieredAdmissionPolicy::kAdmPolicyPlaceholder;
          }
        }
        break;
      case TieredAdmissionPolicy::kAdmPolicyPlaceholder:
      case TieredAdmissionPolicy::kAdmPolicyAllowCacheHits:
      case TieredAdmissionPolicy::kAdmPolicyAllowAll:
        if (opts.nvm_sec_cache) {
          valid_adm_policy = false;
        }
        break;
      case TieredAdmissionPolicy::kAdmPolicyThreeQueue:
        if (!opts.nvm_sec_cache) {
          valid_adm_policy = false;
        }
        break;
      default:
        valid_adm_policy = false;
    }
    if (!valid_adm_policy) {
      return nullptr;
    }
  }

  std::shared_ptr<Cache> cache;
  if (opts.cache_type == PrimaryCacheType::kCacheTypeLRU) {
    LRUCacheOptions cache_opts =
        *(static_cast_with_check<LRUCacheOptions, ShardedCacheOptions>(
            opts.cache_opts));
    cache_opts.capacity = opts.total_capacity;
    cache_opts.secondary_cache = nullptr;
    cache = cache_opts.MakeSharedCache();
  } else if (opts.cache_type == PrimaryCacheType::kCacheTypeHCC) {
    HyperClockCacheOptions cache_opts =
        *(static_cast_with_check<HyperClockCacheOptions, ShardedCacheOptions>(
            opts.cache_opts));
    cache_opts.capacity = opts.total_capacity;
    cache_opts.secondary_cache = nullptr;
    cache = cache_opts.MakeSharedCache();
  } else {
    return nullptr;
  }
  std::shared_ptr<SecondaryCache> sec_cache;
  opts.comp_cache_opts.capacity = static_cast<size_t>(
      opts.total_capacity * opts.compressed_secondary_ratio);
  sec_cache = NewCompressedSecondaryCache(opts.comp_cache_opts);

  if (opts.nvm_sec_cache) {
    if (opts.adm_policy == TieredAdmissionPolicy::kAdmPolicyThreeQueue) {
      sec_cache = std::make_shared<TieredSecondaryCache>(
          sec_cache, opts.nvm_sec_cache,
          TieredAdmissionPolicy::kAdmPolicyThreeQueue);
    } else {
      return nullptr;
    }
  }

  return std::make_shared<CacheWithSecondaryAdapter>(
      cache, sec_cache, opts.adm_policy, /*distribute_cache_res=*/true);
}

Status UpdateTieredCache(const std::shared_ptr<Cache>& cache,
                         int64_t total_capacity,
                         double compressed_secondary_ratio,
                         TieredAdmissionPolicy adm_policy) {
  if (!cache || strcmp(cache->Name(), kTieredCacheName)) {
    return Status::InvalidArgument();
  }
  CacheWithSecondaryAdapter* tiered_cache =
      static_cast<CacheWithSecondaryAdapter*>(cache.get());

  Status s;
  if (total_capacity > 0) {
    tiered_cache->SetCapacity(total_capacity);
  }
  if (compressed_secondary_ratio >= 0.0 && compressed_secondary_ratio <= 1.0) {
    s = tiered_cache->UpdateCacheReservationRatio(compressed_secondary_ratio);
  }
  if (adm_policy < TieredAdmissionPolicy::kAdmPolicyMax) {
    s = tiered_cache->UpdateAdmissionPolicy(adm_policy);
  }
  return s;
}
}  // namespace ROCKSDB_NAMESPACE