上篇：初识 Nebula Graph —— 向量类型支持¶

  Status InsertVerticesValidator::validateImpl() {
    spaceId_ = vctx_->whichSpace().id;
    NG_RETURN_IF_ERROR(check());
    NG_RETURN_IF_ERROR(prepareVertices());
    return Status::OK();
  }

  Status InsertVerticesValidator::toPlan() {
    auto doNode = InsertVertices::make(qctx_,
                                      nullptr,
                                      spaceId_,
                                      std::move(vertices_),
                                      std::move(tagPropNames_),
                                      ifNotExists_,
                                      ignoreExistedIndex_);
    root_ = doNode;
    tail_ = root_;
    return Status::OK();
  }

  folly::Future<Status> AsyncMsgNotifyBasedScheduler::scheduleExecutor(
  std::vector<folly::Future<Status>>&& futures, Executor* exe, folly::Executor* runner) const {
  switch (exe->node()->kind()) {
    // ...
    case PlanNode::Kind::kArgument: {
      return runExecutor(std::move(futures), exe, runner);
    }
    default: {
      if (exe->depends().empty()) {
        return runLeafExecutor(exe, runner);
      } else {
        return runExecutor(std::move(futures), exe, runner);
      }
    }
  }
  }
  folly::Future<Status> AsyncMsgNotifyBasedScheduler::runExecutor(
    std::vector<folly::Future<Status>>&& futures, Executor* exe, folly::Executor* runner) const {
  return folly::collect(futures).via(runner).thenValue(
      [exe, this](auto&& t) mutable -> folly::Future<Status> {
        NG_RETURN_IF_ERROR(checkStatus(std::move(t)));
        // Execute in current thread.
        return execute(exe);
      });
  }

  folly::Future<Status> AsyncMsgNotifyBasedScheduler::runLeafExecutor(Executor* exe,
                                                                    folly::Executor* runner) const {
    return std::move(execute(exe)).via(runner);
  }

template <typename RESP>
void BaseProcessor<RESP>::doPut(GraphSpaceID spaceId,
                                PartitionID partId,
                                std::vector<kvstore::KV>&& data) {
  this->env_->kvstore_->asyncMultiPut(
      spaceId, partId, std::move(data), [spaceId, partId, this](nebula::cpp2::ErrorCode code) {
        handleAsync(spaceId, partId, code);
      });
}

template <typename RESP>
void BaseProcessor<RESP>::handleAsync(GraphSpaceID spaceId,
                                    PartitionID partId,
                                    nebula::cpp2::ErrorCode code) {
  bool finished = false;
  {
    std::lock_guard<std::mutex> lg(this->lock_);
    handleErrorCode(code, spaceId, partId);
    this->callingNum_--;
    if (this->callingNum_ == 0) {
      finished = true;
    }
  }
  if (finished) {
    this->onFinished();
  }
}

virtual void onFinished() {
  memory::MemoryCheckOffGuard guard;
  this->result_.failed_parts_ref() = this->codes_;
  this->resp_.result_ref() = std::move(this->result_);
  this->promise_.setValue(std::move(this->resp_));
  delete this;
}

void RaftPart::appendLogsInternal(AppendLogsIterator iter, TermID termId) {
  do {
    // Process 1
  // Step 1: Write Local WAL
    {
      SCOPED_TIMER(
      if (!wal_->appendLogs(iter)) {
        break;
      }
    }
  } while (false);


  // Step 2: Replicate to followers
  auto* eb = ioThreadPool_->getEventBase();
  replicateLogs(eb, std::move(iter), currTerm, lastId, committed, prevLogTerm, prevLogId);
  return;
}

// process 2 & 3
void RaftPart::replicateLogs(folly::EventBase* eb,
                             AppendLogsIterator iter,
                             TermID currTerm,
                             LogID lastLogId,
                             LogID committedId,
                             TermID prevLogTerm,
                             LogID prevLogId) {
  collectNSucceeded(gen::from(hosts) |
                        gen::map([self = shared_from_this(),
                                  eb,
                                  currTerm,
                                  lastLogId,
                                  prevLogId,
                                  prevLogTerm,
                                  committedId](std::shared_ptr<Host> hostPtr) {
                          return via(eb, [=]() -> Future<cpp2::AppendLogResponse> {
                            // 向所有 follower 发送 AppendLogs 请求
                            return hostPtr->appendLogs(
                                eb, currTerm, lastLogId, committedId, prevLogTerm, prevLogId);
                          });
                        }) |
                        gen::as<std::vector>(),
                    quorum_,
                    [hosts](size_t index, cpp2::AppendLogResponse& resp) {
      // Process 2: 收集到了半数节点通过
                      return resp.get_error_code() == nebula::cpp2::ErrorCode::SUCCEEDED &&
                             !hosts[index]->isLearner();
                    })
      .via(executor_.get())
      .then([self = shared_from_this(),
             eb,
             it = std::move(iter),
             currTerm,
             lastLogId,
             committedId,
             prevLogId,
             prevLogTerm,
             pHosts = std::move(hosts),
             beforeAppendLogUs](folly::Try<AppendLogResponses>&& result) mutable {
        // Process 3: followers 真正开始进行写入
        self->processAppendLogResponses(*result,
                                        eb,
                                        std::move(it),
                                        currTerm,
                                        lastLogId,
                                        committedId,
                                        prevLogTerm,
                                        prevLogId,
                                        std::move(pHosts));
        return *result;
      });
}