Skip to content
This repository has been archived by the owner on Dec 1, 2022. It is now read-only.

Improved data write logic to avoid reading dirty data #517

Merged
merged 3 commits into from
Jul 9, 2021
Merged

Improved data write logic to avoid reading dirty data #517

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6ed7a49
done
bright-starry-sky Jul 6, 2021
21b88e6
addressed panda's comments
bright-starry-sky Jul 7, 2021
987358d
add test case
bright-starry-sky Jul 7, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 20 additions & 19 deletions src/storage/mutate/AddEdgesProcessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,24 @@ void AddEdgesProcessor::doProcessWithIndex(const cpp2::AddEdgesRequest& req) {
visited.reserve(newEdges.size());
for (auto& newEdge : newEdges) {
auto edgeKey = *newEdge.key_ref();
auto l = std::make_tuple(spaceId_,
partId,
(*edgeKey.src_ref()).getStr(),
*edgeKey.edge_type_ref(),
*edgeKey.ranking_ref(),
(*edgeKey.dst_ref()).getStr());
if (std::find(dummyLock.begin(), dummyLock.end(), l) == dummyLock.end()) {
if (!env_->edgesML_->try_lock(l)) {
LOG(ERROR) << folly::format("edge locked : src {}, type {}, rank {}, dst {}",
(*edgeKey.src_ref()).getStr(),
*edgeKey.edge_type_ref(),
*edgeKey.ranking_ref(),
(*edgeKey.dst_ref()).getStr());
code = nebula::cpp2::ErrorCode::E_DATA_CONFLICT_ERROR;
break;
}
}
dummyLock.emplace_back(std::move(l));
panda-sheep marked this conversation as resolved.
Show resolved Hide resolved
VLOG(3) << "PartitionID: " << partId << ", VertexID: " << *edgeKey.src_ref()
<< ", EdgeType: " << *edgeKey.edge_type_ref() << ", EdgeRanking: "
<< *edgeKey.ranking_ref() << ", VertexID: "
Expand Down Expand Up @@ -274,32 +292,15 @@ void AddEdgesProcessor::doProcessWithIndex(const cpp2::AddEdgesRequest& req) {
break;
}
batchHolder->put(std::move(key), std::move(retEnc.value()));
dummyLock.emplace_back(std::make_tuple(spaceId_,
partId,
(*edgeKey.src_ref()).getStr(),
*edgeKey.edge_type_ref(),
*edgeKey.ranking_ref(),
(*edgeKey.dst_ref()).getStr()));
}
if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
env_->edgesML_->unlockBatch(dummyLock);
handleAsync(spaceId_, partId, code);
continue;
}
auto batch = encodeBatchValue(std::move(batchHolder)->getBatch());
DCHECK(!batch.empty());
nebula::MemoryLockGuard<EMLI> lg(env_->edgesML_.get(), std::move(dummyLock), true);
if (!lg) {
auto conflict = lg.conflictKey();
LOG(ERROR) << "edge conflict "
<< std::get<0>(conflict) << ":"
<< std::get<1>(conflict) << ":"
<< std::get<2>(conflict) << ":"
<< std::get<3>(conflict) << ":"
<< std::get<4>(conflict) << ":"
<< std::get<5>(conflict);
handleAsync(spaceId_, partId, nebula::cpp2::ErrorCode::E_DATA_CONFLICT_ERROR);
continue;
}
nebula::MemoryLockGuard<EMLI> lg(env_->edgesML_.get(), std::move(dummyLock), true, false);
env_->kvstore_->asyncAppendBatch(spaceId_, partId, std::move(batch),
[l = std::move(lg), icw = std::move(wrapper), partId, this]
(nebula::cpp2::ErrorCode retCode) {
Expand Down
27 changes: 15 additions & 12 deletions src/storage/mutate/AddVerticesProcessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,16 @@ void AddVerticesProcessor::doProcessWithIndex(const cpp2::AddVerticesRequest& re

for (auto& newTag : newTags) {
auto tagId = newTag.get_tag_id();
auto l = std::make_tuple(spaceId_, partId, tagId, vid);
if (std::find(dummyLock.begin(), dummyLock.end(), l) == dummyLock.end()) {
if (!env_->verticesML_->try_lock(l)) {
LOG(ERROR) << folly::format("The vertex locked : tag {}, vid {}",
tagId, vid);
code = nebula::cpp2::ErrorCode::E_DATA_CONFLICT_ERROR;
break;
}
}
dummyLock.emplace_back(std::move(l));
VLOG(3) << "PartitionID: " << partId << ", VertexID: " << vid
<< ", TagID: " << tagId;

Expand Down Expand Up @@ -277,7 +287,6 @@ void AddVerticesProcessor::doProcessWithIndex(const cpp2::AddVerticesRequest& re
* step 3 , Insert new vertex data
*/
batchHolder->put(std::move(key), std::move(retEnc.value()));
dummyLock.emplace_back(std::make_tuple(spaceId_, partId, tagId, vid));

if (FLAGS_enable_vertex_cache && vertexCache_ != nullptr) {
vertexCache_->evict(std::make_pair(vid, tagId));
Expand All @@ -290,22 +299,16 @@ void AddVerticesProcessor::doProcessWithIndex(const cpp2::AddVerticesRequest& re
}
}
if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
env_->verticesML_->unlockBatch(dummyLock);
handleAsync(spaceId_, partId, code);
continue;
}
auto batch = encodeBatchValue(std::move(batchHolder)->getBatch());
DCHECK(!batch.empty());
nebula::MemoryLockGuard<VMLI> lg(env_->verticesML_.get(), std::move(dummyLock), true);
if (!lg) {
auto conflict = lg.conflictKey();
LOG(ERROR) << "vertex conflict "
<< std::get<0>(conflict) << ":"
<< std::get<1>(conflict) << ":"
<< std::get<2>(conflict) << ":"
<< std::get<3>(conflict);
handleAsync(spaceId_, partId, nebula::cpp2::ErrorCode::E_DATA_CONFLICT_ERROR);
continue;
}
nebula::MemoryLockGuard<VMLI> lg(env_->verticesML_.get(),
std::move(dummyLock),
true,
false);
env_->kvstore_->asyncAppendBatch(spaceId_, partId, std::move(batch),
[l = std::move(lg), icw = std::move(wrapper), partId, this] (
nebula::cpp2::ErrorCode retCode) {
Expand Down
47 changes: 28 additions & 19 deletions src/storage/mutate/DeleteEdgesProcessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,33 +86,42 @@ void DeleteEdgesProcessor::process(const cpp2::DeleteEdgesRequest& req) {
std::vector<EMLI> dummyLock;
dummyLock.reserve(part.second.size());

nebula::cpp2::ErrorCode err = nebula::cpp2::ErrorCode::SUCCEEDED;
for (const auto& edgeKey : part.second) {
dummyLock.emplace_back(std::make_tuple(spaceId_,
partId,
(*edgeKey.src_ref()).getStr(),
*edgeKey.edge_type_ref(),
*edgeKey.ranking_ref(),
(*edgeKey.dst_ref()).getStr()));
auto l = std::make_tuple(spaceId_,
partId,
(*edgeKey.src_ref()).getStr(),
*edgeKey.edge_type_ref(),
*edgeKey.ranking_ref(),
(*edgeKey.dst_ref()).getStr());
if (!env_->edgesML_->try_lock(l)) {
LOG(ERROR) << folly::format("The edge locked : src {}, "
"type {}, tank {}, dst {}",
(*edgeKey.src_ref()).getStr(),
*edgeKey.edge_type_ref(),
*edgeKey.ranking_ref(),
(*edgeKey.dst_ref()).getStr());
err = nebula::cpp2::ErrorCode::E_DATA_CONFLICT_ERROR;
break;
}
dummyLock.emplace_back(std::move(l));
}
if (err != nebula::cpp2::ErrorCode::SUCCEEDED) {
env_->edgesML_->unlockBatch(dummyLock);
handleAsync(spaceId_, partId, err);
continue;
}
auto batch = deleteEdges(partId, std::move(part.second));
if (!nebula::ok(batch)) {
env_->edgesML_->unlockBatch(dummyLock);
handleAsync(spaceId_, partId, nebula::error(batch));
continue;
}
DCHECK(!nebula::value(batch).empty());
nebula::MemoryLockGuard<EMLI> lg(env_->edgesML_.get(), std::move(dummyLock), true);
if (!lg) {
auto conflict = lg.conflictKey();
LOG(ERROR) << "edge conflict "
<< std::get<0>(conflict) << ":"
<< std::get<1>(conflict) << ":"
<< std::get<2>(conflict) << ":"
<< std::get<3>(conflict) << ":"
<< std::get<4>(conflict) << ":"
<< std::get<5>(conflict);
handleAsync(spaceId_, partId, nebula::cpp2::ErrorCode::E_DATA_CONFLICT_ERROR);
continue;
}
nebula::MemoryLockGuard<EMLI> lg(env_->edgesML_.get(),
std::move(dummyLock),
false,
false);
env_->kvstore_->asyncAppendBatch(spaceId_, partId, std::move(nebula::value(batch)),
[l = std::move(lg), icw = std::move(wrapper), partId, this] (
nebula::cpp2::ErrorCode code) {
Expand Down
24 changes: 12 additions & 12 deletions src/storage/mutate/DeleteVerticesProcessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,21 +96,15 @@ void DeleteVerticesProcessor::process(const cpp2::DeleteVerticesRequest& req) {
std::vector<VMLI> dummyLock;
auto batch = deleteVertices(partId, std::move(pv).second, dummyLock);
if (!nebula::ok(batch)) {
env_->verticesML_->unlockBatch(dummyLock);
handleAsync(spaceId_, partId, nebula::error(batch));
continue;
}
DCHECK(!nebula::value(batch).empty());
nebula::MemoryLockGuard<VMLI> lg(env_->verticesML_.get(), std::move(dummyLock), true);
if (!lg) {
auto conflict = lg.conflictKey();
LOG(ERROR) << "vertex conflict "
<< std::get<0>(conflict) << ":"
<< std::get<1>(conflict) << ":"
<< std::get<2>(conflict) << ":"
<< std::get<3>(conflict);
handleAsync(spaceId_, partId, nebula::cpp2::ErrorCode::E_DATA_CONFLICT_ERROR);
continue;
}
nebula::MemoryLockGuard<VMLI> lg(env_->verticesML_.get(),
std::move(dummyLock),
false,
false);
env_->kvstore_->asyncAppendBatch(spaceId_, partId, std::move(nebula::value(batch)),
[l = std::move(lg), icw = std::move(wrapper), partId, this] (
nebula::cpp2::ErrorCode code) {
Expand Down Expand Up @@ -142,6 +136,13 @@ DeleteVerticesProcessor::deleteVertices(PartitionID partId,
while (iter->valid()) {
auto key = iter->key();
auto tagId = NebulaKeyUtils::getTagId(spaceVidLen_, key);
auto l = std::make_tuple(spaceId_, partId, tagId, vertex.getStr());
if (!env_->verticesML_->try_lock(l)) {
LOG(ERROR) << folly::format("The vertex locked : tag {}, vid {}",
tagId, vertex.getStr());
return nebula::cpp2::ErrorCode::E_DATA_CONFLICT_ERROR;
}
target.emplace_back(std::move(l));
RowReaderWrapper reader;
for (auto& index : indexes_) {
if (index->get_schema_id().get_tag_id() == tagId) {
Expand Down Expand Up @@ -186,7 +187,6 @@ DeleteVerticesProcessor::deleteVertices(PartitionID partId,
VLOG(3) << "Evict vertex cache for vertex ID " << vertex << ", tagId " << tagId;
vertexCache_->evict(std::make_pair(vertex.getStr(), tagId));
}
target.emplace_back(std::make_tuple(spaceId_, partId, tagId, vertex.getStr()));
batchHolder->remove(key.str());
iter->next();
}
Expand Down
15 changes: 15 additions & 0 deletions src/storage/test/MemoryLockTest.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,21 @@ TEST_F(MemoryLockTest, MoveTest) {
}
}

TEST_F(MemoryLockTest, PrepTest) {
MemoryLockCore<std::string> mlock;
{
EXPECT_TRUE(mlock.try_lock("1"));
EXPECT_TRUE(mlock.try_lock("2"));
EXPECT_FALSE(mlock.try_lock("1"));
EXPECT_FALSE(mlock.try_lock("2"));
std::vector<std::string> keys{"1", "2"};
auto* lk = new LockGuard(&mlock, keys, false, false);
EXPECT_TRUE(lk);
delete lk;
}
EXPECT_EQ(0, mlock.size());
}

} // namespace storage
} // namespace nebula

Expand Down
19 changes: 13 additions & 6 deletions src/utils/MemoryLockWrapper.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,21 @@ class MemoryLockGuard {
MemoryLockGuard(MemoryLockCore<Key>* lock, const Key& key)
: MemoryLockGuard(lock, std::vector<Key>{key}) {}

MemoryLockGuard(MemoryLockCore<Key>* lock, const std::vector<Key>& keys, bool dedup = false)
MemoryLockGuard(MemoryLockCore<Key>* lock,
const std::vector<Key>& keys,
bool dedup = false,
bool prepCheck = true)
: lock_(lock), keys_(keys) {
if (dedup) {
std::sort(keys_.begin(), keys_.end());
auto last = std::unique(keys_.begin(), keys_.end());
std::tie(iter_, locked_) = lock_->lockBatch(keys_.begin(), last);
if (prepCheck) {
if (dedup) {
std::sort(keys_.begin(), keys_.end());
auto last = std::unique(keys_.begin(), keys_.end());
std::tie(iter_, locked_) = lock_->lockBatch(keys_.begin(), last);
} else {
std::tie(iter_, locked_) = lock_->lockBatch(keys_);
}
} else {
std::tie(iter_, locked_) = lock_->lockBatch(keys_);
locked_ = true;
}
}

Expand Down