ceph/src/common/shared_cache.hpp

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #ifndef CEPH_SHAREDCACHE_H
  16 #define CEPH_SHAREDCACHE_H
  17
  18 #include <map>
  19 #include <list>
  20 #ifdef WITH_SEASTAR
  21 #include <boost/smart_ptr/local_shared_ptr.hpp>
  22 #else
  23 #include <memory>
  24 #endif
  25 #include "common/ceph_mutex.h"
  26 #include "common/dout.h"
  27 #include "include/unordered_map.h"
  28
  29 // re-include our assert to clobber the system one; fix dout:
  30 #include "include/ceph_assert.h"
  31
  32 template <class K, class V>
  33 class SharedLRU {
  34   CephContext *cct;
  35 #ifdef WITH_SEASTAR
  36   using VPtr = boost::local_shared_ptr<V>;
  37   using WeakVPtr = boost::weak_ptr<V>;
  38 #else
  39   using VPtr = std::shared_ptr<V>;
  40   using WeakVPtr = std::weak_ptr<V>;
  41 #endif
  42   ceph::mutex lock;
  43   size_t max_size;
  44   ceph::condition_variable cond;
  45   unsigned size;
  46 public:
  47   int waiting;
  48 private:
  49   using C = std::less<K>;
  50   using H = std::hash<K>;
  51   ceph::unordered_map<K, typename std::list<std::pair<K, VPtr> >::iterator, H> contents;
  52   std::list<std::pair<K, VPtr> > lru;
  53
  54   std::map<K, std::pair<WeakVPtr, V*>, C> weak_refs;
  55
  56   void trim_cache(std::list<VPtr> *to_release) {
  57     while (size > max_size) {
  58       to_release->push_back(lru.back().second);
  59       lru_remove(lru.back().first);
  60     }
  61   }
  62
  63   void lru_remove(const K& key) {
  64     auto i = contents.find(key);
  65     if (i == contents.end())
  66       return;
  67     lru.erase(i->second);
  68     --size;
  69     contents.erase(i);
  70   }
  71
  72   void lru_add(const K& key, const VPtr& val, std::list<VPtr> *to_release) {
  73     auto i = contents.find(key);
  74     if (i != contents.end()) {
  75       lru.splice(lru.begin(), lru, i->second);
  76     } else {
  77       ++size;
  78       lru.push_front(make_pair(key, val));
  79       contents[key] = lru.begin();
  80       trim_cache(to_release);
  81     }
  82   }
  83
  84   void remove(const K& key, V *valptr) {
  85     std::lock_guard l{lock};
  86     auto i = weak_refs.find(key);
  87     if (i != weak_refs.end() && i->second.second == valptr) {
  88       weak_refs.erase(i);
  89     }
  90     cond.notify_all();
  91   }
  92
  93   class Cleanup {
  94   public:
  95     SharedLRU<K, V> *cache;
  96     K key;
  97     Cleanup(SharedLRU<K, V> *cache, K key) : cache(cache), key(key) {}
  98     void operator()(V *ptr) {
  99       cache->remove(key, ptr);
 100       delete ptr;
 101     }
 102   };
 103
 104 public:
 105   SharedLRU(CephContext *cct = NULL, size_t max_size = 20)
 106     : cct(cct),
 107       lock{ceph::make_mutex("SharedLRU::lock")},
 108       max_size(max_size),
 109       size(0), waiting(0) {
 110     contents.rehash(max_size);
 111   }
 112
 113   ~SharedLRU() {
 114     contents.clear();
 115     lru.clear();
 116     if (!weak_refs.empty()) {
 117       lderr(cct) << "leaked refs:\n";
 118       dump_weak_refs(*_dout);
 119       *_dout << dendl;
 120       if (cct->_conf.get_val<bool>("debug_asserts_on_shutdown")) {
 121         ceph_assert(weak_refs.empty());
 122       }
 123     }
 124   }
 125
 126   int get_count() {
 127     std::lock_guard locker{lock};
 128     return size;
 129   }
 130
 131   void set_cct(CephContext *c) {
 132     cct = c;
 133   }
 134
 135   void dump_weak_refs() {
 136     lderr(cct) << "leaked refs:\n";
 137     dump_weak_refs(*_dout);
 138     *_dout << dendl;
 139   }
 140
 141   void dump_weak_refs(std::ostream& out) {
 142     for (const auto& [key, ref] : weak_refs) {
 143       out << __func__ << " " << this << " weak_refs: "
 144           << key << " = " << ref.second
 145           << " with " << ref.first.use_count() << " refs"
 146           << std::endl;
 147     }
 148   }
 149
 150   //clear all strong reference from the lru.
 151   void clear() {
 152     while (true) {
 153       VPtr val; // release any ref we have after we drop the lock
 154       std::lock_guard locker{lock};
 155       if (size == 0)
 156         break;
 157
 158       val = lru.back().second;
 159       lru_remove(lru.back().first);
 160     }
 161   }
 162
 163   void clear(const K& key) {
 164     VPtr val; // release any ref we have after we drop the lock
 165     {
 166       std::lock_guard l{lock};
 167       typename map<K, pair<WeakVPtr, V*>, C>::iterator i = weak_refs.find(key);
 168       if (i != weak_refs.end()) {
 169         val = i->second.first.lock();
 170       }
 171       lru_remove(key);
 172     }
 173   }
 174
 175   void purge(const K &key) {
 176     VPtr val; // release any ref we have after we drop the lock
 177     {
 178       std::lock_guard l{lock};
 179       typename map<K, pair<WeakVPtr, V*>, C>::iterator i = weak_refs.find(key);
 180       if (i != weak_refs.end()) {
 181         val = i->second.first.lock();
 182         weak_refs.erase(i);
 183       }
 184       lru_remove(key);
 185     }
 186   }
 187
 188   void set_size(size_t new_size) {
 189     list<VPtr> to_release;
 190     {
 191       std::lock_guard l{lock};
 192       max_size = new_size;
 193       trim_cache(&to_release);
 194     }
 195   }
 196
 197   // Returns K key s.t. key <= k for all currently cached k,v
 198   K cached_key_lower_bound() {
 199     std::lock_guard l{lock};
 200     return weak_refs.begin()->first;
 201   }
 202
 203   VPtr lower_bound(const K& key) {
 204     VPtr val;
 205     list<VPtr> to_release;
 206     {
 207       std::unique_lock l{lock};
 208       ++waiting;
 209       cond.wait(l, [this, &key, &val, &to_release] {
 210         if (weak_refs.empty()) {
 211           return true;
 212         }
 213         auto i = weak_refs.lower_bound(key);
 214         if (i == weak_refs.end()) {
 215           --i;
 216         }
 217         if (val = i->second.first.lock(); val) {
 218           lru_add(i->first, val, &to_release);
 219           return true;
 220         } else {
 221           return false;
 222         }
 223       });
 224       --waiting;
 225     }
 226     return val;
 227   }
 228   bool get_next(const K &key, std::pair<K, VPtr> *next) {
 229     std::pair<K, VPtr> r;
 230     {
 231       std::lock_guard l{lock};
 232       VPtr next_val;
 233       typename std::map<K, std::pair<WeakVPtr, V*>, C>::iterator i = weak_refs.upper_bound(key);
 234
 235       while (i != weak_refs.end() &&
 236              !(next_val = i->second.first.lock()))
 237         ++i;
 238
 239       if (i == weak_refs.end())
 240         return false;
 241
 242       if (next)
 243         r = make_pair(i->first, next_val);
 244     }
 245     if (next)
 246       *next = r;
 247     return true;
 248   }
 249   bool get_next(const K &key, std::pair<K, V> *next) {
 250     std::pair<K, VPtr> r;
 251     bool found = get_next(key, &r);
 252     if (!found || !next)
 253       return found;
 254     next->first = r.first;
 255     ceph_assert(r.second);
 256     next->second = *(r.second);
 257     return found;
 258   }
 259
 260   VPtr lookup(const K& key) {
 261     VPtr val;
 262     std::list<VPtr> to_release;
 263     {
 264       std::unique_lock l{lock};
 265       ++waiting;
 266       cond.wait(l, [this, &key, &val, &to_release] {
 267         if (auto i = weak_refs.find(key); i != weak_refs.end()) {
 268           if (val = i->second.first.lock(); val) {
 269             lru_add(key, val, &to_release);
 270             return true;
 271           } else {
 272             return false;
 273           }
 274         } else {
 275           return true;
 276         }
 277       });
 278       --waiting;
 279     }
 280     return val;
 281   }
 282   VPtr lookup_or_create(const K &key) {
 283     VPtr val;
 284     list<VPtr> to_release;
 285     {
 286       std::unique_lock l{lock};
 287       cond.wait(l, [this, &key, &val] {
 288         if (auto i = weak_refs.find(key); i != weak_refs.end()) {
 289           if (val = i->second.first.lock(); val) {
 290             return true;
 291           } else {
 292             return false;
 293           }
 294         } else {
 295           return true;
 296         }
 297       });
 298       if (!val) {
 299         val = VPtr{new V{}, Cleanup{this, key}};
 300         weak_refs.insert(make_pair(key, make_pair(val, val.get())));
 301       }
 302       lru_add(key, val, &to_release);
 303     }
 304     return val;
 305   }
 306
 307   /**
 308    * empty()
 309    *
 310    * Returns true iff there are no live references left to anything that has been
 311    * in the cache.
 312    */
 313   bool empty() {
 314     std::lock_guard l{lock};
 315     return weak_refs.empty();
 316   }
 317
 318   /***
 319    * Inserts a key if not present, or bumps it to the front of the LRU if
 320    * it is, and then gives you a reference to the value. If the key already
 321    * existed, you are responsible for deleting the new value you tried to
 322    * insert.
 323    *
 324    * @param key The key to insert
 325    * @param value The value that goes with the key
 326    * @param existed Set to true if the value was already in the
 327    * map, false otherwise
 328    * @return A reference to the map's value for the given key
 329    */
 330   VPtr add(const K& key, V *value, bool *existed = NULL) {
 331     VPtr val;
 332     list<VPtr> to_release;
 333     {
 334       typename map<K, pair<WeakVPtr, V*>, C>::iterator actual;
 335       std::unique_lock l{lock};
 336       cond.wait(l, [this, &key, &actual, &val] {
 337           actual = weak_refs.lower_bound(key);
 338           if (actual != weak_refs.end() && actual->first == key) {
 339             val = actual->second.first.lock();
 340             if (val) {
 341               return true;
 342             } else {
 343               return false;
 344             }
 345           } else {
 346             return true;
 347           }
 348       });
 349
 350       if (val) {
 351         if (existed) {
 352           *existed = true;
 353         }
 354       } else {
 355         if (existed) {
 356           *existed = false;
 357         }
 358         val = VPtr(value, Cleanup(this, key));
 359         weak_refs.insert(actual, make_pair(key, make_pair(val, value)));
 360       }
 361       lru_add(key, val, &to_release);
 362     }
 363     return val;
 364   }
 365
 366   friend class SharedLRUTest;
 367 };
 368
 369 #endif