apps/argos/Checkpoint.hpp
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | // <Checkpoint.hpp> -*- C++ -*- | ||
| 2 | |||
| 3 | #pragma once | ||
| 4 | |||
| 5 | #include "simdb/Exceptions.hpp" | ||
| 6 | #include "simdb/apps/argos/CheckpointDeltas.hpp" | ||
| 7 | #include "simdb/apps/argos/CollectedData.hpp" | ||
| 8 | #include "simdb/utils/ValidValue.hpp" | ||
| 9 | |||
| 10 | namespace simdb::argos { | ||
| 11 | |||
| 12 | //! Encodings that C++ and python agree on. The python deserializers interpret | ||
| 13 | //! collected data using the header: | ||
| 14 | //! | ||
| 15 | //! [uint16_t cid] // collectable ID | ||
| 16 | //! [uint8_t action] // encoding | ||
| 17 | //! | ||
| 18 | //! See CheckpointEncodings.hpp for full action / byte layout documentation. | ||
| 19 | enum class Action : uint8_t { | ||
| 20 | CLOSED = 0x00, | ||
| 21 | FULL = 0x01, | ||
| 22 | CARRY = 0x02, | ||
| 23 | |||
| 24 | CONTAINER_SWAP = 0x10, | ||
| 25 | CONTAINER_MULTI_SWAP = 0x11, | ||
| 26 | |||
| 27 | CONTIG_ARRIVE = 0x20, | ||
| 28 | CONTIG_DEPART = 0x21, | ||
| 29 | CONTIG_BOOKENDS = 0x22, | ||
| 30 | CONTIG_MIMO = 0x23, | ||
| 31 | |||
| 32 | SPARSE_REMOVE = 0x30, | ||
| 33 | SPARSE_ADD = 0x31, | ||
| 34 | SPARSE_MULTI_REMOVE = 0x32, | ||
| 35 | }; | ||
| 36 | |||
| 37 | //! \class CollectableCheckpoint | ||
| 38 | class CollectableCheckpoint | ||
| 39 | { | ||
| 40 | public: | ||
| 41 | /// Destroys this checkpoint node. | ||
| 42 | 41873 | virtual ~CollectableCheckpoint() = default; | |
| 43 | |||
| 44 | /// Returns the previous checkpoint in the chain, or nullptr. | ||
| 45 | 38363 | CollectableCheckpoint* getPrev() const { return prev_.get(); } | |
| 46 | |||
| 47 | /// Returns the next checkpoint in the chain, or nullptr. | ||
| 48 | 131727 | CollectableCheckpoint* getNext() const { return next_; } | |
| 49 | |||
| 50 | /// Returns the collection window that created this checkpoint. | ||
| 51 | 221672 | uint64_t getWindowID() const { return window_id_; } | |
| 52 | |||
| 53 | /// Drops the link to the previous checkpoint. | ||
| 54 | 37461 | void detachPrev() | |
| 55 | { | ||
| 56 | 37461 | if (prev_) | |
| 57 | { | ||
| 58 | 37370 | prev_->next_ = nullptr; | |
| 59 | } | ||
| 60 | 37461 | prev_.reset(); | |
| 61 | 37461 | } | |
| 62 | |||
| 63 | /// Returns the shared_ptr to the previous checkpoint, if any. | ||
| 64 | ✗ | const std::shared_ptr<CollectableCheckpoint>& getPrevShared() const { return prev_; } | |
| 65 | |||
| 66 | protected: | ||
| 67 | /// Links \p prev to this node and stores \p window_id. | ||
| 68 | 41873 | CollectableCheckpoint(std::shared_ptr<CollectableCheckpoint> prev, uint64_t window_id) : | |
| 69 | 41873 | prev_(prev), | |
| 70 | 41873 | window_id_(window_id) | |
| 71 | { | ||
| 72 | 41873 | if (prev_) | |
| 73 | { | ||
| 74 | 41782 | prev_->next_ = this; | |
| 75 | } | ||
| 76 | 41873 | } | |
| 77 | |||
| 78 | /// Returns true if a CLOSED event occurred between \p data_prev and \p self. | ||
| 79 | template <typename CheckpointT> | ||
| 80 | ✗ | static bool closedSincePrevDataCheckpoint_(const CheckpointT* self, const CheckpointT* data_prev) | |
| 81 | { | ||
| 82 | ✗ | if (!data_prev) | |
| 83 | { | ||
| 84 | ✗ | return false; | |
| 85 | } | ||
| 86 | ✗ | const CheckpointT* checkpoint = self->prev(); | |
| 87 | ✗ | while (checkpoint && checkpoint != data_prev) | |
| 88 | { | ||
| 89 | ✗ | if (checkpoint->isClosedEvent()) | |
| 90 | { | ||
| 91 | ✗ | return true; | |
| 92 | } | ||
| 93 | ✗ | checkpoint = checkpoint->prev(); | |
| 94 | } | ||
| 95 | ✗ | return false; | |
| 96 | } | ||
| 97 | |||
| 98 | std::shared_ptr<CollectableCheckpoint> prev_; | ||
| 99 | CollectableCheckpoint* next_ = nullptr; | ||
| 100 | const uint64_t window_id_; | ||
| 101 | }; | ||
| 102 | |||
| 103 | class ScalarCheckpoint final : public CollectableCheckpoint | ||
| 104 | { | ||
| 105 | public: | ||
| 106 | /// Creates a data checkpoint holding a full scalar snapshot. | ||
| 107 | 13008 | ScalarCheckpoint(std::shared_ptr<ScalarCheckpoint> prev, uint64_t window_id, const std::vector<char>& full_bytes) : | |
| 108 | CollectableCheckpoint(prev, window_id), | ||
| 109 | 13008 | full_bytes_(full_bytes) | |
| 110 | { | ||
| 111 | 13008 | } | |
| 112 | |||
| 113 | /// Creates a lifecycle checkpoint recording a close transition. | ||
| 114 | 1091 | ScalarCheckpoint(std::shared_ptr<ScalarCheckpoint> prev, uint64_t window_id, bool switched_to_closed) : | |
| 115 | CollectableCheckpoint(prev, window_id), | ||
| 116 | 1091 | lifecycle_change_(switched_to_closed) | |
| 117 | { | ||
| 118 | 1091 | } | |
| 119 | |||
| 120 | /// Returns the previous scalar checkpoint in the chain. | ||
| 121 | 12098 | ScalarCheckpoint* prev() { return static_cast<ScalarCheckpoint*>(getPrev()); } | |
| 122 | |||
| 123 | /// Returns the previous scalar checkpoint in the chain. | ||
| 124 | ✗ | const ScalarCheckpoint* prev() const { return static_cast<const ScalarCheckpoint*>(getPrev()); } | |
| 125 | |||
| 126 | /// Returns the next scalar checkpoint in the chain. | ||
| 127 | 44073 | ScalarCheckpoint* next() { return static_cast<ScalarCheckpoint*>(getNext()); } | |
| 128 | |||
| 129 | /// Returns true when this node holds collected bytes rather than a lifecycle flag. | ||
| 130 | 29095 | bool isDataCheckpoint() const { return !lifecycle_change_.isValid(); } | |
| 131 | |||
| 132 | /// Returns true when this node records a transition to closed. | ||
| 133 | 15906 | bool isClosedEvent() const { return lifecycle_change_.isValid() && lifecycle_change_.getValue(); } | |
| 134 | |||
| 135 | /// Returns true for any lifecycle (close) checkpoint. | ||
| 136 | bool isLifecycleEvent() const { return lifecycle_change_.isValid(); } | ||
| 137 | |||
| 138 | /// Encodes an unconditional FULL snapshot for heartbeat or absent-CID refresh. | ||
| 139 | 16997 | std::unique_ptr<CollectedData> encodeSnapshotForPipeline(uint16_t cid) const | |
| 140 | { | ||
| 141 | 16997 | auto encoded = std::make_unique<CollectedData>(cid); | |
| 142 | 16997 | auto& buf = encoded->getBuffer(); | |
| 143 | 16997 | buf.append(Action::FULL); | |
| 144 | 16997 | buf.append(full_bytes_); | |
| 145 | 16997 | return encoded; | |
| 146 | ✗ | } | |
| 147 | |||
| 148 | /// Encodes CLOSED, FULL, or CARRY by diffing against the prior data checkpoint. | ||
| 149 | 14099 | std::unique_ptr<CollectedData> encodeForPipeline(uint16_t cid, bool force_snapshot) | |
| 150 | { | ||
| 151 | 14099 | if (lifecycle_change_.isValid()) | |
| 152 | { | ||
| 153 | 1091 | return encodeLifecycleEvt_(cid); | |
| 154 | } | ||
| 155 | |||
| 156 | 13008 | if (force_snapshot) | |
| 157 | { | ||
| 158 | 13008 | return encodeFull_(cid); | |
| 159 | } | ||
| 160 | |||
| 161 | ✗ | const auto* data_prev = prevDataCheckpoint_(); | |
| 162 | ✗ | if (!data_prev) | |
| 163 | { | ||
| 164 | ✗ | return encodeFull_(cid); | |
| 165 | } | ||
| 166 | |||
| 167 | ✗ | if (data_prev->full_bytes_ != full_bytes_) | |
| 168 | { | ||
| 169 | ✗ | return encodeFull_(cid); | |
| 170 | } | ||
| 171 | |||
| 172 | ✗ | if (closedSincePrevDataCheckpoint_(this, data_prev)) | |
| 173 | { | ||
| 174 | ✗ | return encodeFull_(cid); | |
| 175 | } | ||
| 176 | |||
| 177 | ✗ | return encodeDelta_(cid); | |
| 178 | } | ||
| 179 | |||
| 180 | private: | ||
| 181 | /// Walks backward to the nearest prior data checkpoint. | ||
| 182 | ✗ | const ScalarCheckpoint* prevDataCheckpoint_() const | |
| 183 | { | ||
| 184 | ✗ | const auto* checkpoint = prev(); | |
| 185 | ✗ | while (checkpoint && !checkpoint->isDataCheckpoint()) | |
| 186 | { | ||
| 187 | ✗ | checkpoint = checkpoint->prev(); | |
| 188 | } | ||
| 189 | ✗ | return checkpoint; | |
| 190 | } | ||
| 191 | |||
| 192 | /// Encodes a CLOSED lifecycle wire record. | ||
| 193 | 1091 | std::unique_ptr<CollectedData> encodeLifecycleEvt_(uint16_t cid) | |
| 194 | { | ||
| 195 | 1091 | simdb_assert(lifecycle_change_.isValid() && lifecycle_change_.getValue()); | |
| 196 | 1091 | auto encoded = std::make_unique<CollectedData>(cid); | |
| 197 | 1091 | encoded->getBuffer().append(Action::CLOSED); | |
| 198 | 1091 | return encoded; | |
| 199 | ✗ | } | |
| 200 | |||
| 201 | /// Encodes a FULL snapshot of the current scalar bytes. | ||
| 202 | 13008 | std::unique_ptr<CollectedData> encodeFull_(uint16_t cid) | |
| 203 | { | ||
| 204 | 13008 | simdb_assert(!full_bytes_.empty()); | |
| 205 | 13008 | auto encoded = std::make_unique<CollectedData>(cid); | |
| 206 | 13008 | auto& buf = encoded->getBuffer(); | |
| 207 | 13008 | buf.append(Action::FULL); | |
| 208 | 13008 | buf.append(full_bytes_); | |
| 209 | 13008 | return encoded; | |
| 210 | ✗ | } | |
| 211 | |||
| 212 | /// Encodes CARRY when bytes are unchanged since the prior data checkpoint. | ||
| 213 | ✗ | std::unique_ptr<CollectedData> encodeDelta_(uint16_t cid) | |
| 214 | { | ||
| 215 | ✗ | const auto* data_prev = prevDataCheckpoint_(); | |
| 216 | ✗ | simdb_assert(data_prev && data_prev->full_bytes_ == full_bytes_ && !full_bytes_.empty()); | |
| 217 | (void)data_prev; | ||
| 218 | ✗ | auto encoded = std::make_unique<CollectedData>(cid); | |
| 219 | ✗ | auto& buf = encoded->getBuffer(); | |
| 220 | ✗ | buf.append(Action::CARRY); | |
| 221 | ✗ | return encoded; | |
| 222 | ✗ | } | |
| 223 | |||
| 224 | std::vector<char> full_bytes_; | ||
| 225 | ValidValue<bool> lifecycle_change_; | ||
| 226 | }; | ||
| 227 | |||
| 228 | class ContigContainerCheckpoint : public CollectableCheckpoint | ||
| 229 | { | ||
| 230 | public: | ||
| 231 | /// Creates a data checkpoint holding a full contig container snapshot. | ||
| 232 | 12015 | ContigContainerCheckpoint(std::shared_ptr<ContigContainerCheckpoint> prev, uint64_t window_id, | |
| 233 | 12015 | const std::vector<std::vector<char>>& full_bytes) : | |
| 234 | CollectableCheckpoint(prev, window_id), | ||
| 235 | 12015 | full_bytes_(full_bytes) | |
| 236 | { | ||
| 237 | 12015 | } | |
| 238 | |||
| 239 | /// Creates a lifecycle checkpoint recording a close transition. | ||
| 240 | 1713 | ContigContainerCheckpoint(std::shared_ptr<ContigContainerCheckpoint> prev, uint64_t window_id, | |
| 241 | 1713 | bool switched_to_closed) : | |
| 242 | CollectableCheckpoint(prev, window_id), | ||
| 243 | 1713 | lifecycle_change_(switched_to_closed) | |
| 244 | { | ||
| 245 | 1713 | } | |
| 246 | |||
| 247 | /// Returns the previous contig checkpoint in the chain. | ||
| 248 | 13395 | ContigContainerCheckpoint* prev() { return static_cast<ContigContainerCheckpoint*>(getPrev()); } | |
| 249 | |||
| 250 | /// Returns the previous contig checkpoint in the chain. | ||
| 251 | ✗ | const ContigContainerCheckpoint* prev() const { return static_cast<const ContigContainerCheckpoint*>(getPrev()); } | |
| 252 | |||
| 253 | /// Returns the next contig checkpoint in the chain. | ||
| 254 | 43677 | ContigContainerCheckpoint* next() { return static_cast<ContigContainerCheckpoint*>(getNext()); } | |
| 255 | |||
| 256 | /// Returns true when this node holds collected bytes rather than a lifecycle flag. | ||
| 257 | 49323 | bool isDataCheckpoint() const { return !lifecycle_change_.isValid(); } | |
| 258 | |||
| 259 | /// Returns true when this node records a transition to closed. | ||
| 260 | 16251 | bool isClosedEvent() const { return lifecycle_change_.isValid() && lifecycle_change_.getValue(); } | |
| 261 | |||
| 262 | /// Returns true for any lifecycle (close) checkpoint. | ||
| 263 | bool isLifecycleEvent() const { return lifecycle_change_.isValid(); } | ||
| 264 | |||
| 265 | /// Encodes an unconditional FULL snapshot for heartbeat or absent-CID refresh. | ||
| 266 | 17964 | std::unique_ptr<CollectedData> encodeSnapshotForPipeline(uint16_t cid) const | |
| 267 | { | ||
| 268 | 17964 | simdb_assert(isDataCheckpoint()); | |
| 269 | 17964 | auto encoded = std::make_unique<CollectedData>(cid); | |
| 270 | 17964 | auto& buf = encoded->getBuffer(); | |
| 271 | 17964 | buf.append(Action::FULL); | |
| 272 | 17964 | appendContigBins_(buf, full_bytes_); | |
| 273 | 17964 | return encoded; | |
| 274 | ✗ | } | |
| 275 | |||
| 276 | /// Encodes CLOSED, FULL, or a contig delta by diffing against the prior data checkpoint. | ||
| 277 | 13728 | std::unique_ptr<CollectedData> encodeForPipeline(uint16_t cid, bool force_snapshot) | |
| 278 | { | ||
| 279 | 13728 | if (lifecycle_change_.isValid()) | |
| 280 | { | ||
| 281 | 1713 | return encodeLifecycleEvt_(cid); | |
| 282 | } | ||
| 283 | |||
| 284 | 12015 | if (force_snapshot) | |
| 285 | { | ||
| 286 | 12015 | return encodeFull_(cid); | |
| 287 | } | ||
| 288 | |||
| 289 | ✗ | auto* data_prev = prevDataCheckpoint_(); | |
| 290 | ✗ | if (!data_prev) | |
| 291 | { | ||
| 292 | ✗ | return encodeFull_(cid); | |
| 293 | } | ||
| 294 | |||
| 295 | ✗ | if (closedSincePrevDataCheckpoint_(this, data_prev)) | |
| 296 | { | ||
| 297 | ✗ | return encodeFull_(cid); | |
| 298 | } | ||
| 299 | |||
| 300 | ✗ | const auto classification = classifyContigChange(data_prev->full_bytes_, full_bytes_); | |
| 301 | ✗ | if (classification.kind == ContigDeltaKind::FULL) | |
| 302 | { | ||
| 303 | ✗ | return encodeFull_(cid); | |
| 304 | } | ||
| 305 | |||
| 306 | ✗ | return encodeDelta_(cid, classification); | |
| 307 | ✗ | } | |
| 308 | |||
| 309 | private: | ||
| 310 | /// Walks backward to the nearest prior data checkpoint. | ||
| 311 | ✗ | const ContigContainerCheckpoint* prevDataCheckpoint_() const | |
| 312 | { | ||
| 313 | ✗ | const auto* checkpoint = prev(); | |
| 314 | ✗ | while (checkpoint && !checkpoint->isDataCheckpoint()) | |
| 315 | { | ||
| 316 | ✗ | checkpoint = checkpoint->prev(); | |
| 317 | } | ||
| 318 | ✗ | return checkpoint; | |
| 319 | } | ||
| 320 | |||
| 321 | /// Maps a contig delta classification to its wire Action byte. | ||
| 322 | ✗ | static Action contigActionFromKind_(ContigDeltaKind kind) | |
| 323 | { | ||
| 324 | ✗ | switch (kind) | |
| 325 | { | ||
| 326 | ✗ | case ContigDeltaKind::CARRY: | |
| 327 | ✗ | return Action::CARRY; | |
| 328 | ✗ | case ContigDeltaKind::SWAP: | |
| 329 | ✗ | return Action::CONTAINER_SWAP; | |
| 330 | ✗ | case ContigDeltaKind::MULTI_SWAP: | |
| 331 | ✗ | return Action::CONTAINER_MULTI_SWAP; | |
| 332 | ✗ | case ContigDeltaKind::ARRIVE: | |
| 333 | ✗ | return Action::CONTIG_ARRIVE; | |
| 334 | ✗ | case ContigDeltaKind::DEPART: | |
| 335 | ✗ | return Action::CONTIG_DEPART; | |
| 336 | ✗ | case ContigDeltaKind::BOOKENDS: | |
| 337 | ✗ | return Action::CONTIG_BOOKENDS; | |
| 338 | ✗ | case ContigDeltaKind::MIMO: | |
| 339 | ✗ | return Action::CONTIG_MIMO; | |
| 340 | ✗ | case ContigDeltaKind::FULL: | |
| 341 | ✗ | break; | |
| 342 | } | ||
| 343 | ✗ | throw DBException("Invalid contig delta kind"); | |
| 344 | } | ||
| 345 | |||
| 346 | /// Appends the FULL tail for the current contig snapshot to \p buf. | ||
| 347 | 12015 | void appendFullTail_(StreamBuffer& buf) const { appendContigBins_(buf, full_bytes_); } | |
| 348 | |||
| 349 | /// Appends contig bins as [count][bin bytes]... to \p buf. | ||
| 350 | 29979 | static void appendContigBins_(StreamBuffer& buf, const std::vector<std::vector<char>>& bins) | |
| 351 | { | ||
| 352 | 29979 | const auto size = countContigElements(bins); | |
| 353 | 29979 | buf.append(size); | |
| 354 | 172368 | for (uint16_t i = 0; i < size; ++i) | |
| 355 | { | ||
| 356 | 142389 | buf.append(bins[i]); | |
| 357 | } | ||
| 358 | 29979 | } | |
| 359 | |||
| 360 | /// Encodes a CLOSED lifecycle wire record. | ||
| 361 | 1713 | std::unique_ptr<CollectedData> encodeLifecycleEvt_(uint16_t cid) | |
| 362 | { | ||
| 363 | 1713 | simdb_assert(lifecycle_change_.isValid() && lifecycle_change_.getValue()); | |
| 364 | (void)cid; | ||
| 365 | 1713 | auto encoded = std::make_unique<CollectedData>(cid); | |
| 366 | 1713 | encoded->getBuffer().append(Action::CLOSED); | |
| 367 | 1713 | return encoded; | |
| 368 | ✗ | } | |
| 369 | |||
| 370 | /// Encodes a FULL snapshot of the current contig container. | ||
| 371 | 12015 | std::unique_ptr<CollectedData> encodeFull_(uint16_t cid) | |
| 372 | { | ||
| 373 | 12015 | auto encoded = std::make_unique<CollectedData>(cid); | |
| 374 | 12015 | auto& buf = encoded->getBuffer(); | |
| 375 | 12015 | buf.append(Action::FULL); | |
| 376 | 12015 | appendFullTail_(buf); | |
| 377 | 12015 | return encoded; | |
| 378 | ✗ | } | |
| 379 | |||
| 380 | /// Encodes a classified contig delta to the wire buffer. | ||
| 381 | ✗ | std::unique_ptr<CollectedData> encodeDelta_(uint16_t cid, const ContigDeltaClassification& classification) | |
| 382 | { | ||
| 383 | ✗ | auto encoded = std::make_unique<CollectedData>(cid); | |
| 384 | ✗ | auto& buf = encoded->getBuffer(); | |
| 385 | ✗ | const auto action = contigActionFromKind_(classification.kind); | |
| 386 | ✗ | buf.append(action); | |
| 387 | |||
| 388 | ✗ | switch (action) | |
| 389 | { | ||
| 390 | ✗ | case Action::CONTAINER_SWAP: | |
| 391 | ✗ | simdb_assert(classification.swap_index.isValid()); | |
| 392 | ✗ | simdb_assert(!classification.payload.empty()); | |
| 393 | ✗ | buf.append(classification.swap_index.getValue()); | |
| 394 | ✗ | buf.append(classification.payload); | |
| 395 | ✗ | break; | |
| 396 | ✗ | case Action::CONTAINER_MULTI_SWAP: | |
| 397 | ✗ | simdb_assert(classification.swap_indices.size() >= 2); | |
| 398 | ✗ | simdb_assert(classification.swap_indices.size() == classification.swap_payloads.size()); | |
| 399 | ✗ | buf.append(static_cast<uint8_t>(classification.swap_indices.size())); | |
| 400 | ✗ | for (size_t i = 0; i < classification.swap_indices.size(); ++i) | |
| 401 | { | ||
| 402 | ✗ | buf.append(classification.swap_indices[i]); | |
| 403 | ✗ | buf.append(classification.swap_payloads[i]); | |
| 404 | } | ||
| 405 | ✗ | break; | |
| 406 | ✗ | case Action::CARRY: | |
| 407 | ✗ | break; | |
| 408 | ✗ | case Action::CONTIG_ARRIVE: | |
| 409 | case Action::CONTIG_BOOKENDS: | ||
| 410 | ✗ | simdb_assert(!classification.payload.empty()); | |
| 411 | ✗ | buf.append(classification.payload); | |
| 412 | ✗ | break; | |
| 413 | ✗ | case Action::CONTIG_MIMO: | |
| 414 | ✗ | buf.append(classification.depart_count); | |
| 415 | ✗ | buf.append(classification.arrive_count); | |
| 416 | ✗ | for (const auto& arrive_payload : classification.arrive_payloads) | |
| 417 | { | ||
| 418 | ✗ | buf.append(arrive_payload); | |
| 419 | } | ||
| 420 | ✗ | break; | |
| 421 | ✗ | case Action::CONTIG_DEPART: | |
| 422 | ✗ | break; | |
| 423 | ✗ | default: | |
| 424 | ✗ | throw DBException("Unexpected contig delta action"); | |
| 425 | } | ||
| 426 | |||
| 427 | ✗ | return encoded; | |
| 428 | ✗ | } | |
| 429 | |||
| 430 | std::vector<std::vector<char>> full_bytes_; | ||
| 431 | ValidValue<bool> lifecycle_change_; | ||
| 432 | }; | ||
| 433 | |||
| 434 | class SparseContainerCheckpoint : public CollectableCheckpoint | ||
| 435 | { | ||
| 436 | public: | ||
| 437 | /// Creates a data checkpoint holding a full sparse container snapshot. | ||
| 438 | 12438 | SparseContainerCheckpoint(std::shared_ptr<SparseContainerCheckpoint> prev, uint64_t window_id, | |
| 439 | 12438 | const std::map<uint16_t, std::vector<char>>& full_bytes) : | |
| 440 | CollectableCheckpoint(prev, window_id), | ||
| 441 | 12438 | full_bytes_(full_bytes) | |
| 442 | { | ||
| 443 | 12438 | } | |
| 444 | |||
| 445 | /// Creates a lifecycle checkpoint recording a close transition. | ||
| 446 | 1608 | SparseContainerCheckpoint(std::shared_ptr<SparseContainerCheckpoint> prev, uint64_t window_id, | |
| 447 | 1608 | bool switched_to_closed) : | |
| 448 | CollectableCheckpoint(prev, window_id), | ||
| 449 | 1608 | lifecycle_change_(switched_to_closed) | |
| 450 | { | ||
| 451 | 1608 | } | |
| 452 | |||
| 453 | /// Returns the previous sparse checkpoint in the chain. | ||
| 454 | 12870 | SparseContainerCheckpoint* prev() { return static_cast<SparseContainerCheckpoint*>(getPrev()); } | |
| 455 | |||
| 456 | /// Returns the previous sparse checkpoint in the chain. | ||
| 457 | ✗ | const SparseContainerCheckpoint* prev() const { return static_cast<const SparseContainerCheckpoint*>(getPrev()); } | |
| 458 | |||
| 459 | /// Returns the next sparse checkpoint in the chain. | ||
| 460 | 43977 | SparseContainerCheckpoint* next() { return static_cast<SparseContainerCheckpoint*>(getNext()); } | |
| 461 | |||
| 462 | /// Returns true when this node holds collected bytes rather than a lifecycle flag. | ||
| 463 | 47916 | bool isDataCheckpoint() const { return !lifecycle_change_.isValid(); } | |
| 464 | |||
| 465 | /// Returns true when this node records a transition to closed. | ||
| 466 | 15915 | bool isClosedEvent() const { return lifecycle_change_.isValid() && lifecycle_change_.getValue(); } | |
| 467 | |||
| 468 | /// Returns true for any lifecycle (close) checkpoint. | ||
| 469 | bool isLifecycleEvent() const { return lifecycle_change_.isValid(); } | ||
| 470 | |||
| 471 | /// Encodes an unconditional FULL snapshot for heartbeat or absent-CID refresh. | ||
| 472 | 17523 | std::unique_ptr<CollectedData> encodeSnapshotForPipeline(uint16_t cid) const | |
| 473 | { | ||
| 474 | 17523 | simdb_assert(isDataCheckpoint()); | |
| 475 | 17523 | auto encoded = std::make_unique<CollectedData>(cid); | |
| 476 | 17523 | auto& buf = encoded->getBuffer(); | |
| 477 | 17523 | buf.append(Action::FULL); | |
| 478 | 17523 | appendFullTail_(buf); | |
| 479 | 17523 | return encoded; | |
| 480 | ✗ | } | |
| 481 | |||
| 482 | /// Encodes CLOSED, FULL, or a sparse delta by diffing against the prior data checkpoint. | ||
| 483 | 14046 | std::unique_ptr<CollectedData> encodeForPipeline(uint16_t cid, bool force_snapshot) | |
| 484 | { | ||
| 485 | 14046 | if (lifecycle_change_.isValid()) | |
| 486 | { | ||
| 487 | 1608 | return encodeLifecycleEvt_(cid); | |
| 488 | } | ||
| 489 | |||
| 490 | 12438 | if (force_snapshot) | |
| 491 | { | ||
| 492 | 12438 | return encodeFull_(cid); | |
| 493 | } | ||
| 494 | |||
| 495 | ✗ | const auto* data_prev = prevDataCheckpoint_(); | |
| 496 | ✗ | if (!data_prev) | |
| 497 | { | ||
| 498 | ✗ | return encodeFull_(cid); | |
| 499 | } | ||
| 500 | |||
| 501 | ✗ | if (closedSincePrevDataCheckpoint_(this, data_prev)) | |
| 502 | { | ||
| 503 | ✗ | return encodeFull_(cid); | |
| 504 | } | ||
| 505 | |||
| 506 | ✗ | const auto classification = classifySparseChange(data_prev->full_bytes_, full_bytes_); | |
| 507 | ✗ | if (classification.kind == SparseDeltaKind::FULL) | |
| 508 | { | ||
| 509 | ✗ | return encodeFull_(cid); | |
| 510 | } | ||
| 511 | |||
| 512 | ✗ | return encodeDelta_(cid, classification); | |
| 513 | ✗ | } | |
| 514 | |||
| 515 | private: | ||
| 516 | /// Walks backward to the nearest prior data checkpoint. | ||
| 517 | ✗ | const SparseContainerCheckpoint* prevDataCheckpoint_() const | |
| 518 | { | ||
| 519 | ✗ | const auto* checkpoint = prev(); | |
| 520 | ✗ | while (checkpoint && !checkpoint->isDataCheckpoint()) | |
| 521 | { | ||
| 522 | ✗ | checkpoint = checkpoint->prev(); | |
| 523 | } | ||
| 524 | ✗ | return checkpoint; | |
| 525 | } | ||
| 526 | |||
| 527 | /// Maps a sparse delta classification to its wire Action byte. | ||
| 528 | ✗ | static Action sparseActionFromKind_(SparseDeltaKind kind) | |
| 529 | { | ||
| 530 | ✗ | switch (kind) | |
| 531 | { | ||
| 532 | ✗ | case SparseDeltaKind::CARRY: | |
| 533 | ✗ | return Action::CARRY; | |
| 534 | ✗ | case SparseDeltaKind::SWAP: | |
| 535 | ✗ | return Action::CONTAINER_SWAP; | |
| 536 | ✗ | case SparseDeltaKind::MULTI_SWAP: | |
| 537 | ✗ | return Action::CONTAINER_MULTI_SWAP; | |
| 538 | ✗ | case SparseDeltaKind::REMOVE: | |
| 539 | ✗ | return Action::SPARSE_REMOVE; | |
| 540 | ✗ | case SparseDeltaKind::ADD: | |
| 541 | ✗ | return Action::SPARSE_ADD; | |
| 542 | ✗ | case SparseDeltaKind::MULTI_REMOVE: | |
| 543 | ✗ | return Action::SPARSE_MULTI_REMOVE; | |
| 544 | ✗ | case SparseDeltaKind::FULL: | |
| 545 | ✗ | break; | |
| 546 | } | ||
| 547 | ✗ | throw DBException("Invalid sparse delta kind"); | |
| 548 | } | ||
| 549 | |||
| 550 | /// Appends the FULL tail for the current sparse snapshot to \p buf. | ||
| 551 | 29961 | void appendFullTail_(StreamBuffer& buf) const { appendSparseBins_(buf, full_bytes_); } | |
| 552 | |||
| 553 | /// Appends sparse bins as [count][idx][bin bytes]... to \p buf. | ||
| 554 | 29961 | static void appendSparseBins_(StreamBuffer& buf, const std::map<uint16_t, std::vector<char>>& bins) | |
| 555 | { | ||
| 556 | 29961 | const auto size = countSparseElements(bins); | |
| 557 | 29961 | buf.append(size); | |
| 558 | 165831 | for (const auto& [bin_idx, bin_bytes] : bins) | |
| 559 | { | ||
| 560 | 135870 | if (!bin_bytes.empty()) | |
| 561 | { | ||
| 562 | 135870 | buf.append(bin_idx); | |
| 563 | 135870 | buf.append(bin_bytes); | |
| 564 | } | ||
| 565 | } | ||
| 566 | 29961 | } | |
| 567 | |||
| 568 | /// Encodes a CLOSED lifecycle wire record. | ||
| 569 | 1608 | std::unique_ptr<CollectedData> encodeLifecycleEvt_(uint16_t cid) | |
| 570 | { | ||
| 571 | 1608 | simdb_assert(lifecycle_change_.isValid() && lifecycle_change_.getValue()); | |
| 572 | (void)cid; | ||
| 573 | 1608 | auto encoded = std::make_unique<CollectedData>(cid); | |
| 574 | 1608 | encoded->getBuffer().append(Action::CLOSED); | |
| 575 | 1608 | return encoded; | |
| 576 | ✗ | } | |
| 577 | |||
| 578 | /// Encodes a FULL snapshot of the current sparse container. | ||
| 579 | 12438 | std::unique_ptr<CollectedData> encodeFull_(uint16_t cid) | |
| 580 | { | ||
| 581 | 12438 | auto encoded = std::make_unique<CollectedData>(cid); | |
| 582 | 12438 | auto& buf = encoded->getBuffer(); | |
| 583 | 12438 | buf.append(Action::FULL); | |
| 584 | 12438 | appendFullTail_(buf); | |
| 585 | 12438 | return encoded; | |
| 586 | ✗ | } | |
| 587 | |||
| 588 | /// Encodes a classified sparse delta to the wire buffer. | ||
| 589 | ✗ | std::unique_ptr<CollectedData> encodeDelta_(uint16_t cid, const SparseDeltaClassification& classification) | |
| 590 | { | ||
| 591 | ✗ | auto encoded = std::make_unique<CollectedData>(cid); | |
| 592 | ✗ | auto& buf = encoded->getBuffer(); | |
| 593 | ✗ | const auto action = sparseActionFromKind_(classification.kind); | |
| 594 | ✗ | buf.append(action); | |
| 595 | |||
| 596 | ✗ | switch (action) | |
| 597 | { | ||
| 598 | ✗ | case Action::CONTAINER_SWAP: | |
| 599 | ✗ | simdb_assert(classification.bin_index.isValid()); | |
| 600 | ✗ | simdb_assert(!classification.payload.empty()); | |
| 601 | ✗ | buf.append(classification.bin_index.getValue()); | |
| 602 | ✗ | buf.append(classification.payload); | |
| 603 | ✗ | break; | |
| 604 | ✗ | case Action::CONTAINER_MULTI_SWAP: | |
| 605 | ✗ | simdb_assert(classification.bin_indices.size() >= 2); | |
| 606 | ✗ | simdb_assert(classification.bin_indices.size() == classification.payloads.size()); | |
| 607 | ✗ | buf.append(static_cast<uint8_t>(classification.bin_indices.size())); | |
| 608 | ✗ | for (size_t i = 0; i < classification.bin_indices.size(); ++i) | |
| 609 | { | ||
| 610 | ✗ | buf.append(classification.bin_indices[i]); | |
| 611 | ✗ | buf.append(classification.payloads[i]); | |
| 612 | } | ||
| 613 | ✗ | break; | |
| 614 | ✗ | case Action::SPARSE_REMOVE: | |
| 615 | ✗ | simdb_assert(classification.bin_index.isValid()); | |
| 616 | ✗ | buf.append(classification.bin_index.getValue()); | |
| 617 | ✗ | break; | |
| 618 | ✗ | case Action::SPARSE_ADD: | |
| 619 | ✗ | simdb_assert(classification.bin_index.isValid()); | |
| 620 | ✗ | simdb_assert(!classification.payload.empty()); | |
| 621 | ✗ | buf.append(classification.bin_index.getValue()); | |
| 622 | ✗ | buf.append(classification.payload); | |
| 623 | ✗ | break; | |
| 624 | ✗ | case Action::SPARSE_MULTI_REMOVE: | |
| 625 | ✗ | simdb_assert(classification.bin_indices.size() >= 2); | |
| 626 | ✗ | buf.append(static_cast<uint8_t>(classification.bin_indices.size())); | |
| 627 | ✗ | for (const auto idx : classification.bin_indices) | |
| 628 | { | ||
| 629 | ✗ | buf.append(idx); | |
| 630 | } | ||
| 631 | ✗ | break; | |
| 632 | ✗ | case Action::CARRY: | |
| 633 | ✗ | break; | |
| 634 | ✗ | default: | |
| 635 | ✗ | throw DBException("Unexpected sparse delta action"); | |
| 636 | } | ||
| 637 | |||
| 638 | ✗ | return encoded; | |
| 639 | ✗ | } | |
| 640 | |||
| 641 | std::map<uint16_t, std::vector<char>> full_bytes_; | ||
| 642 | ValidValue<bool> lifecycle_change_; | ||
| 643 | }; | ||
| 644 | |||
| 645 | } // namespace simdb::argos | ||
| 646 |