| 927 | } |
| 928 | |
| 929 | Status ReplicationThread::parallelFetchFile(const std::string &dir, |
| 930 | const std::vector<std::pair<std::string, uint32_t>> &files) { |
| 931 | size_t concurrency = 1; |
| 932 | if (files.size() > 20) { |
| 933 | // Use 4 threads to download files in parallel |
| 934 | concurrency = 4; |
| 935 | } |
| 936 | std::atomic<uint32_t> fetch_cnt = {0}; |
| 937 | std::atomic<uint32_t> skip_cnt = {0}; |
| 938 | std::vector<std::future<Status>> results; |
| 939 | for (size_t tid = 0; tid < concurrency; ++tid) { |
| 940 | results.push_back( |
| 941 | std::async(std::launch::async, [this, dir, &files, tid, concurrency, &fetch_cnt, &skip_cnt]() -> Status { |
| 942 | if (this->stop_flag_) { |
| 943 | return {Status::NotOK, "replication thread was stopped"}; |
| 944 | } |
| 945 | ssl_st *ssl = nullptr; |
| 946 | #ifdef ENABLE_OPENSSL |
| 947 | if (this->srv_->GetConfig()->tls_replication) { |
| 948 | ssl = SSL_new(this->srv_->ssl_ctx.get()); |
| 949 | } |
| 950 | auto exit = MakeScopeExit([ssl] { SSL_free(ssl); }); |
| 951 | #endif |
| 952 | int sock_fd = GET_OR_RET(util::SockConnect(this->host_, this->port_, ssl, |
| 953 | this->srv_->GetConfig()->replication_connect_timeout_ms, |
| 954 | this->srv_->GetConfig()->replication_recv_timeout_ms) |
| 955 | .Prefixed("connect the server err")); |
| 956 | #ifdef ENABLE_OPENSSL |
| 957 | exit.Disable(); |
| 958 | #endif |
| 959 | UniqueFD unique_fd{sock_fd}; |
| 960 | auto s = this->sendAuth(sock_fd, ssl); |
| 961 | if (!s.IsOK()) { |
| 962 | return s.Prefixed("send the auth command err"); |
| 963 | } |
| 964 | std::vector<std::string> fetch_files; |
| 965 | std::vector<uint32_t> crcs; |
| 966 | for (auto f_idx = tid; f_idx < files.size(); f_idx += concurrency) { |
| 967 | if (this->stop_flag_) { |
| 968 | return {Status::NotOK, "replication thread was stopped"}; |
| 969 | } |
| 970 | const auto &f_name = files[f_idx].first; |
| 971 | const auto &f_crc = files[f_idx].second; |
| 972 | // Don't fetch existing files |
| 973 | if (engine::Storage::ReplDataManager::FileExists(this->storage_, dir, f_name, f_crc)) { |
| 974 | skip_cnt.fetch_add(1); |
| 975 | uint32_t cur_skip_cnt = skip_cnt.load(); |
| 976 | uint32_t cur_fetch_cnt = fetch_cnt.load(); |
| 977 | INFO("[skip] {} {}, skip count: {}, fetch count: {}, progress: {} / {}", f_name, f_crc, cur_skip_cnt, |
| 978 | cur_fetch_cnt, (cur_skip_cnt + cur_fetch_cnt), files.size()); |
| 979 | continue; |
| 980 | } |
| 981 | fetch_files.push_back(f_name); |
| 982 | crcs.push_back(f_crc); |
| 983 | } |
| 984 | unsigned files_count = files.size(); |
| 985 | FetchFileCallback fn = [&fetch_cnt, &skip_cnt, files_count](const std::string &fetch_file, |
| 986 | uint32_t fetch_crc) { |
nothing calls this directly
no test coverage detected