database/rocksdb
1
0
Fork 0
mirror of https://github.com/facebook/rocksdb.git synced 2026-10-23 08:09:02 +03:00
Code Activity

Extract shared CopyEngine [2/4] (#14946) (#14946)

Browse Source 
Summary:
Extracts BackupEngine's parallel file-copy/checksum thread pool into a standalone `CopyEngine` (`utilities/copy_engine/copy_engine.{h,cc}`) so `CheckpointEngine` (later in the stack) can reuse it. Behavior-preserving; all `TEST_SYNC_POINT` names are unchanged.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/14946

Reviewed By: xingbowang

Differential Revision: D111387950

Pulled By: rban1

fbshipit-source-id: b5046adb98735a606fd1ba471731fd316ce6f036
This commit is contained in:
Ranjan Banerjee
2026-07-23 19:07:17 -07:00
committed by meta-codesync[bot]
parent f56112abb7
commit fac47c3072
8 changed files with 648 additions and 529 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
BUCK
View File

@@ -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",

1
CMakeLists.txt
View File

@@ -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

1
src.mk
View File

@@ -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 \

9
util/file_checksum_helper.h
View File

@@ -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 {

19
util/rate_limiter_impl.h
View File

@@ -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<size_t>(rate_limiter->GetSingleBurstBytes()),
remaining_bytes);
rate_limiter->Request(request_bytes, pri, stats, op_type);
remaining_bytes -= request_bytes;
}
}
} // namespace ROCKSDB_NAMESPACE

557
utilities/backup/backup_engine.cc
View File

@@ -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<void()> 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<FileSystem>& 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<WorkItemResult> result;
std::function<void()> 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<void()> _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<WorkItemResult> result;
bool shared;
@@ -874,10 +688,7 @@ class BackupEngineImpl {
};
bool initialized_;
std::mutex byte_report_mutex_;
mutable channel<WorkItem> work_items_;
std::vector<port::Thread> threads_;
std::atomic<CpuPriority> threads_cpu_priority_;
std::unique_ptr<CopyEngine> 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<FSDirectory> meta_directory_;
std::unique_ptr<FSDirectory> private_directory_;
static const size_t kDefaultCopyFileBufferSize = 5 * 1024 * 1024LL; // 5MB
bool read_only_;
BackupStatistics backup_statistics_;
std::unordered_set<std::string> 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<CopyEngine>(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<void()> 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<FSWritableFile> dst_file;
std::unique_ptr<FSSequentialFile> 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<uint64_t>::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<WritableFileWriter> dest_writer(
new WritableFileWriter(std::move(dst_file), dst, dst_file_options));
std::unique_ptr<SequentialFileReader> src_reader;
std::unique_ptr<char[]> 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_t>(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<std::mutex> 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<size_t>(rate_limiter->GetSingleBurstBytes())
: kDefaultCopyFileBufferSize;
}
// fname will always start with "/"
IOStatus BackupEngineImpl::AddBackupFileWorkItem(
std::unordered_set<std::string>& 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<FileSystem>& 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<uint64_t>::max();
}
std::unique_ptr<SequentialFileReader> 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<char[]> 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_t>(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<size_t>(rate_limiter->GetSingleBurstBytes()),
remaining_bytes);
rate_limiter->Request(request_bytes, pri, stats, op_type);
remaining_bytes -= request_bytes;
}
}
void BackupEngineImpl::InferDBFilesToRetainInRestore(
const std::vector<std::pair<const BackupEngineImpl*, const FileInfo*>>&
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));

355
utilities/copy_engine/copy_engine.cc Normal file
View File

@@ -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 <algorithm>
#include <exception>
#include <limits>
#include <string>
#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<void()>& 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<FSWritableFile> dst_file;
std::unique_ptr<FSSequentialFile> 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<uint64_t>::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<WritableFileWriter> dest_writer(
new WritableFileWriter(std::move(dst_file), dst, dst_file_options));
std::unique_ptr<SequentialFileReader> src_reader;
std::unique_ptr<char[]> 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_t>(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<std::mutex> 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<size_t>(rate_limiter->GetSingleBurstBytes())
: kDefaultCopyFileBufferSize;
}
IOStatus CopyEngine::ReadFileAndComputeChecksum(
const std::string& src, const std::shared_ptr<FileSystem>& 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<uint64_t>::max();
}
std::unique_ptr<SequentialFileReader> 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<char[]> 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_t>(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

234
utilities/copy_engine/copy_engine.h Normal file
View File

@@ -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 <atomic>
#include <cstdint>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <string>
#include <utility>
#include <vector>
#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<WorkItemResult> result;
std::function<void()> 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<void()> _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<bool()> 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<void()>& 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<FileSystem>& 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<WorkItem> work_items_;
std::vector<port::Thread> threads_;
std::atomic<CpuPriority> threads_cpu_priority_;
};
} // namespace ROCKSDB_NAMESPACE