Refactored some code. (#1257)

vesoft-inc · Nov 14, 2019 · 9c857f0 · 9c857f0
1 parent b614cab
commit 9c857f0
Show file tree

Hide file tree

Showing 3 changed files with 52 additions and 101 deletions.
diff --git a/src/meta/processors/partsMan/ListHostsProcessor.cpp b/src/meta/processors/partsMan/ListHostsProcessor.cpp
@@ -133,7 +133,7 @@ StatusOr<std::vector<cpp2::HostItem>> ListHostsProcessor::allHostsWithStatus(
  kvstore_->asyncMultiRemove(kDefaultSpaceId,
  kDefaultPartId,
  std::move(removeHostsKey),
- [this] (kvstore::ResultCode code) {
+ [] (kvstore::ResultCode code) {
  if (code != kvstore::ResultCode::SUCCEEDED) {
  LOG(ERROR) << "Async remove long time offline hosts failed: " << code;
  }

diff --git a/src/storage/BaseProcessor.h b/src/storage/BaseProcessor.h
@@ -105,6 +105,9 @@ class BaseProcessor {
  return tHost;
  }
 
+private:
+ void handleAsync(GraphSpaceID spaceId, PartitionID partId, kvstore::ResultCode code);
+
 protected:
  kvstore::KVStore* kvstore_ = nullptr;
  meta::SchemaManager* schemaMan_ = nullptr;

diff --git a/src/storage/BaseProcessor.inl b/src/storage/BaseProcessor.inl
@@ -28,121 +28,69 @@ cpp2::ErrorCode BaseProcessor<RESP>::to(kvstore::ResultCode code) {
  }
 }
 
+template <typename RESP>
+void BaseProcessor<RESP>::handleAsync(GraphSpaceID spaceId,
+ PartitionID partId,
+ kvstore::ResultCode code) {
+ VLOG(3) << "partId:" << partId << ", code:" << static_cast<int32_t>(code);
 
-template<typename RESP>
+ cpp2::ResultCode thriftResult;
+ thriftResult.set_code(to(code));
+ thriftResult.set_part_id(partId);
+ if (code == kvstore::ResultCode::ERR_LEADER_CHANGED) {
+ nebula::cpp2::HostAddr leader;
+ auto addrRet = kvstore_->partLeader(spaceId, partId);
+ CHECK(ok(addrRet));
+ auto addr = value(std::move(addrRet));
+ leader.set_ip(addr.first);
+ leader.set_port(addr.second);
+ thriftResult.set_leader(std::move(leader));
+ }
+ bool finished = false;
+ {
+ std::lock_guard<std::mutex> lg(this->lock_);
+ if (thriftResult.code != cpp2::ErrorCode::SUCCEEDED) {
+ this->codes_.emplace_back(std::move(thriftResult));
+ }
+ this->callingNum_--;
+ if (this->callingNum_ == 0) {
+ finished = true;
+ }
+ }
+ if (finished) {
+ this->onFinished();
+ }
+}
+
+template <typename RESP>
 void BaseProcessor<RESP>::doPut(GraphSpaceID spaceId,
  PartitionID partId,
  std::vector<kvstore::KV> data) {
- this->kvstore_->asyncMultiPut(spaceId,
- partId,
- std::move(data),
- [spaceId, partId, this](kvstore::ResultCode code) {
- VLOG(3) << "partId:" << partId << ", code:" << static_cast<int32_t>(code);
-
- cpp2::ResultCode thriftResult;
- thriftResult.set_code(to(code));
- thriftResult.set_part_id(partId);
- if (code == kvstore::ResultCode::ERR_LEADER_CHANGED) {
- nebula::cpp2::HostAddr leader;
- auto addrRet = kvstore_->partLeader(spaceId, partId);
- CHECK(ok(addrRet));
- auto addr = value(std::move(addrRet));
- leader.set_ip(addr.first);
- leader.set_port(addr.second);
- thriftResult.set_leader(std::move(leader));
- }
- bool finished = false;
- {
- std::lock_guard<std::mutex> lg(this->lock_);
- if (thriftResult.code != cpp2::ErrorCode::SUCCEEDED) {
- this->codes_.emplace_back(std::move(thriftResult));
- }
- this->callingNum_--;
- if (this->callingNum_ == 0) {
- finished = true;
- }
- }
- if (finished) {
- this->onFinished();
- }
- });
+ this->kvstore_->asyncMultiPut(
+ spaceId, partId, std::move(data), [spaceId, partId, this](kvstore::ResultCode code) {
+ handleAsync(spaceId, partId, code);
+ });
 }
 
-template<typename RESP>
+template <typename RESP>
 void BaseProcessor<RESP>::doRemove(GraphSpaceID spaceId,
  PartitionID partId,
  std::vector<std::string> keys) {
- this->kvstore_->asyncMultiRemove(spaceId,
- partId,
- std::move(keys),
- [spaceId, partId, this](kvstore::ResultCode code) {
- VLOG(3) << "partId:" << partId << ", code:" << static_cast<int32_t>(code);
-
- cpp2::ResultCode thriftResult;
- thriftResult.set_code(to(code));
- thriftResult.set_part_id(partId);
- if (code == kvstore::ResultCode::ERR_LEADER_CHANGED) {
- nebula::cpp2::HostAddr leader;
- auto addrRet = kvstore_->partLeader(spaceId, partId);
- CHECK(ok(addrRet));
- auto addr = value(std::move(addrRet));
- leader.set_ip(addr.first);
- leader.set_port(addr.second);
- thriftResult.set_leader(std::move(leader));
- }
- bool finished = false;
- {
- std::lock_guard<std::mutex> lg(this->lock_);
- if (thriftResult.code != cpp2::ErrorCode::SUCCEEDED) {
- this->codes_.emplace_back(std::move(thriftResult));
- }
- this->callingNum_--;
- if (this->callingNum_ == 0) {
- finished = true;
- }
- }
- if (finished) {
- this->onFinished();
- }
- });
+ this->kvstore_->asyncMultiRemove(
+ spaceId, partId, std::move(keys), [spaceId, partId, this](kvstore::ResultCode code) {
+ handleAsync(spaceId, partId, code);
+ });
 }
 
-template<typename RESP>
+template <typename RESP>
 void BaseProcessor<RESP>::doRemoveRange(GraphSpaceID spaceId,
  PartitionID partId,
  std::string start,
  std::string end) {
- this->kvstore_->asyncRemoveRange(spaceId, partId, start, end,
- [spaceId, partId, this](kvstore::ResultCode code) {
- VLOG(3) << "partId:" << partId << ", code:" << static_cast<int32_t>(code);
-
- cpp2::ResultCode thriftResult;
- thriftResult.set_code(to(code));
- thriftResult.set_part_id(partId);
- if (code == kvstore::ResultCode::ERR_LEADER_CHANGED) {
- nebula::cpp2::HostAddr leader;
- auto addrRet = kvstore_->partLeader(spaceId, partId);
- CHECK(ok(addrRet));
- auto addr = value(std::move(addrRet));
- leader.set_ip(addr.first);
- leader.set_port(addr.second);
- thriftResult.set_leader(std::move(leader));
- }
- bool finished = false;
- {
- std::lock_guard<std::mutex> lg(this->lock_);
- if (thriftResult.code != cpp2::ErrorCode::SUCCEEDED) {
- this->codes_.emplace_back(std::move(thriftResult));
- }
- this->callingNum_--;
- if (this->callingNum_ == 0) {
- finished = true;
- }
- }
- if (finished) {
- this->onFinished();
- }
- });
+ this->kvstore_->asyncRemoveRange(
+ spaceId, partId, start, end, [spaceId, partId, this](kvstore::ResultCode code) {
+ handleAsync(spaceId, partId, code);
+ });
 }
 
 } // namespace storage