mirror of
https://github.com/facebook/rocksdb.git
synced 2026-10-23 08:09:02 +03:00
897162a6bc
Summary: Remote compaction writes output SSTs into caller-selected job directories. If the primary crashes before installation or falls back to local compaction, those directories can outlive the DB session and accumulate under the DB directory. This change defines `kSessionTmpDirName` as `session_tmp`, the canonical name for a DB-local temporary directory whose contents have the lifetime of a DB session. The immutable boolean option `use_session_tmp_dir_for_remote_compaction` opts remote compaction into this directory. When enabled, an `OpenAndCompact()` directory name is resolved as `<dbname>/session_tmp/<job>`. During DB open, RocksDB creates the session temporary directory, completes recovery, and then synchronously deletes contents left by the previous DB session before scheduling compactions. `DestroyDB()` removes it as part of database deletion, and normal completion, failure, and cancellation paths clean the resolved per-job directory. The option defaults to false. Disabled callers retain the existing `<dbname>/<job>` layout and no session-directory cleanup. For rollout compatibility, legacy callers that already pass a DB-rooted output path are accepted unchanged only while the option is false. Directory creation and invalid-path errors are delegated to the configured filesystem rather than prevalidated by RocksDB. After D118301531 landed, per-job directory creation and resumption cleanup are kept together in `InitializeCompactionWorkspace()`. `OpenAndCompact()` resolves the client-provided name and expects DB open to have initialized the session temporary directory when the feature is enabled. The boolean is wired through C++ and generated C APIs, option serialization and reconstruction, `OpenAndCompact()`, ldb, db_bench, and db_stress. Tests cover path resolution, cross-session deletion, ownership boundaries, disabled behavior, `DestroyDB()`, configured compaction output, resumption fallback, and option round trips. Pull Request resolved: https://github.com/facebook/rocksdb/pull/15169 Reviewed By: pdillinger Differential Revision: D118227112 fbshipit-source-id: 01aa834b0b3f75638102130a86d95d537a2797bd
82 lines
2.9 KiB
C++
82 lines
2.9 KiB
C++
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
|
|
// 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).
|
|
|
|
#ifdef GFLAGS
|
|
#pragma once
|
|
|
|
#include <string>
|
|
|
|
#include "db/compaction/compaction_job.h"
|
|
#include "db_stress_shared_state.h"
|
|
#include "file/file_util.h"
|
|
#include "rocksdb/options.h"
|
|
#include "utilities/fault_injection_fs.h"
|
|
|
|
namespace ROCKSDB_NAMESPACE {
|
|
|
|
// Service to simulate Remote Compaction in Stress Test
|
|
class DbStressCompactionService : public CompactionService {
|
|
public:
|
|
explicit DbStressCompactionService(SharedState* shared,
|
|
bool failure_should_fall_back_to_local)
|
|
: shared_(shared),
|
|
aborted_(false),
|
|
failure_should_fall_back_to_local_(failure_should_fall_back_to_local) {}
|
|
|
|
static const char* kClassName() { return "DbStressCompactionService"; }
|
|
|
|
const char* Name() const override { return kClassName(); }
|
|
|
|
static constexpr uint64_t kWaitIntervalInMicros = 10 * 1000; // 10ms
|
|
static constexpr const char* kTempOutputDirectoryPrefix = "tmp_output_";
|
|
|
|
CompactionServiceScheduleResponse Schedule(
|
|
const CompactionServiceJobInfo& info,
|
|
const std::string& compaction_service_input) override {
|
|
std::string job_id = info.db_id + "_" + info.db_session_id + "_" +
|
|
std::to_string(info.job_id);
|
|
|
|
if (aborted_.load()) {
|
|
return CompactionServiceScheduleResponse(
|
|
job_id, CompactionServiceJobStatus::kUseLocal);
|
|
}
|
|
|
|
const std::string output_directory_name =
|
|
kTempOutputDirectoryPrefix + Env::Default()->GenerateUniqueId();
|
|
|
|
shared_->EnqueueRemoteCompaction(
|
|
job_id, info, compaction_service_input, output_directory_name,
|
|
false /* was_cancelled */); // Not canceled initially
|
|
CompactionServiceScheduleResponse response(
|
|
job_id, CompactionServiceJobStatus::kSuccess);
|
|
return response;
|
|
}
|
|
|
|
CompactionServiceJobStatus Wait(const std::string& scheduled_job_id,
|
|
std::string* result) override;
|
|
|
|
void OnInstallation(const std::string& scheduled_job_id,
|
|
CompactionServiceJobStatus /*status*/) override {
|
|
std::string serialized;
|
|
CompactionServiceResult result;
|
|
if (shared_->GetRemoteCompactionResult(scheduled_job_id, &serialized)
|
|
.has_value() &&
|
|
CompactionServiceResult::Read(serialized, &result).ok()) {
|
|
DestroyDir(Env::Default(), result.output_path).PermitUncheckedError();
|
|
}
|
|
shared_->RemoveRemoteCompactionResult(scheduled_job_id);
|
|
}
|
|
|
|
void CancelAwaitingJobs() override { aborted_.store(true); }
|
|
|
|
private:
|
|
SharedState* shared_;
|
|
std::atomic_bool aborted_{false};
|
|
bool failure_should_fall_back_to_local_;
|
|
};
|
|
} // namespace ROCKSDB_NAMESPACE
|
|
|
|
#endif // GFLAGS
|