storage/value_store/hybrid_value_store.h
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | #pragma once | ||
| 2 | |||
| 3 | #include <atomic> | ||
| 4 | #include <iomanip> | ||
| 5 | #include <memory> | ||
| 6 | #include <sstream> | ||
| 7 | #include <stdexcept> | ||
| 8 | |||
| 9 | #include "base/factory.h" | ||
| 10 | #include "storage/value_store/dram_value_store.h" | ||
| 11 | #include "storage/value_store/ssd_static_layout_value_store.h" | ||
| 12 | #include "storage/value_store/ssd_value_store.h" | ||
| 13 | |||
| 14 | class HybridValueStore : public ValueStore { | ||
| 15 | public: | ||
| 16 | 342 | explicit HybridValueStore(const BaseKVConfig& config) | |
| 17 | 342 | : dram_store_(BuildDramConfig(config)) { | |
| 18 | 342 | const auto& value = config.json_config_.at("value"); | |
| 19 | 342 | ssd_store_.reset( | |
| 20 | base::Factory<ValueStore, const BaseKVConfig&>::NewInstance( | ||
| 21 | 684 | SsdStoreType(value), BuildSsdConfig(config))); | |
| 22 | 342 | ssd_static_ = ssd_store_ && !ssd_store_->GetLayoutId().plain(); | |
| 23 | 342 | const auto& v = config.json_config_.at("value"); | |
| 24 | 342 | const auto& dram = v.at("dram_allocator"); | |
| 25 | 342 | dram_capacity_bytes_ = dram.at("capacity_bytes").get<uint64_t>(); | |
| 26 | const auto& tiering = | ||
| 27 | 342 | v.contains("tiering") ? v.at("tiering") : json::object(); | |
| 28 | 342 | high_watermark_ratio_ = tiering.value("high_watermark_ratio", 0.85); | |
| 29 | 342 | } | |
| 30 | |||
| 31 | 98590 | uint64_t Alloc(size_t size) override { | |
| 32 | 98590 | if (ShouldUseDram(size)) { | |
| 33 | 98590 | const uint64_t raw = dram_store_.Alloc(size); | |
| 34 | 98590 | if (raw != kValueHandleNone) { | |
| 35 | 98590 | dram_bytes_reserved_.fetch_add( | |
| 36 | dram_store_.SlotCapacity(raw), std::memory_order_relaxed); | ||
| 37 | 98590 | dram_live_allocs_.fetch_add(1, std::memory_order_relaxed); | |
| 38 | 98590 | dram_total_allocs_.fetch_add(1, std::memory_order_relaxed); | |
| 39 | 98590 | return MakeDramHandle(raw); | |
| 40 | } | ||
| 41 | } | ||
| 42 | ✗ | const uint64_t raw = ssd_store_->Alloc(size); | |
| 43 | ✗ | if (raw == kValueHandleNone) { | |
| 44 | ✗ | return kValueHandleNone; | |
| 45 | } | ||
| 46 | ✗ | ssd_live_allocs_.fetch_add(1, std::memory_order_relaxed); | |
| 47 | ✗ | ssd_total_allocs_.fetch_add(1, std::memory_order_relaxed); | |
| 48 | ✗ | return MakeSsdHandle(raw); | |
| 49 | } | ||
| 50 | |||
| 51 | 98614 | void Write(uint64_t handle, const void* data, size_t size) override { | |
| 52 | 98614 | if (handle == kValueHandleNone) { | |
| 53 | ✗ | return; | |
| 54 | } | ||
| 55 | 98614 | if (IsOnSSD(handle)) { | |
| 56 | 22 | ssd_store_->Write(SsdRawHandle(handle), data, size); | |
| 57 | 22 | return; | |
| 58 | } | ||
| 59 | 98592 | dram_store_.Write(DramRawHandle(handle), data, size); | |
| 60 | } | ||
| 61 | |||
| 62 | 98590 | uint64_t AllocAndWrite(const void* data, size_t size) override { | |
| 63 | 98590 | const uint64_t handle = Alloc(size); | |
| 64 | 98590 | if (handle != kValueHandleNone) { | |
| 65 | 98590 | Write(handle, data, size); | |
| 66 | } | ||
| 67 | 98590 | return handle; | |
| 68 | } | ||
| 69 | |||
| 70 | 98240 | uint64_t AllocAndWrite(uint64_t key, const void* data, size_t size) override { | |
| 71 | 98240 | if (!ssd_static_) { | |
| 72 | 98212 | return AllocAndWrite(data, size); | |
| 73 | } | ||
| 74 | 28 | const uint64_t handle = PlaceStatic(key, size); | |
| 75 | 24 | if (handle != kValueHandleNone) { | |
| 76 | 24 | Write(handle, data, size); | |
| 77 | } | ||
| 78 | 24 | return handle; | |
| 79 | } | ||
| 80 | |||
| 81 | 2 | size_t Read(uint64_t handle, void* out_buf, size_t buf_size) override { | |
| 82 | 2 | if (handle == kValueHandleNone) { | |
| 83 | ✗ | return 0; | |
| 84 | } | ||
| 85 | 2 | if (IsOnSSD(handle)) { | |
| 86 | 2 | return ssd_store_->Read(SsdRawHandle(handle), out_buf, buf_size); | |
| 87 | } | ||
| 88 | ✗ | return dram_store_.Read(DramRawHandle(handle), out_buf, buf_size); | |
| 89 | } | ||
| 90 | |||
| 91 | 41756 | void Free(uint64_t handle) override { | |
| 92 | 41756 | if (handle == kValueHandleNone) { | |
| 93 | ✗ | return; | |
| 94 | } | ||
| 95 | 41756 | if (IsOnSSD(handle)) { | |
| 96 | ✗ | ssd_store_->Free(SsdRawHandle(handle)); | |
| 97 | ✗ | ssd_live_allocs_.fetch_sub(1, std::memory_order_relaxed); | |
| 98 | ✗ | return; | |
| 99 | } | ||
| 100 | 41756 | const uint64_t raw = DramRawHandle(handle); | |
| 101 | 41756 | const size_t cap = dram_store_.SlotCapacity(raw); | |
| 102 | 41756 | dram_store_.Free(raw); | |
| 103 | 41756 | dram_live_allocs_.fetch_sub(1, std::memory_order_relaxed); | |
| 104 | 41756 | uint64_t cur = dram_bytes_reserved_.load(std::memory_order_relaxed); | |
| 105 | 86808 | while (cur != 0 && | |
| 106 | 45052 | !dram_bytes_reserved_.compare_exchange_weak( | |
| 107 | 45052 | cur, cur > cap ? cur - cap : 0, std::memory_order_relaxed)) { | |
| 108 | } | ||
| 109 | } | ||
| 110 | |||
| 111 | 73952 | const char* DirectPtr(uint64_t handle) const override { | |
| 112 | 73952 | if (handle == kValueHandleNone || IsOnSSD(handle)) { | |
| 113 | 20 | return nullptr; | |
| 114 | } | ||
| 115 | 73932 | return dram_store_.DirectPtr(DramRawHandle(handle)); | |
| 116 | } | ||
| 117 | |||
| 118 | ✗ | char* RDMABackingData() const override { | |
| 119 | ✗ | return dram_store_.RDMABackingData(); | |
| 120 | } | ||
| 121 | |||
| 122 | ✗ | size_t RDMABackingSize() const override { | |
| 123 | ✗ | return dram_store_.RDMABackingSize(); | |
| 124 | } | ||
| 125 | |||
| 126 | 73934 | size_t SlotCapacity(uint64_t handle) const override { | |
| 127 | 73934 | if (handle == kValueHandleNone) { | |
| 128 | ✗ | return 0; | |
| 129 | } | ||
| 130 | 73934 | if (IsOnSSD(handle)) { | |
| 131 | 2 | return ssd_store_->SlotCapacity(SsdRawHandle(handle)); | |
| 132 | } | ||
| 133 | 73932 | return dram_store_.SlotCapacity(DramRawHandle(handle)); | |
| 134 | } | ||
| 135 | |||
| 136 | ✗ | void BatchWrite(const std::vector<uint64_t>& handles, | |
| 137 | const std::vector<WriteSpec>& specs) override { | ||
| 138 | ✗ | if (handles.size() != specs.size()) { | |
| 139 | ✗ | throw std::invalid_argument("HybridValueStore::BatchWrite size mismatch"); | |
| 140 | } | ||
| 141 | ✗ | std::vector<uint64_t> dram_handles; | |
| 142 | ✗ | std::vector<WriteSpec> dram_specs; | |
| 143 | ✗ | std::vector<uint64_t> ssd_handles; | |
| 144 | ✗ | std::vector<WriteSpec> ssd_specs; | |
| 145 | ✗ | dram_handles.reserve(handles.size()); | |
| 146 | ✗ | dram_specs.reserve(handles.size()); | |
| 147 | ✗ | ssd_handles.reserve(handles.size()); | |
| 148 | ✗ | ssd_specs.reserve(handles.size()); | |
| 149 | ✗ | for (size_t i = 0; i < handles.size(); ++i) { | |
| 150 | ✗ | if (handles[i] == kValueHandleNone) { | |
| 151 | ✗ | continue; | |
| 152 | } | ||
| 153 | ✗ | if (IsOnSSD(handles[i])) { | |
| 154 | ✗ | ssd_handles.push_back(SsdRawHandle(handles[i])); | |
| 155 | ✗ | ssd_specs.push_back(specs[i]); | |
| 156 | } else { | ||
| 157 | ✗ | dram_handles.push_back(DramRawHandle(handles[i])); | |
| 158 | ✗ | dram_specs.push_back(specs[i]); | |
| 159 | } | ||
| 160 | } | ||
| 161 | ✗ | if (!dram_handles.empty()) { | |
| 162 | ✗ | dram_store_.BatchWrite(dram_handles, dram_specs); | |
| 163 | } | ||
| 164 | ✗ | if (!ssd_handles.empty()) { | |
| 165 | ✗ | ssd_store_->BatchWrite(ssd_handles, ssd_specs); | |
| 166 | } | ||
| 167 | ✗ | } | |
| 168 | |||
| 169 | 4 | void BatchRead(const std::vector<uint64_t>& handles, | |
| 170 | std::vector<ReadResult>& out_results) override { | ||
| 171 | 4 | out_results.clear(); | |
| 172 | 4 | out_results.resize(handles.size()); | |
| 173 | |||
| 174 | 4 | std::vector<uint64_t> ssd_handles; | |
| 175 | 4 | std::vector<size_t> ssd_indices; | |
| 176 | 4 | ssd_handles.reserve(handles.size()); | |
| 177 | 4 | ssd_indices.reserve(handles.size()); | |
| 178 | |||
| 179 | 22 | for (size_t i = 0; i < handles.size(); ++i) { | |
| 180 | 18 | const uint64_t handle = handles[i]; | |
| 181 | 18 | if (handle == kValueHandleNone) { | |
| 182 | ✗ | continue; | |
| 183 | } | ||
| 184 | 18 | if (IsOnSSD(handle)) { | |
| 185 | 18 | ssd_handles.push_back(SsdRawHandle(handle)); | |
| 186 | 18 | ssd_indices.push_back(i); | |
| 187 | 18 | continue; | |
| 188 | } | ||
| 189 | |||
| 190 | ✗ | const uint64_t raw = DramRawHandle(handle); | |
| 191 | ✗ | out_results[i].data.resize(dram_store_.SlotCapacity(raw)); | |
| 192 | ✗ | const size_t actual = dram_store_.Read( | |
| 193 | ✗ | raw, out_results[i].data.data(), out_results[i].data.size()); | |
| 194 | ✗ | out_results[i].data.resize(actual); | |
| 195 | } | ||
| 196 | |||
| 197 | 4 | if (ssd_handles.empty()) { | |
| 198 | ✗ | return; | |
| 199 | } | ||
| 200 | |||
| 201 | 4 | std::vector<ReadResult> ssd_results; | |
| 202 | 4 | ssd_store_->BatchRead(ssd_handles, ssd_results); | |
| 203 | 4 | if (ssd_results.size() != ssd_indices.size()) { | |
| 204 | ✗ | throw std::runtime_error( | |
| 205 | ✗ | "HybridValueStore::BatchRead result size mismatch"); | |
| 206 | } | ||
| 207 | 22 | for (size_t i = 0; i < ssd_indices.size(); ++i) { | |
| 208 | 18 | out_results[ssd_indices[i]] = std::move(ssd_results[i]); | |
| 209 | } | ||
| 210 | 4 | } | |
| 211 | |||
| 212 | ✗ | std::string GetInfo() const override { | |
| 213 | ✗ | std::ostringstream os; | |
| 214 | ✗ | os << "HybridValueStore(dram_reserved=" | |
| 215 | ✗ | << dram_bytes_reserved_.load(std::memory_order_relaxed) | |
| 216 | ✗ | << ", dram_capacity=" << dram_capacity_bytes_ << ", dram=" | |
| 217 | ✗ | << dram_store_.GetInfo() << ", ssd=" << ssd_store_->GetInfo() << ")"; | |
| 218 | ✗ | return os.str(); | |
| 219 | ✗ | } | |
| 220 | |||
| 221 | ✗ | void ResetStats() override { | |
| 222 | ✗ | if (ssd_store_) { | |
| 223 | ✗ | ssd_store_->ResetStats(); | |
| 224 | } | ||
| 225 | ✗ | } | |
| 226 | |||
| 227 | ✗ | std::string ExtraResultFields() const override { | |
| 228 | const uint64_t dram_live = | ||
| 229 | ✗ | dram_live_allocs_.load(std::memory_order_relaxed); | |
| 230 | ✗ | const uint64_t ssd_live = ssd_live_allocs_.load(std::memory_order_relaxed); | |
| 231 | ✗ | const uint64_t live = dram_live + ssd_live; | |
| 232 | const uint64_t dram_total = | ||
| 233 | ✗ | dram_total_allocs_.load(std::memory_order_relaxed); | |
| 234 | const uint64_t ssd_total = | ||
| 235 | ✗ | ssd_total_allocs_.load(std::memory_order_relaxed); | |
| 236 | ✗ | const uint64_t total = dram_total + ssd_total; | |
| 237 | ✗ | const double live_ratio = | |
| 238 | ✗ | live == 0 ? 0.0 : static_cast<double>(ssd_live) / live; | |
| 239 | ✗ | const double total_ratio = | |
| 240 | ✗ | total == 0 ? 0.0 : static_cast<double>(ssd_total) / total; | |
| 241 | ✗ | std::ostringstream os; | |
| 242 | ✗ | os << std::fixed << std::setprecision(6) | |
| 243 | ✗ | << " tiered_dram_capacity_bytes=" << dram_capacity_bytes_ | |
| 244 | ✗ | << " tiered_high_watermark_ratio=" << high_watermark_ratio_ | |
| 245 | ✗ | << " tiered_dram_live_allocs=" << dram_live << " tiered_ssd_live_allocs=" | |
| 246 | ✗ | << ssd_live << " tiered_live_allocs=" << live | |
| 247 | ✗ | << " tiered_ssd_live_ratio=" << live_ratio | |
| 248 | ✗ | << " tiered_dram_total_allocs=" << dram_total | |
| 249 | ✗ | << " tiered_ssd_total_allocs=" << ssd_total << " tiered_total_allocs=" | |
| 250 | ✗ | << total << " tiered_ssd_total_ratio=" << total_ratio; | |
| 251 | ✗ | return os.str() + ssd_store_->ExtraResultFields(); | |
| 252 | ✗ | } | |
| 253 | |||
| 254 | 108 | ValueStore::LayoutId GetLayoutId() const override { | |
| 255 | 108 | return ssd_store_ ? ssd_store_->GetLayoutId() : ValueStore::LayoutId{}; | |
| 256 | } | ||
| 257 | |||
| 258 | private: | ||
| 259 | static constexpr uint64_t kSsdFlag = 1ULL << 63; | ||
| 260 | |||
| 261 | 288276 | static bool IsOnSSD(uint64_t handle) { return (handle & kSsdFlag) != 0; } | |
| 262 | |||
| 263 | 98592 | static uint64_t MakeDramHandle(uint64_t raw) { return raw; } | |
| 264 | |||
| 265 | 22 | static uint64_t MakeSsdHandle(uint64_t raw) { return raw | kSsdFlag; } | |
| 266 | |||
| 267 | 288212 | static uint64_t DramRawHandle(uint64_t handle) { return handle; } | |
| 268 | |||
| 269 | 44 | static uint64_t SsdRawHandle(uint64_t handle) { return handle & ~kSsdFlag; } | |
| 270 | |||
| 271 | 342 | static BaseKVConfig BuildDramConfig(const BaseKVConfig& config) { | |
| 272 | 342 | BaseKVConfig out = config; | |
| 273 | 342 | auto& j = out.json_config_; | |
| 274 | 342 | if (!j.contains("value") || !j.at("value").contains("dram_allocator")) { | |
| 275 | ✗ | return out; | |
| 276 | } | ||
| 277 | 342 | json value = j.at("value"); | |
| 278 | 342 | value["type"] = "DRAM_VALUE_STORE"; | |
| 279 | 342 | if (value.at("dram_allocator").contains("path")) { | |
| 280 | 342 | value["path"] = value.at("dram_allocator").at("path"); | |
| 281 | } | ||
| 282 | 342 | value.erase("ssd_allocator"); | |
| 283 | 342 | value.erase("ssd_layout"); | |
| 284 | 342 | value.erase("tiering"); | |
| 285 | 342 | j["value"] = value; | |
| 286 | 342 | return out; | |
| 287 | 342 | } | |
| 288 | |||
| 289 | 684 | static std::string SsdStoreType(const json& value) { | |
| 290 | 684 | if (!value.contains("ssd_layout")) { | |
| 291 | 668 | return "SSD_VALUE_STORE"; | |
| 292 | } | ||
| 293 | const std::string strategy = | ||
| 294 | 16 | value.at("ssd_layout").value("strategy", "plain"); | |
| 295 | 16 | if (strategy.empty() || strategy == "plain") { | |
| 296 | ✗ | return "SSD_VALUE_STORE"; | |
| 297 | } | ||
| 298 | 16 | if (strategy == "bandana" || strategy == "maxembed") { | |
| 299 | 16 | return "SSD_STATIC_LAYOUT_VALUE_STORE"; | |
| 300 | } | ||
| 301 | ✗ | throw std::invalid_argument("unknown SSD layout strategy: " + strategy); | |
| 302 | 16 | } | |
| 303 | |||
| 304 | 342 | static BaseKVConfig BuildSsdConfig(const BaseKVConfig& config) { | |
| 305 | 342 | BaseKVConfig out = config; | |
| 306 | 342 | auto& j = out.json_config_; | |
| 307 | 342 | if (!j.contains("value") || !j.at("value").contains("ssd_allocator")) { | |
| 308 | ✗ | return out; | |
| 309 | } | ||
| 310 | 342 | json value = j.at("value"); | |
| 311 | 342 | value["type"] = SsdStoreType(value); | |
| 312 | 342 | if (value.at("ssd_allocator").contains("path")) { | |
| 313 | 342 | value["path"] = value.at("ssd_allocator").at("path"); | |
| 314 | } | ||
| 315 | 342 | value.erase("dram_allocator"); | |
| 316 | 342 | value.erase("tiering"); | |
| 317 | 342 | j["value"] = value; | |
| 318 | 342 | return out; | |
| 319 | 342 | } | |
| 320 | |||
| 321 | 98614 | bool ShouldUseDram(size_t size) const { | |
| 322 | 98614 | if (dram_capacity_bytes_ == 0) { | |
| 323 | ✗ | return false; | |
| 324 | } | ||
| 325 | const double next = | ||
| 326 | 98614 | static_cast<double>( | |
| 327 | 98614 | dram_bytes_reserved_.load(std::memory_order_relaxed) + size) / | |
| 328 | 98614 | static_cast<double>(dram_capacity_bytes_); | |
| 329 | 98614 | return next <= high_watermark_ratio_; | |
| 330 | } | ||
| 331 | |||
| 332 | 28 | uint64_t PlaceStatic(uint64_t key, size_t size) { | |
| 333 | 28 | if (key > ssd_layout::kMaxKey) { | |
| 334 | ✗ | throw std::invalid_argument("static SSD key is out of range"); | |
| 335 | } | ||
| 336 | 28 | const size_t row_size = ssd_store_->SlotCapacity(key + 1); | |
| 337 | 26 | if (size != row_size) { | |
| 338 | 2 | throw std::invalid_argument("static SSD layout value size mismatch"); | |
| 339 | } | ||
| 340 | 24 | if (ShouldUseDram(size)) { | |
| 341 | 2 | const uint64_t raw = dram_store_.Alloc(size); | |
| 342 | 2 | if (raw != kValueHandleNone) { | |
| 343 | 2 | dram_bytes_reserved_.fetch_add( | |
| 344 | dram_store_.SlotCapacity(raw), std::memory_order_relaxed); | ||
| 345 | 2 | dram_live_allocs_.fetch_add(1, std::memory_order_relaxed); | |
| 346 | 2 | dram_total_allocs_.fetch_add(1, std::memory_order_relaxed); | |
| 347 | 2 | return MakeDramHandle(raw); | |
| 348 | } | ||
| 349 | } | ||
| 350 | 22 | ssd_live_allocs_.fetch_add(1, std::memory_order_relaxed); | |
| 351 | 22 | ssd_total_allocs_.fetch_add(1, std::memory_order_relaxed); | |
| 352 | 22 | return MakeSsdHandle(key + 1); | |
| 353 | } | ||
| 354 | |||
| 355 | DramValueStore dram_store_; | ||
| 356 | std::unique_ptr<ValueStore> ssd_store_; | ||
| 357 | bool ssd_static_ = false; | ||
| 358 | uint64_t dram_capacity_bytes_ = 0; | ||
| 359 | double high_watermark_ratio_ = 0.85; | ||
| 360 | std::atomic<uint64_t> dram_bytes_reserved_{0}; | ||
| 361 | std::atomic<uint64_t> dram_live_allocs_{0}; | ||
| 362 | std::atomic<uint64_t> ssd_live_allocs_{0}; | ||
| 363 | std::atomic<uint64_t> dram_total_allocs_{0}; | ||
| 364 | std::atomic<uint64_t> ssd_total_allocs_{0}; | ||
| 365 | }; | ||
| 366 | |||
| 367 | FACTORY_REGISTER( | ||
| 368 | ValueStore, TIERED_VALUE_STORE, HybridValueStore, const BaseKVConfig&); | ||
| 369 |