Fix race of create tag and edge (#3735)

* fix rece * polish * fix review
vesoft-inc · Jan 22, 2022 · db7a5a3 · db7a5a3
1 parent 8331896
commit db7a5a3
Show file tree

Hide file tree

Showing 15 changed files with 56 additions and 64 deletions.
diff --git a/src/codec/RowReaderV2.h b/src/codec/RowReaderV2.h
@@ -26,7 +26,7 @@ class RowReaderV2 : public RowReader {
   FRIEND_TEST(ScanEdgePropBench, ProcessEdgeProps);
 
  public:
-  virtual ~RowReaderV2() = default;
+  ~RowReaderV2() override = default;
 
   Value getValueByName(const std::string& prop) const noexcept override;
   Value getValueByIndex(const int64_t index) const noexcept override;

diff --git a/src/graph/planner/match/MatchPlanner.h b/src/graph/planner/match/MatchPlanner.h
@@ -14,7 +14,7 @@ namespace graph {
 class MatchPlanner final : public Planner {
  public:
   static std::unique_ptr<MatchPlanner> make() {
-    return std::unique_ptr<MatchPlanner>(new MatchPlanner());
+    return std::make_unique<MatchPlanner>();
   }
 
   static bool match(AstContext* astCtx);

diff --git a/src/meta/processors/Common.h b/src/meta/processors/Common.h
@@ -25,8 +25,7 @@ class LockUtils {
   GENERATE_LOCK(id);
   GENERATE_LOCK(workerId);
   GENERATE_LOCK(localId);
-  GENERATE_LOCK(tag);
-  GENERATE_LOCK(edge);
+  GENERATE_LOCK(tagAndEdge);
   GENERATE_LOCK(tagIndex);
   GENERATE_LOCK(edgeIndex);
   GENERATE_LOCK(service);

diff --git a/src/meta/processors/index/FTIndexProcessor.cpp b/src/meta/processors/index/FTIndexProcessor.cpp
@@ -15,8 +15,8 @@ void CreateFTIndexProcessor::process(const cpp2::CreateFTIndexReq& req) {
   const auto& index = req.get_index();
   const std::string& name = req.get_fulltext_index_name();
   CHECK_SPACE_ID_AND_RETURN(index.get_space_id());
+  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagAndEdgeLock());
   auto isEdge = index.get_depend_schema().getType() == nebula::cpp2::SchemaID::Type::edge_type;
-  folly::SharedMutex::ReadHolder rHolder(isEdge ? LockUtils::edgeLock() : LockUtils::tagLock());
   auto schemaPrefix = isEdge ? MetaKeyUtils::schemaEdgePrefix(
                                    index.get_space_id(), index.get_depend_schema().get_edge_type())
                              : MetaKeyUtils::schemaTagPrefix(

diff --git a/src/meta/processors/parts/CreateSpaceAsProcessor.cpp b/src/meta/processors/parts/CreateSpaceAsProcessor.cpp
@@ -132,7 +132,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
 
 ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcessor::makeNewTags(
     GraphSpaceID oldSpaceId, GraphSpaceID newSpaceId) {
-  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagLock());
+  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagAndEdgeLock());
   auto prefix = MetaKeyUtils::schemaTagsPrefix(oldSpaceId);
   auto tagPrefix = doPrefix(prefix);
   if (!nebula::ok(tagPrefix)) {
@@ -164,7 +164,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
 
 ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcessor::makeNewEdges(
     GraphSpaceID oldSpaceId, GraphSpaceID newSpaceId) {
-  folly::SharedMutex::ReadHolder rHolder(LockUtils::edgeLock());
+  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagAndEdgeLock());
   auto prefix = MetaKeyUtils::schemaEdgesPrefix(oldSpaceId);
   auto edgePrefix = doPrefix(prefix);
   if (!nebula::ok(edgePrefix)) {
@@ -196,7 +196,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcesso
 
 ErrorOr<nebula::cpp2::ErrorCode, std::vector<kvstore::KV>> CreateSpaceAsProcessor::makeNewIndexes(
     GraphSpaceID oldSpaceId, GraphSpaceID newSpaceId) {
-  folly::SharedMutex::ReadHolder rHolder(LockUtils::edgeLock());
+  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagAndEdgeLock());
   auto prefix = MetaKeyUtils::indexPrefix(oldSpaceId);
   auto indexPrefix = doPrefix(prefix);
   if (!nebula::ok(indexPrefix)) {

diff --git a/src/meta/processors/schema/AlterEdgeProcessor.cpp b/src/meta/processors/schema/AlterEdgeProcessor.cpp
@@ -13,10 +13,10 @@ namespace meta {
 void AlterEdgeProcessor::process(const cpp2::AlterEdgeReq& req) {
   GraphSpaceID spaceId = req.get_space_id();
   CHECK_SPACE_ID_AND_RETURN(spaceId);
-  auto edgeName = req.get_edge_name();
+  const auto& edgeName = req.get_edge_name();
 
   folly::SharedMutex::ReadHolder rHolder(LockUtils::snapshotLock());
-  folly::SharedMutex::WriteHolder wHolder(LockUtils::edgeLock());
+  folly::SharedMutex::WriteHolder wHolder(LockUtils::tagAndEdgeLock());
   auto ret = getEdgeType(spaceId, edgeName);
   if (!nebula::ok(ret)) {
     auto retCode = nebula::error(ret);

diff --git a/src/meta/processors/schema/AlterTagProcessor.cpp b/src/meta/processors/schema/AlterTagProcessor.cpp
@@ -13,10 +13,10 @@ namespace meta {
 void AlterTagProcessor::process(const cpp2::AlterTagReq& req) {
   GraphSpaceID spaceId = req.get_space_id();
   CHECK_SPACE_ID_AND_RETURN(spaceId);
-  auto tagName = req.get_tag_name();
+  const auto& tagName = req.get_tag_name();
 
   folly::SharedMutex::ReadHolder rHolder(LockUtils::snapshotLock());
-  folly::SharedMutex::WriteHolder wHolder(LockUtils::tagLock());
+  folly::SharedMutex::WriteHolder wHolder(LockUtils::tagAndEdgeLock());
   auto ret = getTagId(spaceId, tagName);
   if (!nebula::ok(ret)) {
     auto retCode = nebula::error(ret);

diff --git a/src/meta/processors/schema/CreateEdgeProcessor.cpp b/src/meta/processors/schema/CreateEdgeProcessor.cpp
@@ -13,28 +13,25 @@ namespace meta {
 void CreateEdgeProcessor::process(const cpp2::CreateEdgeReq& req) {
   GraphSpaceID spaceId = req.get_space_id();
   CHECK_SPACE_ID_AND_RETURN(spaceId);
-  auto edgeName = req.get_edge_name();
-  {
-    // if there is an tag of the same name
-    // TODO: there exists race condition, we should address it in the future
-    folly::SharedMutex::ReadHolder rHolder(LockUtils::tagLock());
-    auto conflictRet = getTagId(spaceId, edgeName);
-    if (nebula::ok(conflictRet)) {
-      LOG(ERROR) << "Failed to create edge `" << edgeName
-                 << "': some tag with the same name already exists.";
-      resp_.id_ref() = to(nebula::value(conflictRet), EntryType::EDGE);
-      handleErrorCode(nebula::cpp2::ErrorCode::E_CONFLICT);
+  const auto& edgeName = req.get_edge_name();
+  folly::SharedMutex::WriteHolder holder(LockUtils::tagAndEdgeLock());
+  // Check if the tag with same name exists
+  auto conflictRet = getTagId(spaceId, edgeName);
+  if (nebula::ok(conflictRet)) {
+    LOG(ERROR) << "Failed to create edge `" << edgeName
+               << "': some tag with the same name already exists.";
+    resp_.id_ref() = to(nebula::value(conflictRet), EntryType::EDGE);
+    handleErrorCode(nebula::cpp2::ErrorCode::E_CONFLICT);
+    onFinished();
+    return;
+  } else {
+    auto retCode = nebula::error(conflictRet);
+    if (retCode != nebula::cpp2::ErrorCode::E_TAG_NOT_FOUND) {
+      LOG(ERROR) << "Failed to create edge " << edgeName << " error "
+                 << apache::thrift::util::enumNameSafe(retCode);
+      handleErrorCode(retCode);
       onFinished();
       return;
-    } else {
-      auto retCode = nebula::error(conflictRet);
-      if (retCode != nebula::cpp2::ErrorCode::E_TAG_NOT_FOUND) {
-        LOG(ERROR) << "Failed to create edge " << edgeName << " error "
-                   << apache::thrift::util::enumNameSafe(retCode);
-        handleErrorCode(retCode);
-        onFinished();
-        return;
-      }
     }
   }
 
@@ -49,7 +46,6 @@ void CreateEdgeProcessor::process(const cpp2::CreateEdgeReq& req) {
   schema.columns_ref() = std::move(columns);
   schema.schema_prop_ref() = req.get_schema().get_schema_prop();
 
-  folly::SharedMutex::WriteHolder wHolder(LockUtils::edgeLock());
   auto ret = getEdgeType(spaceId, edgeName);
   if (nebula::ok(ret)) {
     if (req.get_if_not_exists()) {

diff --git a/src/meta/processors/schema/CreateTagProcessor.cpp b/src/meta/processors/schema/CreateTagProcessor.cpp
@@ -13,28 +13,26 @@ namespace meta {
 void CreateTagProcessor::process(const cpp2::CreateTagReq& req) {
   GraphSpaceID spaceId = req.get_space_id();
   CHECK_SPACE_ID_AND_RETURN(spaceId);
-  auto tagName = req.get_tag_name();
-  {
-    // if there is an edge of the same name
-    // TODO: there exists race condition, we should address it in the future
-    folly::SharedMutex::ReadHolder rHolder(LockUtils::edgeLock());
-    auto conflictRet = getEdgeType(spaceId, tagName);
-    if (nebula::ok(conflictRet)) {
-      LOG(ERROR) << "Failed to create tag `" << tagName
-                 << "': some edge with the same name already exists.";
-      resp_.id_ref() = to(nebula::value(conflictRet), EntryType::TAG);
-      handleErrorCode(nebula::cpp2::ErrorCode::E_CONFLICT);
+  const auto& tagName = req.get_tag_name();
+  folly::SharedMutex::WriteHolder holder(LockUtils::tagAndEdgeLock());
+
+  // Check if the edge with same name exists
+  auto conflictRet = getEdgeType(spaceId, tagName);
+  if (nebula::ok(conflictRet)) {
+    LOG(ERROR) << "Failed to create tag `" << tagName
+               << "': some edge with the same name already exists.";
+    resp_.id_ref() = to(nebula::value(conflictRet), EntryType::TAG);
+    handleErrorCode(nebula::cpp2::ErrorCode::E_CONFLICT);
+    onFinished();
+    return;
+  } else {
+    auto retCode = nebula::error(conflictRet);
+    if (retCode != nebula::cpp2::ErrorCode::E_EDGE_NOT_FOUND) {
+      LOG(ERROR) << "Failed to create tag " << tagName << " error "
+                 << apache::thrift::util::enumNameSafe(retCode);
+      handleErrorCode(retCode);
       onFinished();
       return;
-    } else {
-      auto retCode = nebula::error(conflictRet);
-      if (retCode != nebula::cpp2::ErrorCode::E_EDGE_NOT_FOUND) {
-        LOG(ERROR) << "Failed to create tag " << tagName << " error "
-                   << apache::thrift::util::enumNameSafe(retCode);
-        handleErrorCode(retCode);
-        onFinished();
-        return;
-      }
     }
   }
 
@@ -49,7 +47,6 @@ void CreateTagProcessor::process(const cpp2::CreateTagReq& req) {
   schema.columns_ref() = std::move(columns);
   schema.schema_prop_ref() = req.get_schema().get_schema_prop();
 
-  folly::SharedMutex::WriteHolder wHolder(LockUtils::tagLock());
   auto ret = getTagId(spaceId, tagName);
   if (nebula::ok(ret)) {
     if (req.get_if_not_exists()) {

diff --git a/src/meta/processors/schema/DropEdgeProcessor.cpp b/src/meta/processors/schema/DropEdgeProcessor.cpp
@@ -13,8 +13,8 @@ void DropEdgeProcessor::process(const cpp2::DropEdgeReq& req) {
   CHECK_SPACE_ID_AND_RETURN(spaceId);
 
   folly::SharedMutex::ReadHolder rHolder(LockUtils::snapshotLock());
-  folly::SharedMutex::WriteHolder wHolder(LockUtils::edgeLock());
-  auto edgeName = req.get_edge_name();
+  folly::SharedMutex::WriteHolder wHolder(LockUtils::tagAndEdgeLock());
+  const auto& edgeName = req.get_edge_name();
 
   EdgeType edgeType;
   auto indexKey = MetaKeyUtils::indexEdgeKey(spaceId, edgeName);

diff --git a/src/meta/processors/schema/DropTagProcessor.cpp b/src/meta/processors/schema/DropTagProcessor.cpp
@@ -13,8 +13,8 @@ void DropTagProcessor::process(const cpp2::DropTagReq& req) {
   CHECK_SPACE_ID_AND_RETURN(spaceId);
 
   folly::SharedMutex::ReadHolder rHolder(LockUtils::snapshotLock());
-  folly::SharedMutex::WriteHolder wHolder(LockUtils::tagLock());
-  auto tagName = req.get_tag_name();
+  folly::SharedMutex::WriteHolder wHolder(LockUtils::tagAndEdgeLock());
+  const auto& tagName = req.get_tag_name();
 
   TagID tagId;
   auto indexKey = MetaKeyUtils::indexTagKey(spaceId, tagName);

diff --git a/src/meta/processors/schema/GetEdgeProcessor.cpp b/src/meta/processors/schema/GetEdgeProcessor.cpp
@@ -11,10 +11,10 @@ namespace meta {
 void GetEdgeProcessor::process(const cpp2::GetEdgeReq& req) {
   GraphSpaceID spaceId = req.get_space_id();
   CHECK_SPACE_ID_AND_RETURN(spaceId);
-  auto edgeName = req.get_edge_name();
+  const auto& edgeName = req.get_edge_name();
   auto ver = req.get_version();
 
-  folly::SharedMutex::ReadHolder rHolder(LockUtils::edgeLock());
+  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagAndEdgeLock());
   auto edgeTypeRet = getEdgeType(spaceId, edgeName);
   if (!nebula::ok(edgeTypeRet)) {
     LOG(ERROR) << "Get edge " << edgeName << " failed.";

diff --git a/src/meta/processors/schema/GetTagProcessor.cpp b/src/meta/processors/schema/GetTagProcessor.cpp
@@ -11,10 +11,10 @@ namespace meta {
 void GetTagProcessor::process(const cpp2::GetTagReq& req) {
   GraphSpaceID spaceId = req.get_space_id();
   CHECK_SPACE_ID_AND_RETURN(spaceId);
-  auto tagName = req.get_tag_name();
+  const auto& tagName = req.get_tag_name();
   auto ver = req.get_version();
 
-  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagLock());
+  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagAndEdgeLock());
   auto tagIdRet = getTagId(spaceId, tagName);
   if (!nebula::ok(tagIdRet)) {
     LOG(ERROR) << "Get tag " << tagName << " failed.";

diff --git a/src/meta/processors/schema/ListEdgesProcessor.cpp b/src/meta/processors/schema/ListEdgesProcessor.cpp
@@ -12,7 +12,7 @@ void ListEdgesProcessor::process(const cpp2::ListEdgesReq &req) {
   GraphSpaceID spaceId = req.get_space_id();
   CHECK_SPACE_ID_AND_RETURN(spaceId);
 
-  folly::SharedMutex::ReadHolder rHolder(LockUtils::edgeLock());
+  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagAndEdgeLock());
   auto prefix = MetaKeyUtils::schemaEdgesPrefix(spaceId);
   auto ret = doPrefix(prefix);
   if (!nebula::ok(ret)) {

diff --git a/src/meta/processors/schema/ListTagsProcessor.cpp b/src/meta/processors/schema/ListTagsProcessor.cpp
@@ -12,7 +12,7 @@ void ListTagsProcessor::process(const cpp2::ListTagsReq &req) {
   GraphSpaceID spaceId = req.get_space_id();
   CHECK_SPACE_ID_AND_RETURN(spaceId);
 
-  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagLock());
+  folly::SharedMutex::ReadHolder rHolder(LockUtils::tagAndEdgeLock());
   auto prefix = MetaKeyUtils::schemaTagsPrefix(spaceId);
   auto ret = doPrefix(prefix);
   if (!nebula::ok(ret)) {