storage/kv_engine/engine_composite.h
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | #pragma once | ||
| 2 | |||
| 3 | #include <algorithm> | ||
| 4 | #include <array> | ||
| 5 | #include <chrono> | ||
| 6 | #include <cstring> | ||
| 7 | #include <filesystem> | ||
| 8 | #include <fstream> | ||
| 9 | #include <limits> | ||
| 10 | #include <memory> | ||
| 11 | #include <shared_mutex> | ||
| 12 | #include <stdexcept> | ||
| 13 | #include <string> | ||
| 14 | #include <unordered_set> | ||
| 15 | #include <utility> | ||
| 16 | #include <vector> | ||
| 17 | |||
| 18 | #include "base/factory.h" | ||
| 19 | #include "storage/index/dram/extendible_hash_index.h" | ||
| 20 | #include "storage/index/dram/pet_hash_index.h" | ||
| 21 | #include "storage/index/dram/unordered_map_index.h" | ||
| 22 | #include "storage/index/utils/hash.h" | ||
| 23 | #include "storage/kv_engine/base_kv.h" | ||
| 24 | #include "storage/value_store/dram_value_store.h" | ||
| 25 | #include "storage/value_store/hybrid_value_store.h" | ||
| 26 | #include "storage/value_store/ssd_value_store.h" | ||
| 27 | |||
| 28 | class KVEngineComposite : public BaseKV { | ||
| 29 | public: | ||
| 30 | KVEngineComposite(std::unique_ptr<Index> index, | ||
| 31 | std::unique_ptr<ValueStore> value_store, | ||
| 32 | int num_threads = 0) | ||
| 33 | : BaseKV(BaseKVConfig{}), | ||
| 34 | index_(std::move(index)), | ||
| 35 | value_store_(std::move(value_store)), | ||
| 36 | num_threads_(std::max(num_threads, 1)), | ||
| 37 | checkpoint_keys_by_thread_(static_cast<size_t>(num_threads_)) {} | ||
| 38 | |||
| 39 | 1146 | explicit KVEngineComposite(const BaseKVConfig& config) : BaseKV(config) { | |
| 40 | 1146 | config_ = config; | |
| 41 | 1146 | const auto& j = config.json_config_; | |
| 42 | 1146 | const std::string index_type = j.at("index").at("type").get<std::string>(); | |
| 43 | 1146 | const std::string value_type = j.at("value").at("type").get<std::string>(); | |
| 44 | using IF = base::Factory<Index, const BaseKVConfig&>; | ||
| 45 | using VF = base::Factory<ValueStore, const BaseKVConfig&>; | ||
| 46 | 1146 | index_.reset(IF::NewInstance(index_type, config)); | |
| 47 | 1146 | value_store_.reset(VF::NewInstance(value_type, config)); | |
| 48 | 1144 | num_threads_ = config.num_threads_; | |
| 49 | 2288 | checkpoint_keys_by_thread_.resize( | |
| 50 | 1144 | static_cast<size_t>(std::max(num_threads_, 1))); | |
| 51 | 1144 | default_value_size_hint_ = | |
| 52 | 1144 | j.at("value").value("default_value_size_hint", 0); | |
| 53 | 1144 | if (!index_ || !value_store_) { | |
| 54 | ✗ | throw std::runtime_error("failed to create KVEngine components"); | |
| 55 | } | ||
| 56 | 1158 | } | |
| 57 | |||
| 58 | 203140 | void Get(const uint64_t key, std::string& value, unsigned tid) override { | |
| 59 | (void)tid; | ||
| 60 | 203140 | Value_t handle = kValueHandleNone; | |
| 61 | 203140 | index_->Get(key, handle); | |
| 62 | 203140 | if (handle == kValueHandleNone) { | |
| 63 | 6434 | value.clear(); | |
| 64 | 151144 | return; | |
| 65 | } | ||
| 66 | 196706 | if (const char* ptr = value_store_->DirectPtr(handle)) { | |
| 67 | 144710 | value.resize(value_store_->SlotCapacity(handle)); | |
| 68 | 144710 | std::memcpy(value.data(), ptr, value.size()); | |
| 69 | 144710 | return; | |
| 70 | } | ||
| 71 | 51996 | value.resize(value_store_->SlotCapacity(handle)); | |
| 72 | const size_t actual = | ||
| 73 | 51996 | value_store_->Read(handle, value.data(), value.size()); | |
| 74 | 51996 | value.resize(actual); | |
| 75 | } | ||
| 76 | |||
| 77 | 1068 | bool Exists(const uint64_t key, unsigned tid) override { | |
| 78 | (void)tid; | ||
| 79 | 1068 | Value_t handle = kValueHandleNone; | |
| 80 | 1068 | index_->Get(key, handle); | |
| 81 | 1068 | return handle != kValueHandleNone; | |
| 82 | } | ||
| 83 | |||
| 84 | 406176 | void Put(const uint64_t key, | |
| 85 | const std::string_view& value, | ||
| 86 | unsigned tid) override { | ||
| 87 | 406176 | std::shared_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 88 | 406176 | PutInternal(key, value.data(), value.size(), tid, true); | |
| 89 | 406176 | } | |
| 90 | |||
| 91 | 328 | void BatchPut(base::ConstArray<uint64_t> keys, | |
| 92 | std::vector<base::ConstArray<float>>* values, | ||
| 93 | unsigned tid) override { | ||
| 94 | 328 | if (values == nullptr || keys.Size() != static_cast<int>(values->size())) { | |
| 95 | ✗ | LOG(FATAL) << "KVEngine::BatchPut size mismatch"; | |
| 96 | } | ||
| 97 | (void)tid; | ||
| 98 | 328 | if (keys.Size() == 0) { | |
| 99 | 4 | return; | |
| 100 | } | ||
| 101 | 328 | std::shared_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 102 | |||
| 103 | 328 | std::unordered_set<uint64_t> seen_keys; | |
| 104 | 328 | seen_keys.reserve(static_cast<size_t>(keys.Size())); | |
| 105 | 328 | bool has_duplicate_key = false; | |
| 106 | 1296 | for (int i = 0; i < keys.Size(); ++i) { | |
| 107 | 972 | if (!seen_keys.insert(keys[i]).second) { | |
| 108 | 4 | has_duplicate_key = true; | |
| 109 | 4 | break; | |
| 110 | } | ||
| 111 | } | ||
| 112 | 328 | if (has_duplicate_key) { | |
| 113 | 12 | for (int i = 0; i < keys.Size(); ++i) { | |
| 114 | 8 | const auto& item = (*values)[i]; | |
| 115 | 8 | PutInternal(keys[i], | |
| 116 | 8 | item.Data(), | |
| 117 | 8 | static_cast<size_t>(item.Size()) * sizeof(float), | |
| 118 | tid, | ||
| 119 | false); | ||
| 120 | } | ||
| 121 | 4 | return; | |
| 122 | } | ||
| 123 | |||
| 124 | struct PutItem { | ||
| 125 | uint64_t key = 0; | ||
| 126 | ValueStore::WriteSpec spec{}; | ||
| 127 | }; | ||
| 128 | 324 | std::vector<PutItem> items; | |
| 129 | 324 | items.reserve(static_cast<size_t>(keys.Size())); | |
| 130 | |||
| 131 | 1288 | for (int i = 0; i < keys.Size(); ++i) { | |
| 132 | 964 | const auto& item = (*values)[i]; | |
| 133 | 964 | const void* data = item.Data(); | |
| 134 | 964 | const size_t size = static_cast<size_t>(item.Size()) * sizeof(float); | |
| 135 | 964 | items.push_back(PutItem{keys[i], ValueStore::WriteSpec{data, size}}); | |
| 136 | } | ||
| 137 | |||
| 138 | 324 | std::vector<uint64_t> batch_keys; | |
| 139 | 324 | std::vector<ValueStore::WriteSpec> specs; | |
| 140 | 324 | batch_keys.reserve(items.size()); | |
| 141 | 324 | specs.reserve(items.size()); | |
| 142 | 1288 | for (const auto& item : items) { | |
| 143 | 964 | batch_keys.push_back(item.key); | |
| 144 | 964 | specs.push_back(item.spec); | |
| 145 | } | ||
| 146 | const auto new_handles = | ||
| 147 | 324 | value_store_->BatchAllocAndWrite(batch_keys, specs); | |
| 148 | 324 | if (new_handles.size() != items.size()) { | |
| 149 | ✗ | LOG(FATAL) << "KVEngine::BatchPut allocation result size mismatch"; | |
| 150 | } | ||
| 151 | 1288 | for (size_t i = 0; i < items.size(); ++i) { | |
| 152 | 964 | if (new_handles[i] == kValueHandleNone) { | |
| 153 | ✗ | LOG(FATAL) << "KVEngine batch value allocation failed, key=" | |
| 154 | ✗ | << items[i].key << " size=" << items[i].spec.size; | |
| 155 | } | ||
| 156 | } | ||
| 157 | |||
| 158 | 1288 | for (size_t i = 0; i < items.size(); ++i) { | |
| 159 | 964 | Value_t old_handle = index_->Put(items[i].key, new_handles[i], tid); | |
| 160 | 964 | if (old_handle == kValueHandleNone) { | |
| 161 | 740 | TrackKey(items[i].key, tid); | |
| 162 | } | ||
| 163 | 964 | if (old_handle != kValueHandleNone && old_handle != new_handles[i]) { | |
| 164 | 224 | value_store_->Retire(old_handle); | |
| 165 | } | ||
| 166 | } | ||
| 167 | 332 | } | |
| 168 | |||
| 169 | 4 | uint64_t BatchDelete(base::ConstArray<uint64_t> keys, unsigned tid) override { | |
| 170 | 4 | if (keys.Size() == 0) { | |
| 171 | ✗ | return 0; | |
| 172 | } | ||
| 173 | 4 | std::unique_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 174 | 4 | return DeleteKeysLocked(keys, tid); | |
| 175 | 4 | } | |
| 176 | |||
| 177 | 1528 | void BatchGet(base::ConstArray<uint64_t> keys, | |
| 178 | std::vector<base::ConstArray<float>>* values, | ||
| 179 | unsigned tid) override { | ||
| 180 | (void)tid; | ||
| 181 | 1528 | values->resize(keys.Size()); | |
| 182 | 1528 | thread_local std::vector<Value_t> handles; | |
| 183 | 1528 | thread_local std::vector<std::vector<float>> buffers; | |
| 184 | 1528 | handles.assign(keys.Size(), kValueHandleNone); | |
| 185 | 1528 | buffers.clear(); | |
| 186 | 1528 | buffers.resize(keys.Size()); | |
| 187 | |||
| 188 | 1528 | if (keys.Size() > 0) { | |
| 189 | 1528 | index_->BatchGet(keys, handles.data(), tid); | |
| 190 | } | ||
| 191 | 1528 | std::vector<uint64_t> batch_handles; | |
| 192 | 1528 | std::vector<size_t> batch_indices; | |
| 193 | 1528 | batch_handles.reserve(static_cast<size_t>(keys.Size())); | |
| 194 | 1528 | batch_indices.reserve(static_cast<size_t>(keys.Size())); | |
| 195 | 68946 | for (int i = 0; i < keys.Size(); ++i) { | |
| 196 | 67418 | if (handles[i] == kValueHandleNone) { | |
| 197 | 324 | (*values)[i] = base::ConstArray<float>(); | |
| 198 | 324 | continue; | |
| 199 | } | ||
| 200 | 67094 | if (const char* ptr = value_store_->DirectPtr(handles[i])) { | |
| 201 | 45522 | const size_t bytes = value_store_->SlotCapacity(handles[i]); | |
| 202 | 45522 | (*values)[i] = base::ConstArray<float>( | |
| 203 | reinterpret_cast<float*>(const_cast<char*>(ptr)), | ||
| 204 | 45522 | bytes / sizeof(float)); | |
| 205 | 45522 | continue; | |
| 206 | 45522 | } | |
| 207 | 21572 | batch_handles.push_back(handles[i]); | |
| 208 | 21572 | batch_indices.push_back(static_cast<size_t>(i)); | |
| 209 | } | ||
| 210 | |||
| 211 | 1528 | std::vector<ValueStore::ReadResult> batch_results; | |
| 212 | 1528 | if (!batch_handles.empty()) { | |
| 213 | 412 | value_store_->BatchRead(batch_handles, batch_results); | |
| 214 | 412 | if (batch_results.size() != batch_indices.size()) { | |
| 215 | ✗ | LOG(FATAL) << "KVEngine::BatchGet read result size mismatch"; | |
| 216 | } | ||
| 217 | 21984 | for (size_t i = 0; i < batch_indices.size(); ++i) { | |
| 218 | 21572 | const size_t idx = batch_indices[i]; | |
| 219 | 21572 | const auto& result = batch_results[i]; | |
| 220 | 21572 | buffers[idx].resize(result.data.size() / sizeof(float)); | |
| 221 | 21572 | if (!result.data.empty()) { | |
| 222 | 43144 | std::memcpy( | |
| 223 | 21572 | buffers[idx].data(), result.data.data(), result.data.size()); | |
| 224 | } | ||
| 225 | 21572 | (*values)[idx] = | |
| 226 | 43144 | base::ConstArray<float>(buffers[idx].data(), buffers[idx].size()); | |
| 227 | } | ||
| 228 | } | ||
| 229 | 1528 | } | |
| 230 | |||
| 231 | 162 | bool BatchGetFlat(base::ConstArray<uint64_t> keys, | |
| 232 | float* values, | ||
| 233 | int64_t num_rows, | ||
| 234 | int64_t embedding_dim, | ||
| 235 | unsigned tid, | ||
| 236 | BatchGetFlatStats* stats = nullptr) override { | ||
| 237 | 324 | if (values == nullptr || num_rows < 0 || embedding_dim <= 0 || | |
| 238 | 162 | keys.Size() != static_cast<size_t>(num_rows)) { | |
| 239 | ✗ | return false; | |
| 240 | } | ||
| 241 | 162 | const size_t row_bytes = static_cast<size_t>(embedding_dim) * sizeof(float); | |
| 242 | 162 | thread_local std::vector<Value_t> handles; | |
| 243 | 162 | handles.assign(keys.Size(), kValueHandleNone); | |
| 244 | const auto index_lookup_start = | ||
| 245 | 56 | stats != nullptr ? std::chrono::steady_clock::now() | |
| 246 | 162 | : std::chrono::steady_clock::time_point{}; | |
| 247 | 162 | if (keys.Size() > 0) { | |
| 248 | 162 | index_->BatchGet(keys, handles.data(), tid); | |
| 249 | } | ||
| 250 | 162 | if (stats != nullptr) { | |
| 251 | 56 | stats->index_lookup_ns = static_cast<std::uint64_t>( | |
| 252 | 56 | std::chrono::duration_cast< std::chrono::nanoseconds>( | |
| 253 | 112 | std::chrono::steady_clock::now() - index_lookup_start) | |
| 254 | 56 | .count()); | |
| 255 | } | ||
| 256 | |||
| 257 | 162 | std::uint64_t missing_zero_fill_ns = 0; | |
| 258 | 162 | std::uint64_t missing_rows = 0; | |
| 259 | const auto row_copy_start = | ||
| 260 | 56 | stats != nullptr ? std::chrono::steady_clock::now() | |
| 261 | 162 | : std::chrono::steady_clock::time_point{}; | |
| 262 | 214 | if (default_value_size_hint_ == row_bytes && | |
| 263 | 104 | value_store_->ReadFlatFixedRows( | |
| 264 | 52 | handles.data(), | |
| 265 | static_cast<size_t>(num_rows), | ||
| 266 | values, | ||
| 267 | row_bytes, | ||
| 268 | &missing_rows)) { | ||
| 269 | 48 | if (stats != nullptr) { | |
| 270 | 2 | stats->zero_fill_ns = 0; | |
| 271 | 2 | stats->row_copy_ns = static_cast<std::uint64_t>( | |
| 272 | 2 | std::chrono::duration_cast< std::chrono::nanoseconds>( | |
| 273 | 2 | std::chrono::steady_clock::now() - row_copy_start) | |
| 274 | 2 | .count()); | |
| 275 | 2 | stats->missing_rows = missing_rows; | |
| 276 | } | ||
| 277 | 48 | return true; | |
| 278 | } | ||
| 279 | 114 | std::vector<uint64_t> batch_handles; | |
| 280 | 114 | std::vector<size_t> batch_rows; | |
| 281 | 114 | batch_handles.reserve(static_cast<size_t>(num_rows)); | |
| 282 | 114 | batch_rows.reserve(static_cast<size_t>(num_rows)); | |
| 283 | 306 | for (int64_t row = 0; row < num_rows; ++row) { | |
| 284 | 230 | const Value_t handle = handles[static_cast<size_t>(row)]; | |
| 285 | 230 | float* dst = values + row * embedding_dim; | |
| 286 | 230 | if (handle == kValueHandleNone) { | |
| 287 | const auto missing_zero_start = | ||
| 288 | 54 | stats != nullptr ? std::chrono::steady_clock::now() | |
| 289 | 54 | : std::chrono::steady_clock::time_point{}; | |
| 290 | 54 | std::memset(dst, 0, row_bytes); | |
| 291 | 54 | if (stats != nullptr) { | |
| 292 | 54 | missing_zero_fill_ns += static_cast<std::uint64_t>( | |
| 293 | 54 | std::chrono::duration_cast< std::chrono::nanoseconds>( | |
| 294 | 108 | std::chrono::steady_clock::now() - missing_zero_start) | |
| 295 | 54 | .count()); | |
| 296 | } | ||
| 297 | 54 | ++missing_rows; | |
| 298 | 54 | continue; | |
| 299 | 54 | } | |
| 300 | 176 | if (const char* ptr = value_store_->DirectPtr(handle)) { | |
| 301 | 110 | if (default_value_size_hint_ != row_bytes) { | |
| 302 | 110 | const size_t slot_bytes = value_store_->SlotCapacity(handle); | |
| 303 | 110 | if (slot_bytes != row_bytes) { | |
| 304 | 76 | LOG(ERROR) << "KVEngine::BatchGetFlat row size mismatch row=" << row | |
| 305 | 38 | << " key=" << keys[static_cast<int>(row)] | |
| 306 | 38 | << " expected_bytes=" << row_bytes | |
| 307 | 38 | << " actual_bytes=" << slot_bytes; | |
| 308 | 38 | return false; | |
| 309 | } | ||
| 310 | } | ||
| 311 | 72 | std::memcpy(dst, ptr, row_bytes); | |
| 312 | } else { | ||
| 313 | 66 | batch_handles.push_back(handle); | |
| 314 | 66 | batch_rows.push_back(static_cast<size_t>(row)); | |
| 315 | } | ||
| 316 | } | ||
| 317 | 76 | if (!batch_handles.empty()) { | |
| 318 | 40 | std::vector<ValueStore::ReadResult> batch_results; | |
| 319 | 40 | value_store_->BatchRead(batch_handles, batch_results); | |
| 320 | 40 | if (batch_results.size() != batch_rows.size()) { | |
| 321 | ✗ | LOG(ERROR) << "KVEngine::BatchGetFlat result size mismatch"; | |
| 322 | ✗ | return false; | |
| 323 | } | ||
| 324 | 88 | for (size_t i = 0; i < batch_rows.size(); ++i) { | |
| 325 | 66 | const size_t row = batch_rows[i]; | |
| 326 | 66 | const auto& result = batch_results[i]; | |
| 327 | 66 | if (result.data.size() != row_bytes) { | |
| 328 | 36 | LOG(ERROR) << "KVEngine::BatchGetFlat read size mismatch row=" << row | |
| 329 | 18 | << " key=" << keys[static_cast<int>(row)] | |
| 330 | 18 | << " expected_bytes=" << row_bytes | |
| 331 | 18 | << " actual_bytes=" << result.data.size(); | |
| 332 | 18 | return false; | |
| 333 | } | ||
| 334 | 48 | std::memcpy( | |
| 335 | 48 | values + row * embedding_dim, result.data.data(), row_bytes); | |
| 336 | } | ||
| 337 | 40 | } | |
| 338 | 58 | if (stats != nullptr) { | |
| 339 | 54 | stats->zero_fill_ns = missing_zero_fill_ns; | |
| 340 | 54 | stats->row_copy_ns = static_cast<std::uint64_t>( | |
| 341 | 54 | std::chrono::duration_cast< std::chrono::nanoseconds>( | |
| 342 | 54 | std::chrono::steady_clock::now() - row_copy_start) | |
| 343 | 54 | .count()); | |
| 344 | 54 | stats->missing_rows = missing_rows; | |
| 345 | } | ||
| 346 | 58 | return true; | |
| 347 | 114 | } | |
| 348 | |||
| 349 | 4 | bool BatchGetFlatRange( | |
| 350 | base::ConstArray<uint64_t> keys, | ||
| 351 | float* values, | ||
| 352 | int64_t num_rows, | ||
| 353 | int64_t row_dim, | ||
| 354 | int64_t value_offset, | ||
| 355 | int64_t value_dim, | ||
| 356 | unsigned tid, | ||
| 357 | BatchGetFlatStats* stats = nullptr, | ||
| 358 | bool collect_profile = true) override { | ||
| 359 | 4 | if (values == nullptr || num_rows < 0 || row_dim <= 0 || value_offset < 0 || | |
| 360 | 4 | value_dim <= 0 || value_offset > row_dim || | |
| 361 | 10 | value_dim > row_dim - value_offset || | |
| 362 | 2 | keys.Size() != static_cast<size_t>(num_rows)) { | |
| 363 | 2 | return false; | |
| 364 | } | ||
| 365 | 2 | const size_t stored_row_bytes = | |
| 366 | 2 | static_cast<size_t>(row_dim) * sizeof(float); | |
| 367 | 2 | if (default_value_size_hint_ != stored_row_bytes) { | |
| 368 | ✗ | return false; | |
| 369 | } | ||
| 370 | 2 | const size_t source_offset_bytes = | |
| 371 | 2 | static_cast<size_t>(value_offset) * sizeof(float); | |
| 372 | 2 | const size_t output_row_bytes = | |
| 373 | 2 | static_cast<size_t>(value_dim) * sizeof(float); | |
| 374 | 2 | thread_local std::vector<Value_t> handles; | |
| 375 | 2 | handles.assign(keys.Size(), kValueHandleNone); | |
| 376 | 2 | const bool profile_enabled = stats != nullptr && collect_profile; | |
| 377 | 2 | if (stats != nullptr) { | |
| 378 | 2 | stats->index_lookup_ns = 0; | |
| 379 | 2 | stats->zero_fill_ns = 0; | |
| 380 | 2 | stats->row_copy_ns = 0; | |
| 381 | 2 | stats->missing_rows = 0; | |
| 382 | } | ||
| 383 | const auto index_lookup_start = | ||
| 384 | 2 | profile_enabled ? std::chrono::steady_clock::now() | |
| 385 | 2 | : std::chrono::steady_clock::time_point{}; | |
| 386 | 2 | if (keys.Size() > 0) { | |
| 387 | 2 | index_->BatchGet(keys, handles.data(), tid); | |
| 388 | } | ||
| 389 | 2 | if (profile_enabled) { | |
| 390 | 2 | stats->index_lookup_ns = static_cast<std::uint64_t>( | |
| 391 | 2 | std::chrono::duration_cast<std::chrono::nanoseconds>( | |
| 392 | 4 | std::chrono::steady_clock::now() - index_lookup_start) | |
| 393 | 2 | .count()); | |
| 394 | } | ||
| 395 | |||
| 396 | 2 | uint64_t missing_rows = 0; | |
| 397 | const auto row_copy_start = | ||
| 398 | 2 | profile_enabled ? std::chrono::steady_clock::now() | |
| 399 | 2 | : std::chrono::steady_clock::time_point{}; | |
| 400 | 4 | if (!value_store_->ReadFlatFixedRowSlices( | |
| 401 | 2 | handles.data(), | |
| 402 | static_cast<size_t>(num_rows), | ||
| 403 | values, | ||
| 404 | stored_row_bytes, | ||
| 405 | source_offset_bytes, | ||
| 406 | output_row_bytes, | ||
| 407 | &missing_rows)) { | ||
| 408 | ✗ | return false; | |
| 409 | } | ||
| 410 | 2 | if (stats != nullptr) { | |
| 411 | 2 | if (profile_enabled) { | |
| 412 | 2 | stats->row_copy_ns = static_cast<std::uint64_t>( | |
| 413 | 2 | std::chrono::duration_cast<std::chrono::nanoseconds>( | |
| 414 | 4 | std::chrono::steady_clock::now() - row_copy_start) | |
| 415 | 2 | .count()); | |
| 416 | } | ||
| 417 | 2 | stats->missing_rows = missing_rows; | |
| 418 | } | ||
| 419 | 2 | return true; | |
| 420 | } | ||
| 421 | |||
| 422 | ✗ | bool BatchGetIndexOnly(base::ConstArray<uint64_t> keys, | |
| 423 | unsigned tid, | ||
| 424 | BatchGetFlatStats* stats = nullptr) override { | ||
| 425 | ✗ | thread_local std::vector<Value_t> handles; | |
| 426 | ✗ | handles.assign(keys.Size(), kValueHandleNone); | |
| 427 | ✗ | if (keys.Size() > 0) { | |
| 428 | ✗ | index_->BatchGet(keys, handles.data(), tid); | |
| 429 | } | ||
| 430 | ✗ | if (stats != nullptr) { | |
| 431 | ✗ | std::uint64_t missing_rows = 0; | |
| 432 | ✗ | for (const Value_t handle : handles) { | |
| 433 | ✗ | if (handle == kValueHandleNone) { | |
| 434 | ✗ | ++missing_rows; | |
| 435 | } | ||
| 436 | } | ||
| 437 | ✗ | stats->missing_rows = missing_rows; | |
| 438 | } | ||
| 439 | ✗ | return true; | |
| 440 | } | ||
| 441 | |||
| 442 | 26 | bool BatchGetDirectFixedRows( | |
| 443 | base::ConstArray<uint64_t> keys, | ||
| 444 | int64_t num_rows, | ||
| 445 | int64_t embedding_dim, | ||
| 446 | unsigned tid, | ||
| 447 | std::vector<DirectFixedRow>* rows, | ||
| 448 | BatchGetFlatStats* stats = nullptr) override { | ||
| 449 | 52 | if (rows == nullptr || num_rows < 0 || embedding_dim <= 0 || | |
| 450 | 26 | keys.Size() != static_cast<size_t>(num_rows)) { | |
| 451 | ✗ | return false; | |
| 452 | } | ||
| 453 | 26 | const size_t row_bytes = static_cast<size_t>(embedding_dim) * sizeof(float); | |
| 454 | 26 | if (default_value_size_hint_ != row_bytes) { | |
| 455 | ✗ | return false; | |
| 456 | } | ||
| 457 | 26 | thread_local std::vector<Value_t> handles; | |
| 458 | 26 | handles.assign(keys.Size(), kValueHandleNone); | |
| 459 | const auto index_lookup_start = | ||
| 460 | ✗ | stats != nullptr ? std::chrono::steady_clock::now() | |
| 461 | 26 | : std::chrono::steady_clock::time_point{}; | |
| 462 | 26 | if (keys.Size() > 0) { | |
| 463 | 26 | index_->BatchGet(keys, handles.data(), tid); | |
| 464 | } | ||
| 465 | 26 | if (stats != nullptr) { | |
| 466 | ✗ | stats->index_lookup_ns = static_cast<std::uint64_t>( | |
| 467 | ✗ | std::chrono::duration_cast< std::chrono::nanoseconds>( | |
| 468 | ✗ | std::chrono::steady_clock::now() - index_lookup_start) | |
| 469 | ✗ | .count()); | |
| 470 | } | ||
| 471 | |||
| 472 | 26 | thread_local std::vector<ValueStore::DirectFixedRow> store_rows; | |
| 473 | 26 | store_rows.resize(static_cast<size_t>(num_rows)); | |
| 474 | 26 | uint64_t missing_rows = 0; | |
| 475 | 52 | if (!value_store_->GetDirectFixedRows( | |
| 476 | 26 | handles.data(), | |
| 477 | static_cast<size_t>(num_rows), | ||
| 478 | row_bytes, | ||
| 479 | store_rows.data(), | ||
| 480 | &missing_rows)) { | ||
| 481 | ✗ | return false; | |
| 482 | } | ||
| 483 | 26 | rows->resize(store_rows.size()); | |
| 484 | 74 | for (size_t i = 0; i < store_rows.size(); ++i) { | |
| 485 | 48 | (*rows)[i] = DirectFixedRow{ | |
| 486 | 48 | store_rows[i].data, store_rows[i].size, store_rows[i].missing}; | |
| 487 | } | ||
| 488 | 26 | if (stats != nullptr) { | |
| 489 | ✗ | stats->missing_rows = missing_rows; | |
| 490 | } | ||
| 491 | 26 | return true; | |
| 492 | } | ||
| 493 | |||
| 494 | ✗ | RDMABackingRegion GetRDMABackingRegion() const override { | |
| 495 | ✗ | if (!value_store_) { | |
| 496 | ✗ | return {}; | |
| 497 | } | ||
| 498 | return RDMABackingRegion{ | ||
| 499 | ✗ | value_store_->RDMABackingData(), value_store_->RDMABackingSize()}; | |
| 500 | } | ||
| 501 | |||
| 502 | 26 | bool ApplySgdUpdateFlat( | |
| 503 | base::ConstArray<uint64_t> keys, | ||
| 504 | const float* grads, | ||
| 505 | int64_t num_rows, | ||
| 506 | int64_t embedding_dim, | ||
| 507 | float learning_rate, | ||
| 508 | uint8_t tag, | ||
| 509 | unsigned tid, | ||
| 510 | UpdateFlatStats* stats = nullptr) override { | ||
| 511 | 26 | if (grads == nullptr || keys.Size() != static_cast<size_t>(num_rows) || | |
| 512 | embedding_dim <= 0) { | ||
| 513 | ✗ | return false; | |
| 514 | } | ||
| 515 | using clock = std::chrono::steady_clock; | ||
| 516 | 26 | std::shared_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 517 | 26 | const int tag_bits = static_cast<int>(sizeof(tag) * 8); | |
| 518 | 26 | const int shift = static_cast<int>(sizeof(uint64_t) * 8) - tag_bits; | |
| 519 | 26 | const uint64_t key_mask = ~0ULL >> tag_bits; | |
| 520 | 26 | const size_t row_bytes = static_cast<size_t>(embedding_dim) * sizeof(float); | |
| 521 | |||
| 522 | 26 | const auto tag_start = clock::now(); | |
| 523 | 26 | thread_local std::vector<uint64_t> tagged_keys; | |
| 524 | 26 | tagged_keys.resize(static_cast<size_t>(num_rows)); | |
| 525 | 74 | for (int64_t r = 0; r < num_rows; ++r) { | |
| 526 | 48 | tagged_keys[static_cast<size_t>(r)] = | |
| 527 | 96 | (static_cast<uint64_t>(tag) << shift) | | |
| 528 | 48 | (keys[static_cast<size_t>(r)] & key_mask); | |
| 529 | } | ||
| 530 | 26 | if (stats != nullptr) { | |
| 531 | ✗ | stats->tag_ns = static_cast<std::uint64_t>( | |
| 532 | ✗ | std::chrono::duration_cast<std::chrono::nanoseconds>( | |
| 533 | ✗ | clock::now() - tag_start) | |
| 534 | ✗ | .count()); | |
| 535 | } | ||
| 536 | |||
| 537 | 26 | thread_local std::vector<DirectFixedRow> rows; | |
| 538 | 26 | base::ConstArray<uint64_t> tagged(tagged_keys); | |
| 539 | 26 | BatchGetFlatStats get_stats; | |
| 540 | 26 | const auto batch_get_start = clock::now(); | |
| 541 | 26 | if (!BatchGetDirectFixedRows( | |
| 542 | tagged, | ||
| 543 | num_rows, | ||
| 544 | embedding_dim, | ||
| 545 | tid, | ||
| 546 | &rows, | ||
| 547 | stats != nullptr ? &get_stats : nullptr)) { | ||
| 548 | ✗ | return false; | |
| 549 | } | ||
| 550 | 26 | if (stats != nullptr) { | |
| 551 | ✗ | stats->batch_get_ns = static_cast<std::uint64_t>( | |
| 552 | ✗ | std::chrono::duration_cast<std::chrono::nanoseconds>( | |
| 553 | ✗ | clock::now() - batch_get_start) | |
| 554 | ✗ | .count()); | |
| 555 | ✗ | stats->index_lookup_ns = get_stats.index_lookup_ns; | |
| 556 | ✗ | stats->store_read_ns = | |
| 557 | ✗ | stats->batch_get_ns > get_stats.index_lookup_ns | |
| 558 | ✗ | ? stats->batch_get_ns - get_stats.index_lookup_ns | |
| 559 | : 0; | ||
| 560 | ✗ | stats->missing_rows = get_stats.missing_rows; | |
| 561 | } | ||
| 562 | |||
| 563 | 26 | const auto loop_start = clock::now(); | |
| 564 | 26 | std::vector<float> missing_row(static_cast<size_t>(embedding_dim)); | |
| 565 | 74 | for (int64_t r = 0; r < num_rows; ++r) { | |
| 566 | 48 | const auto& row = rows[static_cast<size_t>(r)]; | |
| 567 | 48 | const float* grad = grads + r * embedding_dim; | |
| 568 | 48 | if (row.missing) { | |
| 569 | 604 | for (int64_t c = 0; c < embedding_dim; ++c) { | |
| 570 | 584 | missing_row[static_cast<size_t>(c)] = -learning_rate * grad[c]; | |
| 571 | } | ||
| 572 | const auto miss_start = | ||
| 573 | 20 | stats != nullptr ? clock::now() : clock::time_point{}; | |
| 574 | 20 | PutInternalStats miss_stats; | |
| 575 | 40 | PutInternal(tagged_keys[static_cast<size_t>(r)], | |
| 576 | 20 | missing_row.data(), | |
| 577 | row_bytes, | ||
| 578 | tid, | ||
| 579 | false, | ||
| 580 | stats != nullptr ? &miss_stats : nullptr); | ||
| 581 | 20 | if (stats != nullptr) { | |
| 582 | ✗ | stats->missing_write_ns += static_cast<std::uint64_t>( | |
| 583 | ✗ | std::chrono::duration_cast<std::chrono::nanoseconds>( | |
| 584 | ✗ | clock::now() - miss_start) | |
| 585 | ✗ | .count()); | |
| 586 | ✗ | stats->missing_alloc_ns += miss_stats.alloc_ns; | |
| 587 | ✗ | stats->missing_index_ns += miss_stats.index_ns; | |
| 588 | ✗ | stats->missing_track_ns += miss_stats.track_ns; | |
| 589 | } | ||
| 590 | 20 | continue; | |
| 591 | 20 | } | |
| 592 | |||
| 593 | 28 | float* value = reinterpret_cast<float*>(const_cast<char*>(row.data)); | |
| 594 | 28 | #pragma omp simd | |
| 595 | for (int64_t c = 0; c < embedding_dim; ++c) { | ||
| 596 | 616 | value[c] -= learning_rate * grad[c]; | |
| 597 | } | ||
| 598 | } | ||
| 599 | 26 | if (stats != nullptr) { | |
| 600 | const auto loop_ns = static_cast<std::uint64_t>( | ||
| 601 | ✗ | std::chrono::duration_cast<std::chrono::nanoseconds>( | |
| 602 | ✗ | clock::now() - loop_start) | |
| 603 | ✗ | .count()); | |
| 604 | // The miss path is timed separately; subtract it so the hit write-back | ||
| 605 | // bucket does not double count the inserts. | ||
| 606 | ✗ | stats->writeback_ns = | |
| 607 | ✗ | loop_ns > stats->missing_write_ns | |
| 608 | ✗ | ? loop_ns - stats->missing_write_ns | |
| 609 | : 0; | ||
| 610 | } | ||
| 611 | 26 | return true; | |
| 612 | 26 | } | |
| 613 | |||
| 614 | ✗ | bool ApplySgdUpdateFlatRange( | |
| 615 | base::ConstArray<uint64_t> keys, | ||
| 616 | const float* grads, | ||
| 617 | int64_t num_rows, | ||
| 618 | int64_t row_dim, | ||
| 619 | int64_t update_offset, | ||
| 620 | int64_t update_dim, | ||
| 621 | float learning_rate, | ||
| 622 | unsigned tid) override { | ||
| 623 | ✗ | if (grads == nullptr || keys.Size() != static_cast<size_t>(num_rows) || | |
| 624 | ✗ | row_dim <= 0 || update_offset < 0 || update_dim <= 0 || | |
| 625 | ✗ | update_offset > row_dim || update_dim > row_dim - update_offset) { | |
| 626 | ✗ | return false; | |
| 627 | } | ||
| 628 | ✗ | const size_t row_bytes = static_cast<size_t>(row_dim) * sizeof(float); | |
| 629 | ✗ | if (default_value_size_hint_ != row_bytes) { | |
| 630 | ✗ | return false; | |
| 631 | } | ||
| 632 | |||
| 633 | ✗ | std::shared_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 634 | ✗ | thread_local std::vector<DirectFixedRow> rows; | |
| 635 | ✗ | if (!BatchGetDirectFixedRows(keys, num_rows, row_dim, tid, &rows)) { | |
| 636 | ✗ | return false; | |
| 637 | } | ||
| 638 | ✗ | for (int64_t row = 0; row < num_rows; ++row) { | |
| 639 | ✗ | const auto& direct_row = rows[static_cast<size_t>(row)]; | |
| 640 | ✗ | if (direct_row.missing || direct_row.data == nullptr || | |
| 641 | ✗ | direct_row.size != row_bytes) { | |
| 642 | ✗ | return false; | |
| 643 | } | ||
| 644 | ✗ | float* value = | |
| 645 | ✗ | reinterpret_cast<float*>(const_cast<char*>(direct_row.data)) + | |
| 646 | ✗ | update_offset; | |
| 647 | ✗ | const float* grad = grads + row * update_dim; | |
| 648 | ✗ | #pragma omp simd | |
| 649 | for (int64_t column = 0; column < update_dim; ++column) { | ||
| 650 | ✗ | value[column] -= learning_rate * grad[column]; | |
| 651 | } | ||
| 652 | } | ||
| 653 | ✗ | return true; | |
| 654 | ✗ | } | |
| 655 | |||
| 656 | 56 | void BulkLoad(base::ConstArray<uint64_t> keys, const void* value) override { | |
| 657 | 56 | BulkLoadRange(keys, value, 0); | |
| 658 | 56 | } | |
| 659 | |||
| 660 | 56 | void BulkLoadRange(base::ConstArray<uint64_t> keys, | |
| 661 | const void* value, | ||
| 662 | unsigned tid, | ||
| 663 | BulkLoadStats* stats = nullptr) override { | ||
| 664 | using clock = std::chrono::steady_clock; | ||
| 665 | 56 | const auto total_start = clock::now(); | |
| 666 | 56 | const auto& j = config_.json_config_; | |
| 667 | 56 | const size_t value_size = j.at("value").value("default_value_size_hint", 0); | |
| 668 | 56 | if (value_size == 0) { | |
| 669 | ✗ | LOG(FATAL) << "KVEngine::BulkLoad requires value_size hint"; | |
| 670 | } | ||
| 671 | 56 | if (keys.Size() == 0) { | |
| 672 | ✗ | return; | |
| 673 | } | ||
| 674 | 56 | std::shared_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 675 | 56 | const char* data = reinterpret_cast<const char*>(value); | |
| 676 | 56 | std::vector<ValueStore::WriteSpec> specs; | |
| 677 | 56 | specs.reserve(static_cast<size_t>(keys.Size())); | |
| 678 | 490 | for (int i = 0; i < keys.Size(); ++i) { | |
| 679 | 434 | specs.push_back(ValueStore::WriteSpec{data + i * value_size, value_size}); | |
| 680 | } | ||
| 681 | 56 | std::vector<uint64_t> bulk_keys; | |
| 682 | 56 | bulk_keys.reserve(static_cast<size_t>(keys.Size())); | |
| 683 | 490 | for (int i = 0; i < keys.Size(); ++i) { | |
| 684 | 434 | bulk_keys.push_back(keys[i]); | |
| 685 | } | ||
| 686 | 56 | const auto prepare_end = clock::now(); | |
| 687 | std::vector<uint64_t> handles = | ||
| 688 | 56 | value_store_->BatchAllocAndWrite(bulk_keys, specs); | |
| 689 | 56 | const auto alloc_end = clock::now(); | |
| 690 | 56 | if (handles.size() != static_cast<size_t>(keys.Size())) { | |
| 691 | ✗ | LOG(FATAL) << "KVEngine::BulkLoad allocation result size mismatch"; | |
| 692 | } | ||
| 693 | 490 | for (int i = 0; i < keys.Size(); ++i) { | |
| 694 | 434 | if (handles[static_cast<size_t>(i)] == kValueHandleNone) { | |
| 695 | ✗ | LOG(FATAL) << "KVEngine bulk value allocation failed, key=" << keys[i] | |
| 696 | ✗ | << " size=" << value_size; | |
| 697 | } | ||
| 698 | } | ||
| 699 | 56 | const auto validate_end = clock::now(); | |
| 700 | 56 | index_->BatchPut(keys, handles.data(), tid); | |
| 701 | 56 | const auto index_end = clock::now(); | |
| 702 | 56 | TrackKeys(keys, tid); | |
| 703 | 56 | const auto track_end = clock::now(); | |
| 704 | 56 | if (stats != nullptr) { | |
| 705 | ✗ | auto ns = [](clock::duration duration) { | |
| 706 | return static_cast<std::uint64_t>( | ||
| 707 | ✗ | std::chrono::duration_cast<std::chrono::nanoseconds>(duration) | |
| 708 | ✗ | .count()); | |
| 709 | }; | ||
| 710 | ✗ | stats->prepare_ns = ns(prepare_end - total_start); | |
| 711 | ✗ | stats->value_alloc_ns = ns(alloc_end - prepare_end); | |
| 712 | ✗ | stats->handle_validate_ns = ns(validate_end - alloc_end); | |
| 713 | ✗ | stats->index_put_ns = ns(index_end - validate_end); | |
| 714 | ✗ | stats->key_track_ns = ns(track_end - index_end); | |
| 715 | ✗ | stats->total_ns = ns(track_end - total_start); | |
| 716 | } | ||
| 717 | 56 | } | |
| 718 | |||
| 719 | ✗ | bool BulkLoadIndexedFloatRange( | |
| 720 | base::ConstArray<uint64_t> keys, | ||
| 721 | const int64_t* row_indices, | ||
| 722 | const float* values, | ||
| 723 | int64_t num_rows, | ||
| 724 | int64_t value_dim, | ||
| 725 | unsigned tid, | ||
| 726 | BulkLoadStats* stats = nullptr) override { | ||
| 727 | using clock = std::chrono::steady_clock; | ||
| 728 | ✗ | const auto total_start = clock::now(); | |
| 729 | ✗ | if (row_indices == nullptr || values == nullptr || num_rows < 0 || | |
| 730 | ✗ | value_dim <= 0 || keys.Size() != static_cast<size_t>(num_rows)) { | |
| 731 | ✗ | return false; | |
| 732 | } | ||
| 733 | ✗ | const size_t row_bytes = | |
| 734 | ✗ | sizeof(int64_t) + static_cast<size_t>(value_dim) * sizeof(float); | |
| 735 | ✗ | if (default_value_size_hint_ != row_bytes) { | |
| 736 | ✗ | return false; | |
| 737 | } | ||
| 738 | |||
| 739 | ✗ | std::shared_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 740 | std::vector<uint64_t> handles( | ||
| 741 | ✗ | static_cast<size_t>(num_rows), kValueHandleNone); | |
| 742 | ✗ | const auto prepare_end = clock::now(); | |
| 743 | ✗ | for (int64_t row_index = 0; row_index < num_rows; ++row_index) { | |
| 744 | ✗ | const uint64_t handle = value_store_->Alloc(row_bytes); | |
| 745 | ✗ | if (handle == kValueHandleNone) { | |
| 746 | ✗ | LOG(FATAL) << "KVEngine indexed bulk value allocation failed, key=" | |
| 747 | ✗ | << keys[static_cast<int>(row_index)] | |
| 748 | ✗ | << " size=" << row_bytes; | |
| 749 | } | ||
| 750 | ✗ | char* row = const_cast<char*>(value_store_->DirectPtr(handle)); | |
| 751 | ✗ | if (row == nullptr) { | |
| 752 | ✗ | return false; | |
| 753 | } | ||
| 754 | ✗ | std::memcpy(row, row_indices + row_index, sizeof(int64_t)); | |
| 755 | ✗ | std::memcpy(row + sizeof(int64_t), | |
| 756 | ✗ | values + row_index * value_dim, | |
| 757 | ✗ | static_cast<size_t>(value_dim) * sizeof(float)); | |
| 758 | ✗ | handles[static_cast<size_t>(row_index)] = handle; | |
| 759 | } | ||
| 760 | ✗ | const auto alloc_end = clock::now(); | |
| 761 | ✗ | const auto validate_end = alloc_end; | |
| 762 | ✗ | index_->BatchPut(keys, handles.data(), tid); | |
| 763 | ✗ | const auto index_end = clock::now(); | |
| 764 | ✗ | TrackKeys(keys, tid); | |
| 765 | ✗ | const auto track_end = clock::now(); | |
| 766 | ✗ | if (stats != nullptr) { | |
| 767 | ✗ | auto ns = [](clock::duration duration) { | |
| 768 | return static_cast<std::uint64_t>( | ||
| 769 | ✗ | std::chrono::duration_cast<std::chrono::nanoseconds>(duration) | |
| 770 | ✗ | .count()); | |
| 771 | }; | ||
| 772 | ✗ | stats->prepare_ns = ns(prepare_end - total_start); | |
| 773 | ✗ | stats->value_alloc_ns = ns(alloc_end - prepare_end); | |
| 774 | ✗ | stats->handle_validate_ns = ns(validate_end - alloc_end); | |
| 775 | ✗ | stats->index_put_ns = ns(index_end - validate_end); | |
| 776 | ✗ | stats->key_track_ns = ns(track_end - index_end); | |
| 777 | ✗ | stats->total_ns = ns(track_end - total_start); | |
| 778 | } | ||
| 779 | ✗ | return true; | |
| 780 | ✗ | } | |
| 781 | |||
| 782 | 20 | bool SaveCheckpoint(const std::string& file, | |
| 783 | const std::string& metadata) override { | ||
| 784 | 20 | std::unique_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 785 | 20 | if (file.empty()) { | |
| 786 | ✗ | LOG(ERROR) << "KVEngine checkpoint path is empty"; | |
| 787 | ✗ | return false; | |
| 788 | } | ||
| 789 | |||
| 790 | 20 | std::vector<uint64_t> keys = CollectCheckpointKeys(); | |
| 791 | |||
| 792 | 20 | const std::filesystem::path checkpoint_path(file); | |
| 793 | 20 | std::filesystem::path temp_path(file); | |
| 794 | 20 | temp_path += ".tmp"; | |
| 795 | 20 | std::error_code error; | |
| 796 | 20 | std::filesystem::remove(temp_path, error); | |
| 797 | |||
| 798 | 20 | std::ofstream output(temp_path, std::ios::binary | std::ios::trunc); | |
| 799 | 20 | if (!output) { | |
| 800 | ✗ | LOG(ERROR) << "Failed to open checkpoint temp file: " << temp_path; | |
| 801 | ✗ | return false; | |
| 802 | } | ||
| 803 | |||
| 804 | 20 | const uint64_t metadata_size = static_cast<uint64_t>(metadata.size()); | |
| 805 | 20 | const uint64_t record_count = static_cast<uint64_t>(keys.size()); | |
| 806 | 20 | uint64_t checksum = kCheckpointChecksumSeed; | |
| 807 | 20 | UpdateChecksum(&checksum, kCheckpointMagic.data(), kCheckpointMagic.size()); | |
| 808 | 20 | UpdateChecksumPod(&checksum, kCheckpointVersion); | |
| 809 | 20 | UpdateChecksumPod(&checksum, metadata_size); | |
| 810 | 20 | UpdateChecksumPod(&checksum, record_count); | |
| 811 | 20 | const ValueStore::LayoutId layout_id = value_store_->GetLayoutId(); | |
| 812 | 20 | UpdateChecksumPod(&checksum, layout_id.strategy); | |
| 813 | 20 | UpdateChecksumPod(&checksum, layout_id.checksum); | |
| 814 | 20 | UpdateChecksum(&checksum, metadata.data(), metadata.size()); | |
| 815 | bool write_ok = | ||
| 816 | 20 | WriteBytes(output, kCheckpointMagic.data(), kCheckpointMagic.size()) && | |
| 817 | 20 | WritePod(output, kCheckpointVersion) && | |
| 818 | 20 | WritePod(output, metadata_size) && WritePod(output, record_count) && | |
| 819 | 20 | WritePod(output, layout_id.strategy) && | |
| 820 | 60 | WritePod(output, layout_id.checksum) && | |
| 821 | 20 | WriteBytes(output, metadata.data(), metadata.size()); | |
| 822 | 1088 | for (const uint64_t key : keys) { | |
| 823 | 1068 | std::string value; | |
| 824 | 1068 | Get(key, value, 0); | |
| 825 | 1068 | if (!Exists(key, 0)) { | |
| 826 | ✗ | LOG(ERROR) << "KVEngine checkpoint tracked key is missing: " << key; | |
| 827 | ✗ | write_ok = false; | |
| 828 | ✗ | break; | |
| 829 | } | ||
| 830 | 1068 | const uint64_t value_size = static_cast<uint64_t>(value.size()); | |
| 831 | 1068 | UpdateChecksumPod(&checksum, key); | |
| 832 | 1068 | UpdateChecksumPod(&checksum, value_size); | |
| 833 | 1068 | UpdateChecksum(&checksum, value.data(), value.size()); | |
| 834 | 1068 | write_ok = | |
| 835 | 2136 | write_ok && WritePod(output, key) && WritePod(output, value_size) && | |
| 836 | 1068 | WriteBytes(output, value.data(), value.size()); | |
| 837 | 1068 | if (!write_ok) { | |
| 838 | ✗ | break; | |
| 839 | } | ||
| 840 | 1068 | } | |
| 841 | 20 | write_ok = write_ok && WritePod(output, checksum); | |
| 842 | 20 | output.flush(); | |
| 843 | 20 | write_ok = write_ok && output.good(); | |
| 844 | 20 | output.close(); | |
| 845 | 20 | write_ok = write_ok && !output.fail(); | |
| 846 | 20 | if (!write_ok) { | |
| 847 | ✗ | LOG(ERROR) << "Failed to write checkpoint temp file: " << temp_path; | |
| 848 | ✗ | std::filesystem::remove(temp_path, error); | |
| 849 | ✗ | return false; | |
| 850 | } | ||
| 851 | |||
| 852 | 20 | error.clear(); | |
| 853 | 20 | std::filesystem::rename(temp_path, checkpoint_path, error); | |
| 854 | 20 | if (error) { | |
| 855 | ✗ | LOG(ERROR) << "Failed to publish checkpoint " << checkpoint_path << ": " | |
| 856 | ✗ | << error.message(); | |
| 857 | ✗ | std::filesystem::remove(temp_path, error); | |
| 858 | ✗ | return false; | |
| 859 | } | ||
| 860 | 20 | return true; | |
| 861 | 20 | } | |
| 862 | |||
| 863 | 34 | bool LoadCheckpoint(const std::string& file, | |
| 864 | const std::string& expected_metadata) override { | ||
| 865 | 34 | std::unique_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 866 | 34 | if (HasCheckpointRecords()) { | |
| 867 | 2 | LOG(ERROR) << "KVEngine checkpoint load requires an empty engine"; | |
| 868 | 2 | return false; | |
| 869 | } | ||
| 870 | 32 | if (file.empty()) { | |
| 871 | ✗ | LOG(ERROR) << "KVEngine checkpoint path is empty"; | |
| 872 | ✗ | return false; | |
| 873 | } | ||
| 874 | |||
| 875 | 32 | std::error_code error; | |
| 876 | 32 | const std::uintmax_t file_size = std::filesystem::file_size(file, error); | |
| 877 | 32 | if (error || file_size > std::numeric_limits<uint64_t>::max()) { | |
| 878 | ✗ | LOG(ERROR) << "Failed to inspect checkpoint file: " << file; | |
| 879 | ✗ | return false; | |
| 880 | } | ||
| 881 | 32 | uint64_t remaining = static_cast<uint64_t>(file_size); | |
| 882 | 32 | std::ifstream input(file, std::ios::binary); | |
| 883 | 32 | if (!input) { | |
| 884 | ✗ | LOG(ERROR) << "Failed to open checkpoint file: " << file; | |
| 885 | ✗ | return false; | |
| 886 | } | ||
| 887 | |||
| 888 | 32 | std::array<char, kCheckpointMagic.size()> magic{}; | |
| 889 | 32 | uint32_t version = 0; | |
| 890 | 32 | uint64_t metadata_size = 0; | |
| 891 | 32 | uint64_t record_count = 0; | |
| 892 | 32 | if (!ReadBytes(input, magic.data(), magic.size(), &remaining) || | |
| 893 | 32 | !ReadPod(input, &version, &remaining) || | |
| 894 | 32 | !ReadPod(input, &metadata_size, &remaining) || | |
| 895 | 32 | !ReadPod(input, &record_count, &remaining) || | |
| 896 | 32 | magic != kCheckpointMagic || | |
| 897 | 30 | (version != kCheckpointVersion && version != 2) || | |
| 898 | 94 | metadata_size > remaining || | |
| 899 | 30 | metadata_size > std::numeric_limits<size_t>::max()) { | |
| 900 | 2 | LOG(ERROR) << "Invalid or truncated checkpoint header: " << file; | |
| 901 | 2 | return false; | |
| 902 | } | ||
| 903 | 30 | ValueStore::LayoutId file_layout; | |
| 904 | 56 | if (version == kCheckpointVersion && | |
| 905 | 26 | (!ReadPod(input, &file_layout.strategy, &remaining) || | |
| 906 | 26 | !ReadPod(input, &file_layout.checksum, &remaining))) { | |
| 907 | ✗ | LOG(ERROR) << "Truncated checkpoint layout identity: " << file; | |
| 908 | ✗ | return false; | |
| 909 | } | ||
| 910 | |||
| 911 | 30 | std::string metadata(static_cast<size_t>(metadata_size), '\0'); | |
| 912 | 30 | if (!ReadBytes(input, metadata.data(), metadata_size, &remaining)) { | |
| 913 | ✗ | LOG(ERROR) << "Truncated checkpoint metadata: " << file; | |
| 914 | ✗ | return false; | |
| 915 | } | ||
| 916 | 30 | if (metadata != expected_metadata) { | |
| 917 | 2 | LOG(ERROR) << "Checkpoint metadata mismatch: " << file; | |
| 918 | 2 | return false; | |
| 919 | } | ||
| 920 | 28 | const ValueStore::LayoutId layout_id = value_store_->GetLayoutId(); | |
| 921 | 28 | if (version == 2) { | |
| 922 | 4 | if (!layout_id.plain()) { | |
| 923 | 2 | LOG(ERROR) << "Checkpoint has no SSD layout identity: " << file; | |
| 924 | 2 | return false; | |
| 925 | } | ||
| 926 | 24 | } else if (file_layout.strategy != layout_id.strategy || | |
| 927 | 22 | file_layout.checksum != layout_id.checksum) { | |
| 928 | 2 | LOG(ERROR) << "Checkpoint SSD layout identity mismatch: " << file; | |
| 929 | 2 | return false; | |
| 930 | } | ||
| 931 | 24 | if (remaining < sizeof(uint64_t)) { | |
| 932 | ✗ | LOG(ERROR) << "Checkpoint checksum is missing: " << file; | |
| 933 | ✗ | return false; | |
| 934 | } | ||
| 935 | 48 | if (record_count > std::numeric_limits<size_t>::max() || | |
| 936 | 24 | record_count > | |
| 937 | 24 | (remaining - sizeof(uint64_t)) / (2 * sizeof(uint64_t))) { | |
| 938 | ✗ | LOG(ERROR) << "Invalid checkpoint record count: " << file; | |
| 939 | ✗ | return false; | |
| 940 | } | ||
| 941 | |||
| 942 | 24 | uint64_t checksum = kCheckpointChecksumSeed; | |
| 943 | 24 | UpdateChecksum(&checksum, magic.data(), magic.size()); | |
| 944 | 24 | UpdateChecksumPod(&checksum, version); | |
| 945 | 24 | UpdateChecksumPod(&checksum, metadata_size); | |
| 946 | 24 | UpdateChecksumPod(&checksum, record_count); | |
| 947 | 24 | if (version == kCheckpointVersion) { | |
| 948 | 22 | UpdateChecksumPod(&checksum, file_layout.strategy); | |
| 949 | 22 | UpdateChecksumPod(&checksum, file_layout.checksum); | |
| 950 | } | ||
| 951 | 24 | UpdateChecksum(&checksum, metadata.data(), metadata.size()); | |
| 952 | 24 | std::vector<std::pair<uint64_t, std::string>> records; | |
| 953 | 24 | records.reserve(static_cast<size_t>(record_count)); | |
| 954 | 24 | std::unordered_set<uint64_t> seen_keys; | |
| 955 | 24 | seen_keys.reserve(static_cast<size_t>(record_count)); | |
| 956 | 1094 | for (uint64_t i = 0; i < record_count; ++i) { | |
| 957 | 1070 | uint64_t key = 0; | |
| 958 | 1070 | uint64_t value_size = 0; | |
| 959 | 1070 | if (!ReadPod(input, &key, &remaining) || | |
| 960 | 1070 | !ReadPod(input, &value_size, &remaining) || value_size > remaining || | |
| 961 | 3210 | value_size > std::numeric_limits<size_t>::max() || | |
| 962 | 1070 | !seen_keys.insert(key).second) { | |
| 963 | ✗ | LOG(ERROR) << "Invalid or truncated checkpoint record: " << file; | |
| 964 | ✗ | return false; | |
| 965 | } | ||
| 966 | 1070 | std::string value(static_cast<size_t>(value_size), '\0'); | |
| 967 | 1070 | if (!ReadBytes(input, value.data(), value_size, &remaining)) { | |
| 968 | ✗ | LOG(ERROR) << "Truncated checkpoint value: " << file; | |
| 969 | ✗ | return false; | |
| 970 | } | ||
| 971 | 1070 | UpdateChecksumPod(&checksum, key); | |
| 972 | 1070 | UpdateChecksumPod(&checksum, value_size); | |
| 973 | 1070 | UpdateChecksum(&checksum, value.data(), value.size()); | |
| 974 | 1070 | records.emplace_back(key, std::move(value)); | |
| 975 | 1070 | } | |
| 976 | 24 | uint64_t saved_checksum = 0; | |
| 977 | 24 | if (!ReadPod(input, &saved_checksum, &remaining) || remaining != 0) { | |
| 978 | 2 | LOG(ERROR) << "Checkpoint contains trailing data: " << file; | |
| 979 | 2 | return false; | |
| 980 | } | ||
| 981 | 22 | if (saved_checksum != checksum) { | |
| 982 | 2 | LOG(ERROR) << "Checkpoint checksum mismatch: " << file; | |
| 983 | 2 | return false; | |
| 984 | } | ||
| 985 | |||
| 986 | 1086 | for (const auto& record : records) { | |
| 987 | 1066 | PutInternal( | |
| 988 | 1066 | record.first, record.second.data(), record.second.size(), 0, false); | |
| 989 | } | ||
| 990 | 20 | return true; | |
| 991 | 34 | } | |
| 992 | |||
| 993 | 48 | uint64_t CheckpointRecordCount() const override { return ActiveKeyCount(); } | |
| 994 | |||
| 995 | // CollectCheckpointKeys traverses checkpoint_keys_by_thread_ while the put | ||
| 996 | // path (TrackKeys) push_backs into it under a *shared* lock, so collectors | ||
| 997 | // must take the exclusive lock; a shared lock here races the writers. | ||
| 998 | 48 | uint64_t ActiveKeyCount() const override { | |
| 999 | 48 | std::unique_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 1000 | 96 | return static_cast<uint64_t>(CollectCheckpointKeys().size()); | |
| 1001 | 48 | } | |
| 1002 | |||
| 1003 | ✗ | std::vector<uint64_t> SnapshotKeys() const override { | |
| 1004 | ✗ | std::unique_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 1005 | ✗ | return CollectCheckpointKeys(); | |
| 1006 | ✗ | } | |
| 1007 | |||
| 1008 | ✗ | void clear() override { | |
| 1009 | ✗ | std::unique_lock<std::shared_mutex> checkpoint_lock(checkpoint_mu_); | |
| 1010 | ✗ | const std::vector<uint64_t> keys = CollectCheckpointKeys(); | |
| 1011 | ✗ | if (!keys.empty()) { | |
| 1012 | ✗ | DeleteKeysLocked(base::ConstArray<uint64_t>(keys), 0); | |
| 1013 | } | ||
| 1014 | ✗ | } | |
| 1015 | |||
| 1016 | ✗ | void Util() override { | |
| 1017 | ✗ | LOG(INFO) << "KVEngine index utilization=" << index_->Utilization() | |
| 1018 | ✗ | << " value=" << value_store_->GetInfo(); | |
| 1019 | ✗ | } | |
| 1020 | |||
| 1021 | ✗ | void DebugInfo() const override { | |
| 1022 | ✗ | index_->DebugInfo(); | |
| 1023 | ✗ | LOG(INFO) << value_store_->GetInfo(); | |
| 1024 | ✗ | } | |
| 1025 | |||
| 1026 | ✗ | std::string ExtraResultFields() const override { | |
| 1027 | ✗ | return value_store_ ? value_store_->ExtraResultFields() : ""; | |
| 1028 | } | ||
| 1029 | |||
| 1030 | ✗ | void ResetStats() override { | |
| 1031 | ✗ | if (value_store_) { | |
| 1032 | ✗ | value_store_->ResetStats(); | |
| 1033 | } | ||
| 1034 | ✗ | } | |
| 1035 | |||
| 1036 | private: | ||
| 1037 | 3364 | static bool WriteBytes(std::ofstream& output, const void* data, size_t size) { | |
| 1038 | 3364 | if (size > | |
| 1039 | 3364 | static_cast<size_t>((std::numeric_limits<std::streamsize>::max)())) { | |
| 1040 | ✗ | return false; | |
| 1041 | } | ||
| 1042 | 3364 | if (size != 0) { | |
| 1043 | 3364 | output.write(static_cast<const char*>(data), | |
| 1044 | static_cast<std::streamsize>(size)); | ||
| 1045 | } | ||
| 1046 | 3364 | return output.good(); | |
| 1047 | } | ||
| 1048 | |||
| 1049 | template <typename T> | ||
| 1050 | 2256 | static bool WritePod(std::ofstream& output, const T& value) { | |
| 1051 | 2256 | return WriteBytes(output, &value, sizeof(value)); | |
| 1052 | } | ||
| 1053 | |||
| 1054 | 3444 | static bool ReadBytes( | |
| 1055 | std::ifstream& input, void* data, uint64_t size, uint64_t* remaining) { | ||
| 1056 | 6886 | if (remaining == nullptr || size > *remaining || | |
| 1057 | 3442 | size > static_cast<uint64_t>( | |
| 1058 | 3442 | (std::numeric_limits<std::streamsize>::max)())) { | |
| 1059 | 2 | return false; | |
| 1060 | } | ||
| 1061 | 3442 | if (size != 0) { | |
| 1062 | 3442 | input.read(static_cast<char*>(data), static_cast<std::streamsize>(size)); | |
| 1063 | 3442 | if (!input) { | |
| 1064 | ✗ | return false; | |
| 1065 | } | ||
| 1066 | } | ||
| 1067 | 3442 | *remaining -= size; | |
| 1068 | 3442 | return true; | |
| 1069 | } | ||
| 1070 | |||
| 1071 | template <typename T> | ||
| 1072 | 2312 | static bool ReadPod(std::ifstream& input, T* value, uint64_t* remaining) { | |
| 1073 | 2312 | return ReadBytes(input, value, sizeof(*value), remaining); | |
| 1074 | } | ||
| 1075 | |||
| 1076 | static void | ||
| 1077 | 6718 | UpdateChecksum(uint64_t* checksum, const void* data, size_t size) { | |
| 1078 | 6718 | if (size != 0) { | |
| 1079 | 6718 | *checksum = xxhash(data, size, *checksum); | |
| 1080 | } | ||
| 1081 | 6718 | } | |
| 1082 | |||
| 1083 | template <typename T> | ||
| 1084 | 4492 | static void UpdateChecksumPod(uint64_t* checksum, const T& value) { | |
| 1085 | 4492 | UpdateChecksum(checksum, &value, sizeof(value)); | |
| 1086 | 4492 | } | |
| 1087 | |||
| 1088 | 68 | std::vector<uint64_t> CollectCheckpointKeys() const { | |
| 1089 | 68 | size_t count = 0; | |
| 1090 | 204 | for (const auto& keys : checkpoint_keys_by_thread_) { | |
| 1091 | 136 | count += keys.size(); | |
| 1092 | } | ||
| 1093 | 68 | std::vector<uint64_t> keys; | |
| 1094 | 68 | keys.reserve(count); | |
| 1095 | 204 | for (const auto& thread_keys : checkpoint_keys_by_thread_) { | |
| 1096 | 136 | keys.insert(keys.end(), thread_keys.begin(), thread_keys.end()); | |
| 1097 | } | ||
| 1098 | 68 | std::sort(keys.begin(), keys.end()); | |
| 1099 | 68 | keys.erase(std::unique(keys.begin(), keys.end()), keys.end()); | |
| 1100 | 68 | return keys; | |
| 1101 | ✗ | } | |
| 1102 | |||
| 1103 | 34 | bool HasCheckpointRecords() const { | |
| 1104 | 34 | return std::any_of(checkpoint_keys_by_thread_.begin(), | |
| 1105 | checkpoint_keys_by_thread_.end(), | ||
| 1106 | 98 | [](const auto& keys) { return !keys.empty(); }); | |
| 1107 | } | ||
| 1108 | |||
| 1109 | 200180 | std::vector<uint64_t>& CheckpointKeysForThread(unsigned tid) { | |
| 1110 | 200180 | if (tid >= checkpoint_keys_by_thread_.size()) { | |
| 1111 | ✗ | LOG(FATAL) << "KVEngine checkpoint tid out of range: " << tid | |
| 1112 | ✗ | << " >= " << checkpoint_keys_by_thread_.size(); | |
| 1113 | } | ||
| 1114 | 200180 | return checkpoint_keys_by_thread_[tid]; | |
| 1115 | } | ||
| 1116 | |||
| 1117 | 200124 | void TrackKey(uint64_t key, unsigned tid) { | |
| 1118 | 200124 | CheckpointKeysForThread(tid).push_back(key); | |
| 1119 | 200124 | } | |
| 1120 | |||
| 1121 | 56 | void TrackKeys(base::ConstArray<uint64_t> keys, unsigned tid) { | |
| 1122 | 56 | auto& thread_keys = CheckpointKeysForThread(tid); | |
| 1123 | 56 | thread_keys.reserve(thread_keys.size() + static_cast<size_t>(keys.Size())); | |
| 1124 | 490 | for (int i = 0; i < keys.Size(); ++i) { | |
| 1125 | 434 | thread_keys.push_back(keys[i]); | |
| 1126 | } | ||
| 1127 | 56 | } | |
| 1128 | |||
| 1129 | 4 | uint64_t DeleteKeysLocked(base::ConstArray<uint64_t> keys, unsigned tid) { | |
| 1130 | (void)tid; | ||
| 1131 | 4 | std::unordered_set<uint64_t> deleted_keys; | |
| 1132 | 4 | deleted_keys.reserve(static_cast<size_t>(keys.Size())); | |
| 1133 | 16 | for (int i = 0; i < keys.Size(); ++i) { | |
| 1134 | 12 | Value_t handle = kValueHandleNone; | |
| 1135 | 12 | index_->Get(keys[i], handle); | |
| 1136 | 12 | if (handle == kValueHandleNone) { | |
| 1137 | 4 | continue; | |
| 1138 | } | ||
| 1139 | 8 | uint64_t key = keys[i]; | |
| 1140 | 8 | if (!index_->Delete(key)) { | |
| 1141 | ✗ | continue; | |
| 1142 | } | ||
| 1143 | 8 | value_store_->Retire(handle); | |
| 1144 | 8 | deleted_keys.insert(key); | |
| 1145 | } | ||
| 1146 | 4 | if (!deleted_keys.empty()) { | |
| 1147 | 12 | for (auto& tracked_keys : checkpoint_keys_by_thread_) { | |
| 1148 | 16 | tracked_keys.erase( | |
| 1149 | 8 | std::remove_if( | |
| 1150 | tracked_keys.begin(), | ||
| 1151 | tracked_keys.end(), | ||
| 1152 | 16 | [&](uint64_t key) { | |
| 1153 | 16 | return deleted_keys.find(key) != deleted_keys.end(); | |
| 1154 | }), | ||
| 1155 | 16 | tracked_keys.end()); | |
| 1156 | } | ||
| 1157 | } | ||
| 1158 | 8 | return static_cast<uint64_t>(deleted_keys.size()); | |
| 1159 | 4 | } | |
| 1160 | |||
| 1161 | struct PutInternalStats { | ||
| 1162 | std::uint64_t alloc_ns = 0; | ||
| 1163 | std::uint64_t index_ns = 0; | ||
| 1164 | std::uint64_t track_ns = 0; | ||
| 1165 | }; | ||
| 1166 | |||
| 1167 | 407270 | void PutInternal(uint64_t key, | |
| 1168 | const void* data, | ||
| 1169 | size_t size, | ||
| 1170 | unsigned tid, | ||
| 1171 | bool emit_fence, | ||
| 1172 | PutInternalStats* stats = nullptr) { | ||
| 1173 | (void)tid; | ||
| 1174 | (void)emit_fence; | ||
| 1175 | using clock = std::chrono::steady_clock; | ||
| 1176 | const auto alloc_start = | ||
| 1177 | 407270 | stats != nullptr ? clock::now() : clock::time_point{}; | |
| 1178 | 407270 | Value_t new_handle = value_store_->AllocAndWrite(key, data, size); | |
| 1179 | 407270 | if (new_handle == kValueHandleNone) { | |
| 1180 | ✗ | LOG(FATAL) << "KVEngine value allocation failed, key=" << key | |
| 1181 | ✗ | << " size=" << size; | |
| 1182 | return; | ||
| 1183 | } | ||
| 1184 | const auto index_start = | ||
| 1185 | 407270 | stats != nullptr ? clock::now() : clock::time_point{}; | |
| 1186 | 407270 | Value_t old_handle = index_->Put(key, new_handle, tid); | |
| 1187 | 407270 | if (old_handle != kValueHandleNone && old_handle != new_handle) { | |
| 1188 | 207884 | value_store_->Retire(old_handle); | |
| 1189 | } | ||
| 1190 | const auto track_start = | ||
| 1191 | 407270 | stats != nullptr ? clock::now() : clock::time_point{}; | |
| 1192 | // Checkpoint tracking already contains overwritten keys. Delete removes | ||
| 1193 | // them under the exclusive checkpoint lock, so reinsertion tracks again. | ||
| 1194 | 407270 | if (old_handle == kValueHandleNone) { | |
| 1195 | 199384 | TrackKey(key, tid); | |
| 1196 | } | ||
| 1197 | 407270 | if (stats != nullptr) { | |
| 1198 | ✗ | const auto end = clock::now(); | |
| 1199 | ✗ | stats->alloc_ns += static_cast<std::uint64_t>( | |
| 1200 | ✗ | std::chrono::duration_cast<std::chrono::nanoseconds>( | |
| 1201 | ✗ | index_start - alloc_start) | |
| 1202 | ✗ | .count()); | |
| 1203 | ✗ | stats->index_ns += static_cast<std::uint64_t>( | |
| 1204 | ✗ | std::chrono::duration_cast<std::chrono::nanoseconds>( | |
| 1205 | ✗ | track_start - index_start) | |
| 1206 | ✗ | .count()); | |
| 1207 | ✗ | stats->track_ns += static_cast<std::uint64_t>( | |
| 1208 | ✗ | std::chrono::duration_cast<std::chrono::nanoseconds>( | |
| 1209 | ✗ | end - track_start) | |
| 1210 | ✗ | .count()); | |
| 1211 | } | ||
| 1212 | } | ||
| 1213 | |||
| 1214 | inline static constexpr std::array<char, 8> kCheckpointMagic = { | ||
| 1215 | 'R', 'S', 'K', 'V', 'C', 'P', '0', '1'}; | ||
| 1216 | inline static constexpr uint32_t kCheckpointVersion = 3; | ||
| 1217 | inline static constexpr uint64_t kCheckpointChecksumSeed = | ||
| 1218 | 0x9e3779b97f4a7c15ULL; | ||
| 1219 | |||
| 1220 | BaseKVConfig config_; | ||
| 1221 | std::unique_ptr<Index> index_; | ||
| 1222 | std::unique_ptr<ValueStore> value_store_; | ||
| 1223 | int num_threads_ = 0; | ||
| 1224 | size_t default_value_size_hint_ = 0; | ||
| 1225 | mutable std::shared_mutex checkpoint_mu_; | ||
| 1226 | std::vector<std::vector<uint64_t>> checkpoint_keys_by_thread_; | ||
| 1227 | }; | ||
| 1228 | |||
| 1229 | FACTORY_REGISTER( | ||
| 1230 | BaseKV, KVEngineComposite, KVEngineComposite, const BaseKVConfig&); | ||
| 1231 |