Method UpdateTaskStatusRemote

src/nameserver/name_server_impl.cc:1680–1728

}

int NameServerImpl::UpdateTaskStatusRemote(bool is_recover_op) {
    if (mode_.load(std::memory_order_acquire) == kFOLLOWER) {
        return 0;
    }
    std::map<std::string, std::shared_ptr<::fedb::client::NsClient>> client_map;
    {
        std::lock_guard<std::mutex> lock(mu_);
        if (nsc_.empty()) {
            return 0;
        }
        for (auto iter = nsc_.begin(); iter != nsc_.end(); ++iter) {
            if (iter->second->state_.load(std::memory_order_relaxed) != kClusterHealthy) {
                PDLOG(INFO, "cluster[%s] is not Healthy", iter->first.c_str());
                continue;
            }
            client_map.insert(std::make_pair(
                iter->first, std::atomic_load_explicit(&iter->second->client_, std::memory_order_relaxed)));
        }
    }
    uint64_t last_task_rpc_version = task_rpc_version_.load(std::memory_order_acquire);
    for (auto iter = client_map.begin(); iter != client_map.end(); ++iter) {
        ::fedb::api::TaskStatusResponse response;
        // get task status from replica cluster
        if (iter->second->GetTaskStatus(response)) {
            std::lock_guard<std::mutex> lock(mu_);
            if (last_task_rpc_version != task_rpc_version_.load(std::memory_order_acquire)) {
                DEBUGLOG("task_rpc_version mismatch");
                break;
            }
            std::string endpoint = iter->first;
            uint32_t index = 0;
            for (const auto& op_list : task_vec_) {
                index++;
                if (index <= FLAGS_name_server_task_max_concurrency) {
                    continue;
                }
                std::string endpoint_role = "replica cluster";
                if (UpdateTask(op_list, endpoint, endpoint_role, is_recover_op, response) < 0) {
                    continue;
                }
            }
        } else {
            if (response.has_msg()) {
                PDLOG(WARNING, "get task status faild : [%s]", response.msg().c_str());
            }
        }
    }
    return 0;
}

int NameServerImpl::UpdateTask(const std::list<std::shared_ptr<OPData>>& op_list, const std::string& endpoint,
                               const std::string& msg, bool is_recover_op, ::fedb::api::TaskStatusResponse& response) {