Merge pull request ceph#60670 from cbodley/wip-cls-rgw-shard-io

rgw/rados: add async algorithms for concurrent index operations

Reviewed-by: Adam Emerson <[email protected]>

Author: cbodley

src/cls/rgw/cls_rgw_client.cc

@@ -266,98 +266,8 @@ class BucketIndexShardsManager {
 void cls_rgw_bucket_init_index(librados::ObjectWriteOperation& o);
 void cls_rgw_bucket_init_index2(librados::ObjectWriteOperation& o);
 
-class CLSRGWConcurrentIO {
-protected:
-  librados::IoCtx& io_ctx;
-
-  // map of shard # to oid; the shards that are remaining to be processed
-  std::map<int, std::string>& objs_container;
-  // iterator to work through objs_container
-  std::map<int, std::string>::iterator iter;
-
-  uint32_t max_aio;
-  BucketIndexAioManager manager;
-
-  virtual int issue_op(int shard_id, const std::string& oid) = 0;
-
-  virtual void cleanup() {}
-  virtual int valid_ret_code() { return 0; }
-  // Return true if multiple rounds of OPs might be needed, this happens when
-  // OP needs to be re-send until a certain code is returned.
-  virtual bool need_multiple_rounds() { return false; }
-  // Add a new object to the end of the container.
-  virtual void add_object(int shard, const std::string& oid) {}
-  virtual void reset_container(std::map<int, std::string>& objs) {}
-
-public:
-
-  CLSRGWConcurrentIO(librados::IoCtx& ioc,
-		     std::map<int, std::string>& _objs_container,
-		     uint32_t _max_aio) :
-  io_ctx(ioc), objs_container(_objs_container), max_aio(_max_aio)
-  {}
-
-  virtual ~CLSRGWConcurrentIO() {}
-
-  int operator()();
-}; // class CLSRGWConcurrentIO
-
-
-class CLSRGWIssueBucketIndexInit : public CLSRGWConcurrentIO {
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-  int valid_ret_code() override { return -EEXIST; }
-  void cleanup() override;
-public:
-  CLSRGWIssueBucketIndexInit(librados::IoCtx& ioc,
-			     std::map<int, std::string>& _bucket_objs,
-			     uint32_t _max_aio) :
-    CLSRGWConcurrentIO(ioc, _bucket_objs, _max_aio) {}
-  virtual ~CLSRGWIssueBucketIndexInit() override {}
-};
-
-
-class CLSRGWIssueBucketIndexInit2 : public CLSRGWConcurrentIO {
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-  int valid_ret_code() override { return -EEXIST; }
-  void cleanup() override;
-public:
-  CLSRGWIssueBucketIndexInit2(librados::IoCtx& ioc,
-			     std::map<int, std::string>& _bucket_objs,
-			     uint32_t _max_aio) :
-    CLSRGWConcurrentIO(ioc, _bucket_objs, _max_aio) {}
-  virtual ~CLSRGWIssueBucketIndexInit2() override {}
-};
-
-
-class CLSRGWIssueBucketIndexClean : public CLSRGWConcurrentIO {
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-  int valid_ret_code() override {
-    return -ENOENT;
-  }
-
-public:
-  CLSRGWIssueBucketIndexClean(librados::IoCtx& ioc,
-			      std::map<int, std::string>& _bucket_objs,
-			      uint32_t _max_aio) :
-  CLSRGWConcurrentIO(ioc, _bucket_objs, _max_aio)
-  {}
-  virtual ~CLSRGWIssueBucketIndexClean() override {}
-};
-
-
-class CLSRGWIssueSetTagTimeout : public CLSRGWConcurrentIO {
-  uint64_t tag_timeout;
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-public:
-  CLSRGWIssueSetTagTimeout(librados::IoCtx& ioc, std::map<int, std::string>& _bucket_objs,
-                     uint32_t _max_aio, uint64_t _tag_timeout) :
-    CLSRGWConcurrentIO(ioc, _bucket_objs, _max_aio), tag_timeout(_tag_timeout) {}
-  virtual ~CLSRGWIssueSetTagTimeout() override {}
-};
+void cls_rgw_bucket_set_tag_timeout(librados::ObjectWriteOperation& op,
+                                    uint64_t timeout);
 
 void cls_rgw_bucket_update_stats(librados::ObjectWriteOperation& o,
                                  bool absolute,
@@ -424,54 +334,6 @@ int cls_rgw_usage_log_trim(librados::IoCtx& io_ctx, const std::string& oid, cons
                            uint64_t start_epoch, uint64_t end_epoch);
 #endif
 
-
-/**
- * Std::list the bucket with the starting object and filter prefix.
- * NOTE: this method do listing requests for each bucket index shards identified by
- *       the keys of the *list_results* std::map, which means the std::map should be populated
- *       by the caller to fill with each bucket index object id.
- *
- * io_ctx        - IO context for rados.
- * start_obj     - marker for the listing.
- * filter_prefix - filter prefix.
- * num_entries   - number of entries to request for each object (note the total
- *                 amount of entries returned depends on the number of shardings).
- * list_results  - the std::list results keyed by bucket index object id.
- * max_aio       - the maximum number of AIO (for throttling).
- *
- * Return 0 on success, a failure code otherwise.
-*/
-
-class CLSRGWIssueBucketList : public CLSRGWConcurrentIO {
-  cls_rgw_obj_key start_obj;
-  std::string filter_prefix;
-  std::string delimiter;
-  uint32_t num_entries;
-  bool list_versions;
-  std::map<int, rgw_cls_list_ret>& result; // request_id -> return value
-
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-  void reset_container(std::map<int, std::string>& objs) override;
-
-public:
-  CLSRGWIssueBucketList(librados::IoCtx& io_ctx,
-			const cls_rgw_obj_key& _start_obj,
-                        const std::string& _filter_prefix,
-			const std::string& _delimiter,
-			uint32_t _num_entries,
-                        bool _list_versions,
-                        std::map<int, std::string>& oids, // shard_id -> shard_oid
-			// shard_id -> return value
-                        std::map<int, rgw_cls_list_ret>& list_results,
-                        uint32_t max_aio) :
-  CLSRGWConcurrentIO(io_ctx, oids, max_aio),
-    start_obj(_start_obj), filter_prefix(_filter_prefix), delimiter(_delimiter),
-    num_entries(_num_entries), list_versions(_list_versions),
-    result(list_results)
-  {}
-};
-
 void cls_rgw_bucket_list_op(librados::ObjectReadOperation& op,
                             const cls_rgw_obj_key& start_obj,
                             const std::string& filter_prefix,
@@ -484,133 +346,20 @@ void cls_rgw_bilog_list(librados::ObjectReadOperation& op,
                         const std::string& marker, uint32_t max,
                         cls_rgw_bi_log_list_ret *pdata, int *ret = nullptr);
 
-class CLSRGWIssueBILogList : public CLSRGWConcurrentIO {
-  std::map<int, cls_rgw_bi_log_list_ret>& result;
-  BucketIndexShardsManager& marker_mgr;
-  uint32_t max;
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-public:
-  CLSRGWIssueBILogList(librados::IoCtx& io_ctx, BucketIndexShardsManager& _marker_mgr, uint32_t _max,
-                       std::map<int, std::string>& oids,
-                       std::map<int, cls_rgw_bi_log_list_ret>& bi_log_lists, uint32_t max_aio) :
-    CLSRGWConcurrentIO(io_ctx, oids, max_aio), result(bi_log_lists),
-    marker_mgr(_marker_mgr), max(_max) {}
-  virtual ~CLSRGWIssueBILogList() override {}
-};
-
 void cls_rgw_bilog_trim(librados::ObjectWriteOperation& op,
                         const std::string& start_marker,
                         const std::string& end_marker);
 
-class CLSRGWIssueBILogTrim : public CLSRGWConcurrentIO {
-  BucketIndexShardsManager& start_marker_mgr;
-  BucketIndexShardsManager& end_marker_mgr;
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-  // Trim until -ENODATA is returned.
-  int valid_ret_code() override { return -ENODATA; }
-  bool need_multiple_rounds() override { return true; }
-  void add_object(int shard, const std::string& oid) override { objs_container[shard] = oid; }
-  void reset_container(std::map<int, std::string>& objs) override {
-    objs_container.swap(objs);
-    iter = objs_container.begin();
-    objs.clear();
-  }
-public:
-  CLSRGWIssueBILogTrim(librados::IoCtx& io_ctx, BucketIndexShardsManager& _start_marker_mgr,
-      BucketIndexShardsManager& _end_marker_mgr, std::map<int, std::string>& _bucket_objs, uint32_t max_aio) :
-    CLSRGWConcurrentIO(io_ctx, _bucket_objs, max_aio),
-    start_marker_mgr(_start_marker_mgr), end_marker_mgr(_end_marker_mgr) {}
-  virtual ~CLSRGWIssueBILogTrim() override {}
-};
-
-class CLSRGWIssueReshardLogTrim : public CLSRGWConcurrentIO {
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-  // Trim until -ENODATA is returned.
-  int valid_ret_code() override { return -ENODATA; }
-  bool need_multiple_rounds() override { return true; }
-  void add_object(int shard, const std::string& oid) override { objs_container[shard] = oid; }
-  void reset_container(std::map<int, std::string>& objs) override {
-    objs_container.swap(objs);
-    iter = objs_container.begin();
-    objs.clear();
-  }
-public:
-  CLSRGWIssueReshardLogTrim(librados::IoCtx& io_ctx, std::map<int, std::string>& _bucket_objs, uint32_t max_aio) :
-      CLSRGWConcurrentIO(io_ctx, _bucket_objs, max_aio) {}
-};
+void cls_rgw_bucket_check_index(librados::ObjectReadOperation& op,
+                                bufferlist& out);
+// decode the response; may throw buffer::error
+void cls_rgw_bucket_check_index_decode(const bufferlist& out,
+                                       rgw_cls_check_index_ret& result);
 
-/**
- * Check the bucket index.
- *
- * io_ctx          - IO context for rados.
- * bucket_objs_ret - check result for all shards.
- * max_aio         - the maximum number of AIO (for throttling).
- *
- * Return 0 on success, a failure code otherwise.
- */
-class CLSRGWIssueBucketCheck : public CLSRGWConcurrentIO /*<std::map<std::string, rgw_cls_check_index_ret> >*/ {
-  std::map<int, rgw_cls_check_index_ret>& result;
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-public:
-  CLSRGWIssueBucketCheck(librados::IoCtx& ioc, std::map<int, std::string>& oids,
-			 std::map<int, rgw_cls_check_index_ret>& bucket_objs_ret,
-			 uint32_t _max_aio) :
-    CLSRGWConcurrentIO(ioc, oids, _max_aio), result(bucket_objs_ret) {}
-  virtual ~CLSRGWIssueBucketCheck() override {}
-};
+void cls_rgw_bucket_rebuild_index(librados::ObjectWriteOperation& op);
 
-class CLSRGWIssueBucketRebuild : public CLSRGWConcurrentIO {
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-public:
-  CLSRGWIssueBucketRebuild(librados::IoCtx& io_ctx, std::map<int, std::string>& bucket_objs,
-                           uint32_t max_aio) : CLSRGWConcurrentIO(io_ctx, bucket_objs, max_aio) {}
-  virtual ~CLSRGWIssueBucketRebuild() override {}
-};
-
-class CLSRGWIssueGetDirHeader : public CLSRGWConcurrentIO {
-  std::map<int, rgw_cls_list_ret>& result;
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-public:
-  CLSRGWIssueGetDirHeader(librados::IoCtx& io_ctx, std::map<int, std::string>& oids, std::map<int, rgw_cls_list_ret>& dir_headers,
-                          uint32_t max_aio) :
-    CLSRGWConcurrentIO(io_ctx, oids, max_aio), result(dir_headers) {}
-  virtual ~CLSRGWIssueGetDirHeader() override {}
-};
-
-class CLSRGWIssueSetBucketResharding : public CLSRGWConcurrentIO {
-  cls_rgw_bucket_instance_entry entry;
-protected:
-  int issue_op(int shard_id, const std::string& oid) override;
-public:
-  CLSRGWIssueSetBucketResharding(librados::IoCtx& ioc, std::map<int, std::string>& _bucket_objs,
-                                 const cls_rgw_bucket_instance_entry& _entry,
-                                 uint32_t _max_aio) : CLSRGWConcurrentIO(ioc, _bucket_objs, _max_aio), entry(_entry) {}
-  virtual ~CLSRGWIssueSetBucketResharding() override {}
-};
-
-class CLSRGWIssueResyncBucketBILog : public CLSRGWConcurrentIO {
-protected:
-  int issue_op(int shard_id, const std::string& oid);
-public:
-  CLSRGWIssueResyncBucketBILog(librados::IoCtx& io_ctx, std::map<int, std::string>& _bucket_objs, uint32_t max_aio) :
-    CLSRGWConcurrentIO(io_ctx, _bucket_objs, max_aio) {}
-  virtual ~CLSRGWIssueResyncBucketBILog() override {}
-};
-
-class CLSRGWIssueBucketBILogStop : public CLSRGWConcurrentIO {
-protected:
-  int issue_op(int shard_id, const std::string& oid);
-public:
-  CLSRGWIssueBucketBILogStop(librados::IoCtx& io_ctx, std::map<int, std::string>& _bucket_objs, uint32_t max_aio) :
-    CLSRGWConcurrentIO(io_ctx, _bucket_objs, max_aio) {}
-  virtual ~CLSRGWIssueBucketBILogStop() override {}
-};
+void cls_rgw_bilog_start(librados::ObjectWriteOperation& op);
+void cls_rgw_bilog_stop(librados::ObjectWriteOperation& op);
 
 int cls_rgw_get_dir_header_async(librados::IoCtx& io_ctx, const std::string& oid,
                                  boost::intrusive_ptr<RGWGetDirHeader_CB> cb);
@@ -684,12 +433,15 @@ int cls_rgw_reshard_get(librados::IoCtx& io_ctx, const std::string& oid, cls_rgw
 // cls_rgw in the T+4 (X) release.
 void cls_rgw_guard_bucket_resharding(librados::ObjectOperation& op, int ret_err);
 
-// these overloads which call io_ctx.operate() should not be called in the rgw.
-// rgw_rados_operate() should be called after the overloads w/o calls to io_ctx.operate()
-#ifndef CLS_CLIENT_HIDE_IOCTX
-int cls_rgw_set_bucket_resharding(librados::IoCtx& io_ctx, const std::string& oid,
-                                  const cls_rgw_bucket_instance_entry& entry);
-int cls_rgw_clear_bucket_resharding(librados::IoCtx& io_ctx, const std::string& oid);
-int cls_rgw_get_bucket_resharding(librados::IoCtx& io_ctx, const std::string& oid,
-                                  cls_rgw_bucket_instance_entry *entry);
-#endif
+void cls_rgw_set_bucket_resharding(librados::ObjectWriteOperation& op,
+                                   cls_rgw_reshard_status status);
+void cls_rgw_clear_bucket_resharding(librados::ObjectWriteOperation& op);
+void cls_rgw_get_bucket_resharding(librados::ObjectReadOperation& op,
+                                   bufferlist& out);
+// decode the entry; may throw buffer::error
+void cls_rgw_get_bucket_resharding_decode(const bufferlist& out,
+                                          cls_rgw_bucket_instance_entry& entry);
+
+// Try to remove all reshard log entries from the bucket index. Return success
+// if any entries were removed, and -ENODATA once they're all gone.
+void cls_rgw_bucket_reshard_log_trim(librados::ObjectWriteOperation& op);

src/cls/rgw/cls_rgw_client.h

@@ -7,9 +7,7 @@
 
 struct rgw_cls_tag_timeout_op
 {
-  uint64_t tag_timeout;
-
-  rgw_cls_tag_timeout_op() : tag_timeout(0) {}
+  uint64_t tag_timeout = 0;
 
   void encode(ceph::buffer::list &bl) const {
     ENCODE_START(1, 1, bl);

src/cls/rgw/cls_rgw_ops.h

@@ -2468,7 +2468,7 @@ int POSIXBucket::set_acl(const DoutPrefixProvider* dpp,
   return write_attrs(dpp, y);
 }
 
-int POSIXBucket::read_stats(const DoutPrefixProvider *dpp,
+int POSIXBucket::read_stats(const DoutPrefixProvider *dpp, optional_yield y,
 			    const bucket_index_layout_generation& idx_layout,
 			    int shard_id, std::string* bucket_ver, std::string* master_ver,
 			    std::map<RGWObjCategory, RGWStorageStats>& stats,
@@ -2477,14 +2477,14 @@ int POSIXBucket::read_stats(const DoutPrefixProvider *dpp,
   auto& main = stats[RGWObjCategory::Main];
 
   // TODO: bucket stats shouldn't have to list all objects
-  return dir->for_each(dpp, [this, dpp, &main] (const char* name) {
+  return dir->for_each(dpp, [this, dpp, y, &main] (const char* name) {
     if (name[0] == '.') {
       /* Skip dotfiles */
       return 0;
     }
 
     std::unique_ptr<FSEnt> dent;
-    int ret = dir->get_ent(dpp, null_yield, name, std::string(), dent);
+    int ret = dir->get_ent(dpp, y, name, std::string(), dent);
     if (ret < 0) {
       ret = errno;
       ldpp_dout(dpp, 0) << "ERROR: could not get ent for object " << name << ": "
@@ -2617,17 +2617,19 @@ int POSIXBucket::remove_objs_from_index(const DoutPrefixProvider *dpp, std::list
   return 0;
 }
 
-int POSIXBucket::check_index(const DoutPrefixProvider *dpp, std::map<RGWObjCategory, RGWStorageStats>& existing_stats, std::map<RGWObjCategory, RGWStorageStats>& calculated_stats)
+int POSIXBucket::check_index(const DoutPrefixProvider *dpp, optional_yield y,
+                             std::map<RGWObjCategory, RGWStorageStats>& existing_stats,
+                             std::map<RGWObjCategory, RGWStorageStats>& calculated_stats)
 {
   return 0;
 }
 
-int POSIXBucket::rebuild_index(const DoutPrefixProvider *dpp)
+int POSIXBucket::rebuild_index(const DoutPrefixProvider *dpp, optional_yield y)
 {
   return 0;
 }
 
-int POSIXBucket::set_tag_timeout(const DoutPrefixProvider *dpp, uint64_t timeout)
+int POSIXBucket::set_tag_timeout(const DoutPrefixProvider *dpp, optional_yield y, uint64_t timeout)
 {
   return 0;
 }
@@ -2907,6 +2909,12 @@ int POSIXObject::list_parts(const DoutPrefixProvider* dpp, CephContext* cct,
   return -EOPNOTSUPP;
 }
 
+bool POSIXObject::is_sync_completed(const DoutPrefixProvider* dpp, optional_yield y,
+                                    const ceph::real_time& obj_mtime)
+{
+  return false;
+}
+
 int POSIXObject::load_obj_state(const DoutPrefixProvider* dpp, optional_yield y, bool follow_olh)
 {
   int ret = stat(dpp);