| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "pdc.h" |
| 4 | #include "dbengine-compression.h" |
| 5 | |
| 6 | struct extent_page_details_list { |
| 7 | uint32_t extent_block; |
| 8 | uint32_t extent_size; |
| 9 | Pvoid_t page_details_by_metric_id_JudyL; |
| 10 | struct page_details_control *pdc; |
| 11 | struct rrdengine_datafile *datafile; |
| 12 | |
| 13 | struct rrdeng_cmd *cmd; |
| 14 | bool head_to_datafile_extent_queries_pending_for_extent; |
| 15 | |
| 16 | struct { |
| 17 | struct extent_page_details_list *prev; |
| 18 | struct extent_page_details_list *next; |
| 19 | } query; |
| 20 | }; |
| 21 | |
| 22 | typedef struct datafile_extent_offset_list { |
| 23 | uv_file file; |
| 24 | unsigned fileno; |
| 25 | Pvoid_t extent_pd_list_by_extent_offset_JudyL; |
| 26 | } DEOL; |
| 27 | |
| 28 | // ---------------------------------------------------------------------------- |
| 29 | // PDC cache |
| 30 | |
| 31 | static struct { |
| 32 | struct { |
| 33 | ARAL *ar; |
| 34 | } pdc; |
| 35 | |
| 36 | struct { |
| 37 | ARAL *ar; |
| 38 | } pd; |
| 39 | |
| 40 | struct { |
| 41 | ARAL *ar; |
| 42 | } epdl; |
| 43 | |
| 44 | struct { |
| 45 | ARAL *ar; |
| 46 | } deol; |
| 47 | |
| 48 | struct { |
| 49 | ARAL *ar; |
| 50 | } epdl_extent; |
| 51 | } pdc_globals = {}; |
| 52 | |
| 53 | void pdc_init(void) { |
| 54 | pdc_globals.pdc.ar = aral_create( |
| 55 | "dbengine-pdc", |
| 56 | sizeof(PDC), |
| 57 | 0, |
| 58 | 0, |
| 59 | NULL, |
| 60 | NULL, NULL, false, false, true |
| 61 | ); |
| 62 | |
| 63 | pulse_aral_register(pdc_globals.pdc.ar, "pdc"); |
| 64 | } |
| 65 | |
| 66 | ALWAYS_INLINE PDC *pdc_get(void) { |
| 67 | PDC *pdc = aral_mallocz(pdc_globals.pdc.ar); |
| 68 | memset(pdc, 0, sizeof(PDC)); |
| 69 | return pdc; |
| 70 | } |
| 71 | |
| 72 | static ALWAYS_INLINE void pdc_release(PDC *pdc) { |
| 73 | aral_freez(pdc_globals.pdc.ar, pdc); |
| 74 | } |
| 75 | |
| 76 | struct aral_statistics *pdc_aral_stats(void) { |
| 77 | return aral_get_statistics(pdc_globals.pdc.ar); |
| 78 | } |
| 79 | |
| 80 | // ---------------------------------------------------------------------------- |
| 81 | // PD cache |
| 82 | |
| 83 | void page_details_init(void) { |
| 84 | pdc_globals.pd.ar = aral_create( |
| 85 | "dbengine-pd", |
| 86 | sizeof(struct page_details), |
| 87 | 0, |
| 88 | 0, |
| 89 | NULL, |
| 90 | NULL, NULL, false, false, true |
| 91 | ); |
| 92 | pulse_aral_register(pdc_globals.pd.ar, "pd"); |
| 93 | } |
| 94 | |
| 95 | ALWAYS_INLINE struct page_details *page_details_get(void) { |
| 96 | struct page_details *pd = aral_mallocz(pdc_globals.pd.ar); |
| 97 | memset(pd, 0, sizeof(struct page_details)); |
| 98 | return pd; |
| 99 | } |
| 100 | |
| 101 | static ALWAYS_INLINE void page_details_release(struct page_details *pd) { |
| 102 | aral_freez(pdc_globals.pd.ar, pd); |
| 103 | } |
| 104 | |
| 105 | struct aral_statistics *pd_aral_stats(void) { |
| 106 | return aral_get_statistics(pdc_globals.pd.ar); |
| 107 | } |
| 108 | |
| 109 | // ---------------------------------------------------------------------------- |
| 110 | // epdl cache |
| 111 | |
| 112 | void epdl_init(void) { |
| 113 | pdc_globals.epdl.ar = aral_create( |
| 114 | "dbengine-epdl", |
| 115 | sizeof(EPDL), |
| 116 | 0, |
| 117 | 0, |
| 118 | NULL, |
| 119 | NULL, NULL, false, false, true |
| 120 | ); |
| 121 | pulse_aral_register(pdc_globals.epdl.ar, "epdl"); |
| 122 | } |
| 123 | |
| 124 | static ALWAYS_INLINE EPDL *epdl_get(void) { |
| 125 | EPDL *epdl = aral_mallocz(pdc_globals.epdl.ar); |
| 126 | memset(epdl, 0, sizeof(EPDL)); |
| 127 | return epdl; |
| 128 | } |
| 129 | |
| 130 | static ALWAYS_INLINE void epdl_release(EPDL *epdl) { |
| 131 | aral_freez(pdc_globals.epdl.ar, epdl); |
| 132 | } |
| 133 | |
| 134 | struct aral_statistics *epdl_aral_stats(void) { |
| 135 | return aral_get_statistics(pdc_globals.epdl.ar); |
| 136 | } |
| 137 | |
| 138 | // ---------------------------------------------------------------------------- |
| 139 | // deol cache |
| 140 | |
| 141 | void deol_init(void) { |
| 142 | pdc_globals.deol.ar = aral_create( |
| 143 | "dbengine-deol", |
| 144 | sizeof(DEOL), |
| 145 | 0, |
| 146 | 0, |
| 147 | NULL, |
| 148 | NULL, NULL, false, false, true |
| 149 | ); |
| 150 | |
| 151 | pulse_aral_register(pdc_globals.deol.ar, "deol"); |
| 152 | } |
| 153 | |
| 154 | static ALWAYS_INLINE DEOL *deol_get(void) { |
| 155 | DEOL *deol = aral_mallocz(pdc_globals.deol.ar); |
| 156 | memset(deol, 0, sizeof(DEOL)); |
| 157 | return deol; |
| 158 | } |
| 159 | |
| 160 | static ALWAYS_INLINE void deol_release(DEOL *deol) { |
| 161 | aral_freez(pdc_globals.deol.ar, deol); |
| 162 | } |
| 163 | |
| 164 | struct aral_statistics *deol_aral_stats(void) { |
| 165 | return aral_get_statistics(pdc_globals.deol.ar); |
| 166 | } |
| 167 | |
| 168 | // ---------------------------------------------------------------------------- |
| 169 | // epdl_extent cache |
| 170 | |
| 171 | void epdl_extent_init(void) { |
| 172 | pdc_globals.epdl_extent.ar = aral_create( |
| 173 | "dbengine-epdl-extent", |
| 174 | sizeof(EPDL_EXTENT), |
| 175 | 0, |
| 176 | 0, |
| 177 | NULL, |
| 178 | NULL, NULL, false, false, true |
| 179 | ); |
| 180 | |
| 181 | pulse_aral_register(pdc_globals.epdl_extent.ar, "epdl_extent"); |
| 182 | } |
| 183 | |
| 184 | static ALWAYS_INLINE EPDL_EXTENT *epdl_extent_get(void) { |
| 185 | EPDL_EXTENT *e = aral_mallocz(pdc_globals.epdl_extent.ar); |
| 186 | memset(e, 0, sizeof(EPDL_EXTENT)); |
| 187 | return e; |
| 188 | } |
| 189 | |
| 190 | ALWAYS_INLINE void epdl_extent_release(EPDL_EXTENT *e) { |
| 191 | aral_freez(pdc_globals.epdl_extent.ar, e); |
| 192 | } |
| 193 | |
| 194 | struct aral_statistics *epdl_extent_aral_stats(void) { |
| 195 | return aral_get_statistics(pdc_globals.epdl_extent.ar); |
| 196 | } |
| 197 | |
| 198 | // ---------------------------------------------------------------------------- |
| 199 | // extent with buffer cache |
| 200 | |
| 201 | static struct { |
| 202 | struct { |
| 203 | SPINLOCK spinlock; |
| 204 | struct extent_buffer *available_items; |
| 205 | size_t available; |
| 206 | } protected; |
| 207 | |
| 208 | struct { |
| 209 | size_t allocated; |
| 210 | size_t allocated_bytes; |
| 211 | } atomics; |
| 212 | |
| 213 | size_t max_size; |
| 214 | |
| 215 | } extent_buffer_globals = { |
| 216 | .protected = { |
| 217 | .spinlock = SPINLOCK_INITIALIZER, |
| 218 | .available_items = NULL, |
| 219 | .available = 0, |
| 220 | }, |
| 221 | .atomics = { |
| 222 | .allocated = 0, |
| 223 | .allocated_bytes = 0, |
| 224 | }, |
| 225 | .max_size = MAX_EXTENT_UNCOMPRESSED_SIZE |
| 226 | }; |
| 227 | |
| 228 | void extent_buffer_init(void) { |
| 229 | size_t max_extent_uncompressed = MAX_EXTENT_UNCOMPRESSED_SIZE; |
| 230 | size_t max_size = (size_t)LZ4_compressBound(MAX_EXTENT_UNCOMPRESSED_SIZE); |
| 231 | if(max_size < max_extent_uncompressed) |
| 232 | max_size = max_extent_uncompressed; |
| 233 | |
| 234 | extent_buffer_globals.max_size = max_size; |
| 235 | } |
| 236 | |
| 237 | void extent_buffer_cleanup1(void) { |
| 238 | struct extent_buffer *item = NULL; |
| 239 | |
| 240 | if(!spinlock_trylock(&extent_buffer_globals.protected.spinlock)) |
| 241 | return; |
| 242 | |
| 243 | if(extent_buffer_globals.protected.available_items && extent_buffer_globals.protected.available > 1) { |
| 244 | item = extent_buffer_globals.protected.available_items; |
| 245 | DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(extent_buffer_globals.protected.available_items, item, cache.prev, cache.next); |
| 246 | extent_buffer_globals.protected.available--; |
| 247 | } |
| 248 | |
| 249 | spinlock_unlock(&extent_buffer_globals.protected.spinlock); |
| 250 | |
| 251 | if(item) { |
| 252 | size_t bytes = sizeof(struct extent_buffer) + item->bytes; |
| 253 | freez(item); |
| 254 | __atomic_sub_fetch(&extent_buffer_globals.atomics.allocated, 1, __ATOMIC_RELAXED); |
| 255 | __atomic_sub_fetch(&extent_buffer_globals.atomics.allocated_bytes, bytes, __ATOMIC_RELAXED); |
| 256 | } |
| 257 | } |
| 258 | |
| 259 | ALWAYS_INLINE struct extent_buffer *extent_buffer_get(size_t size) { |
| 260 | internal_fatal(size > extent_buffer_globals.max_size, "DBENGINE: extent size is too big"); |
| 261 | |
| 262 | struct extent_buffer *eb = NULL; |
| 263 | |
| 264 | if(size < extent_buffer_globals.max_size) |
| 265 | size = extent_buffer_globals.max_size; |
| 266 | |
| 267 | spinlock_lock(&extent_buffer_globals.protected.spinlock); |
| 268 | if(likely(extent_buffer_globals.protected.available_items)) { |
| 269 | eb = extent_buffer_globals.protected.available_items; |
| 270 | DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(extent_buffer_globals.protected.available_items, eb, cache.prev, cache.next); |
| 271 | extent_buffer_globals.protected.available--; |
| 272 | } |
| 273 | spinlock_unlock(&extent_buffer_globals.protected.spinlock); |
| 274 | |
| 275 | if(unlikely(eb && eb->bytes < size)) { |
| 276 | size_t bytes = sizeof(struct extent_buffer) + eb->bytes; |
| 277 | freez(eb); |
| 278 | eb = NULL; |
| 279 | __atomic_sub_fetch(&extent_buffer_globals.atomics.allocated, 1, __ATOMIC_RELAXED); |
| 280 | __atomic_sub_fetch(&extent_buffer_globals.atomics.allocated_bytes, bytes, __ATOMIC_RELAXED); |
| 281 | } |
| 282 | |
| 283 | if(unlikely(!eb)) { |
| 284 | size_t bytes = sizeof(struct extent_buffer) + size; |
| 285 | eb = mallocz(bytes); |
| 286 | eb->bytes = size; |
| 287 | __atomic_add_fetch(&extent_buffer_globals.atomics.allocated, 1, __ATOMIC_RELAXED); |
| 288 | __atomic_add_fetch(&extent_buffer_globals.atomics.allocated_bytes, bytes, __ATOMIC_RELAXED); |
| 289 | } |
| 290 | |
| 291 | return eb; |
| 292 | } |
| 293 | |
| 294 | ALWAYS_INLINE void extent_buffer_release(struct extent_buffer *eb) { |
| 295 | if(unlikely(!eb)) return; |
| 296 | |
| 297 | spinlock_lock(&extent_buffer_globals.protected.spinlock); |
| 298 | DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(extent_buffer_globals.protected.available_items, eb, cache.prev, cache.next); |
| 299 | extent_buffer_globals.protected.available++; |
| 300 | spinlock_unlock(&extent_buffer_globals.protected.spinlock); |
| 301 | } |
| 302 | |
| 303 | size_t extent_buffer_cache_size(void) { |
| 304 | return __atomic_load_n(&extent_buffer_globals.atomics.allocated_bytes, __ATOMIC_RELAXED); |
| 305 | } |
| 306 | |
| 307 | // ---------------------------------------------------------------------------- |
| 308 | // epdl logic |
| 309 | |
| 310 | static ALWAYS_INLINE void epdl_destroy(EPDL *epdl) |
| 311 | { |
| 312 | Pvoid_t *pd_by_start_time_s_JudyL; |
| 313 | Word_t metric_id_index = 0; |
| 314 | bool metric_id_first = true; |
| 315 | while ((pd_by_start_time_s_JudyL = PDCJudyLFirstThenNext( |
| 316 | epdl->page_details_by_metric_id_JudyL, |
| 317 | &metric_id_index, &metric_id_first))) |
| 318 | PDCJudyLFreeArray(pd_by_start_time_s_JudyL, PJE0); |
| 319 | |
| 320 | PDCJudyLFreeArray(&epdl->page_details_by_metric_id_JudyL, PJE0); |
| 321 | epdl_release(epdl); |
| 322 | } |
| 323 | |
| 324 | static ALWAYS_INLINE void epdl_mark_all_not_loaded_pages_as_failed(EPDL *epdl, PDC_PAGE_STATUS tags, size_t *statistics_counter) |
| 325 | { |
| 326 | size_t pages_matched = 0; |
| 327 | |
| 328 | Word_t metric_id_index = 0; |
| 329 | bool metric_id_first = true; |
| 330 | Pvoid_t *pd_by_start_time_s_JudyL; |
| 331 | while((pd_by_start_time_s_JudyL = PDCJudyLFirstThenNext(epdl->page_details_by_metric_id_JudyL, &metric_id_index, &metric_id_first))) { |
| 332 | |
| 333 | Word_t start_time_index = 0; |
| 334 | bool start_time_first = true; |
| 335 | Pvoid_t *PValue; |
| 336 | while ((PValue = PDCJudyLFirstThenNext(*pd_by_start_time_s_JudyL, &start_time_index, &start_time_first))) { |
| 337 | struct page_details *pd = *PValue; |
| 338 | |
| 339 | if(!pd->page && !pdc_page_status_check(pd, PDC_PAGE_FAILED|PDC_PAGE_READY)) { |
| 340 | pdc_page_status_set(pd, PDC_PAGE_FAILED | tags); |
| 341 | pages_matched++; |
| 342 | } |
| 343 | } |
| 344 | } |
| 345 | |
| 346 | if(pages_matched && statistics_counter) |
| 347 | __atomic_add_fetch(statistics_counter, pages_matched, __ATOMIC_RELAXED); |
| 348 | } |
| 349 | /* |
| 350 | static bool epdl_check_if_pages_are_already_in_cache(struct rrdengine_instance *ctx, EPDL *epdl, PDC_PAGE_STATUS tags) |
| 351 | { |
| 352 | size_t count_remaining = 0; |
| 353 | size_t found = 0; |
| 354 | |
| 355 | Word_t metric_id_index = 0; |
| 356 | bool metric_id_first = true; |
| 357 | Pvoid_t *pd_by_start_time_s_JudyL; |
| 358 | while((pd_by_start_time_s_JudyL = PDCJudyLFirstThenNext(epdl->page_details_by_metric_id_JudyL, &metric_id_index, &metric_id_first))) { |
| 359 | |
| 360 | Word_t start_time_index = 0; |
| 361 | bool start_time_first = true; |
| 362 | Pvoid_t *PValue; |
| 363 | while ((PValue = PDCJudyLFirstThenNext(*pd_by_start_time_s_JudyL, &start_time_index, &start_time_first))) { |
| 364 | struct page_details *pd = *PValue; |
| 365 | if (pd->page) |
| 366 | continue; |
| 367 | |
| 368 | pd->page = pgc_page_get_and_acquire(main_cache, (Word_t) ctx, pd->metric_id, pd->first_time_s, PGC_SEARCH_EXACT); |
| 369 | if (pd->page) { |
| 370 | found++; |
| 371 | pdc_page_status_set(pd, PDC_PAGE_READY | tags); |
| 372 | } |
| 373 | else |
| 374 | count_remaining++; |
| 375 | } |
| 376 | } |
| 377 | |
| 378 | if(found) { |
| 379 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_ok_preloaded, found, __ATOMIC_RELAXED); |
| 380 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_data_source_main_cache, found, __ATOMIC_RELAXED); |
| 381 | } |
| 382 | |
| 383 | return count_remaining == 0; |
| 384 | } |
| 385 | */ |
| 386 | |
| 387 | // ---------------------------------------------------------------------------- |
| 388 | // PDC logic |
| 389 | |
| 390 | static ALWAYS_INLINE void pdc_destroy(PDC *pdc) { |
| 391 | if(pdc->metric) |
| 392 | mrg_metric_release(main_mrg, pdc->metric); |
| 393 | |
| 394 | completion_destroy(&pdc->prep_completion); |
| 395 | completion_destroy(&pdc->page_completion); |
| 396 | |
| 397 | Pvoid_t *PValue; |
| 398 | struct page_details *pd; |
| 399 | Word_t time_index = 0; |
| 400 | bool first_then_next = true; |
| 401 | size_t unroutable = 0, cancelled = 0; |
| 402 | while((PValue = PDCJudyLFirstThenNext(pdc->page_list_JudyL, &time_index, &first_then_next))) { |
| 403 | pd = *PValue; |
| 404 | |
| 405 | // no need for atomics here - we are done... |
| 406 | PDC_PAGE_STATUS status = pd->status; |
| 407 | |
| 408 | if(status & PDC_PAGE_DATAFILE_ACQUIRED) { |
| 409 | datafile_release(pd->datafile.ptr, DATAFILE_ACQUIRE_PAGE_DETAILS); |
| 410 | pd->datafile.ptr = NULL; |
| 411 | } |
| 412 | |
| 413 | internal_fatal(pd->datafile.ptr, "DBENGINE: page details has a datafile.ptr that is not released."); |
| 414 | |
| 415 | if(!pd->page && !(status & (PDC_PAGE_READY | PDC_PAGE_FAILED | PDC_PAGE_RELEASED | PDC_PAGE_SKIP | PDC_PAGE_INVALID | PDC_PAGE_CANCELLED))) { |
| 416 | // pdc_page_status_set(pd, PDC_PAGE_FAILED); |
| 417 | unroutable++; |
| 418 | } |
| 419 | else if(!pd->page && (status & PDC_PAGE_CANCELLED)) |
| 420 | cancelled++; |
| 421 | |
| 422 | if(pd->page && !(status & PDC_PAGE_RELEASED)) { |
| 423 | pgc_page_release(main_cache, pd->page); |
| 424 | // pdc_page_status_set(pd, PDC_PAGE_RELEASED); |
| 425 | } |
| 426 | |
| 427 | page_details_release(pd); |
| 428 | } |
| 429 | |
| 430 | PDCJudyLFreeArray(&pdc->page_list_JudyL, PJE0); |
| 431 | |
| 432 | __atomic_sub_fetch(&rrdeng_cache_efficiency_stats.currently_running_queries, 1, __ATOMIC_RELAXED); |
| 433 | __atomic_sub_fetch(&pdc->ctx->atomic.inflight_queries, 1, __ATOMIC_RELAXED); |
| 434 | pdc_release(pdc); |
| 435 | |
| 436 | if(unroutable) |
| 437 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_fail_unroutable, unroutable, __ATOMIC_RELAXED); |
| 438 | |
| 439 | if(cancelled) |
| 440 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_fail_cancelled, cancelled, __ATOMIC_RELAXED); |
| 441 | } |
| 442 | |
| 443 | ALWAYS_INLINE void pdc_acquire(PDC *pdc) { |
| 444 | spinlock_lock(&pdc->refcount_spinlock); |
| 445 | |
| 446 | if(pdc->refcount < 1) |
| 447 | fatal("DBENGINE: pdc is not referenced and cannot be acquired"); |
| 448 | |
| 449 | pdc->refcount++; |
| 450 | spinlock_unlock(&pdc->refcount_spinlock); |
| 451 | } |
| 452 | |
| 453 | ALWAYS_INLINE bool pdc_release_and_destroy_if_unreferenced(PDC *pdc, bool worker, bool router __maybe_unused) { |
| 454 | if(unlikely(!pdc)) |
| 455 | return true; |
| 456 | |
| 457 | spinlock_lock(&pdc->refcount_spinlock); |
| 458 | |
| 459 | if(pdc->refcount <= 0) |
| 460 | fatal("DBENGINE: pdc is not referenced and cannot be released"); |
| 461 | |
| 462 | pdc->refcount--; |
| 463 | |
| 464 | if (pdc->refcount <= 1 && worker) { |
| 465 | // when 1 refcount is remaining, and we are a worker, |
| 466 | // we can mark the job completed: |
| 467 | // - if the remaining refcount is from the query caller, we will wake it up |
| 468 | // - if the remaining refcount is from another worker, the query thread is already away |
| 469 | completion_mark_complete(&pdc->page_completion); |
| 470 | } |
| 471 | |
| 472 | if (pdc->refcount == 0) { |
| 473 | spinlock_unlock(&pdc->refcount_spinlock); |
| 474 | pdc_destroy(pdc); |
| 475 | return true; |
| 476 | } |
| 477 | |
| 478 | spinlock_unlock(&pdc->refcount_spinlock); |
| 479 | return false; |
| 480 | } |
| 481 | |
| 482 | ALWAYS_INLINE void epdl_cmd_queued(void *epdl_ptr, struct rrdeng_cmd *cmd) { |
| 483 | EPDL *epdl = epdl_ptr; |
| 484 | epdl->cmd = cmd; |
| 485 | } |
| 486 | |
| 487 | ALWAYS_INLINE void epdl_cmd_dequeued(void *epdl_ptr) { |
| 488 | EPDL *epdl = epdl_ptr; |
| 489 | epdl->cmd = NULL; |
| 490 | } |
| 491 | |
| 492 | static ALWAYS_INLINE struct rrdeng_cmd *epdl_get_cmd(void *epdl_ptr) { |
| 493 | EPDL *epdl = epdl_ptr; |
| 494 | return epdl->cmd; |
| 495 | } |
| 496 | |
| 497 | static ALWAYS_INLINE EPDL_EXTENT *epdl_find_extent_base(EPDL *epdl) { |
| 498 | EPDL_EXTENT *e = NULL; |
| 499 | rw_spinlock_read_lock(&epdl->datafile->extent_epdl.spinlock); |
| 500 | Pvoid_t *PValue = JudyLGet(epdl->datafile->extent_epdl.epdl_per_extent, epdl->extent_block, PJE0); |
| 501 | internal_fatal(PValue == PJERR, "DBENGINE: corrupted pending extent judy"); |
| 502 | if(PValue) |
| 503 | e = *PValue; |
| 504 | rw_spinlock_read_unlock(&epdl->datafile->extent_epdl.spinlock); |
| 505 | |
| 506 | if(!e) { |
| 507 | EPDL_EXTENT *e_to_free = NULL; |
| 508 | e = epdl_extent_get(); |
| 509 | |
| 510 | rw_spinlock_write_lock(&epdl->datafile->extent_epdl.spinlock); |
| 511 | PValue = JudyLIns(&epdl->datafile->extent_epdl.epdl_per_extent, epdl->extent_block, PJE0); |
| 512 | internal_fatal(!PValue || PValue == PJERR, "DBENGINE: corrupted pending extent judy"); |
| 513 | if(!*PValue) { |
| 514 | *PValue = e; |
| 515 | spinlock_init(&e->spinlock); |
| 516 | } |
| 517 | else { |
| 518 | e_to_free = e; |
| 519 | e = *PValue; |
| 520 | } |
| 521 | rw_spinlock_write_unlock(&epdl->datafile->extent_epdl.spinlock); |
| 522 | |
| 523 | epdl_extent_release(e_to_free); |
| 524 | } |
| 525 | |
| 526 | return e; |
| 527 | } |
| 528 | |
| 529 | static ALWAYS_INLINE bool epdl_pending_add(EPDL *epdl) { |
| 530 | EPDL_EXTENT *e = epdl_find_extent_base(epdl); |
| 531 | spinlock_lock(&e->spinlock); |
| 532 | |
| 533 | bool added_new; |
| 534 | if(unlikely(e->base)) { |
| 535 | added_new = false; |
| 536 | epdl->head_to_datafile_extent_queries_pending_for_extent = false; |
| 537 | |
| 538 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_extent_merged, 1, __ATOMIC_RELAXED); |
| 539 | |
| 540 | // if(e->base->pdc->priority > epdl->pdc->priority) { |
| 541 | // e->base->pdc->priority = epdl->pdc->priority; |
| 542 | // rrdeng_req_cmd(epdl_get_cmd, e->base, epdl->pdc->priority); |
| 543 | // } |
| 544 | } |
| 545 | else { |
| 546 | added_new = true; |
| 547 | epdl->head_to_datafile_extent_queries_pending_for_extent = true; |
| 548 | } |
| 549 | |
| 550 | DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(e->base, epdl, query.prev, query.next); |
| 551 | spinlock_unlock(&e->spinlock); |
| 552 | |
| 553 | return added_new; |
| 554 | } |
| 555 | |
| 556 | static ALWAYS_INLINE void epdl_pending_del(EPDL *epdl) { |
| 557 | EPDL_EXTENT *e = epdl_find_extent_base(epdl); |
| 558 | spinlock_lock(&e->spinlock); |
| 559 | e->base = NULL; |
| 560 | spinlock_unlock(&e->spinlock); |
| 561 | } |
| 562 | |
| 563 | ALWAYS_INLINE_HOT void pdc_to_epdl_router(struct rrdengine_instance *ctx, PDC *pdc, execute_extent_page_details_list_t exec_first_extent_list, execute_extent_page_details_list_t exec_rest_extent_list) |
| 564 | { |
| 565 | Pvoid_t *PValue; |
| 566 | Pvoid_t *PValue1; |
| 567 | Pvoid_t *PValue2; |
| 568 | Word_t time_index = 0; |
| 569 | struct page_details *pd = NULL; |
| 570 | |
| 571 | // this is the entire page list |
| 572 | // Lets do some deduplication |
| 573 | // 1. Per datafile |
| 574 | // 2. Per extent |
| 575 | // 3. Pages per extent will be added to the cache either as acquired or not |
| 576 | |
| 577 | Pvoid_t JudyL_datafile_list = NULL; |
| 578 | |
| 579 | DEOL *deol; |
| 580 | EPDL *epdl; |
| 581 | |
| 582 | if (pdc->page_list_JudyL) { |
| 583 | bool first_then_next = true; |
| 584 | while((PValue = PDCJudyLFirstThenNext(pdc->page_list_JudyL, &time_index, &first_then_next))) { |
| 585 | pd = *PValue; |
| 586 | |
| 587 | internal_fatal(!pd, |
| 588 | "DBENGINE: pdc page list has an empty page details entry"); |
| 589 | |
| 590 | if (!(pd->status & PDC_PAGE_DISK_PENDING)) |
| 591 | continue; |
| 592 | |
| 593 | internal_fatal(!(pd->status & PDC_PAGE_DATAFILE_ACQUIRED), |
| 594 | "DBENGINE: page details has not acquired the datafile"); |
| 595 | |
| 596 | internal_fatal((pd->status & (PDC_PAGE_READY | PDC_PAGE_FAILED)), |
| 597 | "DBENGINE: page details has disk pending flag but it is ready/failed"); |
| 598 | |
| 599 | internal_fatal(pd->page, |
| 600 | "DBENGINE: page details has a page linked to it, but it is marked for loading"); |
| 601 | |
| 602 | PValue1 = PDCJudyLIns(&JudyL_datafile_list, pd->datafile.ptr->fileno, PJE0); |
| 603 | if (PValue1 && !*PValue1) { |
| 604 | *PValue1 = deol = deol_get(); |
| 605 | deol->extent_pd_list_by_extent_offset_JudyL = NULL; |
| 606 | deol->fileno = pd->datafile.ptr->fileno; |
| 607 | } |
| 608 | else |
| 609 | deol = *PValue1; |
| 610 | |
| 611 | PValue2 = PDCJudyLIns(&deol->extent_pd_list_by_extent_offset_JudyL, pd->datafile.block, PJE0); |
| 612 | if (PValue2 && !*PValue2) { |
| 613 | *PValue2 = epdl = epdl_get(); |
| 614 | epdl->page_details_by_metric_id_JudyL = NULL; |
| 615 | epdl->extent_block = pd->datafile.block; |
| 616 | epdl->extent_size = pd->datafile.bytes; |
| 617 | epdl->datafile = pd->datafile.ptr; |
| 618 | } |
| 619 | else |
| 620 | epdl = *PValue2; |
| 621 | |
| 622 | Pvoid_t *pd_by_first_time_s_judyL = PDCJudyLIns(&epdl->page_details_by_metric_id_JudyL, pd->metric_id, PJE0); |
| 623 | Pvoid_t *pd_pptr = PDCJudyLIns(pd_by_first_time_s_judyL, pd->first_time_s, PJE0); |
| 624 | *pd_pptr = pd; |
| 625 | } |
| 626 | |
| 627 | size_t extent_list_no = 0; |
| 628 | Word_t datafile_no = 0; |
| 629 | first_then_next = true; |
| 630 | while((PValue = PDCJudyLFirstThenNext(JudyL_datafile_list, &datafile_no, &first_then_next))) { |
| 631 | deol = *PValue; |
| 632 | |
| 633 | bool first_then_next_extent = true; |
| 634 | Word_t pos = 0; |
| 635 | while ((PValue = PDCJudyLFirstThenNext(deol->extent_pd_list_by_extent_offset_JudyL, &pos, &first_then_next_extent))) { |
| 636 | epdl = *PValue; |
| 637 | internal_fatal(!epdl, "DBENGINE: extent_list is not populated properly"); |
| 638 | |
| 639 | // The extent page list can be dispatched to a worker |
| 640 | // It will need to populate the cache with "acquired" pages that are in the list (pd) only |
| 641 | // the rest of the extent pages will be added to the cache butnot acquired |
| 642 | |
| 643 | pdc_acquire(pdc); // we do this for the next worker: do_read_extent_work() |
| 644 | epdl->pdc = pdc; |
| 645 | |
| 646 | if(epdl_pending_add(epdl)) { |
| 647 | if (extent_list_no++ == 0) |
| 648 | exec_first_extent_list(ctx, epdl, pdc->priority); |
| 649 | else |
| 650 | exec_rest_extent_list(ctx, epdl, pdc->priority); |
| 651 | } |
| 652 | } |
| 653 | PDCJudyLFreeArray(&deol->extent_pd_list_by_extent_offset_JudyL, PJE0); |
| 654 | deol_release(deol); |
| 655 | } |
| 656 | PDCJudyLFreeArray(&JudyL_datafile_list, PJE0); |
| 657 | } |
| 658 | |
| 659 | pdc_release_and_destroy_if_unreferenced(pdc, true, true); |
| 660 | } |
| 661 | |
| 662 | void collect_page_flags_to_buffer(BUFFER *wb, RRDENG_COLLECT_PAGE_FLAGS flags) { |
| 663 | if(flags & RRDENG_PAGE_PAST_COLLECTION) |
| 664 | buffer_strcat(wb, "PAST_COLLECTION "); |
| 665 | if(flags & RRDENG_PAGE_REPEATED_COLLECTION) |
| 666 | buffer_strcat(wb, "REPEATED_COLLECTION "); |
| 667 | if(flags & RRDENG_PAGE_BIG_GAP) |
| 668 | buffer_strcat(wb, "BIG_GAP "); |
| 669 | if(flags & RRDENG_PAGE_GAP) |
| 670 | buffer_strcat(wb, "GAP "); |
| 671 | if(flags & RRDENG_PAGE_FUTURE_POINT) |
| 672 | buffer_strcat(wb, "FUTURE_POINT "); |
| 673 | if(flags & RRDENG_PAGE_CREATED_IN_FUTURE) |
| 674 | buffer_strcat(wb, "CREATED_IN_FUTURE "); |
| 675 | if(flags & RRDENG_PAGE_COMPLETED_IN_FUTURE) |
| 676 | buffer_strcat(wb, "COMPLETED_IN_FUTURE "); |
| 677 | if(flags & RRDENG_PAGE_UNALIGNED) |
| 678 | buffer_strcat(wb, "UNALIGNED "); |
| 679 | if(flags & RRDENG_PAGE_CONFLICT) |
| 680 | buffer_strcat(wb, "CONFLICT "); |
| 681 | if(flags & RRDENG_PAGE_FULL) |
| 682 | buffer_strcat(wb, "PAGE_FULL"); |
| 683 | if(flags & RRDENG_PAGE_COLLECT_FINALIZE) |
| 684 | buffer_strcat(wb, "COLLECT_FINALIZE"); |
| 685 | if(flags & RRDENG_PAGE_UPDATE_EVERY_CHANGE) |
| 686 | buffer_strcat(wb, "UPDATE_EVERY_CHANGE"); |
| 687 | if(flags & RRDENG_PAGE_STEP_TOO_SMALL) |
| 688 | buffer_strcat(wb, "STEP_TOO_SMALL"); |
| 689 | if(flags & RRDENG_PAGE_STEP_UNALIGNED) |
| 690 | buffer_strcat(wb, "STEP_UNALIGNED"); |
| 691 | } |
| 692 | |
| 693 | ALWAYS_INLINE VALIDATED_PAGE_DESCRIPTOR validate_extent_page_descr(const struct rrdeng_extent_page_descr *descr, time_t now_s, uint32_t overwrite_zero_update_every_s, bool have_read_error) { |
| 694 | time_t start_time_s = (time_t) (descr->start_time_ut / USEC_PER_SEC); |
| 695 | |
| 696 | time_t end_time_s = 0; |
| 697 | size_t entries = 0; |
| 698 | |
| 699 | switch (descr->type) { |
| 700 | case RRDENG_PAGE_TYPE_ARRAY_32BIT: |
| 701 | case RRDENG_PAGE_TYPE_ARRAY_TIER1: |
| 702 | end_time_s = descr->end_time_ut / USEC_PER_SEC; |
| 703 | entries = 0; |
| 704 | break; |
| 705 | case RRDENG_PAGE_TYPE_GORILLA_32BIT: |
| 706 | end_time_s = start_time_s + descr->gorilla.delta_time_s; |
| 707 | entries = descr->gorilla.entries; |
| 708 | break; |
| 709 | default: |
| 710 | // Nothing to do. Validate page will notify the user. |
| 711 | break; |
| 712 | } |
| 713 | |
| 714 | return validate_page( |
| 715 | (nd_uuid_t *)descr->uuid, |
| 716 | start_time_s, |
| 717 | end_time_s, |
| 718 | 0, |
| 719 | descr->page_length, |
| 720 | descr->type, |
| 721 | entries, |
| 722 | now_s, |
| 723 | overwrite_zero_update_every_s, |
| 724 | have_read_error, |
| 725 | "loaded", 0); |
| 726 | } |
| 727 | |
| 728 | static void validate_page_log(nd_uuid_t *uuid, |
| 729 | time_t start_time_s, |
| 730 | time_t end_time_s, |
| 731 | uint32_t update_every_s, |
| 732 | size_t page_length, |
| 733 | size_t entries, |
| 734 | time_t now_s, |
| 735 | const char *msg, |
| 736 | RRDENG_COLLECT_PAGE_FLAGS flags, |
| 737 | VALIDATED_PAGE_DESCRIPTOR vd) { |
| 738 | #ifndef NETDATA_INTERNAL_CHECKS |
| 739 | nd_log_limit_static_global_var(erl, 1, 0); |
| 740 | #endif |
| 741 | char uuid_str[UUID_STR_LEN + 1]; |
| 742 | uuid_unparse(*uuid, uuid_str); |
| 743 | |
| 744 | CLEAN_BUFFER *wb = NULL; // will be automatically freed on function exit |
| 745 | |
| 746 | if(flags) { |
| 747 | wb = buffer_create(0, NULL); |
| 748 | collect_page_flags_to_buffer(wb, flags); |
| 749 | } |
| 750 | |
| 751 | if(!vd.is_valid) { |
| 752 | #ifdef NETDATA_INTERNAL_CHECKS |
| 753 | internal_error(true, |
| 754 | #else |
| 755 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, |
| 756 | #endif |
| 757 | "DBENGINE: metric '%s' %s invalid page of type %u " |
| 758 | "from %ld to %ld (now %ld), update every %u, page length %zu, entries %zu (flags: %s)", |
| 759 | uuid_str, msg, (unsigned)vd.type, |
| 760 | (long)vd.start_time_s, (long)vd.end_time_s, (long)now_s, (unsigned)vd.update_every_s, (size_t)vd.page_length, (size_t)vd.entries, wb?buffer_tostring(wb):"" |
| 761 | ); |
| 762 | } |
| 763 | else { |
| 764 | CLEAN_BUFFER *log = buffer_create(0, NULL); |
| 765 | |
| 766 | buffer_strcat(log, "DBENGINE: metric '"); |
| 767 | buffer_strcat(log, uuid_str); |
| 768 | buffer_strcat(log, "' "); |
| 769 | buffer_strcat(log, msg ? msg : ""); |
| 770 | buffer_strcat(log, " page of type "); |
| 771 | buffer_print_uint64(log, vd.type); |
| 772 | buffer_strcat(log, " from "); |
| 773 | buffer_print_int64(log, vd.start_time_s); |
| 774 | buffer_strcat(log, " to "); |
| 775 | buffer_print_int64(log, vd.end_time_s); |
| 776 | buffer_strcat(log, " (now "); |
| 777 | buffer_print_int64(log, now_s); |
| 778 | buffer_strcat(log, "), update every "); |
| 779 | buffer_print_uint64(log, vd.update_every_s); |
| 780 | buffer_strcat(log, ", page length "); |
| 781 | buffer_print_uint64(log, vd.page_length); |
| 782 | buffer_strcat(log, ", entries "); |
| 783 | buffer_print_uint64(log, vd.entries); |
| 784 | buffer_strcat(log, " (flags: "); |
| 785 | buffer_strcat(log, wb ? buffer_tostring(wb) : ""); |
| 786 | buffer_strcat(log, ")"); |
| 787 | buffer_strcat(log, "found inconsistent - the right is "); |
| 788 | buffer_print_int64(log, vd.start_time_s); |
| 789 | buffer_strcat(log, " to "); |
| 790 | buffer_print_int64(log, vd.end_time_s); |
| 791 | buffer_strcat(log, ", update every "); |
| 792 | buffer_print_uint64(log, vd.update_every_s); |
| 793 | buffer_strcat(log, ", page length "); |
| 794 | buffer_print_uint64(log, vd.page_length); |
| 795 | buffer_strcat(log, ", entries "); |
| 796 | buffer_print_uint64(log, vd.entries); |
| 797 | buffer_strcat(log, (vd.start_time_s == start_time_s) ? "" : "start time updated, "); |
| 798 | buffer_strcat(log, (vd.end_time_s == end_time_s) ? "" : "end time updated, "); |
| 799 | buffer_strcat(log, (vd.update_every_s == update_every_s) ? "" : "update every updated, "); |
| 800 | buffer_strcat(log, (vd.page_length == page_length) ? "" : "page length updated, "); |
| 801 | buffer_strcat(log, (vd.entries == entries) ? "" : "entries updated, "); |
| 802 | buffer_strcat(log, (now_s && vd.end_time_s <= now_s) ? "" : "future end time, "); |
| 803 | |
| 804 | #ifdef NETDATA_INTERNAL_CHECKS |
| 805 | internal_error(true, "%s", buffer_tostring(log)); |
| 806 | #else |
| 807 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, "%s", buffer_tostring(log)); |
| 808 | #endif |
| 809 | } |
| 810 | } |
| 811 | |
| 812 | ALWAYS_INLINE |
| 813 | VALIDATED_PAGE_DESCRIPTOR validate_page( |
| 814 | nd_uuid_t *uuid, |
| 815 | time_t start_time_s, |
| 816 | time_t end_time_s, |
| 817 | uint32_t update_every_s, // can be zero, if unknown |
| 818 | size_t page_length, |
| 819 | uint8_t page_type, |
| 820 | size_t entries, // can be zero, if unknown |
| 821 | time_t now_s, // can be zero, to disable future timestamp check |
| 822 | uint32_t overwrite_zero_update_every_s, // can be zero, if unknown |
| 823 | bool have_read_error, |
| 824 | const char *msg, |
| 825 | RRDENG_COLLECT_PAGE_FLAGS flags) |
| 826 | { |
| 827 | VALIDATED_PAGE_DESCRIPTOR vd = { |
| 828 | .start_time_s = start_time_s, |
| 829 | .end_time_s = end_time_s, |
| 830 | .update_every_s = update_every_s, |
| 831 | .page_length = page_length, |
| 832 | .point_size = page_type_size[page_type], |
| 833 | .type = page_type, |
| 834 | .is_valid = true, |
| 835 | }; |
| 836 | |
| 837 | bool known_page_type = true; |
| 838 | switch (page_type) { |
| 839 | case RRDENG_PAGE_TYPE_ARRAY_32BIT: |
| 840 | case RRDENG_PAGE_TYPE_ARRAY_TIER1: |
| 841 | // always calculate entries by size |
| 842 | vd.entries = page_entries_by_size(vd.page_length, vd.point_size); |
| 843 | |
| 844 | // allow to be called without entries (when loading pages from disk) |
| 845 | if(!entries) |
| 846 | entries = vd.entries; |
| 847 | break; |
| 848 | case RRDENG_PAGE_TYPE_GORILLA_32BIT: |
| 849 | internal_fatal(entries == 0, "0 number of entries found on gorilla page"); |
| 850 | vd.entries = entries; |
| 851 | break; |
| 852 | default: |
| 853 | known_page_type = false; |
| 854 | break; |
| 855 | } |
| 856 | |
| 857 | // allow to be called without update every (when loading pages from disk) |
| 858 | if(!update_every_s) { |
| 859 | vd.update_every_s = (vd.entries > 1) ? ((uint32_t)(vd.end_time_s - vd.start_time_s) / (vd.entries - 1)) |
| 860 | : overwrite_zero_update_every_s; |
| 861 | |
| 862 | update_every_s = vd.update_every_s; |
| 863 | } |
| 864 | |
| 865 | // another such set of checks exists in |
| 866 | // update_metric_retention_and_granularity_by_uuid() |
| 867 | |
| 868 | bool updated = false; |
| 869 | |
| 870 | size_t max_page_length = RRDENG_BLOCK_SIZE; |
| 871 | |
| 872 | // If gorilla can not compress the data we might end up needing slightly more |
| 873 | // than 4KiB. However, gorilla pages extend the page length by increments of |
| 874 | // 512 bytes. |
| 875 | max_page_length += ((page_type == RRDENG_PAGE_TYPE_GORILLA_32BIT) * (2 * RRDENG_GORILLA_32BIT_BUFFER_SIZE)); |
| 876 | |
| 877 | if (!known_page_type || |
| 878 | have_read_error || |
| 879 | vd.page_length == 0 || |
| 880 | vd.page_length > max_page_length || |
| 881 | vd.start_time_s > vd.end_time_s || |
| 882 | (now_s && vd.end_time_s > now_s) || |
| 883 | vd.start_time_s <= 0 || |
| 884 | vd.end_time_s <= 0 || |
| 885 | (vd.start_time_s == vd.end_time_s && vd.entries > 1) || |
| 886 | (vd.update_every_s == 0 && vd.entries > 1)) |
| 887 | { |
| 888 | vd.is_valid = false; |
| 889 | } |
| 890 | else { |
| 891 | if(unlikely(vd.entries != entries || vd.update_every_s != update_every_s)) |
| 892 | updated = true; |
| 893 | |
| 894 | if (likely(vd.update_every_s)) { |
| 895 | size_t entries_by_time = page_entries_by_time(vd.start_time_s, vd.end_time_s, vd.update_every_s); |
| 896 | |
| 897 | if (vd.entries != entries_by_time) { |
| 898 | if (overwrite_zero_update_every_s < vd.update_every_s) |
| 899 | vd.update_every_s = overwrite_zero_update_every_s; |
| 900 | |
| 901 | time_t new_end_time_s = (time_t)(vd.start_time_s + (vd.entries - 1) * vd.update_every_s); |
| 902 | |
| 903 | if(new_end_time_s <= vd.end_time_s) { |
| 904 | // end time is wrong |
| 905 | vd.end_time_s = new_end_time_s; |
| 906 | } |
| 907 | else { |
| 908 | // update every is wrong |
| 909 | vd.update_every_s = overwrite_zero_update_every_s; |
| 910 | vd.end_time_s = (time_t)(vd.start_time_s + (vd.entries - 1) * vd.update_every_s); |
| 911 | } |
| 912 | |
| 913 | updated = true; |
| 914 | } |
| 915 | } |
| 916 | else if(overwrite_zero_update_every_s) { |
| 917 | vd.update_every_s = overwrite_zero_update_every_s; |
| 918 | updated = true; |
| 919 | } |
| 920 | } |
| 921 | |
| 922 | if(unlikely(!vd.is_valid || updated)) |
| 923 | validate_page_log(uuid, start_time_s, end_time_s, update_every_s, page_length, entries, now_s, msg, flags, vd); |
| 924 | |
| 925 | return vd; |
| 926 | } |
| 927 | |
| 928 | static ALWAYS_INLINE struct page_details *epdl_get_pd_load_link_list_from_metric_start_time(EPDL *epdl, Word_t metric_id, time_t start_time_s) { |
| 929 | |
| 930 | if(unlikely(epdl->head_to_datafile_extent_queries_pending_for_extent)) |
| 931 | // stop appending more pages to this epdl |
| 932 | epdl_pending_del(epdl); |
| 933 | |
| 934 | struct page_details *pd_list = NULL; |
| 935 | |
| 936 | for(EPDL *ep = epdl; ep ;ep = ep->query.next) { |
| 937 | Pvoid_t *pd_by_start_time_s_judyL = PDCJudyLGet(ep->page_details_by_metric_id_JudyL, metric_id, PJE0); |
| 938 | internal_fatal(pd_by_start_time_s_judyL == PJERR, "DBENGINE: corrupted extent metrics JudyL"); |
| 939 | |
| 940 | if (unlikely(pd_by_start_time_s_judyL && *pd_by_start_time_s_judyL)) { |
| 941 | Pvoid_t *pd_pptr = PDCJudyLGet(*pd_by_start_time_s_judyL, start_time_s, PJE0); |
| 942 | internal_fatal(pd_pptr == PJERR, "DBENGINE: corrupted metric page details JudyHS"); |
| 943 | |
| 944 | if(likely(pd_pptr && *pd_pptr)) { |
| 945 | struct page_details *pd = *pd_pptr; |
| 946 | internal_fatal(metric_id != pd->metric_id, "DBENGINE: metric ids do not match"); |
| 947 | |
| 948 | if(likely(!pd->page)) { |
| 949 | if (unlikely(__atomic_load_n(&ep->pdc->workers_should_stop, __ATOMIC_RELAXED))) |
| 950 | pdc_page_status_set(pd, PDC_PAGE_FAILED | PDC_PAGE_CANCELLED); |
| 951 | else |
| 952 | DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(pd_list, pd, load.prev, load.next); |
| 953 | } |
| 954 | } |
| 955 | } |
| 956 | } |
| 957 | |
| 958 | return pd_list; |
| 959 | } |
| 960 | |
| 961 | static void epdl_extent_loading_error_log(struct rrdengine_instance *ctx, EPDL *epdl, struct rrdeng_extent_page_descr *descr, const char *msg, ND_LOG_FIELD_PRIORITY priority) { |
| 962 | char uuid[UUID_STR_LEN] = ""; |
| 963 | time_t start_time_s = 0; |
| 964 | time_t end_time_s = 0; |
| 965 | bool used_epdl = false; |
| 966 | bool used_descr = false; |
| 967 | |
| 968 | if (descr) { |
| 969 | start_time_s = (time_t)(descr->start_time_ut / USEC_PER_SEC); |
| 970 | switch (descr->type) { |
| 971 | case RRDENG_PAGE_TYPE_ARRAY_32BIT: |
| 972 | case RRDENG_PAGE_TYPE_ARRAY_TIER1: |
| 973 | end_time_s = (time_t)(descr->end_time_ut / USEC_PER_SEC); |
| 974 | break; |
| 975 | case RRDENG_PAGE_TYPE_GORILLA_32BIT: |
| 976 | end_time_s = (time_t) start_time_s + (descr->gorilla.delta_time_s); |
| 977 | break; |
| 978 | } |
| 979 | uuid_unparse_lower(descr->uuid, uuid); |
| 980 | used_descr = true; |
| 981 | } |
| 982 | else { |
| 983 | struct page_details *pd = NULL; |
| 984 | |
| 985 | Word_t start = 0; |
| 986 | Pvoid_t *pd_by_start_time_s_judyL = PDCJudyLFirst(epdl->page_details_by_metric_id_JudyL, &start, PJE0); |
| 987 | if(pd_by_start_time_s_judyL) { |
| 988 | start = 0; |
| 989 | Pvoid_t *pd_pptr = PDCJudyLFirst(*pd_by_start_time_s_judyL, &start, PJE0); |
| 990 | if(pd_pptr) { |
| 991 | pd = *pd_pptr; |
| 992 | start_time_s = pd->first_time_s; |
| 993 | end_time_s = pd->last_time_s; |
| 994 | METRIC *metric = (METRIC *)pd->metric_id; |
| 995 | nd_uuid_t *u = mrg_metric_uuid(main_mrg, metric); |
| 996 | uuid_unparse_lower(*u, uuid); |
| 997 | used_epdl = true; |
| 998 | } |
| 999 | } |
| 1000 | } |
| 1001 | |
| 1002 | if(!used_epdl && !used_descr && epdl->pdc) { |
| 1003 | start_time_s = epdl->pdc->start_time_s; |
| 1004 | end_time_s = epdl->pdc->end_time_s; |
| 1005 | } |
| 1006 | |
| 1007 | char start_time_str[LOG_DATE_LENGTH + 1] = ""; |
| 1008 | if(start_time_s) |
| 1009 | log_date(start_time_str, LOG_DATE_LENGTH, start_time_s); |
| 1010 | |
| 1011 | char end_time_str[LOG_DATE_LENGTH + 1] = ""; |
| 1012 | if(end_time_s) |
| 1013 | log_date(end_time_str, LOG_DATE_LENGTH, end_time_s); |
| 1014 | |
| 1015 | nd_log_limit_static_global_var(erl, 1, 0); |
| 1016 | nd_log_limit(&erl, NDLS_DAEMON, priority, |
| 1017 | "DBENGINE: error while reading extent from datafile %u of tier %d, at offset %" PRIu64 " (%u bytes) " |
| 1018 | "%s from %ld (%s) to %ld (%s) %s%s: " |
| 1019 | "%s", |
| 1020 | epdl->datafile->fileno, ctx->config.tier, |
| 1021 | BLOCK_TO_OFFSET(epdl->extent_block), epdl->extent_size, |
| 1022 | used_epdl ? "to extract page (PD)" : used_descr ? "expected page (DESCR)" : "part of a query (PDC)", |
| 1023 | start_time_s, start_time_str, end_time_s, end_time_str, |
| 1024 | used_epdl || used_descr ? " of metric " : "", |
| 1025 | used_epdl || used_descr ? uuid : "", |
| 1026 | msg); |
| 1027 | } |
| 1028 | |
| 1029 | static bool epdl_populate_pages_from_extent_data( |
| 1030 | struct rrdengine_instance *ctx, |
| 1031 | void *data, |
| 1032 | size_t data_length, |
| 1033 | EPDL *epdl, |
| 1034 | bool worker, |
| 1035 | PDC_PAGE_STATUS tags, |
| 1036 | bool cached_extent) |
| 1037 | { |
| 1038 | unsigned i, count; |
| 1039 | void *uncompressed_buf = NULL; |
| 1040 | uint64_t payload_length, payload_offset, trailer_offset; |
| 1041 | uint32_t uncompressed_payload_length = 0; |
| 1042 | bool have_read_error = false; |
| 1043 | /* persistent structures */ |
| 1044 | struct rrdeng_df_extent_header *header; |
| 1045 | struct rrdeng_df_extent_trailer *trailer; |
| 1046 | struct extent_buffer *eb = NULL; |
| 1047 | uLong crc; |
| 1048 | |
| 1049 | bool can_use_data = true; |
| 1050 | if(!rrdeng_valid_extent_disk_size(data_length)) { |
| 1051 | can_use_data = false; |
| 1052 | |
| 1053 | // added to satisfy the requirements of older compilers (prevent warnings) |
| 1054 | payload_length = 0; |
| 1055 | payload_offset = 0; |
| 1056 | trailer_offset = 0; |
| 1057 | count = 0; |
| 1058 | header = NULL; |
| 1059 | trailer = NULL; |
| 1060 | } |
| 1061 | else { |
| 1062 | header = data; |
| 1063 | payload_length = header->payload_length; |
| 1064 | count = header->number_of_pages; |
| 1065 | payload_offset = sizeof(*header) + sizeof(header->descr[0]) * count; |
| 1066 | trailer_offset = data_length - sizeof(*trailer); |
| 1067 | trailer = data + trailer_offset; |
| 1068 | } |
| 1069 | |
| 1070 | if( !can_use_data || |
| 1071 | count < 1 || |
| 1072 | count > MAX_PAGES_PER_EXTENT || |
| 1073 | !dbengine_valid_compression_algorithm(header->compression_algorithm) || |
| 1074 | (payload_length != trailer_offset - payload_offset) || |
| 1075 | (data_length != payload_offset + payload_length + sizeof(*trailer)) |
| 1076 | ) { |
| 1077 | epdl_extent_loading_error_log(ctx, epdl, NULL, "header is INVALID", NDLP_ERR); |
| 1078 | return false; |
| 1079 | } |
| 1080 | |
| 1081 | crc = crc32(0L, Z_NULL, 0); |
| 1082 | crc = crc32(crc, data, data_length - sizeof(*trailer)); |
| 1083 | if (unlikely(crc32cmp(trailer->checksum, crc))) { |
| 1084 | ctx_io_error(ctx); |
| 1085 | have_read_error = true; |
| 1086 | epdl_extent_loading_error_log(ctx, epdl, NULL, "CRC32 checksum FAILED", NDLP_ERR); |
| 1087 | } |
| 1088 | |
| 1089 | if(worker) |
| 1090 | worker_is_busy(UV_EVENT_DBENGINE_EXTENT_DECOMPRESSION); |
| 1091 | |
| 1092 | if (likely(!have_read_error && RRDENG_COMPRESSION_NONE != header->compression_algorithm)) { |
| 1093 | // find the uncompressed extent size |
| 1094 | uncompressed_payload_length = 0; |
| 1095 | for (i = 0; i < count; ++i) { |
| 1096 | size_t page_length = header->descr[i].page_length; |
| 1097 | if (page_length > RRDENG_BLOCK_SIZE && |
| 1098 | (header->descr[i].type != RRDENG_PAGE_TYPE_GORILLA_32BIT || |
| 1099 | (header->descr[i].type == RRDENG_PAGE_TYPE_GORILLA_32BIT && |
| 1100 | (page_length - RRDENG_BLOCK_SIZE) % RRDENG_GORILLA_32BIT_BUFFER_SIZE))) { |
| 1101 | have_read_error = true; |
| 1102 | break; |
| 1103 | } |
| 1104 | |
| 1105 | uncompressed_payload_length += header->descr[i].page_length; |
| 1106 | } |
| 1107 | |
| 1108 | if(unlikely(uncompressed_payload_length > MAX_EXTENT_UNCOMPRESSED_SIZE)) |
| 1109 | have_read_error = true; |
| 1110 | |
| 1111 | if(likely(!have_read_error)) { |
| 1112 | eb = extent_buffer_get(uncompressed_payload_length); |
| 1113 | uncompressed_buf = eb->data; |
| 1114 | |
| 1115 | size_t bytes = dbengine_decompress(uncompressed_buf, data + payload_offset, |
| 1116 | uncompressed_payload_length, payload_length, |
| 1117 | header->compression_algorithm); |
| 1118 | |
| 1119 | if(!bytes) |
| 1120 | have_read_error = true; |
| 1121 | else { |
| 1122 | __atomic_add_fetch(&ctx->stats.before_decompress_bytes, payload_length, __ATOMIC_RELAXED); |
| 1123 | __atomic_add_fetch(&ctx->stats.after_decompress_bytes, bytes, __ATOMIC_RELAXED); |
| 1124 | } |
| 1125 | } |
| 1126 | } |
| 1127 | |
| 1128 | if(worker) |
| 1129 | worker_is_busy(UV_EVENT_DBENGINE_EXTENT_PAGE_LOOKUP); |
| 1130 | |
| 1131 | size_t stats_data_from_main_cache = 0; |
| 1132 | size_t stats_data_from_extent = 0; |
| 1133 | size_t stats_load_compressed = 0; |
| 1134 | size_t stats_load_uncompressed = 0; |
| 1135 | size_t stats_load_invalid_page = 0; |
| 1136 | size_t stats_cache_hit_while_inserting = 0; |
| 1137 | |
| 1138 | uint32_t page_offset = 0, page_length; |
| 1139 | time_t now_s = max_acceptable_collected_time(); |
| 1140 | for (i = 0; i < count; i++, page_offset += page_length) { |
| 1141 | page_length = header->descr[i].page_length; |
| 1142 | time_t start_time_s = (time_t) (header->descr[i].start_time_ut / USEC_PER_SEC); |
| 1143 | |
| 1144 | if(!page_length || !start_time_s) { |
| 1145 | char log[200 + 1]; |
| 1146 | snprintfz(log, sizeof(log) - 1, "page %u (out of %u) is EMPTY", i, count); |
| 1147 | epdl_extent_loading_error_log(ctx, epdl, &header->descr[i], log, NDLP_ERR); |
| 1148 | continue; |
| 1149 | } |
| 1150 | |
| 1151 | METRIC *metric = mrg_metric_get_and_acquire_by_uuid(main_mrg, &header->descr[i].uuid, (Word_t)ctx); |
| 1152 | Word_t metric_id = (Word_t)metric; |
| 1153 | if(!metric) { |
| 1154 | char log[200 + 1]; |
| 1155 | snprintfz(log, sizeof(log) - 1, "page %u (out of %u) has unknown UUID", i, count); |
| 1156 | epdl_extent_loading_error_log(ctx, epdl, &header->descr[i], log, NDLP_DEBUG); |
| 1157 | continue; |
| 1158 | } |
| 1159 | mrg_metric_release(main_mrg, metric); |
| 1160 | |
| 1161 | struct page_details *pd_list = epdl_get_pd_load_link_list_from_metric_start_time(epdl, metric_id, start_time_s); |
| 1162 | if(likely(!pd_list)) |
| 1163 | continue; |
| 1164 | |
| 1165 | VALIDATED_PAGE_DESCRIPTOR vd = validate_extent_page_descr( |
| 1166 | &header->descr[i], now_s, |
| 1167 | (pd_list) ? pd_list->update_every_s : 0, |
| 1168 | have_read_error); |
| 1169 | |
| 1170 | if(worker) |
| 1171 | worker_is_busy(UV_EVENT_DBENGINE_EXTENT_PAGE_ALLOCATION); |
| 1172 | |
| 1173 | PGD *pgd; |
| 1174 | |
| 1175 | if (unlikely(!vd.is_valid)) { |
| 1176 | pgd = PGD_EMPTY; |
| 1177 | stats_load_invalid_page++; |
| 1178 | } |
| 1179 | else { |
| 1180 | if (RRDENG_COMPRESSION_NONE == header->compression_algorithm) { |
| 1181 | if (unlikely(vd.page_length > payload_length || |
| 1182 | page_offset > payload_length - vd.page_length)) { |
| 1183 | char log[200 + 1]; |
| 1184 | snprintfz(log, sizeof(log) - 1, "page %u (out of %u) offset %u + page length %zu, " |
| 1185 | "exceeds the payload size %" PRIu64, |
| 1186 | i, count, page_offset, vd.page_length, payload_length); |
| 1187 | epdl_extent_loading_error_log(ctx, epdl, &header->descr[i], log, NDLP_ERR); |
| 1188 | |
| 1189 | pgd = PGD_EMPTY; |
| 1190 | stats_load_invalid_page++; |
| 1191 | } |
| 1192 | else { |
| 1193 | pgd = pgd_create_from_disk_data(header->descr[i].type, |
| 1194 | data + payload_offset + page_offset, |
| 1195 | vd.page_length); |
| 1196 | stats_load_uncompressed++; |
| 1197 | } |
| 1198 | } |
| 1199 | else { |
| 1200 | if (unlikely(vd.page_length > uncompressed_payload_length || |
| 1201 | page_offset > uncompressed_payload_length - vd.page_length)) { |
| 1202 | char log[200 + 1]; |
| 1203 | snprintfz(log, sizeof(log) - 1, "page %u (out of %u) offset %u + page length %zu, " |
| 1204 | "exceeds the uncompressed buffer size %u", |
| 1205 | i, count, page_offset, vd.page_length, uncompressed_payload_length); |
| 1206 | epdl_extent_loading_error_log(ctx, epdl, &header->descr[i], log, NDLP_ERR); |
| 1207 | |
| 1208 | pgd = PGD_EMPTY; |
| 1209 | stats_load_invalid_page++; |
| 1210 | } |
| 1211 | else { |
| 1212 | pgd = pgd_create_from_disk_data(header->descr[i].type, |
| 1213 | uncompressed_buf + page_offset, |
| 1214 | vd.page_length); |
| 1215 | stats_load_compressed++; |
| 1216 | } |
| 1217 | } |
| 1218 | } |
| 1219 | |
| 1220 | if(worker) |
| 1221 | worker_is_busy(UV_EVENT_DBENGINE_EXTENT_PAGE_POPULATION); |
| 1222 | |
| 1223 | PGC_ENTRY page_entry = { |
| 1224 | .hot = false, |
| 1225 | .section = (Word_t)ctx, |
| 1226 | .metric_id = metric_id, |
| 1227 | .start_time_s = vd.start_time_s, |
| 1228 | .end_time_s = vd.end_time_s, |
| 1229 | .update_every_s = (uint32_t) vd.update_every_s, |
| 1230 | .size = pgd_memory_footprint(pgd), // the footprint of the entire PGD, for accurate memory management |
| 1231 | .data = pgd, |
| 1232 | }; |
| 1233 | |
| 1234 | bool added = true; |
| 1235 | PGC_PAGE *page = pgc_page_add_and_acquire(main_cache, page_entry, &added); |
| 1236 | if (false == added) { |
| 1237 | pgd_free(pgd); |
| 1238 | pgd = pgc_page_data(page); |
| 1239 | stats_cache_hit_while_inserting++; |
| 1240 | stats_data_from_main_cache++; |
| 1241 | } |
| 1242 | else |
| 1243 | stats_data_from_extent++; |
| 1244 | |
| 1245 | struct page_details *pd = pd_list; |
| 1246 | do { |
| 1247 | if(pd != pd_list) |
| 1248 | pgc_page_dup(main_cache, page); |
| 1249 | |
| 1250 | pd->page = page; |
| 1251 | pdc_page_status_set(pd, PDC_PAGE_READY | tags | (pgd_is_empty(pgd) ? PDC_PAGE_EMPTY : 0)); |
| 1252 | |
| 1253 | pd = pd->load.next; |
| 1254 | } while(pd); |
| 1255 | |
| 1256 | if(worker) |
| 1257 | worker_is_busy(UV_EVENT_DBENGINE_EXTENT_PAGE_LOOKUP); |
| 1258 | } |
| 1259 | |
| 1260 | if(stats_data_from_main_cache) |
| 1261 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_data_source_main_cache, stats_data_from_main_cache, __ATOMIC_RELAXED); |
| 1262 | |
| 1263 | if(cached_extent) |
| 1264 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_data_source_extent_cache, stats_data_from_extent, __ATOMIC_RELAXED); |
| 1265 | else { |
| 1266 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_data_source_disk, stats_data_from_extent, __ATOMIC_RELAXED); |
| 1267 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.extents_loaded_from_disk, 1, __ATOMIC_RELAXED); |
| 1268 | } |
| 1269 | |
| 1270 | if(stats_cache_hit_while_inserting) |
| 1271 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_ok_loaded_but_cache_hit_while_inserting, stats_cache_hit_while_inserting, __ATOMIC_RELAXED); |
| 1272 | |
| 1273 | if(stats_load_compressed) |
| 1274 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_ok_compressed, stats_load_compressed, __ATOMIC_RELAXED); |
| 1275 | |
| 1276 | if(stats_load_uncompressed) |
| 1277 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_ok_uncompressed, stats_load_uncompressed, __ATOMIC_RELAXED); |
| 1278 | |
| 1279 | if(stats_load_invalid_page) |
| 1280 | __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_fail_invalid_page_in_extent, stats_load_invalid_page, __ATOMIC_RELAXED); |
| 1281 | |
| 1282 | if(worker) |
| 1283 | worker_is_idle(); |
| 1284 | |
| 1285 | extent_buffer_release(eb); |
| 1286 | |
| 1287 | return true; |
| 1288 | } |
| 1289 | |
| 1290 | static inline void *datafile_extent_read(struct rrdengine_instance *ctx, uv_file file, uint32_t block, unsigned size_bytes) |
| 1291 | { |
| 1292 | if (unlikely(!rrdeng_valid_extent_disk_size(size_bytes))) { |
| 1293 | nd_log_limit_static_global_var(erl, 1, 0); |
| 1294 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, |
| 1295 | "DBENGINE: refusing to read extent at offset %" PRIu64 " with invalid size %u", |
| 1296 | BLOCK_TO_OFFSET(block), size_bytes); |
| 1297 | ctx_io_error(ctx); |
| 1298 | return NULL; |
| 1299 | } |
| 1300 | |
| 1301 | void *buffer = NULL; |
| 1302 | uv_fs_t request; |
| 1303 | |
| 1304 | unsigned real_io_size = ALIGN_BYTES_CEILING(size_bytes); |
| 1305 | (void)posix_memalignz(&buffer, RRDFILE_ALIGNMENT, real_io_size); |
| 1306 | |
| 1307 | uv_buf_t iov = uv_buf_init(buffer, real_io_size); |
| 1308 | int ret = uv_fs_read(NULL, &request, file, &iov, 1, (int64_t) BLOCK_TO_OFFSET(block), NULL); |
| 1309 | if (unlikely(ret < 0 || (unsigned)ret != real_io_size)) { |
| 1310 | ctx_io_error(ctx); |
| 1311 | posix_memalign_freez(buffer); |
| 1312 | buffer = NULL; |
| 1313 | } |
| 1314 | else |
| 1315 | ctx_io_read_op_bytes(ctx, real_io_size); |
| 1316 | |
| 1317 | uv_fs_req_cleanup(&request); |
| 1318 | |
| 1319 | return buffer; |
| 1320 | } |
| 1321 | |
| 1322 | static inline void datafile_extent_read_free(void *buffer) { |
| 1323 | posix_memalign_freez(buffer); |
| 1324 | } |
| 1325 | |
| 1326 | NOT_INLINE_HOT void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *epdl, bool worker) { |
| 1327 | if(worker) |
| 1328 | worker_is_busy(UV_EVENT_DBENGINE_EXTENT_CACHE_LOOKUP); |
| 1329 | |
| 1330 | size_t *statistics_counter = NULL; |
| 1331 | PDC_PAGE_STATUS not_loaded_pages_tag = 0, loaded_pages_tag = 0; |
| 1332 | |
| 1333 | bool should_stop = __atomic_load_n(&epdl->pdc->workers_should_stop, __ATOMIC_RELAXED); |
| 1334 | for(EPDL *ep = epdl->query.next; ep ;ep = ep->query.next) { |
| 1335 | internal_fatal(ep->datafile != epdl->datafile, "DBENGINE: datafiles do not match"); |
| 1336 | internal_fatal(ep->extent_block != epdl->extent_block, "DBENGINE: extent blocks do not match"); |
| 1337 | internal_fatal(ep->extent_size != epdl->extent_size, "DBENGINE: extent sizes do not match"); |
| 1338 | |
| 1339 | if(!__atomic_load_n(&ep->pdc->workers_should_stop, __ATOMIC_RELAXED)) { |
| 1340 | should_stop = false; |
| 1341 | break; |
| 1342 | } |
| 1343 | } |
| 1344 | |
| 1345 | if(unlikely(should_stop)) { |
| 1346 | statistics_counter = &rrdeng_cache_efficiency_stats.pages_load_fail_cancelled; |
| 1347 | not_loaded_pages_tag = PDC_PAGE_CANCELLED; |
| 1348 | goto cleanup; |
| 1349 | } |
| 1350 | |
| 1351 | bool extent_found_in_cache = false; |
| 1352 | |
| 1353 | void *extent_compressed_data = NULL; |
| 1354 | PGC_PAGE *extent_cache_page = pgc_page_get_and_acquire( |
| 1355 | extent_cache, (Word_t)ctx, |
| 1356 | (Word_t)epdl->datafile->fileno, (time_t)epdl->extent_block, |
| 1357 | PGC_SEARCH_EXACT); |
| 1358 | |
| 1359 | if(extent_cache_page) { |
| 1360 | extent_compressed_data = pgc_page_data(extent_cache_page); |
| 1361 | internal_fatal(epdl->extent_size != pgc_page_data_size(extent_cache, extent_cache_page), |
| 1362 | "DBENGINE: cache size does not match the expected size"); |
| 1363 | |
| 1364 | loaded_pages_tag |= PDC_PAGE_EXTENT_FROM_CACHE; |
| 1365 | not_loaded_pages_tag |= PDC_PAGE_EXTENT_FROM_CACHE; |
| 1366 | extent_found_in_cache = true; |
| 1367 | } |
| 1368 | else { |
| 1369 | if(worker) |
| 1370 | worker_is_busy(UV_EVENT_DBENGINE_EXTENT_MMAP); |
| 1371 | |
| 1372 | void *extent_data = datafile_extent_read(ctx, epdl->datafile->file, epdl->extent_block, epdl->extent_size); |
| 1373 | if(extent_data != NULL) { |
| 1374 | |
| 1375 | void *tmp = dbengine_extent_alloc(epdl->extent_size); |
| 1376 | memcpy(tmp, extent_data, epdl->extent_size); |
| 1377 | datafile_extent_read_free(extent_data); |
| 1378 | extent_data = tmp; |
| 1379 | |
| 1380 | if(worker) |
| 1381 | worker_is_busy(UV_EVENT_DBENGINE_EXTENT_CACHE_LOOKUP); |
| 1382 | |
| 1383 | bool added = false; |
| 1384 | extent_cache_page = pgc_page_add_and_acquire(extent_cache, (PGC_ENTRY) { |
| 1385 | .hot = false, |
| 1386 | .section = (Word_t) ctx, |
| 1387 | .metric_id = (Word_t) epdl->datafile->fileno, |
| 1388 | .start_time_s = (time_t) epdl->extent_block, |
| 1389 | .size = epdl->extent_size, |
| 1390 | .end_time_s = 0, |
| 1391 | .update_every_s = 0, |
| 1392 | .data = extent_data, |
| 1393 | }, &added); |
| 1394 | |
| 1395 | if (!added) { |
| 1396 | dbengine_extent_free(extent_data, epdl->extent_size); |
| 1397 | internal_fatal(epdl->extent_size != pgc_page_data_size(extent_cache, extent_cache_page), |
| 1398 | "DBENGINE: cache size does not match the expected size"); |
| 1399 | } |
| 1400 | |
| 1401 | extent_compressed_data = pgc_page_data(extent_cache_page); |
| 1402 | |
| 1403 | loaded_pages_tag |= PDC_PAGE_EXTENT_FROM_DISK; |
| 1404 | not_loaded_pages_tag |= PDC_PAGE_EXTENT_FROM_DISK; |
| 1405 | } |
| 1406 | } |
| 1407 | |
| 1408 | if(extent_compressed_data) { |
| 1409 | // Need to decompress and then process the pagelist |
| 1410 | bool extent_used = epdl_populate_pages_from_extent_data( |
| 1411 | ctx, extent_compressed_data, epdl->extent_size, |
| 1412 | epdl, worker, loaded_pages_tag, extent_found_in_cache); |
| 1413 | |
| 1414 | if(extent_used) { |
| 1415 | // since the extent was used, all the pages that are not |
| 1416 | // loaded from this extent, were not found in the extent |
| 1417 | not_loaded_pages_tag |= PDC_PAGE_FAILED_NOT_IN_EXTENT; |
| 1418 | statistics_counter = &rrdeng_cache_efficiency_stats.pages_load_fail_not_found; |
| 1419 | } |
| 1420 | else { |
| 1421 | not_loaded_pages_tag |= PDC_PAGE_FAILED_INVALID_EXTENT; |
| 1422 | statistics_counter = &rrdeng_cache_efficiency_stats.pages_load_fail_invalid_extent; |
| 1423 | } |
| 1424 | } |
| 1425 | else { |
| 1426 | not_loaded_pages_tag |= PDC_PAGE_FAILED_TO_MAP_EXTENT; |
| 1427 | statistics_counter = &rrdeng_cache_efficiency_stats.pages_load_fail_cant_mmap_extent; |
| 1428 | } |
| 1429 | |
| 1430 | if(extent_cache_page) |
| 1431 | pgc_page_release(extent_cache, extent_cache_page); |
| 1432 | |
| 1433 | cleanup: |
| 1434 | // remove it from the datafile extent_queries |
| 1435 | // this can be called multiple times safely |
| 1436 | epdl_pending_del(epdl); |
| 1437 | |
| 1438 | // mark all pending pages as failed |
| 1439 | for(EPDL *ep = epdl; ep ;ep = ep->query.next) { |
| 1440 | epdl_mark_all_not_loaded_pages_as_failed( |
| 1441 | ep, not_loaded_pages_tag, statistics_counter); |
| 1442 | } |
| 1443 | |
| 1444 | for(EPDL *ep = epdl, *next = NULL; ep ; ep = next) { |
| 1445 | next = ep->query.next; |
| 1446 | |
| 1447 | completion_mark_complete_a_job(&ep->pdc->page_completion); |
| 1448 | pdc_release_and_destroy_if_unreferenced(ep->pdc, true, false); |
| 1449 | |
| 1450 | // Free the Judy that holds the requested pagelist and the extents |
| 1451 | epdl_destroy(ep); |
| 1452 | } |
| 1453 | |
| 1454 | if(worker) |
| 1455 | worker_is_idle(); |
| 1456 | } |