Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions mooncake-common/include/environment_variables.h
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,23 @@ struct ClientAutoPortEnvironmentVariables {
MC_DEFINE_ENV_VAR(int, MC_STORE_CLIENT_MAX_PORT);
};

struct DistributedStorageEnvironmentVariables {
MC_DEFINE_ENV_VAR(std::string, MOONCAKE_DFS_ROOT_DIR);
MC_DEFINE_ENV_VAR(std::string, MOONCAKE_DISTRIBUTED_ROOT_DIR);
MC_DEFINE_ENV_VAR(std::string, MOONCAKE_DFS_FS_ADAPTER);
MC_DEFINE_ENV_VAR(std::string, MOONCAKE_DISTRIBUTED_FS_TYPE);
MC_DEFINE_ENV_VAR(bool, MOONCAKE_DISTRIBUTED_HEALTH_CHECK);
MC_DEFINE_ENV_VAR(int, MOONCAKE_DFS_SHARD_COUNT);
MC_DEFINE_ENV_VAR(uint64_t, MOONCAKE_DFS_SHARD_CAPACITY);
MC_DEFINE_ENV_VAR(uint64_t, MOONCAKE_DFS_ALIGNMENT);
MC_DEFINE_ENV_VAR(bool, MOONCAKE_DFS_SINGLE_TENANT);
MC_DEFINE_ENV_VAR(bool, MOONCAKE_DFS_EVICTION_ENABLED);
MC_DEFINE_ENV_VAR(double, MOONCAKE_DFS_EVICTION_HIGH_WATERMARK);
MC_DEFINE_ENV_VAR(double, MOONCAKE_DFS_EVICTION_LOW_WATERMARK);
MC_DEFINE_ENV_VAR(int, MOONCAKE_DFS_DEFERRED_FREE_SECONDS);
MC_DEFINE_ENV_VAR(int, MOONCAKE_DFS_EVICTION_CHECK_INTERVAL);
};

#undef MC_DEFINE_ENV_VAR

} // namespace mooncake
29 changes: 29 additions & 0 deletions mooncake-store/include/config/distributed_storage_config.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
#pragma once

#include <chrono>
#include <cstdint>
#include <string>

namespace mooncake {

struct DistributedStorageConfig {
std::string fsdir = "/mnt/3fs/mooncake";
std::string fs_adapter_type = "hf3fs";
bool enable_health_check = false;
int shard_count = 64;
uint64_t shard_capacity = 4ULL * 1024 * 1024 * 1024;
uint64_t alignment = 4096;
bool single_tenant = true;
bool eviction_enabled = true;
double eviction_high_watermark = 0.9;
double eviction_low_watermark = 0.7;
std::chrono::seconds deferred_free_duration{30};
std::chrono::seconds eviction_check_interval{5};

bool Validate() const;
bool ValidateForAllocator() const;
static DistributedStorageConfig FromEnvironment();
std::string FormatStr() const;
};

} // namespace mooncake
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
#pragma once

#include <chrono>
#include <cstdint>
#include <memory>
#include <mutex>
#include <string>
#include <vector>

#include "config/distributed_storage_config.h"
#include "fs_adapter.h"
#include "replica.h"
#include "storage/distributed/object_storage_adapter.h"
Expand All @@ -21,26 +21,6 @@ enum class DistributedStorageMode {
kObjectStorage,
};

struct DistributedStorageConfig {
std::string fsdir = "/mnt/3fs/mooncake";
std::string fs_adapter_type = "hf3fs";
bool enable_health_check = false;
int shard_count = 64;
uint64_t shard_capacity = 4ULL * 1024 * 1024 * 1024;
uint64_t alignment = 4096;
bool single_tenant = true;
bool eviction_enabled = true;
double eviction_high_watermark = 0.9;
double eviction_low_watermark = 0.7;
std::chrono::seconds deferred_free_duration{30};
std::chrono::seconds eviction_check_interval{5};

bool Validate() const;
bool ValidateForAllocator() const;
static DistributedStorageConfig FromEnvironment();
std::string FormatStr() const;
};

struct DfsWriteRequest {
std::string key;
DistributedFSDescriptor descriptor;
Expand Down
1 change: 1 addition & 0 deletions mooncake-store/src/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ set(MOONCAKE_STORE_SHARED_SOURCES
thread_pool.cpp
etcd_helper.cpp
uds_transport.cpp
config/distributed_storage_config.cpp
storage/distributed/distributed_storage_backend.cpp
storage/distributed/posix_fs_adapter.cpp
utils/file_util.cpp)
Expand Down
144 changes: 144 additions & 0 deletions mooncake-store/src/config/distributed_storage_config.cpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,144 @@
#include "config/distributed_storage_config.h"

#include <glog/logging.h>
#include <chrono>
#include <filesystem>
#include <sstream>

#include "environ.h"
#include "environment_variables.h"

namespace mooncake {

bool DistributedStorageConfig::Validate() const {
if (fsdir.empty()) {
LOG(ERROR) << "DistributedStorageConfig: fsdir is empty";
return false;
}
if (!std::filesystem::path(fsdir).is_absolute()) {
LOG(ERROR)
<< "DistributedStorageConfig: fsdir must be an absolute path: "
<< fsdir;
return false;
}
if (fs_adapter_type != "hf3fs" && fs_adapter_type != "posix") {
LOG(ERROR) << "DistributedStorageConfig: unsupported fs_adapter_type: "
<< fs_adapter_type;
return false;
}
if (shard_count <= 0) {
LOG(ERROR) << "DistributedStorageConfig: shard_count must > 0";
return false;
}
if (shard_capacity == 0) {
LOG(ERROR) << "DistributedStorageConfig: shard_capacity must > 0";
return false;
}
if (alignment == 0 || (alignment & (alignment - 1)) != 0) {
LOG(ERROR) << "DistributedStorageConfig: alignment must be power of 2";
return false;
}
if (shard_capacity % alignment != 0) {
LOG(ERROR) << "DistributedStorageConfig: shard_capacity must align";
return false;
}
if (!single_tenant) {
LOG(ERROR) << "DistributedStorageConfig: Currently, DFS requires "
"single_tenant=true";
return false;
}
return true;
}

bool DistributedStorageConfig::ValidateForAllocator() const {
if (!Validate()) return false;

if (eviction_low_watermark < 0.0 || eviction_low_watermark > 1.0 ||
eviction_high_watermark < 0.0 || eviction_high_watermark > 1.0 ||
eviction_low_watermark >= eviction_high_watermark) {
LOG(ERROR) << "DistributedStorageConfig: eviction watermarks must "
"satisfy 0 <= low < high <= 1, low="
<< eviction_low_watermark
<< ", high=" << eviction_high_watermark;
return false;
}
if (deferred_free_duration.count() < 0) {
LOG(ERROR) << "DistributedStorageConfig: deferred_free_duration must "
"be non-negative, seconds="
<< deferred_free_duration.count();
return false;
}
if (eviction_enabled && eviction_check_interval.count() <= 0) {
LOG(ERROR) << "DistributedStorageConfig: eviction_check_interval must "
"be positive when eviction is enabled, seconds="
<< eviction_check_interval.count();
return false;
}
return true;
}

DistributedStorageConfig DistributedStorageConfig::FromEnvironment() {
DistributedStorageConfig config;
using Variables = DistributedStorageEnvironmentVariables;

const auto legacy_root_dir =
Environ::ReadOr(Variables::MOONCAKE_DISTRIBUTED_ROOT_DIR, config.fsdir);
config.fsdir =
Environ::ReadOr(Variables::MOONCAKE_DFS_ROOT_DIR, legacy_root_dir);
if (!std::filesystem::path(config.fsdir).is_absolute()) {
config.fsdir = std::filesystem::absolute(config.fsdir).string();
}

const auto legacy_fs_adapter = Environ::ReadOr(
Variables::MOONCAKE_DISTRIBUTED_FS_TYPE, config.fs_adapter_type);
config.fs_adapter_type =
Environ::ReadOr(Variables::MOONCAKE_DFS_FS_ADAPTER, legacy_fs_adapter);
config.enable_health_check =
Environ::ReadOr(Variables::MOONCAKE_DISTRIBUTED_HEALTH_CHECK,
config.enable_health_check);
config.shard_count = Environ::ReadOr(Variables::MOONCAKE_DFS_SHARD_COUNT,
config.shard_count);
config.shard_capacity = Environ::ReadOr(
Variables::MOONCAKE_DFS_SHARD_CAPACITY, config.shard_capacity);
config.alignment =
Environ::ReadOr(Variables::MOONCAKE_DFS_ALIGNMENT, config.alignment);
config.single_tenant = Environ::ReadOr(
Variables::MOONCAKE_DFS_SINGLE_TENANT, config.single_tenant);
config.eviction_enabled = Environ::ReadOr(
Variables::MOONCAKE_DFS_EVICTION_ENABLED, config.eviction_enabled);

// GetDouble silently falls back for an empty value; ReadOr<double> emits a
// warning. Keep the existing diagnostics while this refactor is
// behavior-preserving.
config.eviction_high_watermark =
Environ::GetDouble(Variables::MOONCAKE_DFS_EVICTION_HIGH_WATERMARK.name,
config.eviction_high_watermark);
config.eviction_low_watermark =
Environ::GetDouble(Variables::MOONCAKE_DFS_EVICTION_LOW_WATERMARK.name,
config.eviction_low_watermark);
config.deferred_free_duration = std::chrono::seconds(Environ::ReadOr(
Variables::MOONCAKE_DFS_DEFERRED_FREE_SECONDS,
static_cast<int>(config.deferred_free_duration.count())));
config.eviction_check_interval = std::chrono::seconds(Environ::ReadOr(
Variables::MOONCAKE_DFS_EVICTION_CHECK_INTERVAL,
static_cast<int>(config.eviction_check_interval.count())));
return config;
}

std::string DistributedStorageConfig::FormatStr() const {
std::ostringstream oss;
oss << "fsdir=" << fsdir << ", fs_adapter_type=" << fs_adapter_type
<< ", enable_health_check=" << enable_health_check
<< ", shard_count=" << shard_count
<< ", shard_capacity=" << shard_capacity << ", alignment=" << alignment
<< ", single_tenant=" << single_tenant
<< ", eviction_enabled=" << eviction_enabled
<< ", eviction_high_watermark=" << eviction_high_watermark
<< ", eviction_low_watermark=" << eviction_low_watermark
<< ", deferred_free_seconds=" << deferred_free_duration.count()
<< ", eviction_check_interval_seconds="
<< eviction_check_interval.count();
return oss.str();
}

} // namespace mooncake
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
#include <sstream>
#include <utility>

#include "storage/distributed/distributed_storage_backend.h"
#include "config/distributed_storage_config.h"
#include "storage/distributed/fs_adapter.h"
#include "storage/distributed/posix_fs_adapter.h"
#include "utils.h"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,7 @@
#include <algorithm>
#include <filesystem>
#include <limits>
#include <sstream>

#include "environ.h"
#include "storage/distributed/dfs_global_allocator.h"
#include "types.h"
#include "utils.h"
Expand Down Expand Up @@ -35,126 +33,6 @@ bool IsDfsDescriptorRangeValid(const DistributedFSDescriptor& desc,

} // namespace

bool DistributedStorageConfig::Validate() const {
if (fsdir.empty()) {
LOG(ERROR) << "DistributedStorageConfig: fsdir is empty";
return false;
}
if (!std::filesystem::path(fsdir).is_absolute()) {
LOG(ERROR)
<< "DistributedStorageConfig: fsdir must be an absolute path: "
<< fsdir;
return false;
}
if (fs_adapter_type != "hf3fs" && fs_adapter_type != "posix") {
LOG(ERROR) << "DistributedStorageConfig: unsupported fs_adapter_type: "
<< fs_adapter_type;
return false;
}
if (shard_count <= 0) {
LOG(ERROR) << "DistributedStorageConfig: shard_count must > 0";
return false;
}
if (shard_capacity == 0) {
LOG(ERROR) << "DistributedStorageConfig: shard_capacity must > 0";
return false;
}
if (alignment == 0 || (alignment & (alignment - 1)) != 0) {
LOG(ERROR) << "DistributedStorageConfig: alignment must be power of 2";
return false;
}
if (shard_capacity % alignment != 0) {
LOG(ERROR) << "DistributedStorageConfig: shard_capacity must align";
return false;
}
if (!single_tenant) {
LOG(ERROR) << "DistributedStorageConfig: Currently, DFS requires "
"single_tenant=true";
return false;
}
return true;
}

bool DistributedStorageConfig::ValidateForAllocator() const {
if (!Validate()) return false;

if (eviction_low_watermark < 0.0 || eviction_low_watermark > 1.0 ||
eviction_high_watermark < 0.0 || eviction_high_watermark > 1.0 ||
eviction_low_watermark >= eviction_high_watermark) {
LOG(ERROR) << "DistributedStorageConfig: eviction watermarks must "
"satisfy 0 <= low < high <= 1, low="
<< eviction_low_watermark
<< ", high=" << eviction_high_watermark;
return false;
}
if (deferred_free_duration.count() < 0) {
LOG(ERROR) << "DistributedStorageConfig: deferred_free_duration must "
"be non-negative, seconds="
<< deferred_free_duration.count();
return false;
}
if (eviction_enabled && eviction_check_interval.count() <= 0) {
LOG(ERROR) << "DistributedStorageConfig: eviction_check_interval must "
"be positive when eviction is enabled, seconds="
<< eviction_check_interval.count();
return false;
}
return true;
}

DistributedStorageConfig DistributedStorageConfig::FromEnvironment() {
DistributedStorageConfig config;
config.fsdir = Environ::GetString(
"MOONCAKE_DFS_ROOT_DIR",
Environ::GetString("MOONCAKE_DISTRIBUTED_ROOT_DIR", config.fsdir));
if (!std::filesystem::path(config.fsdir).is_absolute()) {
config.fsdir = std::filesystem::absolute(config.fsdir).string();
}
config.fs_adapter_type =
Environ::GetString("MOONCAKE_DFS_FS_ADAPTER",
Environ::GetString("MOONCAKE_DISTRIBUTED_FS_TYPE",
config.fs_adapter_type));
config.enable_health_check =
Environ::GetBool("MOONCAKE_DISTRIBUTED_HEALTH_CHECK", false);
config.shard_count =
Environ::GetInt("MOONCAKE_DFS_SHARD_COUNT", config.shard_count);
config.shard_capacity = Environ::GetUInt64("MOONCAKE_DFS_SHARD_CAPACITY",
config.shard_capacity);
config.alignment =
Environ::GetUInt64("MOONCAKE_DFS_ALIGNMENT", config.alignment);
config.single_tenant =
Environ::GetBool("MOONCAKE_DFS_SINGLE_TENANT", config.single_tenant);
config.eviction_enabled = Environ::GetBool("MOONCAKE_DFS_EVICTION_ENABLED",
config.eviction_enabled);
config.eviction_high_watermark = Environ::GetDouble(
"MOONCAKE_DFS_EVICTION_HIGH_WATERMARK", config.eviction_high_watermark);
config.eviction_low_watermark = Environ::GetDouble(
"MOONCAKE_DFS_EVICTION_LOW_WATERMARK", config.eviction_low_watermark);
config.deferred_free_duration = std::chrono::seconds(Environ::GetInt(
"MOONCAKE_DFS_DEFERRED_FREE_SECONDS",
static_cast<int>(config.deferred_free_duration.count())));
config.eviction_check_interval = std::chrono::seconds(Environ::GetInt(
"MOONCAKE_DFS_EVICTION_CHECK_INTERVAL",
static_cast<int>(config.eviction_check_interval.count())));
return config;
}

std::string DistributedStorageConfig::FormatStr() const {
std::ostringstream oss;
oss << "fsdir=" << fsdir << ", fs_adapter_type=" << fs_adapter_type
<< ", enable_health_check=" << enable_health_check
<< ", shard_count=" << shard_count
<< ", shard_capacity=" << shard_capacity << ", alignment=" << alignment
<< ", single_tenant=" << single_tenant
<< ", eviction_enabled=" << eviction_enabled
<< ", eviction_high_watermark=" << eviction_high_watermark
<< ", eviction_low_watermark=" << eviction_low_watermark
<< ", deferred_free_seconds=" << deferred_free_duration.count()
<< ", eviction_check_interval_seconds="
<< eviction_check_interval.count();
return oss.str();
}

DistributedStorageBackend::DistributedStorageBackend(
const FileStorageConfig& file_storage_config,
const DistributedStorageConfig& distributed_config,
Expand Down
Loading
Loading