rgw: chain to multiple cache entries in one call

This ensures that chained cache entries that depend on more than one raw
cache entry (bucket info cache depends on both the bucket entry point
and on the bucket info object), are chained and created atomically.

Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>

src/rgw/rgw_cache.cc

@@ -57,26 +57,45 @@ int ObjectCache::get(string& name, ObjectCacheInfo& info, uint32_t mask, rgw_cac
   return 0;
 }
 
-bool ObjectCache::chain_cache_entry(rgw_cache_entry_info& cache_info, RGWChainedCache::Entry *chained_entry)
+bool ObjectCache::chain_cache_entry(list<rgw_cache_entry_info *>& cache_info_entries, RGWChainedCache::Entry *chained_entry)
 {
   RWLock::WLocker l(lock);
 
-  ldout(cct, 10) << "chain_cache_entry: cache_locator=" << cache_info.cache_locator << dendl;
-  map<string, ObjectCacheEntry>::iterator iter = cache_map.find(cache_info.cache_locator);
-  if (iter == cache_map.end()) {
-    ldout(cct, 20) << "chain_cache_entry: couldn't find cachce locator" << dendl;
-    return false;
+  list<rgw_cache_entry_info *>::iterator citer;
+
+  list<ObjectCacheEntry *> cache_entry_list;
+
+  /* first verify that all entries are still valid */
+  for (citer = cache_info_entries.begin(); citer != cache_info_entries.end(); ++citer) {
+    rgw_cache_entry_info *cache_info = *citer;
+
+    ldout(cct, 10) << "chain_cache_entry: cache_locator=" << cache_info->cache_locator << dendl;
+    map<string, ObjectCacheEntry>::iterator iter = cache_map.find(cache_info->cache_locator);
+    if (iter == cache_map.end()) {
+      ldout(cct, 20) << "chain_cache_entry: couldn't find cachce locator" << dendl;
+      return false;
+    }
+
+    ObjectCacheEntry *entry = &iter->second;
+
+    if (entry->gen != cache_info->gen) {
+      ldout(cct, 20) << "chain_cache_entry: entry.gen (" << entry->gen << ") != cache_info.gen (" << cache_info->gen << ")" << dendl;
+      return false;
+    }
+
+    cache_entry_list.push_back(entry);
   }
 
-  ObjectCacheEntry& entry = iter->second;
-  if (entry.gen != cache_info.gen) {
-    ldout(cct, 20) << "chain_cache_entry: entry.gen (" << entry.gen << ") != cache_info.gen (" << cache_info.gen << ")" << dendl;
-    return false;
-  }
 
   chained_entry->cache->chain_cb(chained_entry->key, chained_entry->data);
 
-  entry.chained_entries.push_back(make_pair<RGWChainedCache *, string>(chained_entry->cache, chained_entry->key));
+  list<ObjectCacheEntry *>::iterator liter;
+
+  for (liter = cache_entry_list.begin(); liter != cache_entry_list.end(); ++liter) {
+    ObjectCacheEntry *entry = *liter;
+
+    entry->chained_entries.push_back(make_pair<RGWChainedCache *, string>(chained_entry->cache, chained_entry->key));
+  }
 
   return true;
 }

src/rgw/rgw_cache.h

@@ -154,7 +154,7 @@ public:
     cct = _cct;
     lru_window = cct->_conf->rgw_cache_lru_size / 2;
   }
-  bool chain_cache_entry(rgw_cache_entry_info& cache_info, RGWChainedCache::Entry *chained_entry);
+  bool chain_cache_entry(list<rgw_cache_entry_info *>& cache_info_entries, RGWChainedCache::Entry *chained_entry);
 };
 
 template <class T>
@@ -224,8 +224,8 @@ public:
 
   int delete_obj_impl(void *ctx, const string& bucket_owner, rgw_obj& obj, RGWObjVersionTracker *objv_tracker);
 
-  bool chain_cache_entry(rgw_cache_entry_info& cache_info, RGWChainedCache::Entry *chained_entry) {
-    return cache.chain_cache_entry(cache_info, chained_entry);
+  bool chain_cache_entry(list<rgw_cache_entry_info *>& cache_info_entries, RGWChainedCache::Entry *chained_entry) {
+    return cache.chain_cache_entry(cache_info_entries, chained_entry);
   }
 };
 

src/rgw/rgw_rados.cc

@@ -5522,9 +5522,14 @@ int RGWRados::get_bucket_info(void *ctx, const string& bucket_name, RGWBucketInf
   if (pattrs)
     *pattrs = e.attrs;
 
+  list<rgw_cache_entry_info *> cache_info_entries;
+  cache_info_entries.push_back(&entry_cache_info);
+  cache_info_entries.push_back(&cache_info);
+
+
   /* chain to both bucket entry point and bucket instance */
-  if (binfo_cache.put(this, bucket_name, &e, entry_cache_info)) {
-    binfo_cache.put(this, bucket_name, &e, cache_info);
+  if (!binfo_cache.put(this, bucket_name, &e, cache_info_entries)) {
+    ldout(cct, 20) << "couldn't put binfo cache entry, might have raced with data changes" << dendl;
   }
 
   return 0;

src/rgw/rgw_rados.h

@@ -1667,7 +1667,7 @@ public:
 
   virtual void finish_get_obj(void **handle);
 
-  virtual bool chain_cache_entry(rgw_cache_entry_info& cache_info, RGWChainedCache::Entry *chained_entry) { return false; }
+  virtual bool chain_cache_entry(list<rgw_cache_entry_info *>& cache_info_entries, RGWChainedCache::Entry *chained_entry) { return false; }
 
   int iterate_obj(void *ctx, rgw_obj& obj,
                   off_t ofs, off_t end,
@@ -1957,11 +1957,11 @@ public:
     return true;
   }
 
-  bool put(RGWRados *store, const string& key, T *entry, rgw_cache_entry_info& cache_info) {
+  bool put(RGWRados *store, const string& key, T *entry, list<rgw_cache_entry_info *>& cache_info_entries) {
     Entry chain_entry(this, key, entry);
 
     /* we need the store cache to call us under its lock to maintain lock ordering */
-    return store->chain_cache_entry(cache_info, &chain_entry);
+    return store->chain_cache_entry(cache_info_entries, &chain_entry);
   }
 
   void chain_cb(const string& key, void *data) {

src/rgw/rgw_user.cc

@@ -235,7 +235,10 @@ int rgw_get_user_info_from_index(RGWRados *store, string& key, rgw_bucket& bucke
     return -EIO;
   }
 
-  uinfo_cache.put(store, key, &e, cache_info);
+  list<rgw_cache_entry_info *> cache_info_entries;
+  cache_info_entries.push_back(&cache_info);
+
+  uinfo_cache.put(store, key, &e, cache_info_entries);
 
   info = e.info;
   if (objv_tracker)