Skip to content

Commit 6c7c2a5

Browse files
committed
[fix] enable nfs backend delete checks
1 parent 2d3f88f commit 6c7c2a5

3 files changed

Lines changed: 157 additions & 14 deletions

File tree

kv_cache_manager/data_storage/nfs_backend.cc

Lines changed: 70 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
#include "nfs_backend.h"
22

3+
#include <charconv>
4+
#include <cstdint>
5+
#include <filesystem>
36
#include <memory>
7+
#include <system_error>
48
#include <utility>
59

610
#include "kv_cache_manager/common/hash/hash.h"
@@ -10,6 +14,29 @@
1014

1115
namespace kv_cache_manager {
1216

17+
namespace {
18+
19+
std::pair<ErrorCode, bool> ShouldDeleteFile(const DataStorageUri &storage_uri) {
20+
const std::string blkid = storage_uri.GetParam("blkid");
21+
if (blkid.empty()) {
22+
return {EC_OK, true};
23+
}
24+
25+
std::uint64_t blkid_value = 0;
26+
const auto *begin = blkid.data();
27+
const auto *end = blkid.data() + blkid.size();
28+
const auto [ptr, ec] = std::from_chars(begin, end, blkid_value);
29+
if (ec != std::errc{} || ptr != end) {
30+
KVCM_LOG_WARN("Skip deleting nfs file for invalid blkid [%s], uri: [%s]",
31+
blkid.c_str(),
32+
storage_uri.ToUriString().c_str());
33+
return {EC_ERROR, false};
34+
}
35+
return {EC_OK, blkid_value == 0};
36+
}
37+
38+
} // namespace
39+
1340
NfsBackend::NfsBackend(std::shared_ptr<MetricsRegistry> metrics_registry)
1441
: DataStorageBackend(std::move(metrics_registry)) {}
1542

@@ -79,15 +106,54 @@ std::vector<std::pair<ErrorCode, DataStorageUri>> NfsBackend::Create(const std::
79106
std::vector<ErrorCode> NfsBackend::Delete(const std::vector<DataStorageUri> &storage_uris,
80107
const std::string &trace_id,
81108
std::function<void()> cb) {
82-
std::vector<ErrorCode> result(storage_uris.size(), EC_OK);
83-
// not supported yet
109+
std::vector<ErrorCode> result;
110+
result.reserve(storage_uris.size());
111+
for (const auto &storage_uri : storage_uris) {
112+
const auto [decision, should_delete] = ShouldDeleteFile(storage_uri);
113+
if (decision != EC_OK) {
114+
result.push_back(decision);
115+
continue;
116+
}
117+
if (!should_delete) {
118+
result.push_back(EC_OK);
119+
continue;
120+
}
121+
std::filesystem::path file_path = storage_uri.GetPath();
122+
std::error_code ec;
123+
const bool removed = std::filesystem::remove(file_path, ec);
124+
if (ec) {
125+
KVCM_LOG_ERROR("Failed to delete nfs file [%s]: [%s]", file_path.string().c_str(), ec.message().c_str());
126+
result.push_back(EC_ERROR);
127+
continue;
128+
}
129+
if (!removed) {
130+
KVCM_LOG_WARN("Try delete nfs file not exist, file: [%s]", file_path.string().c_str());
131+
}
132+
result.push_back(EC_OK);
133+
}
134+
if (cb) {
135+
cb();
136+
}
84137
return result;
85138
}
86139
std::vector<bool> NfsBackend::Exist(const std::vector<DataStorageUri> &storage_uris) {
87-
std::vector<bool> result(storage_uris.size(), true);
88-
// not supported yet
140+
std::vector<bool> result;
141+
result.reserve(storage_uris.size());
142+
for (const auto &storage_uri : storage_uris) {
143+
std::error_code ec;
144+
const bool exists = std::filesystem::exists(storage_uri.GetPath(), ec);
145+
if (ec) {
146+
KVCM_LOG_ERROR("Failed to check nfs file [%s]: [%s]", storage_uri.GetPath().c_str(), ec.message().c_str());
147+
result.push_back(false);
148+
continue;
149+
}
150+
result.push_back(exists);
151+
}
89152
return result;
90153
}
154+
std::vector<bool> NfsBackend::MightExist(const std::vector<DataStorageUri> &storage_uris) {
155+
return std::vector<bool>(storage_uris.size(), true);
156+
}
91157
std::vector<ErrorCode> NfsBackend::Lock(const std::vector<DataStorageUri> &storage_uris) {
92158
std::vector<ErrorCode> result(storage_uris.size(), EC_OK);
93159
// not supported yet

kv_cache_manager/data_storage/nfs_backend.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ class NfsBackend : public DataStorageBackend {
2929
const std::string &trace_id,
3030
std::function<void()> cb) override;
3131
std::vector<bool> Exist(const std::vector<DataStorageUri> &storage_uris) override;
32+
std::vector<bool> MightExist(const std::vector<DataStorageUri> &storage_uris) override;
3233
std::vector<ErrorCode> Lock(const std::vector<DataStorageUri> &storage_uris) override;
3334
std::vector<ErrorCode> UnLock(const std::vector<DataStorageUri> &storage_uris) override;
3435

kv_cache_manager/data_storage/test/nfs_backend_test.cc

Lines changed: 86 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
#include <filesystem>
2+
#include <fstream>
13
#include <gtest/gtest.h>
24
#include <memory>
35

@@ -114,31 +116,105 @@ TEST_F(NfsBackendTest, TestDeleteReturnsOkAndSameSize) {
114116
NfsBackend backend(metrics_registry_);
115117
std::shared_ptr<NfsStorageSpec> spec(new NfsStorageSpec);
116118
spec->set_key_count_per_file(1);
117-
spec->set_root_path("/data/");
119+
spec->set_root_path(GetPrivateTestRuntimeDataPath());
118120
StorageConfig storage_config(DataStorageType::DATA_STORAGE_TYPE_NFS, "test", spec);
119121
ASSERT_EQ(EC_OK, backend.Open(storage_config, "fake_trace_id_1"));
120-
std::vector<DataStorageUri> uris(3);
121-
auto res = backend.Delete(uris, "fake_trace_id_2", []() {});
122+
123+
const auto file1 = GetPrivateTestRuntimeDataPath() + "delete-1";
124+
const auto file2 = GetPrivateTestRuntimeDataPath() + "delete-2";
125+
{
126+
std::ofstream(file1) << "1";
127+
std::ofstream(file2) << "2";
128+
}
129+
ASSERT_TRUE(std::filesystem::exists(file1));
130+
ASSERT_TRUE(std::filesystem::exists(file2));
131+
std::vector<DataStorageUri> uris;
132+
uris.emplace_back("file://nfs_test" + file1);
133+
uris.emplace_back("file://nfs_test" + file2);
134+
uris.emplace_back("file://nfs_test" + GetPrivateTestRuntimeDataPath() + "missing");
135+
136+
bool callback_called = false;
137+
auto res = backend.Delete(uris, "fake_trace_id_2", [&callback_called]() { callback_called = true; });
138+
ASSERT_TRUE(callback_called);
122139
ASSERT_EQ(res.size(), uris.size());
123140
for (auto code : res) {
124141
ASSERT_EQ(code, EC_OK);
125142
}
143+
ASSERT_FALSE(std::filesystem::exists(file1));
144+
ASSERT_FALSE(std::filesystem::exists(file2));
145+
}
146+
147+
TEST_F(NfsBackendTest, TestDeleteOnlyRemovesMultiBlockFileForBlockZero) {
148+
NfsBackend backend(metrics_registry_);
149+
std::shared_ptr<NfsStorageSpec> spec(new NfsStorageSpec);
150+
spec->set_key_count_per_file(2);
151+
spec->set_root_path(GetPrivateTestRuntimeDataPath());
152+
StorageConfig storage_config(DataStorageType::DATA_STORAGE_TYPE_NFS, "test", spec);
153+
ASSERT_EQ(EC_OK, backend.Open(storage_config, "fake_trace_id_1"));
154+
155+
const auto file = GetPrivateTestRuntimeDataPath() + "multi-block-delete";
156+
std::ofstream(file) << "shared";
157+
ASSERT_TRUE(std::filesystem::exists(file));
158+
159+
DataStorageUri block_zero("file://nfs_test" + file + "?blkid=0&size=6");
160+
DataStorageUri block_one("file://nfs_test" + file + "?blkid=1&size=6");
161+
DataStorageUri invalid_block("file://nfs_test" + file + "?blkid=bad&size=6");
162+
163+
auto skip_res = backend.Delete({block_one}, "fake_trace_id_2", []() {});
164+
ASSERT_EQ(skip_res.size(), 1u);
165+
ASSERT_EQ(skip_res[0], EC_OK);
166+
ASSERT_TRUE(std::filesystem::exists(file));
167+
168+
auto invalid_res = backend.Delete({invalid_block}, "fake_trace_id_3", []() {});
169+
ASSERT_EQ(invalid_res.size(), 1u);
170+
ASSERT_EQ(invalid_res[0], EC_ERROR);
171+
ASSERT_TRUE(std::filesystem::exists(file));
172+
173+
DataStorageUri missing_blkid("file://nfs_test" + file + "?size=6");
174+
auto missing_blkid_res = backend.Delete({missing_blkid}, "fake_trace_id_4", []() {});
175+
ASSERT_EQ(missing_blkid_res.size(), 1u);
176+
ASSERT_EQ(missing_blkid_res[0], EC_OK);
177+
ASSERT_FALSE(std::filesystem::exists(file));
178+
179+
std::ofstream(file) << "shared";
180+
ASSERT_TRUE(std::filesystem::exists(file));
181+
DataStorageUri empty_blkid("file://nfs_test" + file + "?blkid=&size=6");
182+
auto empty_blkid_res = backend.Delete({empty_blkid}, "fake_trace_id_5", []() {});
183+
ASSERT_EQ(empty_blkid_res.size(), 1u);
184+
ASSERT_EQ(empty_blkid_res[0], EC_OK);
185+
ASSERT_FALSE(std::filesystem::exists(file));
186+
187+
std::ofstream(file) << "shared";
188+
ASSERT_TRUE(std::filesystem::exists(file));
189+
auto delete_res = backend.Delete({block_zero}, "fake_trace_id_6", []() {});
190+
ASSERT_EQ(delete_res.size(), 1u);
191+
ASSERT_EQ(delete_res[0], EC_OK);
192+
ASSERT_FALSE(std::filesystem::exists(file));
126193
}
127194

128-
TEST_F(NfsBackendTest, TestExistReturnsTrues) {
195+
TEST_F(NfsBackendTest, TestExistReturnsActualFileState) {
129196
NfsBackend backend(metrics_registry_);
130197
std::shared_ptr<NfsStorageSpec> spec(new NfsStorageSpec);
131198
spec->set_key_count_per_file(1);
132-
spec->set_root_path("/data/");
199+
spec->set_root_path(GetPrivateTestRuntimeDataPath());
133200
StorageConfig storage_config(DataStorageType::DATA_STORAGE_TYPE_NFS, "test", spec);
134201
ASSERT_EQ(EC_OK, backend.Open(storage_config, "fake_trace_id_1"));
135-
// TODO(qisa.cb) 没实现
136-
std::vector<DataStorageUri> uris(5);
202+
203+
const auto file1 = GetPrivateTestRuntimeDataPath() + "exists-1";
204+
std::ofstream(file1) << "1";
205+
std::vector<DataStorageUri> uris;
206+
uris.emplace_back("file://nfs_test" + file1);
207+
uris.emplace_back("file://nfs_test" + GetPrivateTestRuntimeDataPath() + "missing");
208+
137209
auto res = backend.Exist(uris);
138210
ASSERT_EQ(res.size(), uris.size());
139-
for (bool flag : res) {
140-
ASSERT_TRUE(flag);
141-
}
211+
ASSERT_TRUE(res[0]);
212+
ASSERT_FALSE(res[1]);
213+
214+
auto might_res = backend.MightExist(uris);
215+
ASSERT_EQ(might_res.size(), uris.size());
216+
ASSERT_TRUE(might_res[0]);
217+
ASSERT_TRUE(might_res[1]);
142218
}
143219

144220
TEST_F(NfsBackendTest, TestLockAndUnLockReturnOk) {

0 commit comments

Comments
 (0)