rgw: cache obj version

Also keep bucket objv_tracker on the request state.

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

5 files changed, 97 insertions, 62 deletions

diff --git a/src/rgw/rgw_cache.cc b/src/rgw/rgw_cache.cc
index 97a5e47d0f7..5b96eb45b08 100644
--- a/src/rgw/rgw_cache.cc
+++ b/src/rgw/rgw_cache.cc
@@ -86,6 +86,9 @@ void ObjectCache::put(string& name, ObjectCacheInfo& info)
 
   if (info.flags & CACHE_FLAG_DATA)
     target.data = info.data;
+
+  if (info.flags & CACHE_FLAG_OBJV)
+    target.version = info.version;
 }
 
 void ObjectCache::remove(string& name)
diff --git a/src/rgw/rgw_cache.h b/src/rgw/rgw_cache.h
index d920b3ae1ce..e32a921d1c6 100644
--- a/src/rgw/rgw_cache.h
+++ b/src/rgw/rgw_cache.h
@@ -13,10 +13,11 @@ enum {
   REMOVE_OBJ,
 };
 
-#define CACHE_FLAG_DATA           0x1
-#define CACHE_FLAG_XATTRS         0x2
-#define CACHE_FLAG_META           0x4
-#define CACHE_FLAG_MODIFY_XATTRS  0x8
+#define CACHE_FLAG_DATA           0x01
+#define CACHE_FLAG_XATTRS         0x02
+#define CACHE_FLAG_META           0x04
+#define CACHE_FLAG_MODIFY_XATTRS  0x08
+#define CACHE_FLAG_OBJV           0x10
 
 #define mydout(v) lsubdout(T::cct, rgw, v)
 
@@ -54,11 +55,12 @@ struct ObjectCacheInfo {
   map<string, bufferlist> xattrs;
   map<string, bufferlist> rm_xattrs;
   ObjectMetaInfo meta;
+  obj_version version;
 
   ObjectCacheInfo() : status(0), flags(0), epoch(0) {}
 
   void encode(bufferlist& bl) const {
-    ENCODE_START(4, 3, bl);
+    ENCODE_START(5, 3, bl);
     ::encode(status, bl);
     ::encode(flags, bl);
     ::encode(data, bl);
@@ -66,10 +68,11 @@ struct ObjectCacheInfo {
     ::encode(meta, bl);
     ::encode(rm_xattrs, bl);
     ::encode(epoch, bl);
+    ::encode(version, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator& bl) {
-    DECODE_START_LEGACY_COMPAT_LEN(4, 3, 3, bl);
+    DECODE_START_LEGACY_COMPAT_LEN(5, 3, 3, bl);
     ::decode(status, bl);
     ::decode(flags, bl);
     ::decode(data, bl);
@@ -79,6 +82,8 @@ struct ObjectCacheInfo {
       ::decode(rm_xattrs, bl);
     if (struct_v >= 4)
       ::decode(epoch, bl);
+    if (struct_v >= 5)
+      ::decode(version, bl);
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;
@@ -248,7 +253,12 @@ int RGWCache<T>::get_obj(void *ctx, RGWObjVersionTracker *objv_tracker, void **h
   string name = normal_name(obj.bucket, oid);
 
   ObjectCacheInfo info;
-  if (cache.get(name, info, CACHE_FLAG_DATA) == 0) {
+
+  uint32_t flags = CACHE_FLAG_DATA;
+  if (objv_tracker)
+    flags |= CACHE_FLAG_OBJV;
+  
+  if (cache.get(name, info, flags) == 0) {
     if (info.status < 0)
       return info.status;
 
@@ -259,6 +269,8 @@ int RGWCache<T>::get_obj(void *ctx, RGWObjVersionTracker *objv_tracker, void **h
     obl.clear();
 
     i.copy_all(obl);
+    if (objv_tracker)
+      objv_tracker->read_version = info.version;
     return bl.length();
   }
   int r = T::get_obj(ctx, objv_tracker, handle, obj, obl, ofs, end);
@@ -281,7 +293,10 @@ int RGWCache<T>::get_obj(void *ctx, RGWObjVersionTracker *objv_tracker, void **h
   bufferlist::iterator o = obl.begin();
   o.copy_all(bl);
   info.status = 0;
-  info.flags = CACHE_FLAG_DATA;
+  info.flags = flags;
+  if (objv_tracker) {
+    info.version = objv_tracker->read_version;
+  }
   cache.put(name, info);
   return r;
 }
@@ -299,6 +314,10 @@ int RGWCache<T>::set_attr(void *ctx, rgw_obj& obj, const char *attr_name, buffer
     info.xattrs[attr_name] = bl;
     info.status = 0;
     info.flags = CACHE_FLAG_MODIFY_XATTRS;
+    if (objv_tracker) {
+      info.version = objv_tracker->write_version;
+      info.flags |= CACHE_FLAG_OBJV;
+    }
   }
   int ret = T::set_attr(ctx, obj, attr_name, bl, objv_tracker);
   if (cacheable) {
@@ -334,6 +353,10 @@ int RGWCache<T>::set_attrs(void *ctx, rgw_obj& obj,
       info.rm_xattrs = *rmattrs;
     info.status = 0;
     info.flags = CACHE_FLAG_MODIFY_XATTRS;
+    if (objv_tracker) {
+      info.version = objv_tracker->write_version;
+      info.flags |= CACHE_FLAG_OBJV;
+    }
   }
   int ret = T::set_attrs(ctx, obj, attrs, rmattrs, objv_tracker);
   if (cacheable) {
@@ -372,6 +395,10 @@ int RGWCache<T>::put_obj_meta_impl(void *ctx, rgw_obj& obj, uint64_t size, time_
       info.data = *data;
       info.flags |= CACHE_FLAG_DATA;
     }
+    if (objv_tracker) {
+      info.version = objv_tracker->write_version;
+      info.flags |= CACHE_FLAG_OBJV;
+    }
   }
   int ret = T::put_obj_meta_impl(ctx, obj, size, mtime, attrs, category, flags, rmattrs, data, manifest, ptag, remove_objs,
                                  modify_version, objv_tracker);
@@ -443,7 +470,10 @@ int RGWCache<T>::obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmti
   uint64_t epoch;
 
   ObjectCacheInfo info;
-  int r = cache.get(name, info, CACHE_FLAG_META | CACHE_FLAG_XATTRS);
+  uint32_t flags = CACHE_FLAG_META | CACHE_FLAG_XATTRS;
+  if (objv_tracker)
+    flags |= CACHE_FLAG_OBJV;
+  int r = cache.get(name, info, flags);
   if (r == 0) {
     if (info.status < 0)
       return info.status;
@@ -451,6 +481,8 @@ int RGWCache<T>::obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmti
     size = info.meta.size;
     mtime = info.meta.mtime;
     epoch = info.epoch;
+    if (objv_tracker)
+      objv_tracker->read_version = info.version;
     goto done;
   }
   r = T::obj_stat(ctx, obj, &size, &mtime, &epoch, &info.xattrs, first_chunk, objv_tracker);
@@ -466,6 +498,10 @@ int RGWCache<T>::obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmti
   info.meta.mtime = mtime;
   info.meta.size = size;
   info.flags = CACHE_FLAG_META | CACHE_FLAG_XATTRS;
+  if (objv_tracker) {
+    info.flags |= CACHE_FLAG_OBJV;
+    info.version = objv_tracker->read_version;
+  }
   cache.put(name, info);
 done:
   if (psize)
diff --git a/src/rgw/rgw_common.h b/src/rgw/rgw_common.h
index 67b1074d64a..1d8c32da98c 100644
--- a/src/rgw/rgw_common.h
+++ b/src/rgw/rgw_common.h
@@ -29,6 +29,8 @@
 #include "include/utime.h"
 #include "rgw_acl.h"
 #include "rgw_cors.h"
+#include "cls/version/cls_version_types.h"
+#include "include/rados/librados.hpp"
 
 using namespace std;
 
@@ -602,6 +604,37 @@ struct RGWEnv;
 
 class RGWClientIO;
 
+struct RGWObjVersionTracker {
+  obj_version read_version;
+  obj_version write_version;
+
+  obj_version *version_for_read() {
+    return &read_version;
+  }
+
+  obj_version *version_for_write() {
+    if (write_version.ver == 0)
+      return NULL;
+
+    return &write_version;
+  }
+
+  obj_version *version_for_check() {
+    if (read_version.ver == 0)
+      return NULL;
+
+    return &read_version;
+  }
+
+  void prepare_op_for_read(librados::ObjectReadOperation *op);
+  void prepare_op_for_write(librados::ObjectWriteOperation *op);
+
+  void apply_write() {
+    read_version = write_version;
+    write_version = obj_version();
+  }
+};
+
 /** Store all the state necessary to complete and respond to an HTTP request*/
 struct req_state {
    CephContext *cct;
@@ -639,6 +672,8 @@ struct req_state {
    ACLOwner bucket_owner;
    ACLOwner owner;
 
+   RGWObjVersionTracker objv_tracker;
+
    map<string, string> x_meta_map;
    bool has_bad_meta;
 
diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index d4a38eb775a..799971d1f19 100644
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -187,14 +187,14 @@ static int policy_from_attrset(CephContext *cct, map<string, bufferlist>& attrse
  * object: name of the object to get the ACL for.
  * Returns: 0 on success, -ERR# otherwise.
  */
-static int get_policy_from_attr(CephContext *cct, RGWRados *store, void *ctx, RGWAccessControlPolicy *policy, rgw_obj& obj)
+static int get_policy_from_attr(CephContext *cct, RGWRados *store, void *ctx, RGWAccessControlPolicy *policy, rgw_obj& obj,
+                                RGWObjVersionTracker *objv_tracker)
 {
   bufferlist bl;
   int ret = 0;
 
   if (obj.bucket.name.size()) {
-    RGWObjVersionTracker objv_tracker;
-    ret = store->get_attr(ctx, obj, RGW_ATTR_ACL, bl, &objv_tracker);
+    ret = store->get_attr(ctx, obj, RGW_ATTR_ACL, bl, objv_tracker);
 
     if (ret >= 0) {
       bufferlist::iterator iter = bl.begin();
@@ -214,7 +214,7 @@ static int get_policy_from_attr(CephContext *cct, RGWRados *store, void *ctx, RG
       /* object exists, but policy is broken */
       RGWBucketInfo info;
       RGWUserInfo uinfo;
-      int r = store->get_bucket_info(ctx, obj.bucket.name, info, &objv_tracker);
+      int r = store->get_bucket_info(ctx, obj.bucket.name, info, objv_tracker);
       if (r < 0)
         goto done;
       r = rgw_get_user_info_by_uid(store, info.owner, uinfo);
@@ -258,14 +258,14 @@ static int read_policy(RGWRados *store, struct req_state *s, RGWBucketInfo& buck
   } else {
     obj.init(bucket, oid);
   }
-  int ret = get_policy_from_attr(s->cct, store, s->obj_ctx, policy, obj);
+  int ret = get_policy_from_attr(s->cct, store, s->obj_ctx, policy, obj, &s->objv_tracker);
   if (ret == -ENOENT && object.size()) {
     /* object does not exist checking the bucket's ACL to make sure
        that we send a proper error code */
     RGWAccessControlPolicy bucket_policy(s->cct);
     string no_object;
     rgw_obj no_obj(bucket, no_object);
-    ret = get_policy_from_attr(s->cct, store, s->obj_ctx, &bucket_policy, no_obj);
+    ret = get_policy_from_attr(s->cct, store, s->obj_ctx, &bucket_policy, no_obj, &s->objv_tracker);
     if (ret < 0)
       return ret;
     string& owner = bucket_policy.get_owner().get_id();
@@ -297,8 +297,7 @@ int rgw_build_policies(RGWRados *store, struct req_state *s, bool only_bucket, b
 
   RGWBucketInfo bucket_info;
   if (s->bucket_name_str.size()) {
-    RGWObjVersionTracker objv_tracker;
-    ret = store->get_bucket_info(s->obj_ctx, s->bucket_name_str, bucket_info, &objv_tracker);
+    ret = store->get_bucket_info(s->obj_ctx, s->bucket_name_str, bucket_info, &s->objv_tracker);
     if (ret < 0) {
       ldout(s->cct, 0) << "NOTICE: couldn't get bucket from bucket_name (name=" << s->bucket_name_str << ")" << dendl;
       return ret;
@@ -499,8 +498,7 @@ int RGWGetObj::handle_user_manifest(const char *prefix)
 
   if (bucket_name.compare(s->bucket.name) != 0) {
     RGWBucketInfo bucket_info;
-    RGWObjVersionTracker objv_tracker;
-    int r = store->get_bucket_info(NULL, bucket_name, bucket_info, &objv_tracker);
+    int r = store->get_bucket_info(NULL, bucket_name, bucket_info, &s->objv_tracker);
     if (r < 0) {
       ldout(s->cct, 0) << "could not get bucket info for bucket=" << bucket_name << dendl;
       return r;
@@ -855,7 +853,7 @@ void RGWCreateBucket::execute()
 
   s->bucket_owner.set_id(s->user.user_id);
   s->bucket_owner.set_name(s->user.display_name);
-  r = get_policy_from_attr(s->cct, store, s->obj_ctx, &old_policy, obj);
+  r = get_policy_from_attr(s->cct, store, s->obj_ctx, &old_policy, obj, &s->objv_tracker);
   if (r >= 0)  {
     if (old_policy.get_owner().get_id().compare(s->user.user_id) != 0) {
       ret = -EEXIST;
@@ -885,8 +883,7 @@ void RGWCreateBucket::execute()
      */
     RGWBucketInfo info;
     map<string, bufferlist> attrs;
-    RGWObjVersionTracker objv_tracker;
-    int r = store->get_bucket_info(NULL, s->bucket.name, info, &objv_tracker, &attrs);
+    int r = store->get_bucket_info(NULL, s->bucket.name, info, &s->objv_tracker, &attrs);
     if (r < 0) {
       ldout(s->cct, 0) << "ERROR: get_bucket_info on bucket=" << s->bucket.name << " returned err=" << r << " after create_bucket returned -EEXIST" << dendl;
       ret = r;
@@ -1584,10 +1581,8 @@ void RGWPutMetadata::execute()
 
   rgw_get_request_metadata(s, attrs);
 
-  RGWObjVersionTracker objv_tracker;
-
   /* no need to track object versioning, need it for bucket's data only */
-  RGWObjVersionTracker *ptracker = (s->object ? NULL : &objv_tracker);
+  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->objv_tracker);
 
   /* check if obj exists, read orig attrs */
   ret = get_obj_attrs(store, s, obj, orig_attrs, NULL, ptracker);
@@ -1893,8 +1888,7 @@ void RGWPutACLs::execute()
     *_dout << dendl;
   }
 
-  RGWObjVersionTracker objv_tracker;
-  RGWObjVersionTracker *ptracker = (s->object ? NULL : &objv_tracker);
+  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->objv_tracker);
 
   new_policy.encode(bl);
   obj.init(s->bucket, s->object_str);
@@ -1972,8 +1966,7 @@ void RGWPutCORS::execute()
     *_dout << dendl;
   }
 
-  RGWObjVersionTracker objv_tracker;
-  RGWObjVersionTracker *ptracker = (s->object ? NULL : &objv_tracker);
+  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->objv_tracker);
 
   string no_obj;
   cors_config->encode(bl);
@@ -2005,8 +1998,7 @@ void RGWDeleteCORS::execute()
   map<string, bufferlist> orig_attrs, attrs, rmattrs;
   map<string, bufferlist>::iterator iter;
 
-  RGWObjVersionTracker objv_tracker;
-  RGWObjVersionTracker *ptracker = (s->object ? NULL : &objv_tracker);
+  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->objv_tracker);
 
   /* check if obj exists, read orig attrs */
   ret = get_obj_attrs(store, s, obj, orig_attrs, NULL, ptracker);
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index 444fdecf1a2..f487a98b7d5 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -42,37 +42,6 @@ static inline void get_obj_bucket_and_oid_key(rgw_obj& obj, rgw_bucket& bucket,
   prepend_bucket_marker(bucket, obj.key, key);
 }
 
-struct RGWObjVersionTracker {
-  obj_version read_version;
-  obj_version write_version;
-
-  obj_version *version_for_read() {
-    return &read_version;
-  }
-
-  obj_version *version_for_write() {
-    if (write_version.ver == 0)
-      return NULL;
-
-    return &write_version;
-  }
-
-  obj_version *version_for_check() {
-    if (read_version.ver == 0)
-      return NULL;
-
-    return &read_version;
-  }
-
-  void prepare_op_for_read(librados::ObjectReadOperation *op);
-  void prepare_op_for_write(librados::ObjectWriteOperation *op);
-
-  void apply_write() {
-    read_version = write_version;
-    write_version = obj_version();
-  }
-};
-
 struct RGWUsageBatch {
   map<utime_t, rgw_usage_log_entry> m;