From 81f492c03319390edab17bb45aff33b3e0a294d9 Mon Sep 17 00:00:00 2001 From: "Sgt.Pepper" <1303471564@qq.com> Date: Thu, 14 Aug 2025 14:34:51 +0800 Subject: [PATCH] [Store]feat: Migrate Persistence Metadata from Client to Master Service (#690) * initial commit * fix client::query return fault * fix isexist return fault * fix test bug * fix clearinvalidhandles problem * add file description for 3fs * change ssd function start from client to master * fix naming error * edit doc description * edit doc * clang format * fix as the review comment * fix formmat * add master service test for ssd * fix format * add log and cli * fix putend test --- doc/en/mooncake-store-preview.md | 20 +- doc/zh/mooncake-store-preview.md | 17 +- docs/source/design/mooncake-store-preview.md | 20 +- mooncake-store/include/client.h | 8 +- mooncake-store/include/config.h | 1 + mooncake-store/include/ha_helper.h | 2 + mooncake-store/include/master_client.h | 7 +- mooncake-store/include/master_service.h | 64 ++++- mooncake-store/include/rpc_service.h | 7 +- mooncake-store/include/storage_backend.h | 58 +--- mooncake-store/include/transfer_task.h | 4 +- mooncake-store/include/types.h | 156 +++++++--- mooncake-store/src/client.cpp | 206 +++++++------- mooncake-store/src/client_buffer.cpp | 4 +- mooncake-store/src/ha_helper.cpp | 5 +- mooncake-store/src/hf3fs/README.md | 5 +- mooncake-store/src/hf3fs/hf3fs_file.cpp | 26 +- mooncake-store/src/master.cpp | 13 +- mooncake-store/src/master_client.cpp | 12 +- mooncake-store/src/master_service.cpp | 168 ++++++++--- mooncake-store/src/posix_file.cpp | 20 -- mooncake-store/src/rpc_service.cpp | 27 +- mooncake-store/src/storage_backend.cpp | 134 ++------- mooncake-store/src/transfer_task.cpp | 4 +- mooncake-store/tests/CMakeLists.txt | 4 + mooncake-store/tests/client_buffer_test.cpp | 4 +- .../tests/client_integration_test.cpp | 11 + mooncake-store/tests/master_metrics_test.cpp | 8 +- .../tests/master_service_ssd_test.cpp | 267 ++++++++++++++++++ mooncake-store/tests/master_service_test.cpp | 71 ++--- .../tests/stress_cluster_benchmark.py | 4 - .../tests/test_distributed_object_store.py | 2 - .../tests/test_ssd_offload_in_evict.py | 4 - scripts/run_tests.sh | 4 +- 34 files changed, 875 insertions(+), 492 deletions(-) create mode 100644 mooncake-store/tests/master_service_ssd_test.cpp diff --git a/doc/en/mooncake-store-preview.md b/doc/en/mooncake-store-preview.md index fde7b3ba..0de0fe95 100644 --- a/doc/en/mooncake-store-preview.md +++ b/doc/en/mooncake-store-preview.md @@ -89,7 +89,7 @@ tl::expected Put(const ObjectKey& key, ![mooncake-store-simple-put](../../image/mooncake-store-simple-put.png) -Used to store the value corresponding to `key`. The required number of replicas can be set via the `config` parameter.​​(When persistence is enabled, after a successful in-memory put request, an asynchronous persistence operation to SSD will be initiated.)​ The data structure details of `ReplicateConfig` are as follows: +Used to store the value corresponding to `key`. The required number of replicas can be set via the `config` parameter.​​(When persistence is enabled, Put not only writes to the memory pool but also asynchronously initiates a data persistence operation to the SSD.)​ The data structure details of `ReplicateConfig` are as follows: ```C++ struct ReplicateConfig { @@ -455,13 +455,21 @@ This system provides support for a hierarchical cache architecture, enabling eff #### Enabling Persistence Functionality -When a user specifies the environment variable `MOONCAKE_STORAGE_ROOT_DIR` at client startup, and the path is a valid existing directory, the client-side data persistence feature will be activated. During initialization, the client requests a `cluster_id` from the master. This ID can be specified when initializing the master; if not provided, the default value `mooncake_cluster` will be used. The root directory for persistence is then set to `/`. Note that when using DFS, each client must specify the corresponding DFS mount directory to enable data sharing across SSDs. +When the user specifies `--root_fs_dir=/path/to/dir` when starting the master, and this path is a valid DFS-mounted directory on all machines where the clients reside, Mooncake Store's tiered caching functionality will work properly. Additionally, during master initialization, a `cluster_id` is loaded. This ID can be specified during master initialization (`--cluster_id=xxxx`). If not specified, the default value `mooncake_cluster` will be used. Subsequently, the root directory for client persistence will be `/`. + +​Note​​: When enabling this feature, the user must ensure that the DFS-mounted directory (`root_fs_dir=/path/to/dir`) is valid and consistent across all client hosts. If some clients have invalid or incorrect mount paths, it may cause abnormal behavior in Mooncake Store. #### Data Access Mechanism -In the current implementation, all operations on kvcache objects (e.g., read/write/query) are performed entirely on the client side, with no awareness by the master. The file system maintains the key-to-kvcache-object mapping through a fixed indexing mechanism, where each file corresponds to one kvcache object (the filename is the associated key). +The persistence feature also follows Mooncake Store's design principle of separating control flow from data flow. The read/write operations of kvcache objects are completed on the client side, while the query and management functions of kvcache objects are handled on the master side. In the file system, the key -> kvcache object index information is maintained by a fixed indexing mechanism, with each file corresponding to one kvcache object (the filename serves as the associated key name). -When persistence is enabled, every successful `Put`or`BatchPut` operation in memory triggers an asynchronous persistence write to DFS. During subsequent `Get`or `BatchGet` operations, if the requested kvcache is not found in the memory pool, the system attempts to read the corresponding file from DFS and returns the data to the user. +After enabling the persistence feature: + +- For each `Put` or `BatchPut` operation, both a synchronous memory pool write operation and an asynchronous DFS persistence operation will be initiated. +- For each `Get` or `BatchGet` operation, if the corresponding kvcache is not found in the memory pool, the system will attempt to read the file data from DFS and return it to the user. + +#### 3FS USRBIO Plugin +If you need to use 3FS's native API (USRBIO) to achieve high-performance persistent file reads and writes, you can refer to the configuration instructions in this document [3FS USRBIO Plugin](/mooncake-store/src/hf3fs/READMD.md). ## Mooncake Store Python API @@ -565,13 +573,9 @@ retcode = store.setup( 2. Run `ROLE=prefill python3 ./stress_cluster_benchmark.py` on one machine to start the Prefill node. For "rdma" protocol, you can also enable topology auto discovery and filters, e.g., `ROLE=prefill MC_MS_AUTO_DISC=1 MC_MS_FILTERS="mlx5_1,mlx5_2" python3 ./stress_cluster_benchmark.py`. - To enable the persistence feature, run: -`ROLE=prefill MOONCAKE_STORAGE_ROOT_DIR=/path/to/dir python3 ./stress_cluster_benchmark.py` 3. Run `ROLE=decode python3 ./stress_cluster_benchmark.py` on another machine to start the Decode node. For "rdma" protocol, you can also enable topology auto discovery and filters, e.g., `ROLE=decode MC_MS_AUTO_DISC=1 MC_MS_FILTERS="mlx5_1,mlx5_2" python3 ./stress_cluster_benchmark.py`. - To enable the persistence feature, run: -`ROLE=decode MOONCAKE_STORAGE_ROOT_DIR=/path/to/dir python3 ./stress_cluster_benchmark.py` The absence of error messages indicates successful data transfer. diff --git a/doc/zh/mooncake-store-preview.md b/doc/zh/mooncake-store-preview.md index 3ecd5dbf..f74f0233 100644 --- a/doc/zh/mooncake-store-preview.md +++ b/doc/zh/mooncake-store-preview.md @@ -94,7 +94,7 @@ tl::expected Put(const ObjectKey& key, ![mooncake-store-simple-put](../../image/mooncake-store-simple-put.png) -用于存储 `key` 对应的值。可通过 `config` 参数设置所需的副本数量。（当启用了持久化功能时，在内存的 `Put` 请求成功后，会异步发起一次数据向SSD的持久化操作） +用于存储 `key` 对应的值。可通过 `config` 参数设置所需的副本数量。（当启用了持久化功能时，`Put`除了对memory pool的写入之外，还会异步发起一次向SSD的数据持久化操作） 其中`ReplicateConfig` 的数据结构细节如下： ```C++ @@ -462,12 +462,17 @@ struct ReplicateConfig { #### 持久化功能启用方法 -当用户在启动client时指定了`MOONCAKE_STORAGE_ROOT_DIR`的环境变量，且该路径为一个已存在的有效路径时，则client端的数据持久化功能就会开始工作。同时在client启动时，会向master请求一个`cluster_id`，该id可以在初始化master进行指定，若未指定则会使用默认值`mooncake_cluster`，之后client执行持久化的根目录即为`/`。注意在使用DFS时，需要在各client上分别指定DFS对应的挂载目录，以实现SSD上数据之间的共享。 +当用户在启动master时指定了`--root_fs_dir=/path/to/dir`，且该路径在各client所属的机器上都是有效的DFS挂载目录时，mooncake store的分级缓存功能即可正常工作。此外master初始化时会加载一个`cluster_id`，该id可以在初始化master进行指定（`--cluster_id=xxxx`），若未指定则会使用默认值`mooncake_cluster`，之后client执行持久化的根目录即为`/`。 + +注意在开启该功能时，用户需要保证各client所在主机的DFS挂载目录都是有效且相同的（`root_fs_dir=/path/to/dir`），如果存在部分client挂载目录无效或错误，会导致mooncake store运行出现一些异常情况。 #### 数据访问机制 -在目前的实现版本中，kvcache object的读\写\查询等操作都是完全在client端完成的，master对其无感知。在文件系统中key -> kvcache object的索引信息是由固定的索引机制来维护的，每个文件对应一个kvcache object（文件名即为对应的key名称）。 +持久化功能同样遵循了mooncake store中控制流和数据流分离的设计。kvcache object的读\写操作在client端完成，kvcache object的查询和管理功能在master端完成。在文件系统中key -> kvcache object的索引信息是由固定的索引机制维护，每个文件对应一个kvcache object（文件名即为对应的key名称）。 -启用持久化功能后，对于每次成功写入memory的 `Put`或`BatchPut` 操作，都会异步地发起一次持久化操作，写入DFS当中。之后执行 `Get`或 `BatchGet` 时，如果在memory pool中没有找到对应的kvcache，则会尝试从DFS中读取该文件数据，并返回给用户。 +启用持久化功能后，对于每次 `Put`或`BatchPut` 操作，都会发起一次同步的memory pool写入操作和一次异步的DFS持久化操作。之后执行 `Get`或 `BatchGet` 时，如果在memory pool中没有找到对应的kvcache，则会尝试从DFS中读取该文件数据，并返回给用户。 + +#### 3FS USRBIO 插件 +如需通过3FS原生接口（USRBIO）实现高性能持久化文件读写，请参阅本文档的配置说明。[3FS USRBIO 插件配置](/mooncake-store/src/hf3fs/READMD.md)。 ## Mooncake Store Python API @@ -576,9 +581,9 @@ retcode = store.setup( ``` 2. 在一台机器上运行 `ROLE=prefill python3 ./stress_cluster_benchmark.py`，启动 Prefill 节点。 - 对于 rdma 协议, 你可以开启自动探索 topology 和设置网卡白名单, e.g., `ROLE=prefill MC_MS_AUTO_DISC=1 MC_MS_FILTERS="mlx5_1,mlx5_2" python3 ./stress_cluster_benchmark.py`。如需启用持久化功能，可运行`ROLE=prefill MOONCAKE_STORAGE_ROOT_DIR=/path/to/dir python3 ./stress_cluster_benchmark.py`。 + 对于 rdma 协议, 你可以开启自动探索 topology 和设置网卡白名单, e.g., `ROLE=prefill MC_MS_AUTO_DISC=1 MC_MS_FILTERS="mlx5_1,mlx5_2" python3 ./stress_cluster_benchmark.py`。 3. 在另一台机器上运行 `ROLE=decode python3 ./stress_cluster_benchmark.py`，启动 Decode 节点。 - 对于 rdma 协议, 你可以开启自动探索 topology 和设置网卡白名单, e.g., `ROLE=decode MC_MS_AUTO_DISC=1 MC_MS_FILTERS="mlx5_1,mlx5_2" python3 ./stress_cluster_benchmark.py`。如需启用持久化功能，可运行`ROLE=decode MOONCAKE_STORAGE_ROOT_DIR=/path/to/dir python3 ./stress_cluster_benchmark.py`。 + 对于 rdma 协议, 你可以开启自动探索 topology 和设置网卡白名单, e.g., `ROLE=decode MC_MS_AUTO_DISC=1 MC_MS_FILTERS="mlx5_1,mlx5_2" python3 ./stress_cluster_benchmark.py`。 无报错信息表示数据传输成功。 diff --git a/docs/source/design/mooncake-store-preview.md b/docs/source/design/mooncake-store-preview.md index 65d8a3dd..5a4fcaf8 100644 --- a/docs/source/design/mooncake-store-preview.md +++ b/docs/source/design/mooncake-store-preview.md @@ -89,7 +89,7 @@ tl::expected Put(const ObjectKey& key, ![mooncake-store-simple-put](../image/mooncake-store-simple-put.png) -Used to store the value corresponding to `key`. The required number of replicas can be set via the `config` parameter.​​(When persistence is enabled, after a successful in-memory put request, an asynchronous persistence operation to SSD will be initiated.)​ The data structure details of `ReplicateConfig` are as follows: +Used to store the value corresponding to `key`. The required number of replicas can be set via the `config` parameter.​​​(When persistence is enabled, Put not only writes to the memory pool but also asynchronously initiates a data persistence operation to the SSD.)​ The data structure details of `ReplicateConfig` are as follows: ```C++ struct ReplicateConfig { @@ -455,13 +455,21 @@ This system provides support for a hierarchical cache architecture, enabling eff #### Enabling Persistence Functionality -When a user specifies the environment variable `MOONCAKE_STORAGE_ROOT_DIR` at client startup, and the path is a valid existing directory, the client-side data persistence feature will be activated. During initialization, the client requests a `cluster_id` from the master. This ID can be specified when initializing the master; if not provided, the default value `mooncake_cluster` will be used. The root directory for persistence is then set to `/`. Note that when using DFS, each client must specify the corresponding DFS mount directory to enable data sharing across SSDs. +When the user specifies `--root_fs_dir=/path/to/dir` when starting the master, and this path is a valid DFS-mounted directory on all machines where the clients reside, Mooncake Store's tiered caching functionality will work properly. Additionally, during master initialization, a `cluster_id` is loaded. This ID can be specified during master initialization (`--cluster_id=xxxx`). If not specified, the default value `mooncake_cluster` will be used. Subsequently, the root directory for client persistence will be `/`. + +​Note​​: When enabling this feature, the user must ensure that the DFS-mounted directory (`root_fs_dir=/path/to/dir`) is valid and consistent across all client hosts. If some clients have invalid or incorrect mount paths, it may cause abnormal behavior in Mooncake Store. #### Data Access Mechanism -In the current implementation, all operations on kvcache objects (e.g., read/write/query) are performed entirely on the client side, with no awareness by the master. The file system maintains the key-to-kvcache-object mapping through a fixed indexing mechanism, where each file corresponds to one kvcache object (the filename is the associated key). +The persistence feature also follows Mooncake Store's design principle of separating control flow from data flow. The read/write operations of kvcache objects are completed on the client side, while the query and management functions of kvcache objects are handled on the master side. In the file system, the key -> kvcache object index information is maintained by a fixed indexing mechanism, with each file corresponding to one kvcache object (the filename serves as the associated key name). -When persistence is enabled, every successful `Put`or`BatchPut` operation in memory triggers an asynchronous persistence write to DFS. During subsequent `Get`or `BatchGet` operations, if the requested kvcache is not found in the memory pool, the system attempts to read the corresponding file from DFS and returns the data to the user. +After enabling the persistence feature: + +- For each `Put` or `BatchPut` operation, both a synchronous memory pool write operation and an asynchronous DFS persistence operation will be initiated. +- For each `Get` or `BatchGet` operation, if the corresponding kvcache is not found in the memory pool, the system will attempt to read the file data from DFS and return it to the user. + +#### 3FS USRBIO Plugin +If you need to use 3FS's native API (USRBIO) to achieve high-performance persistent file reads and writes, you can refer to the configuration instructions in this document [3FS USRBIO Plugin](/mooncake-store/src/hf3fs/READMD.md). ## Mooncake Store Python API @@ -565,13 +573,9 @@ retcode = store.setup( 2. Run `ROLE=prefill python3 ./stress_cluster_benchmark.py` on one machine to start the Prefill node. For "rdma" protocol, you can also enable topology auto discovery and filters, e.g., `ROLE=prefill MC_MS_AUTO_DISC=1 MC_MS_FILTERS="mlx5_1,mlx5_2" python3 ./stress_cluster_benchmark.py`. - To enable the persistence feature, run: -`ROLE=prefill MOONCAKE_STORAGE_ROOT_DIR=/path/to/dir python3 ./stress_cluster_benchmark.py` 3. Run `ROLE=decode python3 ./stress_cluster_benchmark.py` on another machine to start the Decode node. For "rdma" protocol, you can also enable topology auto discovery and filters, e.g., `ROLE=decode MC_MS_AUTO_DISC=1 MC_MS_FILTERS="mlx5_1,mlx5_2" python3 ./stress_cluster_benchmark.py`. - To enable the persistence feature, run: -`ROLE=decode MOONCAKE_STORAGE_ROOT_DIR=/path/to/dir python3 ./stress_cluster_benchmark.py` The absence of error messages indicates successful data transfer. diff --git a/mooncake-store/include/client.h b/mooncake-store/include/client.h index c1b190ee..43ac219c 100644 --- a/mooncake-store/include/client.h +++ b/mooncake-store/include/client.h @@ -221,8 +221,7 @@ class Client { * @brief Private constructor to enforce creation through Create() method */ Client(const std::string& local_hostname, - const std::string& metadata_connstring, - const std::string& storage_root_dir); + const std::string& metadata_connstring); /** * @brief Internal helper functions for initialization and data transfer @@ -247,7 +246,8 @@ class Client { const std::string& fsdir); void PutToLocalFile(const std::string& object_key, - const std::vector& slices); + const std::vector& slices, + const DiskDescriptor& disk_descriptor); /** * @brief Find the first complete replica from a replica list @@ -271,7 +271,6 @@ class Client { void SubmitTransfers(std::vector& ops); void WaitForTransfers(std::vector& ops); void FinalizeBatchPut(std::vector& ops); - void BatchPuttoLocalFile(std::vector& ops); std::vector> CollectResults( const std::vector& ops); @@ -290,7 +289,6 @@ class Client { // Configuration const std::string local_hostname_; const std::string metadata_connstring_; - const std::string storage_root_dir_; // Client persistent thread pool for async operations ThreadPool write_thread_pool_; diff --git a/mooncake-store/include/config.h b/mooncake-store/include/config.h index e0422ce1..eafa3395 100644 --- a/mooncake-store/include/config.h +++ b/mooncake-store/include/config.h @@ -26,6 +26,7 @@ struct MasterConfig { std::string etcd_endpoints; std::string cluster_id; + std::string root_fs_dir; std::string memory_allocator; }; diff --git a/mooncake-store/include/ha_helper.h b/mooncake-store/include/ha_helper.h index 494129aa..b009e118 100644 --- a/mooncake-store/include/ha_helper.h +++ b/mooncake-store/include/ha_helper.h @@ -90,6 +90,7 @@ class MasterServiceSupervisor { 0), // Client connection timeout. 0 = no timeout (infinite) bool rpc_enable_tcp_no_delay = true, const std::string& cluster_id = DEFAULT_CLUSTER_ID, + const std::string& root_fs_dir = DEFAULT_ROOT_FS_DIR, BufferAllocatorType memory_allocator = BufferAllocatorType::CACHELIB); MasterServiceSupervisor(const MasterConfig& master_config); int Start(); @@ -124,6 +125,7 @@ class MasterServiceSupervisor { std::string local_hostname_; std::string cluster_id_; + std::string root_fs_dir_; BufferAllocatorType memory_allocator_; }; diff --git a/mooncake-store/include/master_client.h b/mooncake-store/include/master_client.h index b576eeef..f6158611 100644 --- a/mooncake-store/include/master_client.h +++ b/mooncake-store/include/master_client.h @@ -96,9 +96,11 @@ class MasterClient { /** * @brief Ends a put operation * @param key Object key + * @param replica_type Type of replica (memory or disk) * @return tl::expected indicating success/failure */ - [[nodiscard]] tl::expected PutEnd(const std::string& key); + [[nodiscard]] tl::expected PutEnd( + const std::string& key, ReplicaType replica_type); /** * @brief Ends a put operation for a batch of objects @@ -111,10 +113,11 @@ class MasterClient { /** * @brief Revokes a put operation * @param key Object key + * @param replica_type Type of replica (memory or disk) * @return tl::expected indicating success/failure */ [[nodiscard]] tl::expected PutRevoke( - const std::string& key); + const std::string& key, ReplicaType replica_type); /** * @brief Revokes a put operation for a batch of objects diff --git a/mooncake-store/include/master_service.h b/mooncake-store/include/master_service.h index fb9ae831..ef2ae3b5 100644 --- a/mooncake-store/include/master_service.h +++ b/mooncake-store/include/master_service.h @@ -15,6 +15,7 @@ #include #include #include +#include #include "allocation_strategy.h" #include "master_metric_manager.h" @@ -72,6 +73,7 @@ class MasterService { int64_t client_live_ttl_sec = DEFAULT_CLIENT_LIVE_TTL_SEC, bool enable_ha = false, const std::string& cluster_id = DEFAULT_CLUSTER_ID, + const std::string& root_fs_dir = DEFAULT_ROOT_FS_DIR, BufferAllocatorType memory_allocator = BufferAllocatorType::CACHELIB); ~MasterService(); @@ -180,18 +182,22 @@ class MasterService { -> tl::expected, ErrorCode>; /** - * @brief Complete a put operation + * @brief Complete a put operation, replica_type indicates the type of + * replica to complete (memory or disk) * @return ErrorCode::OK on success, ErrorCode::OBJECT_NOT_FOUND if not * found, ErrorCode::INVALID_WRITE if replica status is invalid */ - auto PutEnd(const std::string& key) -> tl::expected; + auto PutEnd(const std::string& key, ReplicaType replica_type) + -> tl::expected; /** - * @brief Revoke a put operation + * @brief Revoke a put operation, replica_type indicates the type of + * replica to revoke (memory or disk) * @return ErrorCode::OK on success, ErrorCode::OBJECT_NOT_FOUND if not * found, ErrorCode::INVALID_WRITE if replica status is invalid */ - auto PutRevoke(const std::string& key) -> tl::expected; + auto PutRevoke(const std::string& key, ReplicaType replica_type) + -> tl::expected; /** * @brief Complete a batch of put operations @@ -245,6 +251,10 @@ class MasterService { tl::expected GetFsdir() const; private: + // Resolve the key to a sanitized format for storage + std::string SanitizeKey(const std::string& key) const; + std::string ResolvePath(const std::string& key) const; + // GC thread function void GCThreadFunc(); @@ -304,9 +314,10 @@ class MasterService { // given value. If there are, return the status of the first replica // that is not equal to the given value. Otherwise, return false. std::optional HasDiffRepStatus( - ReplicaStatus status) const { + ReplicaStatus status, ReplicaType replica_type) const { for (const auto& replica : replicas) { - if (replica.status() != status) { + if (replica.status() != status && + replica.type() == replica_type) { return replica.status(); } } @@ -327,6 +338,32 @@ class MasterService { } } + // Erase all replicas of the given type + void EraseReplica(ReplicaType replica_type) { + replicas.erase( + std::remove_if(replicas.begin(), replicas.end(), + [replica_type](const Replica& replica) { + return replica.type() == replica_type; + }), + replicas.end()); + } + + // Check if there is a memory replica + bool HasMemReplica() const { + return std::any_of(replicas.begin(), replicas.end(), + [](const Replica& replica) { + return replica.type() == ReplicaType::MEMORY; + }); + } + + // Get the count of memory replicas + int GetMemReplicaCount() const { + return std::count_if( + replicas.begin(), replicas.end(), [](const Replica& replica) { + return replica.type() == ReplicaType::MEMORY; + }); + } + // Check if the lease has expired bool IsLeaseExpired() const { return std::chrono::steady_clock::now() >= lease_timeout; @@ -347,6 +384,17 @@ class MasterService { bool IsSoftPinned(std::chrono::steady_clock::time_point& now) const { return soft_pin_timeout && now < *soft_pin_timeout; } + + // Check if the metadata is valid + // Valid means it has at least one replica and size is greater than 0 + bool IsValid() const { return !replicas.empty() && size > 0; } + + bool IsAllReplicasComplete() const { + return std::all_of( + replicas.begin(), replicas.end(), [](const Replica& replica) { + return replica.status() == ReplicaStatus::COMPLETE; + }); + } }; static constexpr size_t kNumShards = 1024; // Number of metadata shards @@ -455,6 +503,10 @@ class MasterService { // cluster id for persistent sub directory const std::string cluster_id_; + // root filesystem directory for persistent storage + const std::string root_fs_dir_; + + bool use_disk_replica_{false}; // Segment management SegmentManager segment_manager_; diff --git a/mooncake-store/include/rpc_service.h b/mooncake-store/include/rpc_service.h index 77cb9995..2602375d 100644 --- a/mooncake-store/include/rpc_service.h +++ b/mooncake-store/include/rpc_service.h @@ -29,6 +29,7 @@ class WrappedMasterService { int64_t client_live_ttl_sec = DEFAULT_CLIENT_LIVE_TTL_SEC, bool enable_ha = false, const std::string& cluster_id = DEFAULT_CLUSTER_ID, + const std::string& root_fs_dir = DEFAULT_ROOT_FS_DIR, BufferAllocatorType memory_allocator = BufferAllocatorType::CACHELIB); ~WrappedMasterService(); @@ -50,9 +51,11 @@ class WrappedMasterService { const std::string& key, const std::vector& slice_lengths, const ReplicateConfig& config); - tl::expected PutEnd(const std::string& key); + tl::expected PutEnd(const std::string& key, + ReplicaType replica_type); - tl::expected PutRevoke(const std::string& key); + tl::expected PutRevoke(const std::string& key, + ReplicaType replica_type); std::vector, ErrorCode>> BatchPutStart(const std::vector& keys, diff --git a/mooncake-store/include/storage_backend.h b/mooncake-store/include/storage_backend.h index 120bd697..8337ca63 100644 --- a/mooncake-store/include/storage_backend.h +++ b/mooncake-store/include/storage_backend.h @@ -81,82 +81,57 @@ class StorageBackend { /** * @brief Stores an object composed of multiple slices - * @param key Object identifier + * @param path path for the object * @param slices Vector of data slices to store * @return tl::expected indicating operation status */ - tl::expected StoreObject(const ObjectKey& key, + tl::expected StoreObject(const std::string& path, const std::vector& slices); /** * @brief Stores an object from a string - * @param key Object identifier + * @param path path for the object * @param str String containing object data * @return tl::expected indicating operation status */ - tl::expected StoreObject(const ObjectKey& key, + tl::expected StoreObject(const std::string& path, const std::string& str); /** * @brief Stores an object from a span of data - * @param key Object identifier + * @param path path for the object * @param data Span containing object data * @return tl::expected indicating operation status */ - tl::expected StoreObject(const ObjectKey& key, + tl::expected StoreObject(const std::string& path, std::span data); /** * @brief Loads an object into slices - * @param path KVCache File path to load from + * @param path path for the object * @param slices Output vector for loaded data slices * @param length Expected length of data to read * @return tl::expected indicating operation status */ - tl::expected LoadObject(std::string& path, + tl::expected LoadObject(const std::string& path, std::vector& slices, size_t length); /** * @brief Loads an object as a string - * @param path KVCache File path to load from + * @param path path for the object * @param str Output string for loaded data * @param length Expected length of data to read * @return tl::expected indicating operation status */ - tl::expected LoadObject(std::string& path, + tl::expected LoadObject(const std::string& path, std::string& str, size_t length); - /** - * @brief Checks if an object with the given key exists - * @param key Object identifier - * @return bool indicating whether the object exists - */ - bool Existkey(const ObjectKey& key); - - /** - * @brief Queries metadata for an object by key - * @param key Object identifier - * @return Optional Replica::Descriptor containing object metadata, or empty - * if not found - * - * This method retrieves the file path and size for the given object key. - */ - std::optional Querykey(const ObjectKey& key); - - /** - * @brief Batch queries metadata for multiple object keys - * @param keys Vector of object identifiers - * @return unordered_map mapping ObjectKey to Replica::Descriptor - */ - std::unordered_map BatchQueryKey( - const std::vector& keys); - /** * @brief Deletes the physical file associated with the given object key - * @param key Object identifier + * @param path Path to the file to remove */ - void RemoveFile(const ObjectKey& key); + void RemoveFile(const std::string& path); /** * @brief Deletes all objects from the storage backend @@ -178,14 +153,9 @@ class StorageBackend { private: /** - * @brief Sanitizes object key for filesystem safety + * @brief Make sure the path is valid and create necessary directories */ - std::string SanitizeKey(const ObjectKey& key) const; - - /** - * @brief Resolves full filesystem path for an object - */ - std::string ResolvePath(const ObjectKey& key) const; + void ResolvePath(const std::string& path) const; /** * @brief Creates a file object for the specified path and mode diff --git a/mooncake-store/include/transfer_task.h b/mooncake-store/include/transfer_task.h index 423317b9..bac0bdb7 100644 --- a/mooncake-store/include/transfer_task.h +++ b/mooncake-store/include/transfer_task.h @@ -293,7 +293,7 @@ class MemcpyWorkerPool { */ struct FilereadTask { std::string file_path; - size_t file_size; + size_t object_size; std::vector slices; std::shared_ptr state; @@ -301,7 +301,7 @@ struct FilereadTask { const std::vector& slices_ref, std::shared_ptr s) : file_path(path), - file_size(size), + object_size(size), slices(slices_ref), state(std::move(s)) {} }; diff --git a/mooncake-store/include/types.h b/mooncake-store/include/types.h index df9816cf..4383ab6f 100644 --- a/mooncake-store/include/types.h +++ b/mooncake-store/include/types.h @@ -35,6 +35,7 @@ static constexpr double DEFAULT_EVICTION_HIGH_WATERMARK_RATIO = 1.0; static constexpr int64_t ETCD_MASTER_VIEW_LEASE_TTL = 5; // in seconds static constexpr int64_t DEFAULT_CLIENT_LIVE_TTL_SEC = 10; // in seconds static const std::string DEFAULT_CLUSTER_ID = "mooncake_cluster"; +static const std::string DEFAULT_ROOT_FS_DIR = ""; // Forward declarations class BufferAllocatorBase; @@ -178,6 +179,29 @@ inline std::ostream& operator<<(std::ostream& os, class BufferAllocator; +/** + * @brief Type of buffer allocator used in the system + */ +enum class ReplicaType { + MEMORY, // Memory replica + DISK, // Disk replica +}; + +/** + * @brief Stream operator for ReplicaType + */ +inline std::ostream& operator<<(std::ostream& os, + const ReplicaType& replicaType) noexcept { + static const std::unordered_map + replica_type_strings{{ReplicaType::MEMORY, "MEMORY"}, + {ReplicaType::DISK, "DISK"}}; + + os << (replica_type_strings.count(replicaType) + ? replica_type_strings.at(replicaType) + : "UNKNOWN"); + return os; +} + /** * @brief Status of a replica in the system */ @@ -303,6 +327,15 @@ inline std::ostream& operator<<(std::ostream& os, << "buffer_ptr: " << static_cast(buffer.data()) << " }"; } +struct MemoryReplicaData { + std::vector> buffers; +}; + +struct DiskReplicaData { + std::string file_path; + uint64_t object_size = 0; +}; + struct MemoryDescriptor { std::vector buffer_descriptors; YLT_REFL(MemoryDescriptor, buffer_descriptors); @@ -310,40 +343,60 @@ struct MemoryDescriptor { struct DiskDescriptor { std::string file_path{}; - uint64_t file_size = 0; - YLT_REFL(DiskDescriptor, file_path, file_size); + uint64_t object_size = 0; + YLT_REFL(DiskDescriptor, file_path, object_size); }; class Replica { public: struct Descriptor; - Replica() = default; + // memory replica constructor Replica(std::vector> buffers, ReplicaStatus status) - : buffers_(std::move(buffers)), status_(status) {} + : data_(MemoryReplicaData{std::move(buffers)}), status_(status) {} - void reset() noexcept { - buffers_.clear(); - status_ = ReplicaStatus::UNDEFINED; - } + // disk replica constructor + Replica(std::string file_path, uint64_t object_size, ReplicaStatus status) + : data_(DiskReplicaData{std::move(file_path), object_size}), + status_(status) {} [[nodiscard]] Descriptor get_descriptor() const; [[nodiscard]] ReplicaStatus status() const { return status_; } - [[nodiscard]] bool has_invalid_handle() const { - return std::any_of(buffers_.begin(), buffers_.end(), - [](const std::unique_ptr& buf_ptr) { - return !buf_ptr->isAllocatorValid(); - }); + [[nodiscard]] ReplicaType type() const { + return std::visit(ReplicaTypeVisitor{}, data_); + } + + [[nodiscard]] bool is_memory_replica() const { + return std::holds_alternative(data_); + } + + [[nodiscard]] bool is_disk_replica() const { + return std::holds_alternative(data_); + } + + [[nodiscard]] bool has_invalid_mem_handle() const { + if (is_memory_replica()) { + const auto& mem_data = std::get(data_); + return std::any_of( + mem_data.buffers.begin(), mem_data.buffers.end(), + [](const std::unique_ptr& buf_ptr) { + return !buf_ptr->isAllocatorValid(); + }); + } + return false; // DiskReplicaData does not have handles } void mark_complete() { if (status_ == ReplicaStatus::PROCESSING) { status_ = ReplicaStatus::COMPLETE; - for (const auto& buf_ptr : buffers_) { - buf_ptr->mark_complete(); + if (is_memory_replica()) { + auto& mem_data = std::get(data_); + for (const auto& buf_ptr : mem_data.buffers) { + buf_ptr->mark_complete(); + } } } else if (status_ == ReplicaStatus::COMPLETE) { LOG(WARNING) << "Replica already marked as complete"; @@ -354,6 +407,15 @@ class Replica { friend std::ostream& operator<<(std::ostream& os, const Replica& replica); + struct ReplicaTypeVisitor { + ReplicaType operator()(const MemoryReplicaData&) const { + return ReplicaType::MEMORY; + } + ReplicaType operator()(const DiskReplicaData&) const { + return ReplicaType::DISK; + } + }; + struct Descriptor { std::variant descriptor_variant; ReplicaStatus status; @@ -368,6 +430,14 @@ class Replica { return std::holds_alternative(descriptor_variant); } + bool is_disk_replica() noexcept { + return std::holds_alternative(descriptor_variant); + } + + bool is_disk_replica() const noexcept { + return std::holds_alternative(descriptor_variant); + } + MemoryDescriptor& get_memory_descriptor() { if (auto* desc = std::get_if(&descriptor_variant)) { @@ -400,33 +470,55 @@ class Replica { }; private: - std::vector> buffers_; + std::variant data_; ReplicaStatus status_{ReplicaStatus::UNDEFINED}; }; inline Replica::Descriptor Replica::get_descriptor() const { Replica::Descriptor desc; desc.status = status_; - MemoryDescriptor mem_desc; - mem_desc.buffer_descriptors.reserve(buffers_.size()); - for (const auto& buf_ptr : buffers_) { - if (buf_ptr) { - mem_desc.buffer_descriptors.push_back(buf_ptr->get_descriptor()); + + if (is_memory_replica()) { + const auto& mem_data = std::get(data_); + MemoryDescriptor mem_desc; + mem_desc.buffer_descriptors.reserve(mem_data.buffers.size()); + for (const auto& buf_ptr : mem_data.buffers) { + if (buf_ptr) { + mem_desc.buffer_descriptors.push_back( + buf_ptr->get_descriptor()); + } } + desc.descriptor_variant = std::move(mem_desc); + } else if (is_disk_replica()) { + const auto& disk_data = std::get(data_); + DiskDescriptor disk_desc; + disk_desc.file_path = disk_data.file_path; + disk_desc.object_size = disk_data.object_size; + desc.descriptor_variant = std::move(disk_desc); } - desc.descriptor_variant = std::move(mem_desc); + return desc; } inline std::ostream& operator<<(std::ostream& os, const Replica& replica) { - os << "Replica: { " << "status: " << replica.status_ << ", " - << "buffers: ["; - for (const auto& buf_ptr : replica.buffers_) { - if (buf_ptr) { - os << *buf_ptr; + os << "Replica: { status: " << replica.status_ << ", "; + + if (replica.is_memory_replica()) { + const auto& mem_data = std::get(replica.data_); + os << "type: MEMORY, buffers: ["; + for (const auto& buf_ptr : mem_data.buffers) { + if (buf_ptr) { + os << *buf_ptr; + } } + os << "]"; + } else if (replica.is_disk_replica()) { + const auto& disk_data = std::get(replica.data_); + os << "type: DISK, file_path: " << disk_data.file_path + << ", object_size: " << disk_data.object_size; } - os << "] }"; + + os << " }"; return os; } @@ -447,8 +539,8 @@ const static uint64_t kMaxSliceSize = */ struct Segment { UUID id{0, 0}; - std::string name{}; // The name of the segment, also might be the hostname - // of the server that owns the segment + std::string name{}; // The name of the segment, also might be the + // hostname of the server that owns the segment uintptr_t base{0}; size_t size{0}; Segment() = default; @@ -464,8 +556,8 @@ YLT_REFL(Segment, id, name, base, size); enum class ClientStatus { UNDEFINED = 0, // Uninitialized OK, // Client is alive, no need to remount for now - NEED_REMOUNT, // Ping ttl expired, or the first time connect to master, so - // need to remount + NEED_REMOUNT, // Ping ttl expired, or the first time connect to master, + // so need to remount }; /** diff --git a/mooncake-store/src/client.cpp b/mooncake-store/src/client.cpp index b190d48c..b04a7966 100644 --- a/mooncake-store/src/client.cpp +++ b/mooncake-store/src/client.cpp @@ -36,13 +36,11 @@ namespace mooncake { } Client::Client(const std::string& local_hostname, - const std::string& metadata_connstring, - const std::string& storage_root_dir) + const std::string& metadata_connstring) : metrics_(ClientMetric::Create()), master_client_(metrics_ ? &metrics_->master_client_metric : nullptr), local_hostname_(local_hostname), metadata_connstring_(metadata_connstring), - storage_root_dir_(storage_root_dir), write_thread_pool_(2) { client_id_ = generate_uuid(); LOG(INFO) << "client_id=" << client_id_; @@ -260,32 +258,34 @@ std::optional> Client::Create( const std::string& local_hostname, const std::string& metadata_connstring, const std::string& protocol, void** protocol_args, const std::string& master_server_entry) { - // If MOONCAKE_STORAGE_ROOT_DIR is set, use it as the storage root directory - std::string storage_root_dir = - std::getenv("MOONCAKE_STORAGE_ROOT_DIR") - ? std::getenv("MOONCAKE_STORAGE_ROOT_DIR") - : ""; - auto client = std::shared_ptr( - new Client(local_hostname, metadata_connstring, storage_root_dir)); + new Client(local_hostname, metadata_connstring)); ErrorCode err = client->ConnectToMaster(master_server_entry); if (err != ErrorCode::OK) { return std::nullopt; } - // Initialize storage backend if storage_root_dir is provided + // Initialize storage backend if storage_root_dir is valid auto response = client->master_client_.GetFsdir(); if (!response) { LOG(ERROR) << "Failed to get fsdir from master"; - } else if (storage_root_dir.empty()) { + } else if (response.value().empty()) { LOG(INFO) << "Storage root directory is not set. persisting data is " "disabled."; } else { - LOG(INFO) << "Storage root directory is: " << storage_root_dir; - LOG(INFO) << "Fs subdir is: " << response.value(); - // Initialize storage backend - client->PrepareStorageBackend(storage_root_dir, response.value()); + auto dir_string = response.value(); + size_t pos = dir_string.find_last_of('/'); + if (pos != std::string::npos) { + std::string storage_root_dir = dir_string.substr(0, pos); + std::string fs_subdir = dir_string.substr(pos + 1); + LOG(INFO) << "Storage root directory is: " << storage_root_dir; + LOG(INFO) << "Fs subdir is: " << fs_subdir; + // Initialize storage backend + client->PrepareStorageBackend(storage_root_dir, fs_subdir); + } else { + LOG(ERROR) << "Invalid fsdir format: " << dir_string; + } } // Initialize transfer engine @@ -359,16 +359,7 @@ std::vector> Client::BatchGet( tl::expected, ErrorCode> Client::Query( const std::string& object_key) { auto result = master_client_.GetReplicaList(object_key); - if (!result) { - // Check storage backend if master query fails - if (storage_backend_) { - if (auto desc_opt = storage_backend_->Querykey(object_key)) { - return std::vector{std::move(*desc_opt)}; - } - } - return tl::unexpected(result.error()); - } - return result.value(); + return result; } std::vector, ErrorCode>> @@ -388,20 +379,6 @@ Client::BatchQuery(const std::vector& object_keys) { } return results; } - - // For failed queries, check storage backend if available - if (storage_backend_) { - for (size_t i = 0; i < response.size(); ++i) { - if (!response[i]) { - if (auto desc_opt = - storage_backend_->Querykey(object_keys[i])) { - response[i] = - std::vector{std::move(*desc_opt)}; - } - } - } - } - return response; } @@ -556,17 +533,35 @@ tl::expected Client::Put(const ObjectKey& key, // Record Put transfer latency (all replicas) auto t0_put = std::chrono::steady_clock::now(); - // Transfer data using allocated handles from all replicas - for (const auto& replica : start_result.value()) { - ErrorCode transfer_err = TransferWrite(replica, slices); - if (transfer_err != ErrorCode::OK) { - // Revoke put operation - auto revoke_result = master_client_.PutRevoke(key); - if (!revoke_result) { - LOG(ERROR) << "Failed to revoke put operation"; - return tl::unexpected(revoke_result.error()); + // We must deal with disk replica first, then the disk putrevoke/putend can + // be called surely + if (storage_backend_) { + for (auto it = start_result.value().rbegin(); + it != start_result.value().rend(); ++it) { + const auto& replica = *it; + if (replica.is_disk_replica()) { + // Store to local file if storage backend is available + auto disk_descriptor = replica.get_disk_descriptor(); + PutToLocalFile(key, slices, disk_descriptor); + break; // Only one disk replica is needed + } + } + } + + for (const auto& replica : start_result.value()) { + if (replica.is_memory_replica()) { + // Transfer data using allocated handles from all replicas + ErrorCode transfer_err = TransferWrite(replica, slices); + if (transfer_err != ErrorCode::OK) { + // Revoke put operation + auto revoke_result = + master_client_.PutRevoke(key, ReplicaType::MEMORY); + if (!revoke_result) { + LOG(ERROR) << "Failed to revoke put operation"; + return tl::unexpected(revoke_result.error()); + } + return tl::unexpected(transfer_err); } - return tl::unexpected(transfer_err); } } @@ -578,16 +573,13 @@ tl::expected Client::Put(const ObjectKey& key, } // End put operation - auto end_result = master_client_.PutEnd(key); + auto end_result = master_client_.PutEnd(key, ReplicaType::MEMORY); if (!end_result) { ErrorCode err = end_result.error(); LOG(ERROR) << "Failed to end put operation: " << err; return tl::unexpected(err); } - // Store to local file if storage backend is available - PutToLocalFile(key, slices); - return {}; } @@ -740,21 +732,37 @@ void Client::SubmitTransfers(std::vector& ops) { bool all_transfers_submitted = true; std::string failure_context; + // We must deal with disk replica first, then the disk putrevoke/putend + // can be called surely + if (storage_backend_) { + for (auto it = op.replicas.rbegin(); it != op.replicas.rend(); + ++it) { + const auto& replica = *it; + if (replica.is_disk_replica()) { + auto disk_descriptor = replica.get_disk_descriptor(); + PutToLocalFile(op.key, op.slices, disk_descriptor); + break; // Only one disk replica is needed + } + } + } + for (size_t replica_idx = 0; replica_idx < op.replicas.size(); ++replica_idx) { const auto& replica = op.replicas[replica_idx]; + if (replica.is_memory_replica()) { + auto submit_result = transfer_submitter_->submit( + replica, op.slices, TransferRequest::WRITE); - auto submit_result = transfer_submitter_->submit( - replica, op.slices, TransferRequest::WRITE); + if (!submit_result) { + failure_context = "Failed to submit transfer for replica " + + std::to_string(replica_idx); + all_transfers_submitted = false; + break; + } - if (!submit_result) { - failure_context = "Failed to submit transfer for replica " + - std::to_string(replica_idx); - all_transfers_submitted = false; - break; + op.pending_transfers.emplace_back( + std::move(submit_result.value())); } - - op.pending_transfers.emplace_back(std::move(submit_result.value())); } if (!all_transfers_submitted) { @@ -957,22 +965,6 @@ std::vector> Client::CollectResults( return results; } -void Client::BatchPuttoLocalFile(std::vector& ops) { - if (!storage_backend_) { - return; // No storage backend initialized - } - - for (const auto& op : ops) { - if (op.IsSuccessful()) { - // Store to local file if operation was successful - PutToLocalFile(op.key, op.slices); - } else { - LOG(ERROR) << "Skipping local file storage for key " << op.key - << " due to failure: " << toString(op.result.error()); - } - } -} - std::vector> Client::BatchPut( const std::vector& keys, std::vector>& batched_slices, @@ -991,15 +983,14 @@ std::vector> Client::BatchPut( } FinalizeBatchPut(ops); - BatchPuttoLocalFile(ops); return CollectResults(ops); } tl::expected Client::Remove(const ObjectKey& key) { auto result = master_client_.Remove(key); - if (storage_backend_) { - storage_backend_->RemoveFile(key); - } + // if (storage_backend_) { + // storage_backend_->RemoveFile(key); + // } if (!result) { return tl::unexpected(result.error()); } @@ -1007,9 +998,9 @@ tl::expected Client::Remove(const ObjectKey& key) { } tl::expected Client::RemoveAll() { - if (storage_backend_) { - storage_backend_->RemoveAll(); - } + // if (storage_backend_) { + // storage_backend_->RemoveAll(); + // } return master_client_.RemoveAll(); } @@ -1129,16 +1120,7 @@ tl::expected Client::unregisterLocalMemory( tl::expected Client::IsExist(const std::string& key) { auto result = master_client_.ExistKey(key); - if (!result) { - if (storage_backend_) { - // If master query fails, check storage backend - if (storage_backend_->Existkey(key)) { - return true; // Key exists in storage backend - } - } - return tl::unexpected(result.error()); - } - return result.value(); + return result; } std::vector> Client::BatchIsExist( @@ -1173,7 +1155,8 @@ void Client::PrepareStorageBackend(const std::string& storage_root_dir, } void Client::PutToLocalFile(const std::string& key, - const std::vector& slices) { + const std::vector& slices, + const DiskDescriptor& disk_descriptor) { if (!storage_backend_) return; size_t total_size = 0; @@ -1181,6 +1164,7 @@ void Client::PutToLocalFile(const std::string& key, total_size += slice.size; } + std::string path = disk_descriptor.file_path; // Currently, persistence is achieved through asynchronous writes, but // before asynchronous writing in 3FS, significant performance degradation // may occur due to data copying. Profiling reveals that the number of page @@ -1194,10 +1178,28 @@ void Client::PutToLocalFile(const std::string& key, value.append(static_cast(slice.ptr), slice.size); } - write_thread_pool_.enqueue( - [backend = storage_backend_, key, value = std::move(value)] { - backend->StoreObject(key, value); - }); + write_thread_pool_.enqueue([this, backend = storage_backend_, key, + value = std::move(value), path] { + // Store the object + auto store_result = backend->StoreObject(path, value); + ReplicaType replica_type = ReplicaType::DISK; + + if (!store_result) { + // If storage failed, revoke the put operation + LOG(ERROR) << "Failed to store object for key: " << key; + auto revoke_result = master_client_.PutRevoke(key, replica_type); + if (!revoke_result) { + LOG(ERROR) << "Failed to revoke put operation for key: " << key; + } + return; + } + + // If storage succeeded, end the put operation + auto end_result = master_client_.PutEnd(key, replica_type); + if (!end_result) { + LOG(ERROR) << "Failed to end put operation for key: " << key; + } + }); } ErrorCode Client::TransferData(const Replica::Descriptor& replica_descriptor, @@ -1235,7 +1237,7 @@ ErrorCode Client::TransferRead(const Replica::Descriptor& replica_descriptor, } } else { auto& disk_desc = replica_descriptor.get_disk_descriptor(); - total_size = disk_desc.file_size; + total_size = disk_desc.object_size; } size_t slices_size = CalculateSliceSize(slices); diff --git a/mooncake-store/src/client_buffer.cpp b/mooncake-store/src/client_buffer.cpp index 795b821f..f715e7d2 100644 --- a/mooncake-store/src/client_buffer.cpp +++ b/mooncake-store/src/client_buffer.cpp @@ -67,7 +67,7 @@ uint64_t calculate_total_size(const Replica::Descriptor& replica) { uint64_t total_length = 0; if (replica.is_memory_replica() == false) { auto& disk_descriptor = replica.get_disk_descriptor(); - total_length = disk_descriptor.file_size; + total_length = disk_descriptor.object_size; } else { for (auto& handle : replica.get_memory_descriptor().buffer_descriptors) { @@ -83,7 +83,7 @@ int allocateSlices(std::vector& slices, uint64_t offset = 0; if (replica.is_memory_replica() == false) { // For disk-based replica, split into slices based on file size - uint64_t total_length = replica.get_disk_descriptor().file_size; + uint64_t total_length = replica.get_disk_descriptor().object_size; while (offset < total_length) { auto chunk_size = std::min(total_length - offset, kMaxSliceSize); void* chunk_ptr = static_cast(buffer_handle.ptr()) + offset; diff --git a/mooncake-store/src/ha_helper.cpp b/mooncake-store/src/ha_helper.cpp index 0aa95f8c..3ec6b5bd 100644 --- a/mooncake-store/src/ha_helper.cpp +++ b/mooncake-store/src/ha_helper.cpp @@ -111,7 +111,8 @@ MasterServiceSupervisor::MasterServiceSupervisor( etcd_endpoints_(master_config.etcd_endpoints), local_hostname_(master_config.rpc_address + ":" + std::to_string(master_config.rpc_port)), - cluster_id_(master_config.cluster_id) { + cluster_id_(master_config.cluster_id), + root_fs_dir_(master_config.root_fs_dir) { if (master_config.memory_allocator == "cachelib") { memory_allocator_ = BufferAllocatorType::CACHELIB; } else { @@ -156,7 +157,7 @@ int MasterServiceSupervisor::Start() { enable_gc_, default_kv_lease_ttl_, default_kv_soft_pin_ttl_, allow_evict_soft_pinned_objects_, enable_metric_reporting_, metrics_port_, eviction_ratio_, eviction_high_watermark_ratio_, - version, client_live_ttl_sec_, enable_ha, cluster_id_, + version, client_live_ttl_sec_, enable_ha, cluster_id_, root_fs_dir_, memory_allocator_); mooncake::RegisterRpcService(server, wrapped_master_service); // Metric reporting is now handled by WrappedMasterService. diff --git a/mooncake-store/src/hf3fs/README.md b/mooncake-store/src/hf3fs/README.md index d629dc43..a542b05a 100644 --- a/mooncake-store/src/hf3fs/README.md +++ b/mooncake-store/src/hf3fs/README.md @@ -23,10 +23,11 @@ cmake -DUSE_3FS=ON ... ## Usage ### Basic Operation -Set the environment variable to specify the 3FS mount point: +Start master server and specify the 3FS mount point: ```bash -MOONCAKE_STORAGE_ROOT_DIR=/path/to/3fs_mount_point python3 ... +./build/mooncake-store/src/mooncake_master \ + --root_fs_dir=/path/to/3fs_mount_point ``` ### Important Notes 1. The specified directory **must** be a 3FS mount point diff --git a/mooncake-store/src/hf3fs/hf3fs_file.cpp b/mooncake-store/src/hf3fs/hf3fs_file.cpp index c50b094f..1da03af4 100644 --- a/mooncake-store/src/hf3fs/hf3fs_file.cpp +++ b/mooncake-store/src/hf3fs/hf3fs_file.cpp @@ -49,13 +49,7 @@ tl::expected ThreeFSFile::write(std::span data, return make_error(ErrorCode::FILE_OPEN_FAIL); } - // 3. Acquire write lock - auto lock = acquire_write_lock(); - if (!lock.is_locked()) { - return make_error(ErrorCode::FILE_LOCK_FAIL); - } - - // 4. Write in chunks + // 3. Write in chunks auto& threefs_iov = resource->iov_; auto& ior_write = resource->ior_write_; const char* data_ptr = data.data(); @@ -121,13 +115,7 @@ tl::expected ThreeFSFile::read(std::string& buffer, return make_error(ErrorCode::FILE_OPEN_FAIL); } - // 3. Acquire read lock - auto lock = acquire_read_lock(); - if (!lock.is_locked()) { - return make_error(ErrorCode::FILE_LOCK_FAIL); - } - - // 4. Prepare buffer + // 3. Prepare buffer buffer.clear(); buffer.reserve(length); size_t total_bytes_read = 0; @@ -194,11 +182,6 @@ tl::expected ThreeFSFile::vector_write(const iovec* iov, auto& threefs_iov = resource->iov_; auto& ior_write = resource->ior_write_; - auto lock = acquire_write_lock(); - if (!lock.is_locked()) { - return make_error(ErrorCode::FILE_LOCK_FAIL); - } - // 1. Calculate total length size_t total_length = 0; for (int i = 0; i < iovcnt; ++i) { @@ -286,11 +269,6 @@ tl::expected ThreeFSFile::vector_read(const iovec* iov, auto& threefs_iov = resource->iov_; auto& ior_read = resource->ior_read_; - auto lock = acquire_read_lock(); - if (!lock.is_locked()) { - return make_error(ErrorCode::FILE_LOCK_FAIL); - } - // Calculate total length size_t total_length = 0; for (int i = 0; i < iovcnt; ++i) { diff --git a/mooncake-store/src/master.cpp b/mooncake-store/src/master.cpp index 21bb6dc4..3d347aba 100644 --- a/mooncake-store/src/master.cpp +++ b/mooncake-store/src/master.cpp @@ -66,6 +66,8 @@ DEFINE_int64(client_ttl, mooncake::DEFAULT_CLIENT_LIVE_TTL_SEC, "How long a client is considered alive after the last ping, only " "used in HA mode"); +DEFINE_string(root_fs_dir, mooncake::DEFAULT_ROOT_FS_DIR, + "Root directory for storage backend, used in HA mode"); DEFINE_string(cluster_id, mooncake::DEFAULT_CLUSTER_ID, "Cluster ID for the master service, used for kvcache persistence " "in HA mode"); @@ -119,6 +121,8 @@ void InitMasterConf(const mooncake::DefaultConfig& default_config, FLAGS_etcd_endpoints); default_config.GetString("cluster_id", &master_config.cluster_id, FLAGS_cluster_id); + default_config.GetString("root_fs_dir", &master_config.root_fs_dir, + FLAGS_root_fs_dir); default_config.GetString("memory_allocator", &master_config.memory_allocator, FLAGS_memory_allocator); @@ -250,6 +254,11 @@ void LoadConfigFromCmdline(mooncake::MasterConfig& master_config, !conf_set) { master_config.cluster_id = FLAGS_cluster_id; } + if ((google::GetCommandLineFlagInfo("root_fs_dir", &info) && + !info.is_default) || + !conf_set) { + master_config.root_fs_dir = FLAGS_root_fs_dir; + } if ((google::GetCommandLineFlagInfo("memory_allocator", &info) && !info.is_default) || !conf_set) { @@ -318,6 +327,7 @@ int main(int argc, char* argv[]) { << ", rpc_enable_tcp_no_delay=" << master_config.rpc_enable_tcp_no_delay << ", cluster_id=" << master_config.cluster_id + << ", root_fs_dir=" << master_config.root_fs_dir << ", memory_allocator=" << master_config.memory_allocator; if (master_config.enable_ha) { @@ -347,7 +357,8 @@ int main(int argc, char* argv[]) { master_config.eviction_ratio, master_config.eviction_high_watermark_ratio, version, master_config.client_live_ttl_sec, master_config.enable_ha, - master_config.cluster_id, allocator_type); + master_config.cluster_id, master_config.root_fs_dir, + allocator_type); mooncake::RegisterRpcService(server, wrapped_master_service); return server.start(); diff --git a/mooncake-store/src/master_client.cpp b/mooncake-store/src/master_client.cpp index 3287f71b..d8825f83 100644 --- a/mooncake-store/src/master_client.cpp +++ b/mooncake-store/src/master_client.cpp @@ -308,11 +308,13 @@ MasterClient::BatchPutStart( return result; } -tl::expected MasterClient::PutEnd(const std::string& key) { +tl::expected MasterClient::PutEnd(const std::string& key, + ReplicaType replica_type) { ScopedVLogTimer timer(1, "MasterClient::PutEnd"); timer.LogRequest("key=", key); - auto result = invoke_rpc<&WrappedMasterService::PutEnd, void>(key); + auto result = + invoke_rpc<&WrappedMasterService::PutEnd, void>(key, replica_type); timer.LogResponseExpected(result); return result; } @@ -328,11 +330,13 @@ std::vector> MasterClient::BatchPutEnd( return result; } -tl::expected MasterClient::PutRevoke(const std::string& key) { +tl::expected MasterClient::PutRevoke( + const std::string& key, ReplicaType replica_type) { ScopedVLogTimer timer(1, "MasterClient::PutRevoke"); timer.LogRequest("key=", key); - auto result = invoke_rpc<&WrappedMasterService::PutRevoke, void>(key); + auto result = + invoke_rpc<&WrappedMasterService::PutRevoke, void>(key, replica_type); timer.LogResponseExpected(result); return result; } diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index 17e3c8ad..491390a8 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -17,7 +17,8 @@ MasterService::MasterService( uint64_t default_kv_soft_pin_ttl, bool allow_evict_soft_pinned_objects, double eviction_ratio, double eviction_high_watermark_ratio, ViewVersionId view_version, int64_t client_live_ttl_sec, bool enable_ha, - const std::string& cluster_id, BufferAllocatorType memory_allocator) + const std::string& cluster_id, const std::string& root_fs_dir, + BufferAllocatorType memory_allocator) : enable_gc_(enable_gc), default_kv_lease_ttl_(default_kv_lease_ttl), default_kv_soft_pin_ttl_(default_kv_soft_pin_ttl), @@ -27,6 +28,7 @@ MasterService::MasterService( client_live_ttl_sec_(client_live_ttl_sec), enable_ha_(enable_ha), cluster_id_(cluster_id), + root_fs_dir_(root_fs_dir), segment_manager_(memory_allocator), allocation_strategy_(std::make_shared()) { if (eviction_ratio_ < 0.0 || eviction_ratio_ > 1.0) { @@ -51,6 +53,10 @@ MasterService::MasterService( std::thread(&MasterService::ClientMonitorFunc, this); VLOG(1) << "action=start_client_monitor_thread"; } + + if (!root_fs_dir_.empty()) { + use_disk_replica_ = true; + } } MasterService::~MasterService() { @@ -165,6 +171,7 @@ void MasterService::ClearInvalidHandles() { auto it = shard.metadata.begin(); while (it != shard.metadata.end()) { if (CleanupStaleHandles(it->second)) { + // If the object is empty, we need to erase the iterator it = shard.metadata.erase(it); } else { ++it; @@ -216,16 +223,17 @@ auto MasterService::ExistKey(const std::string& key) } auto& metadata = accessor.Get(); - if (auto status = metadata.HasDiffRepStatus(ReplicaStatus::COMPLETE)) { - LOG(WARNING) << "key=" << key << ", status=" << *status - << ", error=replica_not_ready"; - return tl::make_unexpected(ErrorCode::REPLICA_IS_NOT_READY); + for (const auto& replica : metadata.replicas) { + if (replica.status() == ReplicaStatus::COMPLETE) { + // Grant a lease to the object as it may be further used by the + // client. + metadata.GrantLease(default_kv_lease_ttl_, + default_kv_soft_pin_ttl_); + return true; + } } - // Grant a lease to the object as it may be further used by the client. - metadata.GrantLease(default_kv_lease_ttl_, default_kv_soft_pin_ttl_); - - return true; + return false; // If no complete replica is found, return false } std::vector> MasterService::BatchExistKey( @@ -280,16 +288,18 @@ auto MasterService::GetReplicaList(std::string_view key) return tl::make_unexpected(ErrorCode::OBJECT_NOT_FOUND); } auto& metadata = accessor.Get(); - if (auto status = metadata.HasDiffRepStatus(ReplicaStatus::COMPLETE)) { - LOG(WARNING) << "key=" << key << ", status=" << *status - << ", error=replica_not_ready"; - return tl::make_unexpected(ErrorCode::REPLICA_IS_NOT_READY); - } std::vector replica_list; replica_list.reserve(metadata.replicas.size()); for (const auto& replica : metadata.replicas) { - replica_list.emplace_back(replica.get_descriptor()); + if (replica.status() == ReplicaStatus::COMPLETE) { + replica_list.emplace_back(replica.get_descriptor()); + } + } + + if (replica_list.empty()) { + LOG(WARNING) << "key=" << key << ", error=replica_not_ready"; + return tl::make_unexpected(ErrorCode::REPLICA_IS_NOT_READY); } // Only mark for GC if enabled @@ -357,7 +367,7 @@ auto MasterService::PutStart(const std::string& key, // Allocate replicas std::vector replicas; - replicas.reserve(config.replica_num); + replicas.reserve(config.replica_num + use_disk_replica_); { ScopedAllocatorAccess allocator_access = segment_manager_.getAllocatorAccess(); @@ -393,6 +403,14 @@ auto MasterService::PutStart(const std::string& key, } } + // If disk replica is enabled, allocate a disk replica + if (use_disk_replica_) { + // Allocate a file path for the disk replica + std::string file_path = ResolvePath(key); + replicas.emplace_back(file_path, total_length, + ReplicaStatus::PROCESSING); + } + std::vector replica_list; replica_list.reserve(replicas.size()); for (const auto& replica : replicas) { @@ -408,7 +426,7 @@ auto MasterService::PutStart(const std::string& key, return replica_list; } -auto MasterService::PutEnd(const std::string& key) +auto MasterService::PutEnd(const std::string& key, ReplicaType replica_type) -> tl::expected { MetadataAccessor accessor(this, key); if (!accessor.Exists()) { @@ -418,7 +436,9 @@ auto MasterService::PutEnd(const std::string& key) auto& metadata = accessor.Get(); for (auto& replica : metadata.replicas) { - replica.mark_complete(); + if (replica.type() == replica_type) { + replica.mark_complete(); + } } // 1. Set lease timeout to now, indicating that the object has no lease // at beginning. 2. If this object has soft pin enabled, set it to be soft @@ -427,7 +447,7 @@ auto MasterService::PutEnd(const std::string& key) return {}; } -auto MasterService::PutRevoke(const std::string& key) +auto MasterService::PutRevoke(const std::string& key, ReplicaType replica_type) -> tl::expected { MetadataAccessor accessor(this, key); if (!accessor.Exists()) { @@ -436,13 +456,16 @@ auto MasterService::PutRevoke(const std::string& key) } auto& metadata = accessor.Get(); - if (auto status = metadata.HasDiffRepStatus(ReplicaStatus::PROCESSING)) { + if (auto status = metadata.HasDiffRepStatus(ReplicaStatus::PROCESSING, + replica_type)) { LOG(ERROR) << "key=" << key << ", status=" << *status << ", error=invalid_replica_status"; return tl::make_unexpected(ErrorCode::INVALID_WRITE); } - - accessor.Erase(); + metadata.EraseReplica(replica_type); + if (metadata.IsValid() == false) { + accessor.Erase(); + } return {}; } @@ -451,7 +474,7 @@ std::vector> MasterService::BatchPutEnd( std::vector> results; results.reserve(keys.size()); for (const auto& key : keys) { - results.emplace_back(PutEnd(key)); + results.emplace_back(PutEnd(key, ReplicaType::MEMORY)); } return results; } @@ -461,7 +484,7 @@ std::vector> MasterService::BatchPutRevoke( std::vector> results; results.reserve(keys.size()); for (const auto& key : keys) { - results.emplace_back(PutRevoke(key)); + results.emplace_back(PutRevoke(key, ReplicaType::MEMORY)); } return results; } @@ -481,9 +504,8 @@ auto MasterService::Remove(const std::string& key) return tl::make_unexpected(ErrorCode::OBJECT_HAS_LEASE); } - if (auto status = metadata.HasDiffRepStatus(ReplicaStatus::COMPLETE)) { - LOG(ERROR) << "key=" << key << ", status=" << *status - << ", error=invalid_replica_status"; + if (!metadata.IsAllReplicasComplete()) { + LOG(ERROR) << "key=" << key << ", error=replica_not_ready"; return tl::make_unexpected(ErrorCode::REPLICA_IS_NOT_READY); } @@ -510,7 +532,7 @@ long MasterService::RemoveAll() { while (it != shard.metadata.end()) { if (it->second.IsLeaseExpired(now)) { total_freed_size += - it->second.size * it->second.replicas.size(); + it->second.size * it->second.GetMemReplicaCount(); it = shard.metadata.erase(it); removed_count++; } else { @@ -544,10 +566,10 @@ bool MasterService::CleanupStaleHandles(ObjectMetadata& metadata) { auto replica_it = metadata.replicas.begin(); while (replica_it != metadata.replicas.end()) { // Use any_of algorithm to check if any handle has an invalid allocator - bool has_invalid_handle = replica_it->has_invalid_handle(); + bool has_invalid_mem_handle = replica_it->has_invalid_mem_handle(); // Remove replicas with invalid handles using erase-remove idiom - if (has_invalid_handle) { + if (has_invalid_mem_handle) { replica_it = metadata.replicas.erase(replica_it); } else { ++replica_it; @@ -593,11 +615,11 @@ auto MasterService::Ping(const UUID& client_id) } tl::expected MasterService::GetFsdir() const { - if (cluster_id_.empty()) { - LOG(ERROR) << "Cluster ID is not initialized"; + if (root_fs_dir_.empty() || cluster_id_.empty()) { + LOG(ERROR) << "root_fs_dir or cluster_id is not set"; return tl::make_unexpected(ErrorCode::INVALID_PARAMS); } - return cluster_id_; + return root_fs_dir_ + "/" + cluster_id_; } void MasterService::GCThreadFunc() { @@ -699,7 +721,8 @@ void MasterService::BatchEvict(double evict_ratio_target, it++) { // Skip objects that are not expired or have incomplete replicas if (!it->second.IsLeaseExpired(now) || - it->second.HasDiffRepStatus(ReplicaStatus::COMPLETE)) { + it->second.HasDiffRepStatus(ReplicaStatus::COMPLETE, + ReplicaType::MEMORY)) { continue; } if (!it->second.IsSoftPinned(now)) { @@ -733,15 +756,23 @@ void MasterService::BatchEvict(double evict_ratio_target, // pass if (!it->second.IsLeaseExpired(now) || it->second.IsSoftPinned(now) || - it->second.HasDiffRepStatus(ReplicaStatus::COMPLETE)) { + it->second.HasDiffRepStatus(ReplicaStatus::COMPLETE, + ReplicaType::MEMORY) || + !it->second.HasMemReplica()) { ++it; continue; } if (it->second.lease_timeout <= target_timeout) { // Evict this object total_freed_size += - it->second.size * it->second.replicas.size(); - it = shard.metadata.erase(it); + it->second.size * it->second.GetMemReplicaCount(); + it->second.EraseReplica( + ReplicaType::MEMORY); // Erase memory replicas + if (it->second.IsValid() == false) { + it = shard.metadata.erase(it); + } else { + ++it; + } shard_evicted_count++; } else { // second pass candidates @@ -789,11 +820,19 @@ void MasterService::BatchEvict(double evict_ratio_target, while (it != shard.metadata.end() && target_evict_num > 0) { if (it->second.lease_timeout <= target_timeout && !it->second.IsSoftPinned(now) && - !it->second.HasDiffRepStatus(ReplicaStatus::COMPLETE)) { + !it->second.HasDiffRepStatus(ReplicaStatus::COMPLETE, + ReplicaType::MEMORY) && + it->second.HasMemReplica()) { // Evict this object total_freed_size += - it->second.size * it->second.replicas.size(); - it = shard.metadata.erase(it); + it->second.size * it->second.GetMemReplicaCount(); + it->second.EraseReplica( + ReplicaType::MEMORY); // Erase memory replicas + if (it->second.IsValid() == false) { + it = shard.metadata.erase(it); + } else { + ++it; + } evicted_count++; target_evict_num--; } else { @@ -828,7 +867,9 @@ void MasterService::BatchEvict(double evict_ratio_target, // Skip objects that are not expired or have incomplete // replicas if (!it->second.IsLeaseExpired(now) || - it->second.HasDiffRepStatus(ReplicaStatus::COMPLETE)) { + it->second.HasDiffRepStatus(ReplicaStatus::COMPLETE, + ReplicaType::MEMORY) || + !it->second.HasMemReplica()) { ++it; continue; } @@ -837,8 +878,14 @@ void MasterService::BatchEvict(double evict_ratio_target, if (!it->second.IsSoftPinned(now) || it->second.lease_timeout <= soft_target_timeout) { total_freed_size += - it->second.size * it->second.replicas.size(); - it = shard.metadata.erase(it); + it->second.size * it->second.GetMemReplicaCount(); + it->second.EraseReplica( + ReplicaType::MEMORY); // Erase memory replicas + if (it->second.IsValid() == false) { + it = shard.metadata.erase(it); + } else { + ++it; + } evicted_count++; target_evict_num--; } else { @@ -969,4 +1016,39 @@ void MasterService::ClientMonitorFunc() { } } +std::string MasterService::SanitizeKey(const std::string& key) const { + // Set of invalid filesystem characters to be replaced + constexpr std::string_view kInvalidChars = "/\\:*?\"<>|"; + std::string sanitized_key; + sanitized_key.reserve(key.size()); + + for (char c : key) { + // Replace invalid characters with underscore + sanitized_key.push_back( + kInvalidChars.find(c) != std::string_view::npos ? '_' : c); + } + return sanitized_key; +} + +std::string MasterService::ResolvePath(const std::string& key) const { + // Compute hash of the key + size_t hash = std::hash{}(key); + + // Use low 8 bits to create 2-level directory structure (e.g. "a1/b2") + char dir1 = + static_cast('a' + (hash & 0x0F)); // Lower 4 bits -> 16 dirs + char dir2 = static_cast( + 'a' + ((hash >> 4) & 0x0F)); // Next 4 bits -> 16 subdirs + + // Safely construct path using std::filesystem + namespace fs = std::filesystem; + fs::path dir_path = fs::path(std::string(1, dir1)) / std::string(1, dir2); + + // Combine directory path with sanitized filename + fs::path full_path = + fs::path(root_fs_dir_) / cluster_id_ / dir_path / SanitizeKey(key); + + return full_path.lexically_normal().string(); +} + } // namespace mooncake diff --git a/mooncake-store/src/posix_file.cpp b/mooncake-store/src/posix_file.cpp index 4ebbae7c..17611258 100644 --- a/mooncake-store/src/posix_file.cpp +++ b/mooncake-store/src/posix_file.cpp @@ -50,11 +50,6 @@ tl::expected PosixFile::write(std::span data, return make_error(ErrorCode::FILE_INVALID_BUFFER); } - auto lock = acquire_write_lock(); - if (!lock.is_locked()) { - return make_error(ErrorCode::FILE_LOCK_FAIL); - } - size_t remaining = length; size_t written_bytes = 0; const char *ptr = data.data(); @@ -86,11 +81,6 @@ tl::expected PosixFile::read(std::string &buffer, return make_error(ErrorCode::FILE_INVALID_BUFFER); } - auto lock = acquire_read_lock(); - if (!lock.is_locked()) { - return make_error(ErrorCode::FILE_LOCK_FAIL); - } - buffer.resize(length); size_t read_bytes = 0; char *ptr = buffer.data(); @@ -121,11 +111,6 @@ tl::expected PosixFile::vector_write(const iovec *iov, return make_error(ErrorCode::FILE_NOT_FOUND); } - auto lock = acquire_write_lock(); - if (!lock.is_locked()) { - return make_error(ErrorCode::FILE_LOCK_FAIL); - } - ssize_t ret = ::pwritev(fd_, iov, iovcnt, offset); if (ret < 0) { return make_error(ErrorCode::FILE_WRITE_FAIL); @@ -141,11 +126,6 @@ tl::expected PosixFile::vector_read(const iovec *iov, return make_error(ErrorCode::FILE_NOT_FOUND); } - auto lock = acquire_read_lock(); - if (!lock.is_locked()) { - return make_error(ErrorCode::FILE_LOCK_FAIL); - } - ssize_t ret = ::preadv(fd_, iov, iovcnt, offset); if (ret < 0) { return make_error(ErrorCode::FILE_READ_FAIL); diff --git a/mooncake-store/src/rpc_service.cpp b/mooncake-store/src/rpc_service.cpp index a65a8ded..00e9df94 100644 --- a/mooncake-store/src/rpc_service.cpp +++ b/mooncake-store/src/rpc_service.cpp @@ -29,11 +29,12 @@ WrappedMasterService::WrappedMasterService( bool enable_metric_reporting, uint16_t http_port, double eviction_ratio, double eviction_high_watermark_ratio, ViewVersionId view_version, int64_t client_live_ttl_sec, bool enable_ha, const std::string& cluster_id, - BufferAllocatorType memory_allocator) + const std::string& root_fs_dir, BufferAllocatorType memory_allocator) : master_service_(enable_gc, default_kv_lease_ttl, default_kv_soft_pin_ttl, allow_evict_soft_pinned_objects, eviction_ratio, eviction_high_watermark_ratio, view_version, - client_live_ttl_sec, enable_ha, cluster_id, + client_live_ttl_sec, enable_ha, cluster_id, root_fs_dir, + memory_allocator), http_server_(4, http_port), metric_report_running_(enable_metric_reporting) { @@ -289,19 +290,24 @@ WrappedMasterService::PutStart(const std::string& key, } tl::expected WrappedMasterService::PutEnd( - const std::string& key) { + const std::string& key, ReplicaType replica_type) { return execute_rpc( - "PutEnd", [&] { return master_service_.PutEnd(key); }, - [&](auto& timer) { timer.LogRequest("key=", key); }, + "PutEnd", [&] { return master_service_.PutEnd(key, replica_type); }, + [&](auto& timer) { + timer.LogRequest("key=", key, ", replica_type=", replica_type); + }, [] { MasterMetricManager::instance().inc_put_end_requests(); }, [] { MasterMetricManager::instance().inc_put_end_failures(); }); } tl::expected WrappedMasterService::PutRevoke( - const std::string& key) { + const std::string& key, ReplicaType replica_type) { return execute_rpc( - "PutRevoke", [&] { return master_service_.PutRevoke(key); }, - [&](auto& timer) { timer.LogRequest("key=", key); }, + "PutRevoke", + [&] { return master_service_.PutRevoke(key, replica_type); }, + [&](auto& timer) { + timer.LogRequest("key=", key, ", replica_type=", replica_type); + }, [] { MasterMetricManager::instance().inc_put_revoke_requests(); }, [] { MasterMetricManager::instance().inc_put_revoke_failures(); }); } @@ -359,7 +365,7 @@ std::vector> WrappedMasterService::BatchPutEnd( results.reserve(keys.size()); for (const auto& key : keys) { - results.emplace_back(master_service_.PutEnd(key)); + results.emplace_back(master_service_.PutEnd(key, ReplicaType::MEMORY)); } size_t failure_count = 0; @@ -396,7 +402,8 @@ std::vector> WrappedMasterService::BatchPutRevoke( results.reserve(keys.size()); for (const auto& key : keys) { - results.emplace_back(master_service_.PutRevoke(key)); + results.emplace_back( + master_service_.PutRevoke(key, ReplicaType::MEMORY)); } size_t failure_count = 0; diff --git a/mooncake-store/src/storage_backend.cpp b/mooncake-store/src/storage_backend.cpp index e72d613c..b1880d48 100644 --- a/mooncake-store/src/storage_backend.cpp +++ b/mooncake-store/src/storage_backend.cpp @@ -10,13 +10,8 @@ namespace mooncake { tl::expected StorageBackend::StoreObject( - const ObjectKey& key, const std::vector& slices) { - std::string path = ResolvePath(key); - - if (std::filesystem::exists(path)) { - return tl::make_unexpected(ErrorCode::FILE_OPEN_FAIL); - } - + const std::string& path, const std::vector& slices) { + ResolvePath(path); auto file = create_file(path, FileMode::Write); if (!file) { LOG(INFO) << "Failed to open file for writing: " << path; @@ -50,18 +45,13 @@ tl::expected StorageBackend::StoreObject( } tl::expected StorageBackend::StoreObject( - const ObjectKey& key, const std::string& str) { - return StoreObject(key, std::span(str.data(), str.size())); + const std::string& path, const std::string& str) { + return StoreObject(path, std::span(str.data(), str.size())); } tl::expected StorageBackend::StoreObject( - const ObjectKey& key, std::span data) { - std::string path = ResolvePath(key); - - if (std::filesystem::exists(path)) { - return tl::make_unexpected(ErrorCode::FILE_OPEN_FAIL); - } - + const std::string& path, std::span data) { + ResolvePath(path); auto file = create_file(path, FileMode::Write); if (!file) { LOG(INFO) << "Failed to open file for writing: " << path; @@ -87,7 +77,8 @@ tl::expected StorageBackend::StoreObject( } tl::expected StorageBackend::LoadObject( - std::string& path, std::vector& slices, size_t length) { + const std::string& path, std::vector& slices, size_t length) { + ResolvePath(path); auto file = create_file(path, FileMode::Read); if (!file) { LOG(INFO) << "Failed to open file for reading: " << path; @@ -116,9 +107,9 @@ tl::expected StorageBackend::LoadObject( return {}; } -tl::expected StorageBackend::LoadObject(std::string& path, - std::string& str, - size_t length) { +tl::expected StorageBackend::LoadObject( + const std::string& path, std::string& str, size_t length) { + ResolvePath(path); auto file = create_file(path, FileMode::Read); if (!file) { LOG(INFO) << "Failed to open file for reading: " << path; @@ -140,65 +131,7 @@ tl::expected StorageBackend::LoadObject(std::string& path, return {}; } -bool StorageBackend::Existkey(const ObjectKey& key) { - std::string path = ResolvePath(key); - namespace fs = std::filesystem; - - // Check if the file exists - if (fs::exists(path)) { - return true; - } else { - return false; - } -} - -std::optional StorageBackend::Querykey( - const ObjectKey& key) { - std::string path = ResolvePath(key); - namespace fs = std::filesystem; - - // Check if the file exists - if (!fs::exists(path)) { - return std::nullopt; // File does not exist - } - - // Populate object_info with file metadata - Replica::Descriptor desc; - auto& disk_desc = desc.descriptor_variant.emplace(); - disk_desc.file_path = path; - disk_desc.file_size = fs::file_size(path); - desc.status = ReplicaStatus::COMPLETE; - - return desc; -} - -std::unordered_map -StorageBackend::BatchQueryKey(const std::vector& keys) { - namespace fs = std::filesystem; - std::unordered_map result; - - for (const auto& key : keys) { - std::string path = ResolvePath(key); - - if (!fs::exists(path)) { - LOG(WARNING) << "Key not found: " << key << ", skipping..."; - return {}; - } - - Replica::Descriptor desc; - auto& disk_desc = desc.descriptor_variant.emplace(); - disk_desc.file_path = path; - disk_desc.file_size = fs::file_size(path); - desc.status = ReplicaStatus::COMPLETE; - - result.emplace(key, std::move(desc)); - } - - return result; -} - -void StorageBackend::RemoveFile(const ObjectKey& key) { - std::string path = ResolvePath(key); +void StorageBackend::RemoveFile(const std::string& path) { namespace fs = std::filesystem; // TODO: attention: this function is not thread-safe, need to add lock if // used in multi-thread environment Check if the file exists before @@ -232,49 +165,20 @@ void StorageBackend::RemoveAll() { } } -std::string StorageBackend::SanitizeKey(const ObjectKey& key) const { - // Set of invalid filesystem characters to be replaced - constexpr std::string_view kInvalidChars = "/\\:*?\"<>|"; - std::string sanitized_key; - sanitized_key.reserve(key.size()); - - for (char c : key) { - // Replace invalid characters with underscore - sanitized_key.push_back( - kInvalidChars.find(c) != std::string_view::npos ? '_' : c); - } - return sanitized_key; -} - -std::string StorageBackend::ResolvePath(const ObjectKey& key) const { - // Compute hash of the key - size_t hash = std::hash{}(key); - - // Use low 8 bits to create 2-level directory structure (e.g. "a1/b2") - char dir1 = - static_cast('a' + (hash & 0x0F)); // Lower 4 bits -> 16 dirs - char dir2 = static_cast( - 'a' + ((hash >> 4) & 0x0F)); // Next 4 bits -> 16 subdirs - +void StorageBackend::ResolvePath(const std::string& path) const { // Safely construct path using std::filesystem namespace fs = std::filesystem; - fs::path dir_path = fs::path(root_dir_) / fsdir_ / std::string(1, dir1) / - std::string(1, dir2); + fs::path full_path = path; - // Create directory if not exists + // Create all parent directories if they don't exist std::error_code ec; - if (!fs::exists(dir_path)) { - if (!fs::create_directories(dir_path, ec) && ec) { - LOG(INFO) << "Failed to create directory: " << dir_path + fs::path parent_path = full_path.parent_path(); + if (!parent_path.empty() && !fs::exists(parent_path)) { + if (!fs::create_directories(parent_path, ec) && ec) { + LOG(INFO) << "Failed to create directories: " << parent_path << ", error: " << ec.message(); - return ""; // Empty string indicates failure } } - - // Combine directory path with sanitized filename - fs::path full_path = dir_path / SanitizeKey(key); - - return full_path.lexically_normal().string(); } std::unique_ptr StorageBackend::create_file( diff --git a/mooncake-store/src/transfer_task.cpp b/mooncake-store/src/transfer_task.cpp index 98861420..27b550b7 100644 --- a/mooncake-store/src/transfer_task.cpp +++ b/mooncake-store/src/transfer_task.cpp @@ -93,7 +93,7 @@ void FilereadWorkerPool::workerThread() { } auto load_result = backend_->LoadObject( - task.file_path, task.slices, task.file_size); + task.file_path, task.slices, task.object_size); if (load_result) { VLOG(2) << "Fileread task completed successfully with " << task.file_path; @@ -540,7 +540,7 @@ std::optional TransferSubmitter::submitFileReadOperation( auto state = std::make_shared(); auto disk_replica = replica.get_disk_descriptor(); std::string file_path = disk_replica.file_path; - size_t file_length = disk_replica.file_size; + size_t file_length = disk_replica.object_size; // Submit memcpy operations to worker pool for async execution FilereadTask task(file_path, file_length, slices, state); diff --git a/mooncake-store/tests/CMakeLists.txt b/mooncake-store/tests/CMakeLists.txt index 0e1ca581..ea2e862d 100644 --- a/mooncake-store/tests/CMakeLists.txt +++ b/mooncake-store/tests/CMakeLists.txt @@ -14,6 +14,10 @@ add_executable(master_service_test master_service_test.cpp) target_link_libraries(master_service_test PUBLIC mooncake_store cachelib_memory_allocator ${ETCD_WRAPPER_LIB} glog gtest gtest_main pthread) add_test(NAME master_service_test COMMAND master_service_test) +add_executable(master_service_ssd_test master_service_ssd_test.cpp) +target_link_libraries(master_service_ssd_test PUBLIC mooncake_store cachelib_memory_allocator ${ETCD_WRAPPER_LIB} glog gtest gtest_main pthread) +add_test(NAME master_service_ssd_test COMMAND master_service_ssd_test) + add_executable(client_integration_test client_integration_test.cpp) target_link_libraries(client_integration_test PUBLIC mooncake_store diff --git a/mooncake-store/tests/client_buffer_test.cpp b/mooncake-store/tests/client_buffer_test.cpp index 1df5a9fa..f91c15eb 100644 --- a/mooncake-store/tests/client_buffer_test.cpp +++ b/mooncake-store/tests/client_buffer_test.cpp @@ -301,7 +301,7 @@ TEST_F(ClientBufferTest, CalculateTotalSizeDiskReplica) { // Create a disk replica descriptor Replica::Descriptor replica; DiskDescriptor disk_desc; - disk_desc.file_size = 4096; + disk_desc.object_size = 4096; replica.descriptor_variant = disk_desc; replica.status = ReplicaStatus::COMPLETE; @@ -386,7 +386,7 @@ TEST_F(ClientBufferTest, AllocateSlicesDiskReplica) { // Create a disk replica descriptor Replica::Descriptor replica; DiskDescriptor disk_desc; - disk_desc.file_size = 8192; + disk_desc.object_size = 8192; replica.descriptor_variant = disk_desc; replica.status = ReplicaStatus::COMPLETE; diff --git a/mooncake-store/tests/client_integration_test.cpp b/mooncake-store/tests/client_integration_test.cpp index 5a468a39..694bd4ea 100644 --- a/mooncake-store/tests/client_integration_test.cpp +++ b/mooncake-store/tests/client_integration_test.cpp @@ -246,6 +246,17 @@ TEST_F(ClientIntegrationTest, RemoveOperation) { ASSERT_TRUE(remove_result.has_value()) << "Remove operation failed: " << toString(remove_result.error()); + // Verify that the data is removed using Query operation + auto query_result = test_client_->Query(key); + ASSERT_FALSE(query_result.has_value()) + << "Query should not find the removed key: " << key; + + // Check if the key exists using IsExist + auto exist_result = test_client_->IsExist(key); + ASSERT_TRUE(exist_result.has_value()); + ASSERT_FALSE(exist_result.value()) + << "IsExist should return false for removed key: " << key; + // Try to get the removed data - should fail buffer = client_buffer_allocator_->allocate(test_data.size()); slices.clear(); diff --git a/mooncake-store/tests/master_metrics_test.cpp b/mooncake-store/tests/master_metrics_test.cpp index ea0b4334..0c303663 100644 --- a/mooncake-store/tests/master_metrics_test.cpp +++ b/mooncake-store/tests/master_metrics_test.cpp @@ -129,7 +129,7 @@ TEST_F(MasterMetricsTest, BasicRequestTest) { ASSERT_EQ(metrics.get_allocated_size(), value_length); ASSERT_EQ(metrics.get_put_start_requests(), 1); ASSERT_EQ(metrics.get_put_start_failures(), 0); - auto put_revoke_result = service_.PutRevoke(key); + auto put_revoke_result = service_.PutRevoke(key, ReplicaType::MEMORY); ASSERT_TRUE(put_revoke_result.has_value()); ASSERT_EQ(metrics.get_key_count(), 0); ASSERT_EQ(metrics.get_allocated_size(), 0); @@ -143,7 +143,7 @@ TEST_F(MasterMetricsTest, BasicRequestTest) { ASSERT_EQ(metrics.get_allocated_size(), value_length); ASSERT_EQ(metrics.get_put_start_requests(), 2); ASSERT_EQ(metrics.get_put_start_failures(), 0); - auto put_end_result = service_.PutEnd(key); + auto put_end_result = service_.PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); ASSERT_EQ(metrics.get_key_count(), 1); ASSERT_EQ(metrics.get_allocated_size(), value_length); @@ -175,7 +175,7 @@ TEST_F(MasterMetricsTest, BasicRequestTest) { // Test RemoveAll request auto put_start_result3 = service_.PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result3.has_value()); - auto put_end_result2 = service_.PutEnd(key); + auto put_end_result2 = service_.PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result2.has_value()); ASSERT_EQ(metrics.get_key_count(), 1); ASSERT_EQ(1, service_.RemoveAll()); @@ -187,7 +187,7 @@ TEST_F(MasterMetricsTest, BasicRequestTest) { // Test UnmountSegment request auto put_start_result4 = service_.PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result4.has_value()); - auto put_end_result3 = service_.PutEnd(key); + auto put_end_result3 = service_.PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result3.has_value()); auto unmount_result = service_.UnmountSegment(segment_id, client_id); ASSERT_TRUE(unmount_result.has_value()); diff --git a/mooncake-store/tests/master_service_ssd_test.cpp b/mooncake-store/tests/master_service_ssd_test.cpp new file mode 100644 index 00000000..f247e602 --- /dev/null +++ b/mooncake-store/tests/master_service_ssd_test.cpp @@ -0,0 +1,267 @@ +#include "master_service.h" + +#include +#include + +#include +#include +#include +#include +#include + +#include "types.h" + +namespace mooncake::test { + +std::unique_ptr CreateMasterServiceWithSSDFeat( + const std::string& root_fs_dir) { + return std::make_unique( + false, // enable_gc + DEFAULT_DEFAULT_KV_LEASE_TTL, // default_kv_lease_ttl + DEFAULT_KV_SOFT_PIN_TTL_MS, // default_kv_soft_pin_ttl + DEFAULT_ALLOW_EVICT_SOFT_PINNED_OBJECTS, DEFAULT_EVICTION_RATIO, + DEFAULT_EVICTION_HIGH_WATERMARK_RATIO, + 0, // view_version + DEFAULT_CLIENT_LIVE_TTL_SEC, + false, // enable_ha + DEFAULT_CLUSTER_ID, root_fs_dir, BufferAllocatorType::CACHELIB); +} + +class MasterServiceSSDTest : public ::testing::Test { + protected: + void SetUp() override { + google::InitGoogleLogging("MasterServiceTest"); + FLAGS_logtostderr = true; + } + + void TearDown() override { google::ShutdownGoogleLogging(); } +}; + +TEST_F(MasterServiceSSDTest, PutEndBothReplica) { + auto service_ = CreateMasterServiceWithSSDFeat("/mnt/ssd"); + + constexpr size_t buffer = 0x300000000; + constexpr size_t size = 1024 * 1024 * 64; + std::string segment_name = "test_segment"; + Segment segment(generate_uuid(), segment_name, buffer, size); + UUID client_id = generate_uuid(); + + auto mount_result = service_->MountSegment(segment, client_id); + ASSERT_TRUE(mount_result.has_value()); + + std::string key = "disk_key"; + std::vector slice_lengths = {1024}; + ReplicateConfig config; + config.replica_num = 1; + + auto put_start_result = service_->PutStart(key, slice_lengths, config); + ASSERT_TRUE(put_start_result.has_value()); + auto replicas = put_start_result.value(); + ASSERT_EQ(2, replicas.size()); + + bool has_mem = false, has_disk = false; + for (const auto& r : replicas) { + if (r.is_memory_replica()) has_mem = true; + if (r.is_disk_replica()) has_disk = true; + } + EXPECT_TRUE(has_mem); + EXPECT_TRUE(has_disk); + + auto get_result = service_->GetReplicaList(key); + ASSERT_FALSE(get_result.has_value()); + EXPECT_EQ(ErrorCode::REPLICA_IS_NOT_READY, get_result.error()); + + // PutEnd for both memory and disk + EXPECT_TRUE(service_->PutEnd(key, ReplicaType::MEMORY).has_value()); + EXPECT_TRUE(service_->PutEnd(key, ReplicaType::DISK).has_value()); + + get_result = service_->GetReplicaList(key); + ASSERT_TRUE(get_result.has_value()); + EXPECT_EQ(2, get_result.value().size()); + + for (const auto& r : get_result.value()) { + EXPECT_EQ(ReplicaStatus::COMPLETE, r.status); + } +} + +TEST_F(MasterServiceSSDTest, PutRevokeDiskReplica) { + auto service_ = CreateMasterServiceWithSSDFeat("/mnt/ssd"); + + constexpr size_t buffer = 0x300000000; + constexpr size_t size = 1024 * 1024 * 64; + std::string segment_name = "test_segment"; + Segment segment(generate_uuid(), segment_name, buffer, size); + UUID client_id = generate_uuid(); + + auto mount_result = service_->MountSegment(segment, client_id); + ASSERT_TRUE(mount_result.has_value()); + + std::string key = "revoke_key"; + std::vector slice_lengths = {1024}; + ReplicateConfig config; + config.replica_num = 1; + + ASSERT_TRUE(service_->PutStart(key, slice_lengths, config).has_value()); + EXPECT_TRUE(service_->PutEnd(key, ReplicaType::MEMORY).has_value()); + + auto get_result = service_->GetReplicaList(key); + ASSERT_TRUE(get_result.has_value()); + EXPECT_EQ(1, get_result.value().size()); + ASSERT_TRUE(get_result.value()[0].is_memory_replica()); + + EXPECT_TRUE(service_->PutRevoke(key, ReplicaType::DISK).has_value()); + + get_result = service_->GetReplicaList(key); + ASSERT_TRUE(get_result.has_value()); + EXPECT_EQ(1, get_result.value().size()); + ASSERT_TRUE(get_result.value()[0].is_memory_replica()); +} + +TEST_F(MasterServiceSSDTest, PutRevokeMemoryReplica) { + auto service_ = CreateMasterServiceWithSSDFeat("/mnt/ssd"); + + constexpr size_t buffer = 0x300000000; + constexpr size_t size = 1024 * 1024 * 64; + std::string segment_name = "test_segment"; + Segment segment(generate_uuid(), segment_name, buffer, size); + UUID client_id = generate_uuid(); + + auto mount_result = service_->MountSegment(segment, client_id); + ASSERT_TRUE(mount_result.has_value()); + + std::string key = "revoke_key"; + std::vector slice_lengths = {1024}; + ReplicateConfig config; + config.replica_num = 1; + + ASSERT_TRUE(service_->PutStart(key, slice_lengths, config).has_value()); + EXPECT_TRUE(service_->PutRevoke(key, ReplicaType::MEMORY).has_value()); + + auto get_result = service_->GetReplicaList(key); + ASSERT_FALSE(get_result.has_value()); + EXPECT_EQ(ErrorCode::REPLICA_IS_NOT_READY, get_result.error()); + + EXPECT_TRUE(service_->PutEnd(key, ReplicaType::DISK).has_value()); + get_result = service_->GetReplicaList(key); + ASSERT_TRUE(get_result.has_value()); + EXPECT_EQ(1, get_result.value().size()); + ASSERT_TRUE(get_result.value()[0].is_disk_replica()); +} + +TEST_F(MasterServiceSSDTest, PutRevokeBothReplica) { + auto service_ = CreateMasterServiceWithSSDFeat("/mnt/ssd"); + + constexpr size_t buffer = 0x300000000; + constexpr size_t size = 1024 * 1024 * 64; + std::string segment_name = "test_segment"; + Segment segment(generate_uuid(), segment_name, buffer, size); + UUID client_id = generate_uuid(); + + auto mount_result = service_->MountSegment(segment, client_id); + ASSERT_TRUE(mount_result.has_value()); + + std::string key = "revoke_key"; + std::vector slice_lengths = {1024}; + ReplicateConfig config; + config.replica_num = 1; + + ASSERT_TRUE(service_->PutStart(key, slice_lengths, config).has_value()); + EXPECT_TRUE(service_->PutRevoke(key, ReplicaType::DISK).has_value()); + + auto get_result = service_->GetReplicaList(key); + ASSERT_FALSE(get_result.has_value()); + EXPECT_EQ(ErrorCode::REPLICA_IS_NOT_READY, get_result.error()); + + EXPECT_TRUE(service_->PutRevoke(key, ReplicaType::MEMORY).has_value()); + get_result = service_->GetReplicaList(key); + ASSERT_FALSE(get_result.has_value()); + EXPECT_EQ(ErrorCode::OBJECT_NOT_FOUND, get_result.error()); +} + +TEST_F(MasterServiceSSDTest, RemoveKey) { + auto service_ = CreateMasterServiceWithSSDFeat("/mnt/ssd"); + + constexpr size_t buffer = 0x300000000; + constexpr size_t size = 1024 * 1024 * 64; + std::string segment_name = "test_segment"; + Segment segment(generate_uuid(), segment_name, buffer, size); + UUID client_id = generate_uuid(); + + auto mount_result = service_->MountSegment(segment, client_id); + ASSERT_TRUE(mount_result.has_value()); + + std::string key = "remove_key"; + std::vector slice_lengths = {1024}; + ReplicateConfig config; + config.replica_num = 1; + + ASSERT_TRUE(service_->PutStart(key, slice_lengths, config).has_value()); + EXPECT_TRUE(service_->PutEnd(key, ReplicaType::MEMORY).has_value()); + EXPECT_TRUE(service_->PutEnd(key, ReplicaType::DISK).has_value()); + + EXPECT_TRUE(service_->Remove(key).has_value()); + + auto get_result = service_->GetReplicaList(key); + EXPECT_FALSE(get_result.has_value()); + EXPECT_EQ(ErrorCode::OBJECT_NOT_FOUND, get_result.error()); +} + +TEST_F(MasterServiceSSDTest, EvictObject) { + auto service_ = CreateMasterServiceWithSSDFeat("/mnt/ssd"); + // Mount a segment that can hold about 1024 * 16 objects. + // As the eviction is processed separately for each shard, + // we need to fill each shard with enough objects to thoroughly + // test the eviction process. + constexpr size_t buffer = 0x300000000; + constexpr size_t size = 1024 * 1024 * 16 * 15; + constexpr size_t object_size = 1024 * 15; + std::string segment_name = "test_segment"; + Segment segment(generate_uuid(), segment_name, buffer, size); + UUID client_id = generate_uuid(); + auto mount_result = service_->MountSegment(segment, client_id); + ASSERT_TRUE(mount_result.has_value()); + + // Verify if we can put objects more than the segment can hold + int success_puts = 0; + for (int i = 0; i < 1024 * 16 + 50; ++i) { + std::string key = "test_key" + std::to_string(i); + std::vector slice_lengths = {object_size}; + ReplicateConfig config; + config.replica_num = 1; + auto put_start_result = service_->PutStart(key, slice_lengths, config); + if (put_start_result.has_value()) { + auto put_end_mem_result = + service_->PutEnd(key, ReplicaType::MEMORY); + auto put_end_disk_result = service_->PutEnd(key, ReplicaType::DISK); + ASSERT_TRUE(put_end_mem_result.has_value()); + ASSERT_TRUE(put_end_disk_result.has_value()); + success_puts++; + } else { + // wait for gc thread to work + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + } + } + ASSERT_GT(success_puts, 1024 * 16); + + // Verify if we can get objects more than the segment can hold + int success_gets = 0; + for (int i = 0; i < 1024 * 16 + 50; ++i) { + std::string key = "test_key" + std::to_string(i); + auto get_result = service_->GetReplicaList(key); + if (get_result.has_value()) { + success_gets++; + } + } + ASSERT_GT(success_gets, 1024 * 16); + + std::this_thread::sleep_for( + std::chrono::milliseconds(DEFAULT_DEFAULT_KV_LEASE_TTL)); + service_->RemoveAll(); +} + +} // namespace mooncake::test + +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} \ No newline at end of file diff --git a/mooncake-store/tests/master_service_test.cpp b/mooncake-store/tests/master_service_test.cpp index b0ddbf06..c3c2afe0 100644 --- a/mooncake-store/tests/master_service_test.cpp +++ b/mooncake-store/tests/master_service_test.cpp @@ -54,7 +54,7 @@ std::string GenerateKeyForSegment(const std::unique_ptr& service, throw std::runtime_error("PutStart failed with code: " + std::to_string(static_cast(code))); } - auto put_end_result = service->PutEnd(key); + auto put_end_result = service->PutEnd(key, ReplicaType::MEMORY); if (!put_end_result.has_value()) { throw std::runtime_error("PutEnd failed"); } @@ -275,7 +275,7 @@ TEST_F(MasterServiceTest, PutStartEndFlow) { EXPECT_EQ(ErrorCode::REPLICA_IS_NOT_READY, remove_result.error()); // Test PutEnd - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); EXPECT_TRUE(put_end_result.has_value()); // Verify replica list after PutEnd @@ -321,7 +321,7 @@ TEST_F(MasterServiceTest, RandomPutStartEndFlow) { EXPECT_FALSE(remove_result.has_value()); EXPECT_EQ(ErrorCode::REPLICA_IS_NOT_READY, remove_result.error()); // Test PutEnd - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); EXPECT_TRUE(put_end_result.has_value()); // Verify replica list after PutEnd auto get_result2 = service_->GetReplicaList(key); @@ -357,7 +357,7 @@ TEST_F(MasterServiceTest, GetReplicaList) { config.replica_num = 1; auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); // Test getting existing key @@ -386,7 +386,7 @@ TEST_F(MasterServiceTest, RemoveObject) { config.replica_num = 1; auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); // Test removing the object @@ -427,7 +427,7 @@ TEST_F(MasterServiceTest, RandomRemoveObject) { config.replica_num = 1; auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); // Test removing the object @@ -463,7 +463,7 @@ TEST_F(MasterServiceTest, RemoveAll) { config.replica_num = 1; auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); auto exist_result = service_->ExistKey(key); ASSERT_TRUE(exist_result.has_value()); @@ -553,7 +553,7 @@ TEST_F(MasterServiceTest, MultiSliceMultiReplicaFlow) { EXPECT_EQ(ErrorCode::REPLICA_IS_NOT_READY, get_result.error()); // Complete the put operation - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); // Test GetReplicaList after completion @@ -625,7 +625,8 @@ TEST_F(MasterServiceTest, ConcurrentGarbageCollectionTest) { auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = + service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); // Add the key to the tracking list @@ -691,7 +692,7 @@ TEST_F(MasterServiceTest, CleanupStaleHandlesTest) { // Create the object auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); // Verify object exists @@ -718,7 +719,7 @@ TEST_F(MasterServiceTest, CleanupStaleHandlesTest) { std::string key2 = "another_segment_object"; auto put_start_result2 = service_->PutStart(key2, slice_lengths, config); ASSERT_TRUE(put_start_result2.has_value()); - auto put_end_result2 = service_->PutEnd(key2); + auto put_end_result2 = service_->PutEnd(key2, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result2.has_value()); // Verify we can get it @@ -766,7 +767,8 @@ TEST_F(MasterServiceTest, ConcurrentWriteAndRemoveAll) { auto put_start_result = service_->PutStart(key, slice_lengths, config); if (put_start_result.has_value()) { - auto put_end_result = service_->PutEnd(key); + auto put_end_result = + service_->PutEnd(key, ReplicaType::MEMORY); if (put_end_result.has_value()) { success_writes++; } @@ -832,7 +834,7 @@ TEST_F(MasterServiceTest, ConcurrentReadAndRemoveAll) { auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); } @@ -913,7 +915,7 @@ TEST_F(MasterServiceTest, ConcurrentRemoveAllOperations) { auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); } @@ -987,7 +989,7 @@ TEST_F(MasterServiceTest, UnmountSegmentImmediateCleanup) { auto put_start_result = service_->PutStart(key1, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); replica_list = put_start_result.value(); - auto put_end_result = service_->PutEnd(key1); + auto put_end_result = service_->PutEnd(key1, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); auto get_result3 = service_->GetReplicaList(key1); ASSERT_TRUE(get_result3.has_value()); @@ -1027,7 +1029,7 @@ TEST_F(MasterServiceTest, ReadableAfterPartialUnmountWithReplication) { auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); ASSERT_EQ(2u, put_start_result->size()); - ASSERT_TRUE(service_->PutEnd(key).has_value()); + ASSERT_TRUE(service_->PutEnd(key, ReplicaType::MEMORY).has_value()); // Verify two replicas exist and they are on distinct segments auto get_result = service_->GetReplicaList(key); @@ -1134,7 +1136,7 @@ TEST_F(MasterServiceTest, RemoveLeasedObject) { // Verify lease is granted on ExistsKey auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); auto exist_result = service_->ExistKey(key); ASSERT_TRUE(exist_result.has_value()); @@ -1148,7 +1150,7 @@ TEST_F(MasterServiceTest, RemoveLeasedObject) { // Verify lease is extended on successive ExistsKey auto put_start_result2 = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result2.has_value()); - auto put_end_result2 = service_->PutEnd(key); + auto put_end_result2 = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result2.has_value()); auto exist_result2 = service_->ExistKey(key); ASSERT_TRUE(exist_result2.has_value()); @@ -1165,7 +1167,7 @@ TEST_F(MasterServiceTest, RemoveLeasedObject) { // Verify lease is granted on GetReplicaList auto put_start_result3 = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result3.has_value()); - auto put_end_result3 = service_->PutEnd(key); + auto put_end_result3 = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result3.has_value()); auto get_result = service_->GetReplicaList(key); ASSERT_TRUE(get_result.has_value()); @@ -1179,7 +1181,7 @@ TEST_F(MasterServiceTest, RemoveLeasedObject) { // Verify lease is extended on successive GetReplicaList auto put_start_result4 = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result4.has_value()); - auto put_end_result4 = service_->PutEnd(key); + auto put_end_result4 = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result4.has_value()); auto get_result2 = service_->GetReplicaList(key); ASSERT_TRUE(get_result2.has_value()); @@ -1218,7 +1220,7 @@ TEST_F(MasterServiceTest, RemoveAllLeasedObject) { config.replica_num = 1; auto put_start_result = service_->PutStart(key, slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); if (i >= 5) { auto exist_result = service_->ExistKey(key); @@ -1268,7 +1270,7 @@ TEST_F(MasterServiceTest, EvictObject) { config.replica_num = 1; auto put_start_result = service_->PutStart(key, slice_lengths, config); if (put_start_result.has_value()) { - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); success_puts++; } else { @@ -1306,7 +1308,7 @@ TEST_F(MasterServiceTest, TryEvictLeasedObject) { config.replica_num = 1; auto put_start_result = service_->PutStart(key, slice_lengths, config); if (put_start_result.has_value()) { - auto put_end_result = service_->PutEnd(key); + auto put_end_result = service_->PutEnd(key, ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); // the object is leased auto get_result = service_->GetReplicaList(key); @@ -1353,12 +1355,12 @@ TEST_F(MasterServiceTest, RemoveSoftPinObject) { // Verify soft pin does not block remove ASSERT_TRUE(service_->PutStart(key, slice_lengths, config).has_value()); - ASSERT_TRUE(service_->PutEnd(key).has_value()); + ASSERT_TRUE(service_->PutEnd(key, ReplicaType::MEMORY).has_value()); EXPECT_TRUE(service_->Remove(key).has_value()); // Verify soft pin does not block RemoveAll ASSERT_TRUE(service_->PutStart(key, slice_lengths, config).has_value()); - ASSERT_TRUE(service_->PutEnd(key).has_value()); + ASSERT_TRUE(service_->PutEnd(key, ReplicaType::MEMORY).has_value()); EXPECT_EQ(1, service_->RemoveAll()); } @@ -1394,7 +1396,8 @@ TEST_F(MasterServiceTest, SoftPinObjectsNotEvictedBeforeOtherObjects) { ASSERT_TRUE( service_->PutStart(pin_key, slice_lengths, soft_pin_config) .has_value()); - ASSERT_TRUE(service_->PutEnd(pin_key).has_value()); + ASSERT_TRUE( + service_->PutEnd(pin_key, ReplicaType::MEMORY).has_value()); } // Fill the segment to trigger eviction @@ -1405,7 +1408,8 @@ TEST_F(MasterServiceTest, SoftPinObjectsNotEvictedBeforeOtherObjects) { ReplicateConfig config; config.replica_num = 1; if (service_->PutStart(key, slice_lengths, config).has_value()) { - ASSERT_TRUE(service_->PutEnd(key).has_value()); + ASSERT_TRUE( + service_->PutEnd(key, ReplicaType::MEMORY).has_value()); } else { failed_puts++; } @@ -1453,7 +1457,7 @@ TEST_F(MasterServiceTest, SoftPinObjectsCanBeEvicted) { config.replica_num = 1; config.with_soft_pin = true; if (service_->PutStart(key, slice_lengths, config).has_value()) { - ASSERT_TRUE(service_->PutEnd(key).has_value()); + ASSERT_TRUE(service_->PutEnd(key, ReplicaType::MEMORY).has_value()); success_puts++; } else { // wait for gc thread to work @@ -1500,7 +1504,8 @@ TEST_F(MasterServiceTest, SoftPinExtendedOnGet) { ASSERT_TRUE( service_->PutStart(pin_key, slice_lengths, soft_pin_config)); - ASSERT_TRUE(service_->PutEnd(pin_key).has_value()); + ASSERT_TRUE( + service_->PutEnd(pin_key, ReplicaType::MEMORY).has_value()); } // Wait for the soft pin to expire @@ -1520,7 +1525,8 @@ TEST_F(MasterServiceTest, SoftPinExtendedOnGet) { ReplicateConfig config; config.replica_num = 1; if (service_->PutStart(key, slice_lengths, config).has_value()) { - ASSERT_TRUE(service_->PutEnd(key).has_value()); + ASSERT_TRUE( + service_->PutEnd(key, ReplicaType::MEMORY).has_value()); } else { failed_puts++; } @@ -1571,7 +1577,7 @@ TEST_F(MasterServiceTest, SoftPinObjectsNotAllowEvict) { config.replica_num = 1; config.with_soft_pin = true; if (service_->PutStart(key, slice_lengths, config).has_value()) { - ASSERT_TRUE(service_->PutEnd(key).has_value()); + ASSERT_TRUE(service_->PutEnd(key, ReplicaType::MEMORY).has_value()); success_keys.push_back(key); } else { // wait for gc thread to work @@ -1610,7 +1616,8 @@ TEST_F(MasterServiceTest, BatchExistKeyTest) { auto put_start_result = service_->PutStart(test_keys[i], slice_lengths, config); ASSERT_TRUE(put_start_result.has_value()); - auto put_end_result = service_->PutEnd(test_keys[i]); + auto put_end_result = + service_->PutEnd(test_keys[i], ReplicaType::MEMORY); ASSERT_TRUE(put_end_result.has_value()); } diff --git a/mooncake-store/tests/stress_cluster_benchmark.py b/mooncake-store/tests/stress_cluster_benchmark.py index 5e7ac56c..cfd806d5 100644 --- a/mooncake-store/tests/stress_cluster_benchmark.py +++ b/mooncake-store/tests/stress_cluster_benchmark.py @@ -177,9 +177,6 @@ class TestInstance: def setup(self): """Initialize the MooncakeDistributedStore and allocate registered memory.""" - if self.args.root_dir: - os.environ["MOONCAKE_STORAGE_ROOT_DIR"] = self.args.root_dir - logger.info(f"Set storage root directory to: {self.args.root_dir}") self.store = MooncakeDistributedStore() self.performance_tracker.start_timer() @@ -445,7 +442,6 @@ def parse_arguments(): parser.add_argument("--value-length", type=int, default=4*1024*1024, help="Size of each value in bytes") parser.add_argument("--batch-size", type=int, default=1, help="Batch size for operations") parser.add_argument("--wait-time", type=int, default=20, help="Wait time in seconds after operations complete") - parser.add_argument("--root-dir", type=str, default="", help="Root directory for storage (sets MOONCAKE_STORAGE_ROOT_DIR)") # Multi-threading parameters parser.add_argument("--num-workers", type=int, default=1, diff --git a/mooncake-wheel/tests/test_distributed_object_store.py b/mooncake-wheel/tests/test_distributed_object_store.py index de97e597..9a2c2b17 100644 --- a/mooncake-wheel/tests/test_distributed_object_store.py +++ b/mooncake-wheel/tests/test_distributed_object_store.py @@ -64,8 +64,6 @@ class TestDistributedObjectStore(unittest.TestCase): cls.store = MooncakeDistributedStore() get_client(cls.store) - @unittest.skipIf(os.getenv("MOONCAKE_STORAGE_ROOT_DIR"), - "Skipping test_client_tear_down because SSD environment variable is set") def test_client_tear_down(self): """Test client tear down and re-initialization.""" test_data = b"Hello, World!" diff --git a/mooncake-wheel/tests/test_ssd_offload_in_evict.py b/mooncake-wheel/tests/test_ssd_offload_in_evict.py index 30d4eec3..389fee60 100644 --- a/mooncake-wheel/tests/test_ssd_offload_in_evict.py +++ b/mooncake-wheel/tests/test_ssd_offload_in_evict.py @@ -201,8 +201,6 @@ class TestDistributedObjectStore(unittest.TestCase): cls.store = MooncakeDistributedStore() get_client(cls.store) - # @unittest.skipUnless(os.getenv("MOONCAKE_STORAGE_ROOT_DIR"), - # "Skipping test_put_get_in_evict_operations because SSD environment variable is not set") def test_put_get_in_evict_operations(self): """Test basic Put/Get operations with eviction scenario @@ -296,8 +294,6 @@ class TestDistributedObjectStore(unittest.TestCase): index = index + 1 print("Cleanup completed") - # @unittest.skipUnless(os.getenv("MOONCAKE_STORAGE_ROOT_DIR"), - # "Skipping test_concurrent_stress because SSD environment variable is not set") def test_concurrent_stress(self): """Multi-threaded stress test for Put/Get operations diff --git a/scripts/run_tests.sh b/scripts/run_tests.sh index 41813b0b..7ce19b74 100755 --- a/scripts/run_tests.sh +++ b/scripts/run_tests.sh @@ -43,10 +43,10 @@ if [ -n "$TEST_SSD_OFFLOAD_IN_EVICT" ]; then echo "Running with ssd offload in evict tests..." # Set a small kv lease ttl to make the test faster. # Must be consistent with the client test parameters. - mooncake_master --default_kv_lease_ttl=500 & + mooncake_master --default_kv_lease_ttl=500 --root_fs_dir=$TEST_ROOT_DIR & MASTER_PID=$! sleep 1 - MC_METADATA_SERVER=http://127.0.0.1:8080/metadata MOONCAKE_STORAGE_ROOT_DIR=$TEST_ROOT_DIR DEFAULT_KV_LEASE_TTL=500 python test_ssd_offload_in_evict.py + MC_METADATA_SERVER=http://127.0.0.1:8080/metadata DEFAULT_KV_LEASE_TTL=500 python test_ssd_offload_in_evict.py kill $MASTER_PID || true rm -rf $TEST_ROOT_DIR else