diff --git a/BUCK b/BUCK index b9e3d043a5..c1aa22ba50 100644 --- a/BUCK +++ b/BUCK @@ -306,6 +306,7 @@ cpp_library_wrapper(name="rocksdb_lib", srcs=[ "utilities/compaction_filters.cc", "utilities/compaction_filters/remove_emptyvalue_compactionfilter.cc", "utilities/convenience/info_log_finder.cc", + "utilities/copy_engine/copy_engine.cc", "utilities/counted_fs.cc", "utilities/debug.cc", "utilities/env_mirror.cc", diff --git a/CMakeLists.txt b/CMakeLists.txt index d901493b43..a276239f7d 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1023,6 +1023,7 @@ set(SOURCES util/xxhash.cc utilities/agg_merge/agg_merge.cc utilities/backup/backup_engine.cc + utilities/copy_engine/copy_engine.cc utilities/blob_db/blob_compaction_filter.cc utilities/blob_db/blob_db.cc utilities/blob_db/blob_db_impl.cc diff --git a/src.mk b/src.mk index c9d36518bc..9157056016 100644 --- a/src.mk +++ b/src.mk @@ -287,6 +287,7 @@ LIB_SOURCES = \ utilities/cassandra/format.cc \ utilities/cassandra/merge_operator.cc \ utilities/checkpoint/checkpoint_impl.cc \ + utilities/copy_engine/copy_engine.cc \ utilities/compaction_filters.cc \ utilities/sorted_run_builder/sorted_run_builder.cc \ utilities/compaction_filters/remove_emptyvalue_compactionfilter.cc \ diff --git a/util/file_checksum_helper.h b/util/file_checksum_helper.h index 6c3349ee52..2f7e7ceb27 100644 --- a/util/file_checksum_helper.h +++ b/util/file_checksum_helper.h @@ -14,6 +14,15 @@ namespace ROCKSDB_NAMESPACE { +// Converts a 32-bit checksum value (e.g. a CRC32C) to its hex string +// representation, matching the on-disk/backup-metadata encoding. Shared by the +// backup and copy engines. +inline std::string ChecksumInt32ToHex(uint32_t checksum_value) { + std::string checksum_str; + PutFixed32(&checksum_str, EndianSwapValue(checksum_value)); + return Slice(checksum_str).ToString(/*hex=*/true); +} + // This is the class to generate the file checksum based on Crc32. It // will be used as the default checksum method for SST file checksum class FileChecksumGenCrc32c : public FileChecksumGenerator { diff --git a/util/rate_limiter_impl.h b/util/rate_limiter_impl.h index a16ba0eeb4..04684836c0 100644 --- a/util/rate_limiter_impl.h +++ b/util/rate_limiter_impl.h @@ -157,4 +157,23 @@ class GenericRateLimiter : public RateLimiter { std::chrono::microseconds tuned_time_; }; +// Requests `total_bytes_to_request` bytes from `rate_limiter`, split into +// GetSingleBurstBytes()-sized chunks and blocking until all are granted. Shared +// by the backup and copy engines. +inline void LoopRateLimitRequestHelper(size_t total_bytes_to_request, + RateLimiter* rate_limiter, + Env::IOPriority pri, Statistics* stats, + RateLimiter::OpType op_type) { + assert(rate_limiter != nullptr); + size_t remaining_bytes = total_bytes_to_request; + size_t request_bytes = 0; + while (remaining_bytes > 0) { + request_bytes = + std::min(static_cast(rate_limiter->GetSingleBurstBytes()), + remaining_bytes); + rate_limiter->Request(request_bytes, pri, stats, op_type); + remaining_bytes -= request_bytes; + } +} + } // namespace ROCKSDB_NAMESPACE diff --git a/utilities/backup/backup_engine.cc b/utilities/backup/backup_engine.cc index 00275a01f4..1230a6ba3b 100644 --- a/utilities/backup/backup_engine.cc +++ b/utilities/backup/backup_engine.cc @@ -47,11 +47,13 @@ #include "util/channel.h" #include "util/coding.h" #include "util/crc32c.h" +#include "util/file_checksum_helper.h" #include "util/math.h" #include "util/rate_limiter_impl.h" #include "util/string_util.h" #include "utilities/backup/backup_engine_impl.h" #include "utilities/checkpoint/checkpoint_impl.h" +#include "utilities/copy_engine/copy_engine.h" namespace ROCKSDB_NAMESPACE { @@ -68,11 +70,6 @@ inline uint32_t ChecksumHexToInt32(const std::string& checksum_hex) { inline std::string ChecksumStrToHex(const std::string& checksum_str) { return Slice(checksum_str).ToString(true); } -inline std::string ChecksumInt32ToHex(const uint32_t& checksum_value) { - std::string checksum_str; - PutFixed32(&checksum_str, EndianSwapValue(checksum_value)); - return ChecksumStrToHex(checksum_str); -} const std::string kPrivateDirName = "private"; const std::string kMetaDirName = "meta"; @@ -249,15 +246,6 @@ class BackupEngineImpl { } }; - // TODO: deprecate this function once we migrate all BackupEngine's rate - // limiting to lower-level ones (i.e, ones in file access wrapper level like - // `WritableFileWriter`) - static void LoopRateLimitRequestHelper(const size_t total_bytes_to_request, - RateLimiter* rate_limiter, - const Env::IOPriority pri, - Statistics* stats, - const RateLimiter::OpType op_type); - static inline std::string WithoutTrailingSlash(const std::string& path) { if (path.empty() || path.back() != '/') { return path; @@ -587,34 +575,6 @@ class BackupEngineImpl { std::to_string(backup_id) + (tmp ? ".tmp" : ""); } - // If size_limit == 0, there is no size limit, copy everything. - // - // Exactly one of src and contents must be non-empty. - // - // @param src If non-empty, the file is copied from this pathname. - // @param contents If non-empty, the file will be created with these contents. - // @param src_temperature Pass in expected temperature of src, return back - // temperature reported by FileSystem - IOStatus CopyOrCreateFile(const std::string& src, const std::string& dst, - const std::string& contents, uint64_t size_limit, - Env* src_env, Env* dst_env, - const EnvOptions& src_env_options, bool sync, - RateLimiter* rate_limiter, - std::function progress_callback, - Temperature* src_temperature, - Temperature dst_temperature, - uint64_t* bytes_toward_next_callback, - uint64_t* size, std::string* checksum_hex); - - uint64_t CalculateIOBufferSize(RateLimiter* rate_limiter) const; - - IOStatus ReadFileAndComputeChecksum(const std::string& src, - const std::shared_ptr& src_fs, - const EnvOptions& src_env_options, - uint64_t size_limit, - std::string* checksum_hex, - const Temperature src_temperature) const; - // Helper method to check if backup should be stopped. Can be overridden // via sync points for testing. bool ShouldStopBackup() const; @@ -626,152 +586,6 @@ class BackupEngineImpl { std::string* db_id, std::string* db_session_id) const; - struct WorkItemResult { - WorkItemResult() - : size(0), - expected_src_temperature(Temperature::kUnknown), - current_src_temperature(Temperature::kUnknown) {} - - WorkItemResult(const WorkItemResult& other) = delete; - WorkItemResult& operator=(const WorkItemResult& other) = delete; - - WorkItemResult(WorkItemResult&& o) noexcept { *this = std::move(o); } - - WorkItemResult& operator=(WorkItemResult&& o) noexcept { - size = o.size; - checksum_hex = std::move(o.checksum_hex); - db_id = std::move(o.db_id); - db_session_id = std::move(o.db_session_id); - io_status = std::move(o.io_status); - expected_src_temperature = o.expected_src_temperature; - current_src_temperature = o.current_src_temperature; - return *this; - } - - ~WorkItemResult() { - // The Status needs to be ignored here for two reasons. - // First, if the BackupEngineImpl shuts down with jobs outstanding, then - // it is possible that the Status in the future/promise is never read, - // resulting in an unchecked Status. Second, if there are items in the - // channel when the BackupEngineImpl is shutdown, these will also have - // Status that have not been checked. This - // TODO: Fix those issues so that the Status - io_status.PermitUncheckedError(); - } - uint64_t size; - std::string checksum_hex; - std::string db_id; - std::string db_session_id; - IOStatus io_status; - Temperature expected_src_temperature = Temperature::kUnknown; - Temperature current_src_temperature = Temperature::kUnknown; - }; - - enum WorkItemType : uint64_t { - CopyOrCreate = 1U, - ComputeChecksum = 2U, - }; - - // Exactly one of src_path and contents must be non-empty. If src_path is - // non-empty, the file is copied from this pathname. Otherwise, if contents is - // non-empty, the file will be created at dst_path with these contents. - struct WorkItem { - std::string src_path; - std::string dst_path; - Temperature src_temperature; - Temperature dst_temperature; - std::string contents; - Env* src_env; - Env* dst_env; - EnvOptions src_env_options; - bool sync; - RateLimiter* rate_limiter; - uint64_t size_limit; - Statistics* stats; - std::promise result; - std::function progress_callback; - std::string src_checksum_func_name; - std::string src_checksum_hex; - std::string db_id; - std::string db_session_id; - WorkItemType type; - - WorkItem() - : src_temperature(Temperature::kUnknown), - dst_temperature(Temperature::kUnknown), - src_env(nullptr), - dst_env(nullptr), - src_env_options(), - sync(false), - rate_limiter(nullptr), - size_limit(0), - stats(nullptr), - src_checksum_func_name(kUnknownFileChecksumFuncName), - type(WorkItemType::CopyOrCreate) {} - - WorkItem(const WorkItem&) = delete; - WorkItem& operator=(const WorkItem&) = delete; - - WorkItem(WorkItem&& o) noexcept { *this = std::move(o); } - - WorkItem& operator=(WorkItem&& o) noexcept { - src_path = std::move(o.src_path); - dst_path = std::move(o.dst_path); - src_temperature = std::move(o.src_temperature); - dst_temperature = std::move(o.dst_temperature); - contents = std::move(o.contents); - src_env = o.src_env; - dst_env = o.dst_env; - src_env_options = std::move(o.src_env_options); - sync = o.sync; - rate_limiter = o.rate_limiter; - size_limit = o.size_limit; - stats = o.stats; - result = std::move(o.result); - progress_callback = std::move(o.progress_callback); - src_checksum_func_name = std::move(o.src_checksum_func_name); - src_checksum_hex = std::move(o.src_checksum_hex); - db_id = std::move(o.db_id); - db_session_id = std::move(o.db_session_id); - src_temperature = o.src_temperature; - type = std::move(o.type); - return *this; - } - - WorkItem(std::string _src_path, std::string _dst_path, - const Temperature _src_temperature, - const Temperature _dst_temperature, std::string _contents, - Env* _src_env, Env* _dst_env, EnvOptions _src_env_options, - bool _sync, RateLimiter* _rate_limiter, uint64_t _size_limit, - Statistics* _stats, std::function _progress_callback = {}, - const std::string& _src_checksum_func_name = - kUnknownFileChecksumFuncName, - const std::string& _src_checksum_hex = "", - const std::string& _db_id = "", - const std::string& _db_session_id = "", - WorkItemType _type = WorkItemType::CopyOrCreate) - : src_path(std::move(_src_path)), - dst_path(std::move(_dst_path)), - src_temperature(_src_temperature), - dst_temperature(_dst_temperature), - contents(std::move(_contents)), - src_env(_src_env), - dst_env(_dst_env), - src_env_options(std::move(_src_env_options)), - sync(_sync), - rate_limiter(_rate_limiter), - size_limit(_size_limit), - stats(_stats), - progress_callback(_progress_callback), - src_checksum_func_name(_src_checksum_func_name), - src_checksum_hex(_src_checksum_hex), - db_id(_db_id), - db_session_id(_db_session_id), - type(_type) {} - - ~WorkItem() = default; - }; - struct BackupAfterCopyOrCreateWorkItem { std::future result; bool shared; @@ -874,10 +688,7 @@ class BackupEngineImpl { }; bool initialized_; - std::mutex byte_report_mutex_; - mutable channel work_items_; - std::vector threads_; - std::atomic threads_cpu_priority_; + std::unique_ptr copy_engine_; // Certain operations like PurgeOldBackups and DeleteBackup will trigger // automatic GarbageCollect (true) unless we've already done one in this @@ -929,7 +740,6 @@ class BackupEngineImpl { std::unique_ptr meta_directory_; std::unique_ptr private_directory_; - static const size_t kDefaultCopyFileBufferSize = 5 * 1024 * 1024LL; // 5MB bool read_only_; BackupStatistics backup_statistics_; std::unordered_set reported_ignored_fields_; @@ -1106,7 +916,6 @@ namespace { BackupEngineImpl::BackupEngineImpl(const BackupEngineOptions& options, Env* db_env, bool read_only) : initialized_(false), - threads_cpu_priority_(), latest_backup_id_(0), latest_valid_backup_id_(0), stop_backup_(false), @@ -1129,10 +938,8 @@ BackupEngineImpl::BackupEngineImpl(const BackupEngineOptions& options, } BackupEngineImpl::~BackupEngineImpl() { - work_items_.sendEof(); - for (auto& t : threads_) { - t.join(); - } + // copy_engine_'s destructor drains and joins its threads. + copy_engine_.reset(); LogFlush(options_.info_log); for (const auto& it : corrupt_backups_) { it.second.first.PermitUncheckedError(); @@ -1328,98 +1135,20 @@ IOStatus BackupEngineImpl::Initialize() { ROCKS_LOG_INFO(options_.info_log, "Latest valid backup is %u", latest_valid_backup_id_); - // set up threads to perform file creation / copy or checksum computations - // from work_items_ in the background. - threads_cpu_priority_ = CpuPriority::kNormal; - threads_.reserve(options_.max_background_operations); - for (int t = 0; t < options_.max_background_operations; t++) { - threads_.emplace_back([this]() { -#if defined(_GNU_SOURCE) && defined(__GLIBC_PREREQ) -#if __GLIBC_PREREQ(2, 12) - pthread_setname_np(pthread_self(), "backup_engine"); -#endif -#endif - CpuPriority current_priority = CpuPriority::kNormal; - WorkItem work_item; - uint64_t bytes_toward_next_callback = 0; - while (work_items_.read(work_item)) { - CpuPriority priority = threads_cpu_priority_; - if (current_priority != priority) { - TEST_SYNC_POINT_CALLBACK( - "BackupEngineImpl::Initialize:SetCpuPriority", &priority); - port::SetCpuPriority(0, priority); - current_priority = priority; - } - // `bytes_read` and `bytes_written` stats are enabled based on - // compile-time support and cannot be dynamically toggled. So we do not - // need to worry about `PerfLevel` here, unlike many other - // `IOStatsContext` / `PerfContext` stats. - uint64_t prev_bytes_read = IOSTATS(bytes_read); - uint64_t prev_bytes_written = IOSTATS(bytes_written); + // Create the shared CopyEngine pool now that options_ (incl. rate limiters) + // is finalized. + CopyEngineOptions copy_options; + copy_options.max_background_operations = options_.max_background_operations; + copy_options.io_buffer_size = options_.io_buffer_size; + copy_options.callback_trigger_interval_size = + options_.callback_trigger_interval_size; + copy_options.checksum_rate_limiter = options_.backup_rate_limiter.get(); + copy_options.info_log = options_.info_log; + copy_options.thread_name = "backup_engine"; + copy_options.should_abort = [this]() { return ShouldStopBackup(); }; + copy_options.abort_status_message = "Backup stopped"; + copy_engine_ = std::make_unique(std::move(copy_options)); - WorkItemResult result; - if (work_item.type == WorkItemType::CopyOrCreate) { - Temperature temp = work_item.src_temperature; - result.io_status = CopyOrCreateFile( - work_item.src_path, work_item.dst_path, work_item.contents, - work_item.size_limit, work_item.src_env, work_item.dst_env, - work_item.src_env_options, work_item.sync, work_item.rate_limiter, - work_item.progress_callback, &temp, work_item.dst_temperature, - &bytes_toward_next_callback, &result.size, &result.checksum_hex); - - RecordTick(work_item.stats, BACKUP_READ_BYTES, - IOSTATS(bytes_read) - prev_bytes_read); - RecordTick(work_item.stats, BACKUP_WRITE_BYTES, - IOSTATS(bytes_written) - prev_bytes_written); - - result.db_id = work_item.db_id; - result.db_session_id = work_item.db_session_id; - result.expected_src_temperature = work_item.src_temperature; - result.current_src_temperature = temp; - if (result.io_status.ok() && !work_item.src_checksum_hex.empty()) { - // unknown checksum function name implies no db table file checksum - // in db manifest; work_item.src_checksum_hex not empty means backup - // engine has calculated its crc32c checksum for the table file; - // therefore, we are able to compare the checksums. - if (work_item.src_checksum_func_name == - kUnknownFileChecksumFuncName || - work_item.src_checksum_func_name == kDbFileChecksumFuncName) { - if (work_item.src_checksum_hex != result.checksum_hex) { - std::string checksum_info( - "Expected checksum is " + work_item.src_checksum_hex + - " while computed checksum is " + result.checksum_hex); - result.io_status = IOStatus::Corruption( - "Checksum mismatch after copying to " + work_item.dst_path + - ": " + checksum_info); - } - } else { - // FIXME(peterd): dead code? - std::string checksum_function_info( - "Existing checksum function is " + - work_item.src_checksum_func_name + - " while provided checksum function is " + - kBackupFileChecksumFuncName); - ROCKS_LOG_INFO( - options_.info_log, - "Unable to verify checksum after copying to %s: %s\n", - work_item.dst_path.c_str(), checksum_function_info.c_str()); - } - } - } else if (work_item.type == ComputeChecksum) { - result.io_status = ReadFileAndComputeChecksum( - work_item.src_path, work_item.src_env->GetFileSystem(), - work_item.src_env_options, work_item.size_limit, - &result.checksum_hex, work_item.src_temperature); - result.db_id = work_item.db_id; - result.db_session_id = work_item.db_session_id; - } else { - result.io_status = IOStatus::InvalidArgument( - "Unknown work item type: " + std::to_string(work_item.type)); - } - work_item.result.set_value(std::move(result)); - } - }); - } ROCKS_LOG_INFO(options_.info_log, "Initialized BackupEngine"); return IOStatus::OK(); } @@ -1443,9 +1172,8 @@ IOStatus BackupEngineImpl::CreateNewBackupWithMetadata( } if (options.decrease_background_thread_cpu_priority) { - if (options.background_thread_cpu_priority < threads_cpu_priority_) { - threads_cpu_priority_.store(options.background_thread_cpu_priority); - } + copy_engine_->MaybeDecreaseCpuPriority( + options.background_thread_cpu_priority); } BackupID new_backup_id = latest_backup_id_ + 1; @@ -1637,7 +1365,7 @@ IOStatus BackupEngineImpl::CreateNewBackupWithMetadata( if (maybe_exclude_files[i].exclude_decision) { new_backup.get()->AddExcludedFile(e.second.dst_relative); } else { - work_items_.write(std::move(e.first)); + copy_engine_->Submit(std::move(e.first)); backup_items_to_finish.push_back(std::move(e.second)); } } @@ -2154,7 +1882,7 @@ IOStatus BackupEngineImpl::RestoreDBFromBackup( RestoreAfterCopyOrCreateWorkItem after_copy_or_create_work_item( copy_or_create_work_item.result.get_future(), file, dst, file_info->checksum_hex); - work_items_.write(std::move(copy_or_create_work_item)); + copy_engine_->Submit(std::move(copy_or_create_work_item)); restore_items_to_finish.push_back( std::move(after_copy_or_create_work_item)); } @@ -2290,7 +2018,7 @@ IOStatus BackupEngineImpl::VerifyBackup(BackupID backup_id, ROCKS_LOG_INFO(options_.info_log, "Scheduling checksum evaluation for %s...\n", abs_path.c_str()); - work_items_.write(std::move(backup_file_work_item)); + copy_engine_->Submit(std::move(backup_file_work_item)); backup_verification_checksum_work_items.push_back( std::move(backup_file_checksum_work_item)); } @@ -2355,162 +2083,6 @@ IOStatus BackupEngineImpl::VerifyBackup(BackupID backup_id, return io_s; } -IOStatus BackupEngineImpl::CopyOrCreateFile( - const std::string& src, const std::string& dst, const std::string& contents, - uint64_t size_limit, Env* src_env, Env* dst_env, - const EnvOptions& src_env_options, bool sync, RateLimiter* rate_limiter, - std::function progress_callback, Temperature* src_temperature, - Temperature dst_temperature, uint64_t* bytes_toward_next_callback, - uint64_t* size, std::string* checksum_hex) { - assert(src.empty() != contents.empty()); - if (ShouldStopBackup()) { - return status_to_io_status(Status::Incomplete("Backup stopped")); - } - - IOStatus io_s; - std::unique_ptr dst_file; - std::unique_ptr src_file; - FileOptions dst_file_options; - dst_file_options.use_mmap_writes = false; - dst_file_options.temperature = dst_temperature; - // TODO:(gzh) maybe use direct reads/writes here if possible - if (size != nullptr) { - *size = 0; - } - uint32_t checksum_value = 0; - - // Check if size limit is set. if not, set it to very big number - if (size_limit == 0) { - size_limit = std::numeric_limits::max(); - } - - io_s = dst_env->GetFileSystem()->NewWritableFile(dst, dst_file_options, - &dst_file, nullptr); - if (!io_s.ok()) { - return io_s; - } - - if (!src.empty()) { - auto src_file_options = FileOptions(src_env_options); - src_file_options.temperature = *src_temperature; - io_s = src_env->GetFileSystem()->NewSequentialFile(src, src_file_options, - &src_file, nullptr); - } - if (io_s.IsPathNotFound() && *src_temperature != Temperature::kUnknown) { - // Retry without temperature hint in case the FileSystem is strict with - // non-kUnknown temperature option - io_s = src_env->GetFileSystem()->NewSequentialFile( - src, FileOptions(src_env_options), &src_file, nullptr); - } - if (!io_s.ok()) { - return io_s; - } - - size_t buf_size = CalculateIOBufferSize(rate_limiter); - TEST_SYNC_POINT_CALLBACK( - "BackupEngineImpl::CopyOrCreateFile:CalculateIOBufferSize", &buf_size); - - // TODO: pass in Histograms if the destination file is sst or blob - std::unique_ptr dest_writer( - new WritableFileWriter(std::move(dst_file), dst, dst_file_options)); - std::unique_ptr src_reader; - std::unique_ptr buf; - if (!src.empty()) { - // Return back current temperature in FileSystem - *src_temperature = src_file->GetTemperature(); - - src_reader.reset(new SequentialFileReader( - std::move(src_file), src, nullptr /* io_tracer */, {}, rate_limiter)); - buf.reset(new char[buf_size]); - } - - Slice data; - const IOOptions opts; - do { - if (ShouldStopBackup()) { - return status_to_io_status(Status::Incomplete("Backup stopped")); - } - if (!src.empty()) { - size_t buffer_to_read = - (buf_size < size_limit) ? buf_size : static_cast(size_limit); - io_s = src_reader->Read(buffer_to_read, &data, buf.get(), - Env::IO_LOW /* rate_limiter_priority */); - *bytes_toward_next_callback += data.size(); - } else { - data = contents; - } - size_limit -= data.size(); - TEST_SYNC_POINT_CALLBACK( - "BackupEngineImpl::CopyOrCreateFile:CorruptionDuringBackup", - (src.length() > 4 && src.rfind(".sst") == src.length() - 4) ? &data - : nullptr); - - if (!io_s.ok()) { - return io_s; - } - - if (size != nullptr) { - *size += data.size(); - } - if (checksum_hex != nullptr) { - checksum_value = crc32c::Extend(checksum_value, data.data(), data.size()); - } - - io_s = dest_writer->Append(opts, data); - - if (rate_limiter != nullptr) { - if (!src.empty()) { - rate_limiter->Request(data.size(), Env::IO_LOW, nullptr /* stats */, - RateLimiter::OpType::kWrite); - } else { - LoopRateLimitRequestHelper(data.size(), rate_limiter, Env::IO_LOW, - nullptr /* stats */, - RateLimiter::OpType::kWrite); - } - } - while (*bytes_toward_next_callback >= - options_.callback_trigger_interval_size) { - *bytes_toward_next_callback -= options_.callback_trigger_interval_size; - if (progress_callback) { - std::lock_guard lock(byte_report_mutex_); - try { - progress_callback(); - } catch (const std::exception& exn) { - io_s = IOStatus::Aborted("Exception in progress_callback: " + - std::string(exn.what())); - break; - } catch (...) { - io_s = IOStatus::Aborted("Unknown exception in progress_callback"); - break; - } - } - } - } while (io_s.ok() && contents.empty() && data.size() > 0 && size_limit > 0); - - // Convert uint32_t checksum to hex checksum - if (checksum_hex != nullptr) { - checksum_hex->assign(ChecksumInt32ToHex(checksum_value)); - } - - if (io_s.ok() && sync) { - io_s = dest_writer->Sync(opts, false); - } - if (io_s.ok()) { - io_s = dest_writer->Close(opts); - } - return io_s; -} - -uint64_t BackupEngineImpl::CalculateIOBufferSize( - RateLimiter* rate_limiter) const { - if (options_.io_buffer_size > 0) { - return options_.io_buffer_size; - } - return rate_limiter != nullptr - ? static_cast(rate_limiter->GetSingleBurstBytes()) - : kDefaultCopyFileBufferSize; -} - // fname will always start with "/" IOStatus BackupEngineImpl::AddBackupFileWorkItem( std::unordered_set& live_dst_paths, @@ -2577,7 +2149,7 @@ IOStatus BackupEngineImpl::AddBackupFileWorkItem( // since the session id should suffice to avoid file name collision in // the shared_checksum directory. if (checksum_hex.empty() && db_session_id.empty()) { - IOStatus io_s = ReadFileAndComputeChecksum( + IOStatus io_s = copy_engine_->ReadFileAndComputeChecksum( src_path, db_fs_, src_env_options, size_limit, &checksum_hex, src_temperature); if (!io_s.ok()) { @@ -2697,7 +2269,7 @@ IOStatus BackupEngineImpl::AddBackupFileWorkItem( // ID, but even in that case, we double check the file sizes in // BackupMeta::AddFile. } else { - IOStatus io_s = ReadFileAndComputeChecksum( + IOStatus io_s = copy_engine_->ReadFileAndComputeChecksum( src_path, db_fs_, src_env_options, size_limit, &checksum_hex, src_temperature); if (!io_s.ok()) { @@ -2742,7 +2314,7 @@ IOStatus BackupEngineImpl::AddBackupFileWorkItem( // the checkpoint ROCKS_LOG_INFO(options_.info_log, "Copying %s to %s", fname.c_str(), copy_dest_path->c_str()); - work_items_.write(std::move(copy_or_create_work_item)); + copy_engine_->Submit(std::move(copy_or_create_work_item)); backup_items_to_finish.push_back( std::move(after_copy_or_create_work_item)); } @@ -2769,63 +2341,6 @@ bool BackupEngineImpl::ShouldStopBackup() const { return should_stop; } -IOStatus BackupEngineImpl::ReadFileAndComputeChecksum( - const std::string& src, const std::shared_ptr& src_fs, - const EnvOptions& src_env_options, uint64_t size_limit, - std::string* checksum_hex, const Temperature src_temperature) const { - if (checksum_hex == nullptr) { - return status_to_io_status(Status::Aborted("Checksum pointer is null")); - } - if (ShouldStopBackup()) { - return status_to_io_status(Status::Incomplete("Backup stopped")); - } - uint32_t checksum_value = 0; - if (size_limit == 0) { - size_limit = std::numeric_limits::max(); - } - - std::unique_ptr src_reader; - auto file_options = FileOptions(src_env_options); - file_options.temperature = src_temperature; - RateLimiter* rate_limiter = options_.backup_rate_limiter.get(); - IOStatus io_s = SequentialFileReader::Create( - src_fs, src, file_options, &src_reader, nullptr /* dbg */, rate_limiter); - if (io_s.IsPathNotFound() && src_temperature != Temperature::kUnknown) { - // Retry without temperature hint in case the FileSystem is strict with - // non-kUnknown temperature option - file_options.temperature = Temperature::kUnknown; - io_s = SequentialFileReader::Create(src_fs, src, file_options, &src_reader, - nullptr /* dbg */, rate_limiter); - } - if (!io_s.ok()) { - return io_s; - } - - size_t buf_size = CalculateIOBufferSize(rate_limiter); - std::unique_ptr buf(new char[buf_size]); - Slice data; - - do { - if (ShouldStopBackup()) { - return status_to_io_status(Status::Incomplete("Backup stopped")); - } - size_t buffer_to_read = - (buf_size < size_limit) ? buf_size : static_cast(size_limit); - io_s = src_reader->Read(buffer_to_read, &data, buf.get(), - Env::IO_LOW /* rate_limiter_priority */); - if (!io_s.ok()) { - return io_s; - } - - size_limit -= data.size(); - checksum_value = crc32c::Extend(checksum_value, data.data(), data.size()); - } while (data.size() > 0 && size_limit > 0); - - checksum_hex->assign(ChecksumInt32ToHex(checksum_value)); - - return io_s; -} - Status BackupEngineImpl::GetFileDbIdentities(Env* src_env, const EnvOptions& src_env_options, const std::string& file_path, @@ -2891,22 +2406,6 @@ Status BackupEngineImpl::GetFileDbIdentities(Env* src_env, } } -void BackupEngineImpl::LoopRateLimitRequestHelper( - const size_t total_bytes_to_request, RateLimiter* rate_limiter, - const Env::IOPriority pri, Statistics* stats, - const RateLimiter::OpType op_type) { - assert(rate_limiter != nullptr); - size_t remaining_bytes = total_bytes_to_request; - size_t request_bytes = 0; - while (remaining_bytes > 0) { - request_bytes = - std::min(static_cast(rate_limiter->GetSingleBurstBytes()), - remaining_bytes); - rate_limiter->Request(request_bytes, pri, stats, op_type); - remaining_bytes -= request_bytes; - } -} - void BackupEngineImpl::InferDBFilesToRetainInRestore( const std::vector>& restore_file_infos, @@ -3071,7 +2570,7 @@ void BackupEngineImpl::InferDBFilesToRetainInRestore( backup_file_work_item.result.get_future(), backup_file_info->filename, number); - work_items_.write(std::move(backup_file_work_item)); + copy_engine_->Submit(std::move(backup_file_work_item)); backup_files_compute_checksum_work_items.push_back( std::move(backup_file_checksum_work_item)); @@ -3103,7 +2602,7 @@ void BackupEngineImpl::InferDBFilesToRetainInRestore( ComputeChecksumWorkItem db_file_checksum_work_item( db_file_work_item.result.get_future(), db_file_path, number); - work_items_.write(std::move(db_file_work_item)); + copy_engine_->Submit(std::move(db_file_work_item)); db_files_compute_checksum_work_items.push_back( std::move(db_file_checksum_work_item)); diff --git a/utilities/copy_engine/copy_engine.cc b/utilities/copy_engine/copy_engine.cc new file mode 100644 index 0000000000..648f0c8db4 --- /dev/null +++ b/utilities/copy_engine/copy_engine.cc @@ -0,0 +1,355 @@ +// Copyright (c) Meta Platforms, Inc. and affiliates. +// This source code is licensed under both the GPLv2 (found in the +// COPYING file in the root directory) and Apache 2.0 License +// (found in the LICENSE.Apache file in the root directory). + +#include "utilities/copy_engine/copy_engine.h" + +#include +#include +#include +#include + +#include "file/sequence_file_reader.h" +#include "file/writable_file_writer.h" +#include "logging/logging.h" +#include "monitoring/iostats_context_imp.h" +#include "rocksdb/file_system.h" +#include "rocksdb/slice.h" +#include "rocksdb/utilities/backup_engine.h" +#include "test_util/sync_point.h" +#include "util/crc32c.h" +#include "util/file_checksum_helper.h" +#include "util/rate_limiter_impl.h" + +namespace ROCKSDB_NAMESPACE { + +CopyEngine::CopyEngine(CopyEngineOptions options) + : options_(std::move(options)), + threads_cpu_priority_(CpuPriority::kNormal) { + threads_.reserve(options_.max_background_operations); + for (int t = 0; t < options_.max_background_operations; t++) { + threads_.emplace_back([this]() { ThreadBody(); }); + } +} + +CopyEngine::~CopyEngine() { + work_items_.sendEof(); + for (auto& t : threads_) { + t.join(); + } +} + +void CopyEngine::Submit(WorkItem&& item) { work_items_.write(std::move(item)); } + +void CopyEngine::MaybeDecreaseCpuPriority(CpuPriority priority) { + if (priority < threads_cpu_priority_) { + threads_cpu_priority_.store(priority); + } +} + +void CopyEngine::ThreadBody() { +#if defined(_GNU_SOURCE) && defined(__GLIBC_PREREQ) +#if __GLIBC_PREREQ(2, 12) + pthread_setname_np(pthread_self(), options_.thread_name.c_str()); +#endif +#endif + CpuPriority current_priority = CpuPriority::kNormal; + WorkItem work_item; + uint64_t bytes_toward_next_callback = 0; + while (work_items_.read(work_item)) { + CpuPriority priority = threads_cpu_priority_; + if (current_priority != priority) { + TEST_SYNC_POINT_CALLBACK("BackupEngineImpl::Initialize:SetCpuPriority", + &priority); + port::SetCpuPriority(0, priority); + current_priority = priority; + } + // `bytes_read` and `bytes_written` stats are enabled based on compile-time + // support and cannot be dynamically toggled. So we do not need to worry + // about `PerfLevel` here, unlike many other `IOStatsContext` / + // `PerfContext` stats. + uint64_t prev_bytes_read = IOSTATS(bytes_read); + uint64_t prev_bytes_written = IOSTATS(bytes_written); + + WorkItemResult result; + if (work_item.type == WorkItemType::CopyOrCreate) { + Temperature temp = work_item.src_temperature; + result.io_status = CopyOrCreateFile( + work_item.src_path, work_item.dst_path, work_item.contents, + work_item.size_limit, work_item.src_env, work_item.dst_env, + work_item.src_env_options, work_item.sync, work_item.rate_limiter, + work_item.progress_callback, &temp, work_item.dst_temperature, + &bytes_toward_next_callback, &result.size, &result.checksum_hex); + + RecordTick(work_item.stats, options_.read_bytes_ticker, + IOSTATS(bytes_read) - prev_bytes_read); + RecordTick(work_item.stats, options_.write_bytes_ticker, + IOSTATS(bytes_written) - prev_bytes_written); + + result.db_id = work_item.db_id; + result.db_session_id = work_item.db_session_id; + result.expected_src_temperature = work_item.src_temperature; + result.current_src_temperature = temp; + if (result.io_status.ok() && !work_item.src_checksum_hex.empty()) { + // unknown checksum function name implies no db table file checksum in + // db manifest; work_item.src_checksum_hex not empty means backup engine + // has calculated its crc32c checksum for the table file; therefore, we + // are able to compare the checksums. + if (work_item.src_checksum_func_name == kUnknownFileChecksumFuncName || + work_item.src_checksum_func_name == kDbFileChecksumFuncName) { + if (work_item.src_checksum_hex != result.checksum_hex) { + std::string checksum_info( + "Expected checksum is " + work_item.src_checksum_hex + + " while computed checksum is " + result.checksum_hex); + result.io_status = + IOStatus::Corruption("Checksum mismatch after copying to " + + work_item.dst_path + ": " + checksum_info); + } + } else { + // FIXME(peterd): dead code? + std::string checksum_function_info( + "Existing checksum function is " + + work_item.src_checksum_func_name + + " while provided checksum function is " + + kBackupFileChecksumFuncName); + ROCKS_LOG_INFO(options_.info_log, + "Unable to verify checksum after copying to %s: %s\n", + work_item.dst_path.c_str(), + checksum_function_info.c_str()); + } + } + } else if (work_item.type == ComputeChecksum) { + result.io_status = ReadFileAndComputeChecksum( + work_item.src_path, work_item.src_env->GetFileSystem(), + work_item.src_env_options, work_item.size_limit, &result.checksum_hex, + work_item.src_temperature); + result.db_id = work_item.db_id; + result.db_session_id = work_item.db_session_id; + } else { + result.io_status = IOStatus::InvalidArgument( + "Unknown work item type: " + std::to_string(work_item.type)); + } + work_item.result.set_value(std::move(result)); + } +} + +IOStatus CopyEngine::CopyOrCreateFile( + const std::string& src, const std::string& dst, const std::string& contents, + uint64_t size_limit, Env* src_env, Env* dst_env, + const EnvOptions& src_env_options, bool sync, RateLimiter* rate_limiter, + const std::function& progress_callback, + Temperature* src_temperature, Temperature dst_temperature, + uint64_t* bytes_toward_next_callback, uint64_t* size, + std::string* checksum_hex) { + assert(src.empty() != contents.empty()); + if (ShouldAbort()) { + return status_to_io_status( + Status::Incomplete(options_.abort_status_message)); + } + + IOStatus io_s; + std::unique_ptr dst_file; + std::unique_ptr src_file; + FileOptions dst_file_options; + dst_file_options.use_mmap_writes = false; + dst_file_options.temperature = dst_temperature; + // TODO:(gzh) maybe use direct reads/writes here if possible + if (size != nullptr) { + *size = 0; + } + uint32_t checksum_value = 0; + + // Check if size limit is set. if not, set it to very big number + if (size_limit == 0) { + size_limit = std::numeric_limits::max(); + } + + io_s = dst_env->GetFileSystem()->NewWritableFile(dst, dst_file_options, + &dst_file, nullptr); + if (!io_s.ok()) { + return io_s; + } + + auto src_file_options = FileOptions(src_env_options); + if (!src.empty()) { + src_file_options.temperature = *src_temperature; + io_s = src_env->GetFileSystem()->NewSequentialFile(src, src_file_options, + &src_file, nullptr); + } + if (io_s.IsPathNotFound() && *src_temperature != Temperature::kUnknown) { + // Retry without temperature hint in case the FileSystem is strict with + // non-kUnknown temperature option + src_file_options.temperature = Temperature::kUnknown; + io_s = src_env->GetFileSystem()->NewSequentialFile(src, src_file_options, + &src_file, nullptr); + } + if (!io_s.ok()) { + return io_s; + } + + size_t buf_size = CalculateIOBufferSize(rate_limiter); + TEST_SYNC_POINT_CALLBACK( + "BackupEngineImpl::CopyOrCreateFile:CalculateIOBufferSize", &buf_size); + + // TODO: pass in Histograms if the destination file is sst or blob + std::unique_ptr dest_writer( + new WritableFileWriter(std::move(dst_file), dst, dst_file_options)); + std::unique_ptr src_reader; + std::unique_ptr buf; + if (!src.empty()) { + // Return back current temperature in FileSystem + *src_temperature = src_file->GetTemperature(); + + src_reader.reset(new SequentialFileReader( + std::move(src_file), src, nullptr /* io_tracer */, {}, rate_limiter)); + buf.reset(new char[buf_size]); + } + + Slice data; + const IOOptions opts; + do { + if (ShouldAbort()) { + return status_to_io_status( + Status::Incomplete(options_.abort_status_message)); + } + if (!src.empty()) { + size_t buffer_to_read = + (buf_size < size_limit) ? buf_size : static_cast(size_limit); + io_s = src_reader->Read(buffer_to_read, &data, buf.get(), + Env::IO_LOW /* rate_limiter_priority */); + *bytes_toward_next_callback += data.size(); + } else { + data = contents; + } + size_limit -= data.size(); + TEST_SYNC_POINT_CALLBACK( + "BackupEngineImpl::CopyOrCreateFile:CorruptionDuringBackup", + (src.length() > 4 && src.rfind(".sst") == src.length() - 4) ? &data + : nullptr); + + if (!io_s.ok()) { + return io_s; + } + + if (size != nullptr) { + *size += data.size(); + } + if (checksum_hex != nullptr) { + checksum_value = crc32c::Extend(checksum_value, data.data(), data.size()); + } + + io_s = dest_writer->Append(opts, data); + + if (rate_limiter != nullptr) { + if (!src.empty()) { + rate_limiter->Request(data.size(), Env::IO_LOW, nullptr /* stats */, + RateLimiter::OpType::kWrite); + } else { + LoopRateLimitRequestHelper(data.size(), rate_limiter, Env::IO_LOW, + nullptr /* stats */, + RateLimiter::OpType::kWrite); + } + } + while (*bytes_toward_next_callback >= + options_.callback_trigger_interval_size) { + *bytes_toward_next_callback -= options_.callback_trigger_interval_size; + if (progress_callback) { + std::lock_guard lock(byte_report_mutex_); + try { + progress_callback(); + } catch (const std::exception& exn) { + io_s = IOStatus::Aborted("Exception in progress_callback: " + + std::string(exn.what())); + break; + } catch (...) { + io_s = IOStatus::Aborted("Unknown exception in progress_callback"); + break; + } + } + } + } while (io_s.ok() && contents.empty() && data.size() > 0 && size_limit > 0); + + // Convert uint32_t checksum to hex checksum + if (checksum_hex != nullptr) { + checksum_hex->assign(ChecksumInt32ToHex(checksum_value)); + } + + if (io_s.ok() && sync) { + io_s = dest_writer->Sync(opts, false); + } + if (io_s.ok()) { + io_s = dest_writer->Close(opts); + } + return io_s; +} + +uint64_t CopyEngine::CalculateIOBufferSize(RateLimiter* rate_limiter) const { + if (options_.io_buffer_size > 0) { + return options_.io_buffer_size; + } + return rate_limiter != nullptr + ? static_cast(rate_limiter->GetSingleBurstBytes()) + : kDefaultCopyFileBufferSize; +} + +IOStatus CopyEngine::ReadFileAndComputeChecksum( + const std::string& src, const std::shared_ptr& src_fs, + const EnvOptions& src_env_options, uint64_t size_limit, + std::string* checksum_hex, const Temperature src_temperature) const { + if (checksum_hex == nullptr) { + return status_to_io_status(Status::Aborted("Checksum pointer is null")); + } + if (ShouldAbort()) { + return status_to_io_status( + Status::Incomplete(options_.abort_status_message)); + } + uint32_t checksum_value = 0; + if (size_limit == 0) { + size_limit = std::numeric_limits::max(); + } + + std::unique_ptr src_reader; + auto file_options = FileOptions(src_env_options); + file_options.temperature = src_temperature; + RateLimiter* rate_limiter = options_.checksum_rate_limiter; + IOStatus io_s = SequentialFileReader::Create( + src_fs, src, file_options, &src_reader, nullptr /* dbg */, rate_limiter); + if (io_s.IsPathNotFound() && src_temperature != Temperature::kUnknown) { + // Retry without temperature hint in case the FileSystem is strict with + // non-kUnknown temperature option + file_options.temperature = Temperature::kUnknown; + io_s = SequentialFileReader::Create(src_fs, src, file_options, &src_reader, + nullptr /* dbg */, rate_limiter); + } + if (!io_s.ok()) { + return io_s; + } + + size_t buf_size = CalculateIOBufferSize(rate_limiter); + std::unique_ptr buf(new char[buf_size]); + Slice data; + + do { + if (ShouldAbort()) { + return status_to_io_status( + Status::Incomplete(options_.abort_status_message)); + } + size_t buffer_to_read = + (buf_size < size_limit) ? buf_size : static_cast(size_limit); + io_s = src_reader->Read(buffer_to_read, &data, buf.get(), + Env::IO_LOW /* rate_limiter_priority */); + if (!io_s.ok()) { + return io_s; + } + + size_limit -= data.size(); + checksum_value = crc32c::Extend(checksum_value, data.data(), data.size()); + } while (data.size() > 0 && size_limit > 0); + + checksum_hex->assign(ChecksumInt32ToHex(checksum_value)); + + return io_s; +} + +} // namespace ROCKSDB_NAMESPACE diff --git a/utilities/copy_engine/copy_engine.h b/utilities/copy_engine/copy_engine.h new file mode 100644 index 0000000000..e226e347bb --- /dev/null +++ b/utilities/copy_engine/copy_engine.h @@ -0,0 +1,234 @@ +// Copyright (c) Meta Platforms, Inc. and affiliates. +// This source code is licensed under both the GPLv2 (found in the +// COPYING file in the root directory) and Apache 2.0 License +// (found in the LICENSE.Apache file in the root directory). + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "port/port.h" +#include "rocksdb/env.h" +#include "rocksdb/file_checksum.h" +#include "rocksdb/io_status.h" +#include "rocksdb/rate_limiter.h" +#include "rocksdb/rocksdb_namespace.h" +#include "rocksdb/statistics.h" +#include "util/channel.h" + +namespace ROCKSDB_NAMESPACE { + +class FileSystem; + +struct WorkItemResult { + WorkItemResult() = default; + + WorkItemResult(const WorkItemResult& other) = delete; + WorkItemResult& operator=(const WorkItemResult& other) = delete; + + WorkItemResult(WorkItemResult&& o) noexcept { *this = std::move(o); } + + WorkItemResult& operator=(WorkItemResult&& o) noexcept { + size = o.size; + checksum_hex = std::move(o.checksum_hex); + db_id = std::move(o.db_id); + db_session_id = std::move(o.db_session_id); + io_status = std::move(o.io_status); + expected_src_temperature = o.expected_src_temperature; + current_src_temperature = o.current_src_temperature; + return *this; + } + + ~WorkItemResult() { + // The Status needs to be ignored here for two reasons. + // First, if the CopyEngine shuts down with jobs outstanding, then + // it is possible that the Status in the future/promise is never read, + // resulting in an unchecked Status. Second, if there are items in the + // channel when the CopyEngine is shutdown, these will also have + // Status that have not been checked. This + // TODO: Fix those issues so that the Status + io_status.PermitUncheckedError(); + } + uint64_t size = 0; + std::string checksum_hex; + std::string db_id; + std::string db_session_id; + IOStatus io_status; + Temperature expected_src_temperature = Temperature::kUnknown; + Temperature current_src_temperature = Temperature::kUnknown; +}; + +enum WorkItemType : uint64_t { + CopyOrCreate = 1U, + ComputeChecksum = 2U, +}; + +// Exactly one of src_path and contents must be non-empty. If src_path is +// non-empty, the file is copied from this pathname. Otherwise, if contents is +// non-empty, the file will be created at dst_path with these contents. +struct WorkItem { + std::string src_path; + std::string dst_path; + Temperature src_temperature = Temperature::kUnknown; + Temperature dst_temperature = Temperature::kUnknown; + std::string contents; + Env* src_env = nullptr; + Env* dst_env = nullptr; + EnvOptions src_env_options; + bool sync = false; + RateLimiter* rate_limiter = nullptr; + uint64_t size_limit = 0; + Statistics* stats = nullptr; + std::promise result; + std::function progress_callback; + std::string src_checksum_func_name = kUnknownFileChecksumFuncName; + std::string src_checksum_hex; + std::string db_id; + std::string db_session_id; + WorkItemType type = WorkItemType::CopyOrCreate; + + WorkItem() = default; + + WorkItem(const WorkItem&) = delete; + WorkItem& operator=(const WorkItem&) = delete; + + WorkItem(WorkItem&& o) noexcept { *this = std::move(o); } + + WorkItem& operator=(WorkItem&& o) noexcept { + src_path = std::move(o.src_path); + dst_path = std::move(o.dst_path); + src_temperature = o.src_temperature; + dst_temperature = o.dst_temperature; + contents = std::move(o.contents); + src_env = o.src_env; + dst_env = o.dst_env; + src_env_options = std::move(o.src_env_options); + sync = o.sync; + rate_limiter = o.rate_limiter; + size_limit = o.size_limit; + stats = o.stats; + result = std::move(o.result); + progress_callback = std::move(o.progress_callback); + src_checksum_func_name = std::move(o.src_checksum_func_name); + src_checksum_hex = std::move(o.src_checksum_hex); + db_id = std::move(o.db_id); + db_session_id = std::move(o.db_session_id); + type = o.type; + return *this; + } + + WorkItem( + std::string _src_path, std::string _dst_path, + const Temperature _src_temperature, const Temperature _dst_temperature, + std::string _contents, Env* _src_env, Env* _dst_env, + EnvOptions _src_env_options, bool _sync, RateLimiter* _rate_limiter, + uint64_t _size_limit, Statistics* _stats, + std::function _progress_callback = {}, + const std::string& _src_checksum_func_name = kUnknownFileChecksumFuncName, + const std::string& _src_checksum_hex = "", const std::string& _db_id = "", + const std::string& _db_session_id = "", + WorkItemType _type = WorkItemType::CopyOrCreate) + : src_path(std::move(_src_path)), + dst_path(std::move(_dst_path)), + src_temperature(_src_temperature), + dst_temperature(_dst_temperature), + contents(std::move(_contents)), + src_env(_src_env), + dst_env(_dst_env), + src_env_options(std::move(_src_env_options)), + sync(_sync), + rate_limiter(_rate_limiter), + size_limit(_size_limit), + stats(_stats), + progress_callback(std::move(_progress_callback)), + src_checksum_func_name(_src_checksum_func_name), + src_checksum_hex(_src_checksum_hex), + db_id(_db_id), + db_session_id(_db_session_id), + type(_type) {} + + ~WorkItem() = default; +}; + +struct CopyEngineOptions { + int max_background_operations = 1; + uint64_t io_buffer_size = 0; + uint64_t callback_trigger_interval_size = 4 * 1024 * 1024; + // Rate limiter for standalone ComputeChecksum reads; copies use the + // per-WorkItem rate_limiter instead. Not owned; must outlive the CopyEngine. + RateLimiter* checksum_rate_limiter = nullptr; + // Tickers recorded for bytes read/written by copies. Defaults are the backup + // tickers; other callers (e.g. checkpoint) should set their own. + Tickers read_bytes_ticker = BACKUP_READ_BYTES; + Tickers write_bytes_ticker = BACKUP_WRITE_BYTES; + Logger* info_log = nullptr; + // Pool thread name; must be <= 15 chars on Linux or it is left unnamed. + std::string thread_name = "copy_engine"; + // If set, polled to abort work early (e.g. BackupEngine::StopBackup). + std::function should_abort = {}; + std::string abort_status_message = "Operation stopped"; +}; + +// A pool of background threads that copy or checksum files pulled from a shared +// work queue, so BackupEngine and CheckpointEngine share one implementation. +class CopyEngine { + public: + explicit CopyEngine(CopyEngineOptions options); + ~CopyEngine(); + + CopyEngine(const CopyEngine&) = delete; + CopyEngine& operator=(const CopyEngine&) = delete; + CopyEngine(CopyEngine&&) = delete; + CopyEngine& operator=(CopyEngine&&) = delete; + + // Enqueues item (moved in); extract item.result's future before calling. + void Submit(WorkItem&& item); + + void MaybeDecreaseCpuPriority(CpuPriority priority); + + // Synchronous primitives, usable inline or from the pool. + IOStatus CopyOrCreateFile(const std::string& src, const std::string& dst, + const std::string& contents, uint64_t size_limit, + Env* src_env, Env* dst_env, + const EnvOptions& src_env_options, bool sync, + RateLimiter* rate_limiter, + const std::function& progress_callback, + Temperature* src_temperature, + Temperature dst_temperature, + uint64_t* bytes_toward_next_callback, + uint64_t* size, std::string* checksum_hex); + + IOStatus ReadFileAndComputeChecksum(const std::string& src, + const std::shared_ptr& src_fs, + const EnvOptions& src_env_options, + uint64_t size_limit, + std::string* checksum_hex, + Temperature src_temperature) const; + + uint64_t CalculateIOBufferSize(RateLimiter* rate_limiter) const; + + private: + static constexpr size_t kDefaultCopyFileBufferSize = + 5 * 1024 * 1024LL; // 5MB + + void ThreadBody(); + bool ShouldAbort() const { + return options_.should_abort && options_.should_abort(); + } + + CopyEngineOptions options_; + std::mutex byte_report_mutex_; + mutable channel work_items_; + std::vector threads_; + std::atomic threads_cpu_priority_; +}; + +} // namespace ROCKSDB_NAMESPACE