posixdriver: fix cache fill of versioned buckets

This change completes the original intent (hypothesized) to
conditionally set the FLAG_CURRENT bit on just the current
entries during bucket listing cache fill.

This avoids interning 2 copies of the current version of each
object in the listing cache, and also correctly sets the
FLAG_CURRENT bit as required--so the current versions are correctly
reported in versioned listings.

Janky logic to find the current version by explicitly chasing
the symlink target and saving it outside the enumeration scope
has been replaced with proper call to stat() provided by Dang.

Symlink::fill_cache() is no longer used, so removed.

Signed-off-by: Matt Benjamin <mbenjamin@redhat.com>

diff --git a/src/rgw/driver/posix/rgw_sal_posix.cc b/src/rgw/driver/posix/rgw_sal_posix.cc
index 3f5a8f95af5bba7200028122d78b9804f696a646..a4e5cf156624785758d407dad621aec0362209f6 100644 (file)
--- a/src/rgw/driver/posix/rgw_sal_posix.cc
+++ b/src/rgw/driver/posix/rgw_sal_posix.cc
@@ -18,6 +18,7 @@
 #include <sys/stat.h>
 #include <sys/xattr.h>
 #include <unistd.h>
+#include <cstdint>
 #include "rgw_multi.h"
 #include "include/scope_guard.h"
 #include "common/Clock.h" // for ceph_clock_now()
@@ -145,7 +146,7 @@ static bool decode_attr(Attrs &attrs, const char *name, F &f) {
 
 static inline rgw_obj_key decode_obj_key(const char* fname)
 {
-  std::string dname, oname, ns;
+  std::string dname, oname, ns; // XXX ns is unused?
   dname = url_decode(fname);
   rgw_obj_key key;
   rgw_obj_key::parse_raw_oid(dname, &key);
@@ -433,7 +434,7 @@ int FSEnt::read_attrs(const DoutPrefixProvider* dpp, optional_yield y, Attrs& at
   return get_x_attrs(y, dpp, get_fd(), attrs, get_name());
 }
 
-int FSEnt::fill_cache(const DoutPrefixProvider *dpp, optional_yield y, fill_cache_cb_t& cb)
+int FSEnt::fill_cache(const DoutPrefixProvider *dpp, optional_yield y, fill_cache_cb_t& cb, uint32_t flags)
 {
   rgw_bucket_dir_entry bde{};
 
@@ -449,7 +450,7 @@ int FSEnt::fill_cache(const DoutPrefixProvider *dpp, optional_yield y, fill_cach
     case ObjectType::VERSIONED:
       bde.flags = rgw_bucket_dir_entry::FLAG_VER;
       bde.exists = true;
-      if (!key.have_instance()) {
+      if (flags & FSEnt::FLAG_CURRENT) {
          bde.flags |= rgw_bucket_dir_entry::FLAG_CURRENT;
       }
       break;
@@ -1081,7 +1082,7 @@ int Directory::get_ent(const DoutPrefixProvider *dpp, optional_yield y, const st
 }
 
 int Directory::fill_cache(const DoutPrefixProvider *dpp, optional_yield y,
-                          fill_cache_cb_t &cb)
+                          fill_cache_cb_t &cb, uint32_t flags)
 {
   int ret = for_each(dpp, [this, &cb, &dpp, &y](const char *name) {
     std::unique_ptr<FSEnt> ent;
@@ -1097,7 +1098,7 @@ int Directory::fill_cache(const DoutPrefixProvider *dpp, optional_yield y,
 
     ent->stat(dpp); // Stat the object to get the type
 
-    ret = ent->fill_cache(dpp, y, cb);
+    ret = ent->fill_cache(dpp, y, cb, FSEnt::FLAG_NONE);
     if (ret < 0)
       return ret;
     return 0;
@@ -1182,54 +1183,6 @@ int Symlink::stat(const DoutPrefixProvider* dpp, bool force)
   return fill_target(dpp, parent, get_name(), std::string(), target, ctx);
 }
 
-int Symlink::fill_cache(const DoutPrefixProvider *dpp, optional_yield y, fill_cache_cb_t& cb)
-{
-  rgw_bucket_dir_entry bde{};
-  int ret;
-
-  rgw_obj_key key = decode_obj_key(get_name());
-  key.get_index_key(&bde.key);
-  bde.ver.pool = 1;
-  bde.ver.epoch = 1;
-
-  bde.flags = rgw_bucket_dir_entry::FLAG_VER;
-  bde.exists = true;
-  bde.flags |= rgw_bucket_dir_entry::FLAG_CURRENT;
-
-  if (!target) {
-    ret = stat(dpp, /*force=*/false);
-    if (ret < 0)
-      return ret;
-  }
-
-  Attrs attrs;
-  ret = target->read_attrs(dpp, y, attrs);
-  if (ret < 0)
-    return ret;
-
-  POSIXOwner o;
-  ret = decode_owner(attrs, o);
-  if (ret < 0) {
-    bde.meta.owner = "unknown";
-    bde.meta.owner_display_name = "unknown";
-  } else {
-    bde.meta.owner = o.user.to_str();
-    bde.meta.owner_display_name = o.display_name;
-  }
-  bde.meta.category = RGWObjCategory::Main;
-  bde.meta.size = stx.stx_size;
-  bde.meta.accounted_size = stx.stx_size;
-  bde.meta.mtime = from_statx_timestamp(stx.stx_mtime);
-  bde.meta.storage_class = RGW_STORAGE_CLASS_STANDARD;
-  bde.meta.appendable = true;
-  bufferlist etag_bl;
-  if (rgw::sal::get_attr(attrs, RGW_ATTR_ETAG, etag_bl)) {
-    bde.meta.etag = etag_bl.to_str();
-  }
-
-  return cb(dpp, bde);
-}
-
 int Symlink::read_attrs(const DoutPrefixProvider* dpp, optional_yield y, Attrs& attrs)
 {
   if (target)
@@ -1384,13 +1337,13 @@ std::unique_ptr<File> MPDirectory::get_part_file(int partnum)
 }
 
 int MPDirectory::fill_cache(const DoutPrefixProvider *dpp, optional_yield y,
-                          fill_cache_cb_t &cb)
+                            fill_cache_cb_t &cb, uint32_t flags)
 {
-  int ret = FSEnt::fill_cache(dpp, y, cb);
+  int ret = FSEnt::fill_cache(dpp, y, cb, FSEnt::FLAG_NONE);
   if (ret < 0)
     return ret;
 
-  return Directory::fill_cache(dpp, y, cb);
+  return Directory::fill_cache(dpp, y, cb, FSEnt::FLAG_NONE);
 }
 
 int VersionedDirectory::open(const DoutPrefixProvider* dpp)
@@ -1796,8 +1749,11 @@ int VersionedDirectory::remove(const DoutPrefixProvider* dpp, optional_yield y,
 }
 
 int VersionedDirectory::fill_cache(const DoutPrefixProvider *dpp, optional_yield y,
-                          fill_cache_cb_t &cb)
+                                   fill_cache_cb_t &cb, uint32_t flags)
 {
+  /* Fill cur_version */
+  stat(dpp, /*force=*/false);
+
   int ret = for_each(dpp, [this, &cb, &dpp, &y](const char *name) {
     std::unique_ptr<FSEnt> ent;
 
@@ -1812,9 +1768,17 @@ int VersionedDirectory::fill_cache(const DoutPrefixProvider *dpp, optional_yield
 
     ent->stat(dpp); // Stat the object to get the type
 
-    ret = ent->fill_cache(dpp, y, cb);
-    if (ret < 0)
-      return ret;
+    if (ent->get_type() != ObjectType::SYMLINK) {
+      uint32_t fill_flags =
+          (cur_version &&
+           (ent->get_name() == cur_version->get_name())) ?
+        FSEnt::FLAG_CURRENT :
+        FSEnt::FLAG_NONE;
+
+      ret = ent->fill_cache(dpp, y, cb, fill_flags);
+      if (ret < 0)
+        return ret;
+    }
     return 0;
   });
 
@@ -2347,7 +2311,7 @@ std::unique_ptr<Object> POSIXBucket::get_object(const rgw_obj_key& k)
 
 int POSIXObject::fill_cache(const DoutPrefixProvider *dpp, optional_yield y, fill_cache_cb_t& cb)
 {
-  return ent->fill_cache(dpp, y, cb);
+  return ent->fill_cache(dpp, y, cb, FSEnt::FLAG_NONE);
 }
 
 int POSIXDriver::mint_listing_entry(const std::string &bname,
@@ -2414,9 +2378,9 @@ std::unique_ptr<RGWRole> POSIXDriver::get_role(const RGWRoleInfo& info)
 }
 
 int POSIXBucket::fill_cache(const DoutPrefixProvider* dpp, optional_yield y,
-                         fill_cache_cb_t& cb)
+                            fill_cache_cb_t& cb)
 {
-return dir->fill_cache(dpp, y, cb);
+  return dir->fill_cache(dpp, y, cb, FSEnt::FLAG_NONE);
 }
 
 int POSIXBucket::list(const DoutPrefixProvider* dpp, ListParams& params,

diff --git a/src/rgw/driver/posix/rgw_sal_posix.h b/src/rgw/driver/posix/rgw_sal_posix.h
index 1d2e0a0cd2124ac86c8bcbb3a37f4d8c349554ac..357038e963beadf3fd5ab2cc95dbcb6321bdf0ab 100644 (file)
--- a/src/rgw/driver/posix/rgw_sal_posix.h
+++ b/src/rgw/driver/posix/rgw_sal_posix.h
@@ -17,6 +17,7 @@
 
 #include "rgw_sal_filter.h"
 #include "rgw_sal_store.h"
+#include <cstdint>
 #include <memory>
 #include "common/dout.h"
 #include "bucket_cache.h"
@@ -106,6 +107,9 @@ protected:
   CephContext* ctx;
 
 public:
+  static constexpr uint32_t FLAG_NONE =      0x0;
+  static constexpr uint32_t FLAG_CURRENT =   0x2;
+
   FSEnt(std::string _name, Directory* _parent, CephContext* _ctx) : fname(_name), parent(_parent), ctx(_ctx) {}
   FSEnt(std::string _name, Directory* _parent, struct statx& _stx, CephContext* _ctx) : fname(_name), parent(_parent), exist(true), stx(_stx), stat_done(true), ctx(_ctx) {}
   FSEnt(const FSEnt& _e) :
@@ -138,7 +142,7 @@ public:
   virtual int copy(const DoutPrefixProvider *dpp, optional_yield y, Directory* dst_dir, const std::string& name) = 0;
   virtual int link_temp_file(const DoutPrefixProvider* dpp, optional_yield y, std::string target_fname) = 0;
   virtual std::unique_ptr<FSEnt> clone_base() = 0;
-  virtual int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t& cb);
+  virtual int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t& cb, uint32_t flags);
   virtual std::string get_cur_version() { return ""; };
 };
 
@@ -210,7 +214,7 @@ public:
   }
   virtual int copy(const DoutPrefixProvider *dpp, optional_yield y, Directory* dst_dir, const std::string& name) override;
   virtual int link_temp_file(const DoutPrefixProvider* dpp, optional_yield y, std::string target_fname) override;
-  virtual int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t& cb) override;
+  virtual int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t& cb, uint32_t flags) override;
 
   int get_ent(const DoutPrefixProvider *dpp, optional_yield y, const std::string& name, const std::string& version, std::unique_ptr<FSEnt>& ent);
 };
@@ -247,7 +251,6 @@ public:
     return std::make_unique<Symlink>(*this);
   }
   virtual int copy(const DoutPrefixProvider *dpp, optional_yield y, Directory* dst_dir, const std::string& name) override;
-  virtual int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t& cb) override;
 };
 
 class MPDirectory : public Directory {
@@ -283,7 +286,7 @@ public:
   std::unique_ptr<MPDirectory> clone() {
     return std::make_unique<MPDirectory>(*this);
   }
-  virtual int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t& cb) override;
+  virtual int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t& cb, uint32_t flags) override;
 };
 
 class VersionedDirectory : public Directory {
@@ -344,7 +347,7 @@ public:
     return std::make_unique<VersionedDirectory>(*this);
   }
   virtual int copy(const DoutPrefixProvider *dpp, optional_yield y, Directory* dst_dir, const std::string& name) override;
-  virtual int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t& cb) override;
+  virtual int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t& cb, uint32_t flags) override;
 };
 
 std::string get_key_fname(rgw_obj_key& key, bool use_version);