1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 #ifndef CEPH_RGWCACHE_H
5 #define CEPH_RGWCACHE_H
10 #include "include/types.h"
11 #include "include/utime.h"
12 #include "include/assert.h"
13 #include "common/RWLock.h"
20 #define CACHE_FLAG_DATA 0x01
21 #define CACHE_FLAG_XATTRS 0x02
22 #define CACHE_FLAG_META 0x04
23 #define CACHE_FLAG_MODIFY_XATTRS 0x08
24 #define CACHE_FLAG_OBJV 0x10
26 #define mydout(v) lsubdout(T::cct, rgw, v)
28 struct ObjectMetaInfo {
32 ObjectMetaInfo() : size(0) {}
34 void encode(bufferlist& bl) const {
35 ENCODE_START(2, 2, bl);
40 void decode(bufferlist::iterator& bl) {
41 DECODE_START_LEGACY_COMPAT_LEN(2, 2, 2, bl);
46 void dump(Formatter *f) const;
47 static void generate_test_instances(list<ObjectMetaInfo*>& o);
49 WRITE_CLASS_ENCODER(ObjectMetaInfo)
51 struct ObjectCacheInfo {
56 map<string, bufferlist> xattrs;
57 map<string, bufferlist> rm_xattrs;
61 ObjectCacheInfo() : status(0), flags(0), epoch(0), version() {}
63 void encode(bufferlist& bl) const {
64 ENCODE_START(5, 3, bl);
70 ::encode(rm_xattrs, bl);
72 ::encode(version, bl);
75 void decode(bufferlist::iterator& bl) {
76 DECODE_START_LEGACY_COMPAT_LEN(5, 3, 3, bl);
83 ::decode(rm_xattrs, bl);
87 ::decode(version, bl);
90 void dump(Formatter *f) const;
91 static void generate_test_instances(list<ObjectCacheInfo*>& o);
93 WRITE_CLASS_ENCODER(ObjectCacheInfo)
95 struct RGWCacheNotifyInfo {
98 ObjectCacheInfo obj_info;
102 RGWCacheNotifyInfo() : op(0), ofs(0) {}
104 void encode(bufferlist& obl) const {
105 ENCODE_START(2, 2, obl);
108 ::encode(obj_info, obl);
113 void decode(bufferlist::iterator& ibl) {
114 DECODE_START_LEGACY_COMPAT_LEN(2, 2, 2, ibl);
117 ::decode(obj_info, ibl);
122 void dump(Formatter *f) const;
123 static void generate_test_instances(list<RGWCacheNotifyInfo*>& o);
125 WRITE_CLASS_ENCODER(RGWCacheNotifyInfo)
127 struct ObjectCacheEntry {
128 ObjectCacheInfo info;
129 std::list<string>::iterator lru_iter;
130 uint64_t lru_promotion_ts;
132 std::list<pair<RGWChainedCache *, string> > chained_entries;
134 ObjectCacheEntry() : lru_promotion_ts(0), gen(0) {}
138 std::map<string, ObjectCacheEntry> cache_map;
139 std::list<string> lru;
140 unsigned long lru_size;
141 unsigned long lru_counter;
142 unsigned long lru_window;
146 list<RGWChainedCache *> chained_cache;
150 void touch_lru(string& name, ObjectCacheEntry& entry, std::list<string>::iterator& lru_iter);
151 void remove_lru(string& name, std::list<string>::iterator& lru_iter);
153 void do_invalidate_all();
155 ObjectCache() : lru_size(0), lru_counter(0), lru_window(0), lock("ObjectCache"), cct(NULL), enabled(false) { }
156 int get(std::string& name, ObjectCacheInfo& bl, uint32_t mask, rgw_cache_entry_info *cache_info);
157 void put(std::string& name, ObjectCacheInfo& bl, rgw_cache_entry_info *cache_info);
158 void remove(std::string& name);
159 void set_ctx(CephContext *_cct) {
161 lru_window = cct->_conf->rgw_cache_lru_size / 2;
163 bool chain_cache_entry(list<rgw_cache_entry_info *>& cache_info_entries, RGWChainedCache::Entry *chained_entry);
165 void set_enabled(bool status);
167 void chain_cache(RGWChainedCache *cache);
168 void invalidate_all();
172 class RGWCache : public T
176 int list_objects_raw_init(rgw_pool& pool, RGWAccessHandle *handle) {
177 return T::list_objects_raw_init(pool, handle);
179 int list_objects_raw_next(rgw_bucket_dir_entry& obj, RGWAccessHandle *handle) {
180 return T::list_objects_raw_next(obj, handle);
183 string normal_name(rgw_pool& pool, const std::string& oid) {
185 buf.reserve(pool.name.size() + pool.ns.size() + oid.size() + 2);
186 buf.append(pool.name).append("+").append(pool.ns).append("+").append(oid);
190 void normalize_pool_and_obj(rgw_pool& src_pool, const string& src_obj, rgw_pool& dst_pool, string& dst_obj);
191 string normal_name(rgw_raw_obj& obj) {
192 return normal_name(obj.pool, obj.oid);
195 int init_rados() override {
197 cache.set_ctx(T::cct);
198 ret = T::init_rados();
205 bool need_watch_notify() override {
209 int distribute_cache(const string& normal_name, rgw_raw_obj& obj, ObjectCacheInfo& obj_info, int op);
210 int watch_cb(uint64_t notify_id,
212 uint64_t notifier_id,
213 bufferlist& bl) override;
215 void set_cache_enabled(bool state) override {
216 cache.set_enabled(state);
221 void register_chained_cache(RGWChainedCache *cc) override {
222 cache.chain_cache(cc);
225 int system_obj_set_attrs(void *ctx, rgw_raw_obj& obj,
226 map<string, bufferlist>& attrs,
227 map<string, bufferlist>* rmattrs,
228 RGWObjVersionTracker *objv_tracker);
229 int put_system_obj_impl(rgw_raw_obj& obj, uint64_t size, real_time *mtime,
230 map<std::string, bufferlist>& attrs, int flags,
232 RGWObjVersionTracker *objv_tracker,
233 real_time set_mtime) override;
234 int put_system_obj_data(void *ctx, rgw_raw_obj& obj, bufferlist& bl, off_t ofs, bool exclusive,
235 RGWObjVersionTracker *objv_tracker = nullptr) override;
237 int get_system_obj(RGWObjectCtx& obj_ctx, RGWRados::SystemObject::Read::GetObjState& read_state,
238 RGWObjVersionTracker *objv_tracker, rgw_raw_obj& obj,
239 bufferlist& bl, off_t ofs, off_t end,
240 map<string, bufferlist> *attrs,
241 rgw_cache_entry_info *cache_info) override;
243 int raw_obj_stat(rgw_raw_obj& obj, uint64_t *psize, real_time *pmtime, uint64_t *epoch, map<string, bufferlist> *attrs,
244 bufferlist *first_chunk, RGWObjVersionTracker *objv_tracker) override;
246 int delete_system_obj(rgw_raw_obj& obj, RGWObjVersionTracker *objv_tracker) override;
248 bool chain_cache_entry(list<rgw_cache_entry_info *>& cache_info_entries, RGWChainedCache::Entry *chained_entry) override {
249 return cache.chain_cache_entry(cache_info_entries, chained_entry);
254 void RGWCache<T>::normalize_pool_and_obj(rgw_pool& src_pool, const string& src_obj, rgw_pool& dst_pool, string& dst_obj)
256 if (src_obj.size()) {
260 dst_pool = T::get_zone_params().domain_root;
261 dst_obj = src_pool.name;
266 int RGWCache<T>::delete_system_obj(rgw_raw_obj& obj, RGWObjVersionTracker *objv_tracker)
270 normalize_pool_and_obj(obj.pool, obj.oid, pool, oid);
272 string name = normal_name(obj);
275 ObjectCacheInfo info;
276 distribute_cache(name, obj, info, REMOVE_OBJ);
278 return T::delete_system_obj(obj, objv_tracker);
282 int RGWCache<T>::get_system_obj(RGWObjectCtx& obj_ctx, RGWRados::SystemObject::Read::GetObjState& read_state,
283 RGWObjVersionTracker *objv_tracker, rgw_raw_obj& obj,
284 bufferlist& obl, off_t ofs, off_t end,
285 map<string, bufferlist> *attrs,
286 rgw_cache_entry_info *cache_info)
290 normalize_pool_and_obj(obj.pool, obj.oid, pool, oid);
292 return T::get_system_obj(obj_ctx, read_state, objv_tracker, obj, obl, ofs, end, attrs, cache_info);
294 string name = normal_name(obj.pool, oid);
296 ObjectCacheInfo info;
298 uint32_t flags = CACHE_FLAG_DATA;
300 flags |= CACHE_FLAG_OBJV;
302 flags |= CACHE_FLAG_XATTRS;
304 if (cache.get(name, info, flags, cache_info) == 0) {
308 bufferlist& bl = info.data;
310 bufferlist::iterator i = bl.begin();
316 objv_tracker->read_version = info.version;
318 *attrs = info.xattrs;
321 int r = T::get_system_obj(obj_ctx, read_state, objv_tracker, obj, obl, ofs, end, attrs, cache_info);
323 if (r == -ENOENT) { // only update ENOENT, we'd rather retry other errors
325 cache.put(name, info, cache_info);
330 if (obl.length() == end + 1) {
331 /* in this case, most likely object contains more data, we can't cache it */
336 bufferlist& bl = info.data;
338 bufferlist::iterator o = obl.begin();
343 info.version = objv_tracker->read_version;
346 info.xattrs = *attrs;
348 cache.put(name, info, cache_info);
353 int RGWCache<T>::system_obj_set_attrs(void *ctx, rgw_raw_obj& obj,
354 map<string, bufferlist>& attrs,
355 map<string, bufferlist>* rmattrs,
356 RGWObjVersionTracker *objv_tracker)
360 normalize_pool_and_obj(obj.pool, obj.oid, pool, oid);
361 ObjectCacheInfo info;
364 info.rm_xattrs = *rmattrs;
366 info.flags = CACHE_FLAG_MODIFY_XATTRS;
368 info.version = objv_tracker->write_version;
369 info.flags |= CACHE_FLAG_OBJV;
371 int ret = T::system_obj_set_attrs(ctx, obj, attrs, rmattrs, objv_tracker);
372 string name = normal_name(pool, oid);
374 cache.put(name, info, NULL);
375 int r = distribute_cache(name, obj, info, UPDATE_OBJ);
377 mydout(0) << "ERROR: failed to distribute cache for " << obj << dendl;
386 int RGWCache<T>::put_system_obj_impl(rgw_raw_obj& obj, uint64_t size, real_time *mtime,
387 map<std::string, bufferlist>& attrs, int flags,
389 RGWObjVersionTracker *objv_tracker,
394 normalize_pool_and_obj(obj.pool, obj.oid, pool, oid);
395 ObjectCacheInfo info;
399 info.flags = CACHE_FLAG_XATTRS | CACHE_FLAG_DATA | CACHE_FLAG_META;
401 info.version = objv_tracker->write_version;
402 info.flags |= CACHE_FLAG_OBJV;
404 ceph::real_time result_mtime;
405 int ret = T::put_system_obj_impl(obj, size, &result_mtime, attrs, flags, data,
406 objv_tracker, set_mtime);
408 *mtime = result_mtime;
410 info.meta.mtime = result_mtime;
411 info.meta.size = size;
412 string name = normal_name(pool, oid);
414 cache.put(name, info, NULL);
415 int r = distribute_cache(name, obj, info, UPDATE_OBJ);
417 mydout(0) << "ERROR: failed to distribute cache for " << obj << dendl;
426 int RGWCache<T>::put_system_obj_data(void *ctx, rgw_raw_obj& obj, bufferlist& data, off_t ofs, bool exclusive,
427 RGWObjVersionTracker *objv_tracker)
431 normalize_pool_and_obj(obj.pool, obj.oid, pool, oid);
432 ObjectCacheInfo info;
433 bool cacheable = false;
434 if ((ofs == 0) || (ofs == -1)) {
437 info.meta.size = data.length();
439 info.flags = CACHE_FLAG_DATA;
442 info.version = objv_tracker->write_version;
443 info.flags |= CACHE_FLAG_OBJV;
445 int ret = T::put_system_obj_data(ctx, obj, data, ofs, exclusive, objv_tracker);
447 string name = normal_name(pool, oid);
449 cache.put(name, info, NULL);
450 int r = distribute_cache(name, obj, info, UPDATE_OBJ);
452 mydout(0) << "ERROR: failed to distribute cache for " << obj << dendl;
462 int RGWCache<T>::raw_obj_stat(rgw_raw_obj& obj, uint64_t *psize, real_time *pmtime,
463 uint64_t *pepoch, map<string, bufferlist> *attrs,
464 bufferlist *first_chunk, RGWObjVersionTracker *objv_tracker)
468 normalize_pool_and_obj(obj.pool, obj.oid, pool, oid);
470 string name = normal_name(pool, oid);
476 ObjectCacheInfo info;
477 uint32_t flags = CACHE_FLAG_META | CACHE_FLAG_XATTRS;
479 flags |= CACHE_FLAG_OBJV;
480 int r = cache.get(name, info, flags, NULL);
485 size = info.meta.size;
486 mtime = info.meta.mtime;
489 objv_tracker->read_version = info.version;
492 r = T::raw_obj_stat(obj, &size, &mtime, &epoch, &info.xattrs, first_chunk, objv_tracker);
496 cache.put(name, info, NULL);
502 info.meta.mtime = mtime;
503 info.meta.size = size;
504 info.flags = CACHE_FLAG_META | CACHE_FLAG_XATTRS;
506 info.flags |= CACHE_FLAG_OBJV;
507 info.version = objv_tracker->read_version;
509 cache.put(name, info, NULL);
518 *attrs = info.xattrs;
523 int RGWCache<T>::distribute_cache(const string& normal_name, rgw_raw_obj& obj, ObjectCacheInfo& obj_info, int op)
525 RGWCacheNotifyInfo info;
529 info.obj_info = obj_info;
533 return T::distribute(normal_name, bl);
537 int RGWCache<T>::watch_cb(uint64_t notify_id,
539 uint64_t notifier_id,
542 RGWCacheNotifyInfo info;
545 bufferlist::iterator iter = bl.begin();
546 ::decode(info, iter);
547 } catch (buffer::end_of_buffer& err) {
548 mydout(0) << "ERROR: got bad notification" << dendl;
550 } catch (buffer::error& err) {
551 mydout(0) << "ERROR: buffer::error" << dendl;
557 normalize_pool_and_obj(info.obj.pool, info.obj.oid, pool, oid);
558 string name = normal_name(pool, oid);
562 cache.put(name, info.obj_info, NULL);
568 mydout(0) << "WARNING: got unknown notification op: " << info.op << dendl;