rgw: block on bucket index operation when resharding the index

Signed-off-by: Orit Wasserman <owasserm@redhat.com>

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index d01c249f3d171bcdecb21c4b5095095124003060..f8bcc7375b4b3f54cfb978f56fcf90dc1bb8c4e0 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -5381,17 +5381,29 @@ int RGWRados::init_bucket_index(RGWBucketInfo& bucket_info, int num_shards)
 {
   librados::IoCtx index_ctx; // context for new bucket
 
-  int r = open_bucket_index_ctx(bucket_info, index_ctx);
-  if (r < 0)
+  /* handle on going bucket resharding */
+  int r = reshard->block_while_resharding(bucket_info.bucket.oid);
+  if (r < 0) {
     return r;
+  }
 
   string dir_oid =  dir_oid_prefix;
+  r = open_bucket_index_ctx(bucket_info, index_ctx);
+  if (r < 0) {
+    reshard->unlock_bucket_index(bucket_info.bucket.oid);
+    return r;
+  }
+
   dir_oid.append(bucket_info.bucket.bucket_id);
 
   map<int, string> bucket_objs;
   get_bucket_index_objects(dir_oid, num_shards, bucket_objs);
 
-  return CLSRGWIssueBucketIndexInit(index_ctx, bucket_objs, cct->_conf->rgw_bucket_index_max_aio)();
+  r =  CLSRGWIssueBucketIndexInit(index_ctx, bucket_objs, cct->_conf->rgw_bucket_index_max_aio)();
+
+  reshard->unlock_bucket_index(bucket_info.bucket.oid);
+
+  return r;
 }
 
 void RGWRados::create_bucket_id(string *bucket_id)
@@ -8206,13 +8218,24 @@ int RGWRados::bucket_check_index(RGWBucketInfo& bucket_info,
   // value - bucket index check OP returned result with the given bucket index object (shard)
   map<int, string> oids;
   map<int, struct rgw_cls_check_index_ret> bucket_objs_ret;
-  int ret = open_bucket_index(bucket_info, index_ctx, oids, bucket_objs_ret);
-  if (ret < 0)
+
+  /* handle on going bucket resharding */
+  int ret = reshard->block_while_resharding(bucket_info.bucket.oid);
+  if (ret < 0) {
     return ret;
+  }
+
+  ret = open_bucket_index(bucket_info, index_ctx, oids, bucket_objs_ret);
+  if (ret < 0) {
+      reshard->unlock_bucket_index(bucket_info.bucket.oid);
+      return ret;
+  }
 
   ret = CLSRGWIssueBucketCheck(index_ctx, oids, bucket_objs_ret, cct->_conf->rgw_bucket_index_max_aio)();
-  if (ret < 0)
-    return ret;
+  if (ret < 0) {
+      reshard->unlock_bucket_index(bucket_info.bucket.oid);
+      return ret;
+  }
 
   // Aggregate results (from different shards if there is any)
   map<int, struct rgw_cls_check_index_ret>::iterator iter;
@@ -8226,15 +8249,26 @@ int RGWRados::bucket_check_index(RGWBucketInfo& bucket_info,
 
 int RGWRados::bucket_rebuild_index(RGWBucketInfo& bucket_info)
 {
-  /* hande on going bucket resharding */
-
   librados::IoCtx index_ctx;
   map<int, string> bucket_objs;
-  int r = open_bucket_index(bucket_info, index_ctx, bucket_objs);
-  if (r < 0)
+
+  /* handle on going bucket resharding */
+  int r = reshard->block_while_resharding(bucket_info.bucket.oid);
+  if (r < 0) {
+    return r;
+  }
+
+  r = open_bucket_index(bucket_info, index_ctx, bucket_objs);
+  if (r < 0) {
+    reshard->unlock_bucket_index(bucket_info.bucket.oid);
     return r;
+  }
+
+  r =  CLSRGWIssueBucketRebuild(index_ctx, bucket_objs, cct->_conf->rgw_bucket_index_max_aio)();
+
+  reshard->unlock_bucket_index(bucket_info.bucket.oid);
 
-  return CLSRGWIssueBucketRebuild(index_ctx, bucket_objs, cct->_conf->rgw_bucket_index_max_aio)();
+  return r;
 }
 
 
@@ -9497,9 +9531,17 @@ int RGWRados::Bucket::UpdateIndex::prepare(RGWModifyOp op, const string *write_t
   }
   RGWRados *store = target->get_store();
   BucketShard *bs;
-  int ret = get_bucket_shard(&bs);
+
+  /* handle on going bucket resharding */
+  int ret = store->reshard->block_while_resharding(target->get_bucket().oid);
+  if (ret < 0) {
+    return ret;
+  }
+
+  ret = get_bucket_shard(&bs);
   if (ret < 0) {
     ldout(store->ctx(), 5) << "failed to get BucketShard object: ret=" << ret << dendl;
+    store->reshard->unlock_bucket_index(target->get_bucket().oid);
     return ret;
   }
 
@@ -9513,9 +9555,12 @@ int RGWRados::Bucket::UpdateIndex::prepare(RGWModifyOp op, const string *write_t
 
   int r = store->cls_obj_prepare_op(*bs, op, optag, obj, bilog_flags, zones_trace);
   if (r < 0) {
+    store->reshard->unlock_bucket_index(target->get_bucket().oid);
     return r;
   }
   prepared = true;
+
+  store->reshard->unlock_bucket_index(target->get_bucket().oid);
   return 0;
 }
 
@@ -9532,9 +9577,17 @@ int RGWRados::Bucket::UpdateIndex::complete(int64_t poolid, uint64_t epoch,
   }
   RGWRados *store = target->get_store();
   BucketShard *bs;
-  int ret = get_bucket_shard(&bs);
+
+  /* handle on going bucket resharding */
+  int ret = store->reshard->block_while_resharding(target->get_bucket().oid);
+  if (ret < 0) {
+    return ret;
+  }
+
+  ret = get_bucket_shard(&bs);
   if (ret < 0) {
     ldout(store->ctx(), 5) << "failed to get BucketShard object: ret=" << ret << dendl;
+    store->reshard->unlock_bucket_index(target->get_bucket().oid);
     return ret;
   }
 
@@ -9565,6 +9618,8 @@ int RGWRados::Bucket::UpdateIndex::complete(int64_t poolid, uint64_t epoch,
     lderr(store->ctx()) << "ERROR: failed writing data log" << dendl;
   }
 
+  store->reshard->unlock_bucket_index(target->get_bucket().oid);
+
   return ret;
 }
 
@@ -9577,9 +9632,17 @@ int RGWRados::Bucket::UpdateIndex::complete_del(int64_t poolid, uint64_t epoch,
   }
   RGWRados *store = target->get_store();
   BucketShard *bs;
-  int ret = get_bucket_shard(&bs);
+
+  /* handle on going bucket resharding */
+  int ret = store->reshard->block_while_resharding(target->get_bucket().oid);
+  if (ret < 0) {
+    return ret;
+  }
+
+  ret = get_bucket_shard(&bs);
   if (ret < 0) {
     ldout(store->ctx(), 5) << "failed to get BucketShard object: ret=" << ret << dendl;
+    store->reshard->unlock_bucket_index(target->get_bucket().oid);
     return ret;
   }
 
@@ -9590,6 +9653,7 @@ int RGWRados::Bucket::UpdateIndex::complete_del(int64_t poolid, uint64_t epoch,
     lderr(store->ctx()) << "ERROR: failed writing data log" << dendl;
   }
 
+  store->reshard->unlock_bucket_index(target->get_bucket().oid);
   return ret;
 }
 
@@ -9601,9 +9665,17 @@ int RGWRados::Bucket::UpdateIndex::cancel()
   }
   RGWRados *store = target->get_store();
   BucketShard *bs;
-  int ret = get_bucket_shard(&bs);
+
+  /* handle on going bucket resharding */
+  int ret = store->reshard->block_while_resharding(target->get_bucket().oid);
+  if (ret < 0) {
+    return ret;
+  }
+
+  ret = get_bucket_shard(&bs);
   if (ret < 0) {
     ldout(store->ctx(), 5) << "failed to get BucketShard object: ret=" << ret << dendl;
+    store->reshard->unlock_bucket_index(target->get_bucket().oid);
     return ret;
   }
 
@@ -9619,6 +9691,7 @@ int RGWRados::Bucket::UpdateIndex::cancel()
     lderr(store->ctx()) << "ERROR: failed writing data log" << dendl;
   }
 
+  store->reshard->unlock_bucket_index(target->get_bucket().oid);
   return ret;
 }
 
@@ -10418,11 +10491,18 @@ int RGWRados::bucket_index_link_olh(const RGWBucketInfo& bucket_info, RGWObjStat
     return r;
   }
 
+  /* handle on going bucket resharding */
+  r = reshard->block_while_resharding(bucket_info.bucket.oid);
+  if (r < 0) {
+    return r;
+  }
+
   BucketShard bs(this);
-  int ret = bs.init(obj_instance.bucket, obj_instance);
-  if (ret < 0) {
-    ldout(cct, 5) << "bs.init() returned ret=" << ret << dendl;
-    return ret;
+  r = bs.init(obj_instance.bucket, obj_instance);
+  if (r < 0) {
+    ldout(cct, 5) << "bs.init() returned ret=" << r << dendl;
+    reshard->unlock_bucket_index(bucket_info.bucket.oid);
+    return r;
   }
   
   rgw_zone_set zones_trace;
@@ -10434,14 +10514,12 @@ int RGWRados::bucket_index_link_olh(const RGWBucketInfo& bucket_info, RGWObjStat
   }
 
   cls_rgw_obj_key key(obj_instance.key.get_index_key_name(), obj_instance.key.instance);
-  ret = cls_rgw_bucket_link_olh(bs.index_ctx, bs.bucket_obj, key, olh_state.olh_tag, delete_marker, op_tag, meta, olh_epoch,
+  r = cls_rgw_bucket_link_olh(bs.index_ctx, bs.bucket_obj, key, olh_state.olh_tag, delete_marker, op_tag, meta, olh_epoch,
                                 unmod_since, high_precision_time,
                                 get_zone().log_data, zones_trace);
-  if (ret < 0) {
-    return ret;
-  }
 
-  return 0;
+  reshard->unlock_bucket_index(bucket_info.bucket.oid);
+    return r;
 }
 
 void RGWRados::bucket_index_guard_olh_op(RGWObjState& olh_state, ObjectOperation& op)
@@ -11860,21 +11938,39 @@ int RGWRados::trim_bi_log_entries(RGWBucketInfo& bucket_info, int shard_id, stri
 {
   librados::IoCtx index_ctx;
   map<int, string> bucket_objs;
-  int r = open_bucket_index(bucket_info, index_ctx, bucket_objs, shard_id);
-  if (r < 0)
+
+  /* handle on going bucket resharding */
+  int r = reshard->block_while_resharding(bucket_info.bucket.oid);
+  if (r < 0) {
     return r;
+  }
 
   BucketIndexShardsManager start_marker_mgr;
+  BucketIndexShardsManager end_marker_mgr;
+
+  r = open_bucket_index(bucket_info, index_ctx, bucket_objs, shard_id);
+  if (r < 0) {
+    reshard->unlock_bucket_index(bucket_info.bucket.oid);
+    return r;
+  }
+
   r = start_marker_mgr.from_string(start_marker, shard_id);
-  if (r < 0)
+  if (r < 0) {
+    reshard->unlock_bucket_index(bucket_info.bucket.oid);
     return r;
-  BucketIndexShardsManager end_marker_mgr;
+  }
+
   r = end_marker_mgr.from_string(end_marker, shard_id);
-  if (r < 0)
+  if (r < 0) {
+    reshard->unlock_bucket_index(bucket_info.bucket.oid);
     return r;
+  }
 
-  return CLSRGWIssueBILogTrim(index_ctx, start_marker_mgr, end_marker_mgr, bucket_objs,
-      cct->_conf->rgw_bucket_index_max_aio)();
+  r = CLSRGWIssueBILogTrim(index_ctx, start_marker_mgr, end_marker_mgr, bucket_objs,
+                          cct->_conf->rgw_bucket_index_max_aio)();
+
+  reshard->unlock_bucket_index(bucket_info.bucket.oid);
+  return r;
 }
 
 int RGWRados::bi_get_instance(const RGWBucketInfo& bucket_info, rgw_obj& obj, rgw_bucket_dir_entry *dirent)