// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
	// vim: ts=8 sw=2 smarttab
	
	#include "common/debug.h"
	#include "SimplePolicy.h"
	
	#define dout_context g_ceph_context
	#define dout_subsys ceph_subsys_immutable_obj_cache
	#undef dout_prefix
	#define dout_prefix *_dout << "ceph::cache::SimplePolicy: " << this << " " \
	                           << __func__ << ": "
	
	namespace ceph {
	namespace immutable_obj_cache {
	
	SimplePolicy::SimplePolicy(CephContext *cct, uint64_t cache_size,
	                           uint64_t max_inflight, double watermark)
	  : cct(cct), m_watermark(watermark), m_max_inflight_ops(max_inflight),
	    m_max_cache_size(cache_size) {
	
	  ldout(cct, 20) << "max cache size= " << m_max_cache_size
	                 << " ,watermark= " << m_watermark
	                 << " ,max inflight ops= " << m_max_inflight_ops << dendl;
	
	  m_cache_size = 0;
	
	}
	
	SimplePolicy::~SimplePolicy() {
	  ldout(cct, 20) << dendl;
	
	  for (auto it : m_cache_map) {
	    Entry* entry = (it.second);
	    delete entry;
	  }
	}
	
	cache_status_t SimplePolicy::alloc_entry(std::string file_name) {
	  ldout(cct, 20) << "alloc entry for: " << file_name << dendl;
	
	  std::unique_lock wlocker{m_cache_map_lock};
	
	  // cache hit when promoting
	  if (m_cache_map.find(file_name) != m_cache_map.end()) {
	    ldout(cct, 20) << "object is under promoting: " << file_name << dendl;
	    return OBJ_CACHE_SKIP;
	  }
	
	  if ((m_cache_size < m_max_cache_size) &&
	      (inflight_ops < m_max_inflight_ops)) {
	    Entry* entry = new Entry();
	    ceph_assert(entry != nullptr);
	    m_cache_map[file_name] = entry;
	    wlocker.unlock();
	    update_status(file_name, OBJ_CACHE_SKIP);
	    return OBJ_CACHE_NONE;  // start promotion request
	  }
	
	  // if there's no free entry, return skip to read from rados
	  return OBJ_CACHE_SKIP;
	}
	
	cache_status_t SimplePolicy::lookup_object(std::string file_name) {
	  ldout(cct, 20) << "lookup: " << file_name << dendl;
	
	  std::shared_lock rlocker{m_cache_map_lock};
	
	  auto entry_it = m_cache_map.find(file_name);
	  // simply promote on first lookup
	  if (entry_it == m_cache_map.end()) {
	      rlocker.unlock();
	      return alloc_entry(file_name);
	  }
	
	  Entry* entry = entry_it->second;
	
	  if (entry->status == OBJ_CACHE_PROMOTED) {
	    // bump pos in lru on hit
	    m_promoted_lru.lru_touch(entry);
	  }
	
	  return entry->status;
	}
	
	void SimplePolicy::update_status(std::string file_name,
	                                 cache_status_t new_status, uint64_t size) {
	  ldout(cct, 20) << "update status for: " << file_name
	                 << " new status = " << new_status << dendl;
	
	  std::unique_lock locker{m_cache_map_lock};
	
	  auto entry_it = m_cache_map.find(file_name);
	  if (entry_it == m_cache_map.end()) {
	    return;
	  }
	
	  ceph_assert(entry_it != m_cache_map.end());
	  Entry* entry = entry_it->second;
	
	  // to promote
	  if (entry->status == OBJ_CACHE_NONE && new_status== OBJ_CACHE_SKIP) {
	    entry->status = new_status;
	    entry->file_name = file_name;
	    inflight_ops++;
	    return;
	  }
	
	  // promoting done
	  if (entry->status == OBJ_CACHE_SKIP && new_status== OBJ_CACHE_PROMOTED) {
	    m_promoted_lru.lru_insert_top(entry);
	    entry->status = new_status;
	    entry->size = size;
	    m_cache_size += entry->size;
	    inflight_ops--;
	    return;
	  }
	
	  // promoting failed
	  if (entry->status == OBJ_CACHE_SKIP && new_status== OBJ_CACHE_NONE) {
	    // mark this entry as free
	    entry->file_name = "";
	    entry->status = new_status;
	
	    m_cache_map.erase(entry_it);
	    inflight_ops--;
	    delete entry;
	    return;
	  }
	
	  // to evict
	  if (entry->status == OBJ_CACHE_PROMOTED && new_status== OBJ_CACHE_NONE) {
	    // mark this entry as free

	    uint64_t size = entry->size;
	    entry->file_name = "";
	    entry->size = 0;
	    entry->status = new_status;
	
	    m_promoted_lru.lru_remove(entry);
	    m_cache_map.erase(entry_it);
	    m_cache_size -= size;
	    delete entry;
	    return;
	  }
	}
	
	int SimplePolicy::evict_entry(std::string file_name) {
	  ldout(cct, 20) << "to evict: " << file_name << dendl;
	
	  update_status(file_name, OBJ_CACHE_NONE);
	
	  return 0;
	}
	
	cache_status_t SimplePolicy::get_status(std::string file_name) {
	  ldout(cct, 20) << file_name << dendl;
	
	  std::shared_lock locker{m_cache_map_lock};
	  auto entry_it = m_cache_map.find(file_name);
	  if (entry_it == m_cache_map.end()) {
	    return OBJ_CACHE_NONE;
	  }
	
	  return entry_it->second->status;
	}
	
	void SimplePolicy::get_evict_list(std::list<std::string>* obj_list) {
	  ldout(cct, 20) << dendl;
	
	  std::unique_lock locker{m_cache_map_lock};
	  // check free ratio, pop entries from LRU
	  if ((double)m_cache_size / m_max_cache_size > (1 - m_watermark)) {
	    // TODO(dehao): make this configurable
	    int evict_num = m_cache_map.size() * 0.1;
	    for (int i = 0; i < evict_num; i++) {
	      Entry* entry = reinterpret_cast<Entry*>(m_promoted_lru.lru_expire());
	      if (entry == nullptr) {
	        continue;
	      }
	      std::string file_name = entry->file_name;
	      obj_list->push_back(file_name);
	    }
	  }
	}
	
	// for unit test
	uint64_t SimplePolicy::get_free_size() {
	  return m_max_cache_size - m_cache_size;
	}
	
	uint64_t SimplePolicy::get_promoting_entry_num() {
	  uint64_t index = 0;
	  std::shared_lock rlocker{m_cache_map_lock};
	  for (auto it : m_cache_map) {
	    if (it.second->status == OBJ_CACHE_SKIP) {
	      index++;
	    }
	  }
	  return index;
	}
	
	uint64_t SimplePolicy::get_promoted_entry_num() {
	  return m_promoted_lru.lru_get_size();
	}
	
	std::string SimplePolicy::get_evict_entry() {
	  Entry* entry = reinterpret_cast<Entry*>(m_promoted_lru.lru_get_next_expire());
	  if (entry == nullptr) {
	    return "";
	  }
	  return entry->file_name;
	}
	
	}  // namespace immutable_obj_cache
	}  // namespace ceph