@cryptotaxi247 / netdata-1 / commits / 3a430c181

DBENGINE v2 - improvements part 8 (#14319)

* cache 100 pages for each size our tiers need * smarter page caching * account the caching structures * dynamic max number of cached pages * make variables const to ensure they are not changed * make sure replication timestamps do not go to the future * replication now sends chart and dimension states atomically; replication receivers ignores chart and dimension states when rbegin is also ignored * make sure all pages are flushed on shutdown * take into account empty points too * when recalculating retention update first_time_s on metrics only when they are bigger * Report the datafile number we use to recalculate retention * Report the datafile number we use to recalculate retention * rotate db at startup * make query plans overlap * Calculate properly first time s * updated event labels * negative page caching fix * Atempt to create missing tables on query failure * Atempt to create missing tables on query failure (part 2) * negative page caching for all gaps, to eliminate jv2 scans * Fix unittest Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>

Costa Tsaousis committed Jan 25, 2023 at 01:56 UTC 3a430c181e7655a8460b40e9864395694f223e46
20 files changed +595 -306
collectors/plugins.d/pluginsd_parser.c
+6
@@ -1125,6 +1125,9 @@ PARSER_RC pluginsd_replay_set(char **words, size_t num_words, void *user)
1125
1126 PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, size_t num_words, void *user)
1127 {
1128 + if(((PARSER_USER_OBJECT *) user)->replay.rset_enabled == false)
1129 + return PARSER_RC_OK;
1130 +
1131 char *dimension = get_word(words, num_words, 1);
1132 char *last_collected_ut_str = get_word(words, num_words, 2);
1133 char *last_collected_value_str = get_word(words, num_words, 3);
@@ -1157,6 +1160,9 @@ PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, size_t num_words
1160
1161 PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, size_t num_words, void *user)
1162 {
1163 + if(((PARSER_USER_OBJECT *) user)->replay.rset_enabled == false)
1164 + return PARSER_RC_OK;
1165 +
1166 char *last_collected_ut_str = get_word(words, num_words, 1);
1167 char *last_updated_ut_str = get_word(words, num_words, 2);
1168
daemon/event_loop.c
+39 -24
@@ -13,31 +13,46 @@ void register_libuv_worker_jobs() {
13 registered = true;
14
15 worker_register("LIBUV");
16 - worker_register_job_name(UV_EVENT_PAGE_LOOKUP, "page lookup");
17 - worker_register_job_name(UV_EVENT_PAGE_POPULATION, "populate page");
18 - worker_register_job_name(UV_EVENT_EXT_DECOMPRESSION, "extent decompression");
19 - worker_register_job_name(UV_EVENT_METADATA_STORE, "store host metadata");
20 - worker_register_job_name(UV_EVENT_JOURNAL_INDEX_WAIT, "journal v2 wait");
21 - worker_register_job_name(UV_EVENT_JOURNAL_INDEX, "journal v2 indexing");
22 - worker_register_job_name(UV_EVENT_SCHEDULE_CMD, "schedule command");
23 - worker_register_job_name(UV_EVENT_METADATA_CLEANUP, "metadata cleanup");
24 - worker_register_job_name(UV_EVENT_EXTENT_CACHE, "extent cache");
25 - worker_register_job_name(UV_EVENT_EXTENT_MMAP, "extent mmap");
26 - worker_register_job_name(UV_EVENT_FLUSH_MAIN, "flush main");
27 - worker_register_job_name(UV_EVENT_EVICT_MAIN, "evict main");
28 - worker_register_job_name(UV_EVENT_ANALYZE_V2, "analyze journalfile");
29 - worker_register_job_name(UV_EVENT_RETENTION_V2, "calculate retention");
30 - worker_register_job_name(UV_EVENT_RETENTION_UPDATE, "update retention");
31 - worker_register_job_name(UV_EVENT_DATAFILE_ACQUIRE, "datafile acquire");
32 - worker_register_job_name(UV_EVENT_DATAFILE_DELETE, "datafile deletion");
33 - worker_register_job_name(UV_EVENT_FLUSHED_TO_OPEN, "flushed to open");
34 - worker_register_job_name(UV_EVENT_PREP_QUERY, "prep query");
16 +
17 + // generic
18 worker_register_job_name(UV_EVENT_WORKER_INIT, "worker init");
36 - worker_register_job_name(UV_EVENT_FLUSH_PAGES, "flush pages");
37 - worker_register_job_name(UV_EVENT_BUFFERS_CLEANUP, "buffers cleanup");
38 - worker_register_job_name(UV_EVENT_QUIESCE, "quiesce");
39 - worker_register_job_name(UV_EVENT_POPULATE_MRG, "populate mrg");
40 - worker_register_job_name(UV_EVENT_SHUTDOWN, "shutdown");
19 +
20 + // query related
21 + worker_register_job_name(UV_EVENT_DBENGINE_QUERY, "query");
22 + worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_CACHE_LOOKUP, "extent cache");
23 + worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_MMAP, "extent mmap");
24 + worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_DECOMPRESSION, "extent decompression");
25 + worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_PAGE_LOOKUP, "page lookup");
26 + worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_PAGE_POPULATION, "page populate");
27 +
28 + // flushing related
29 + worker_register_job_name(UV_EVENT_DBENGINE_FLUSH_MAIN_CACHE, "flush main");
30 + worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_WRITE, "extent write");
31 + worker_register_job_name(UV_EVENT_DBENGINE_FLUSHED_TO_OPEN, "flushed to open");
32 +
33 + // datafile full
34 + worker_register_job_name(UV_EVENT_DBENGINE_JOURNAL_INDEX_WAIT, "jv2 index wait");
35 + worker_register_job_name(UV_EVENT_DBENGINE_JOURNAL_INDEX, "jv2 indexing");
36 +
37 + // db rotation related
38 + worker_register_job_name(UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT, "datafile delete wait");
39 + worker_register_job_name(UV_EVENT_DBENGINE_DATAFILE_DELETE, "datafile deletion");
40 + worker_register_job_name(UV_EVENT_DBENGINE_FIND_ROTATED_METRICS, "find rotated metrics");
41 + worker_register_job_name(UV_EVENT_DBENGINE_FIND_REMAINING_RETENTION, "find remaining retention");
42 + worker_register_job_name(UV_EVENT_DBENGINE_POPULATE_MRG, "update retention");
43 +
44 + // other dbengine events
45 + worker_register_job_name(UV_EVENT_DBENGINE_EVICT_MAIN_CACHE, "evict main");
46 + worker_register_job_name(UV_EVENT_DBENGINE_BUFFERS_CLEANUP, "dbengine buffers cleanup");
47 + worker_register_job_name(UV_EVENT_DBENGINE_QUIESCE, "dbengine quiesce");
48 + worker_register_job_name(UV_EVENT_DBENGINE_SHUTDOWN, "dbengine shutdown");
49 +
50 + // metadata
51 + worker_register_job_name(UV_EVENT_METADATA_STORE, "metadata store host");
52 + worker_register_job_name(UV_EVENT_METADATA_CLEANUP, "metadata cleanup");
53 +
54 + // netdatacli
55 + worker_register_job_name(UV_EVENT_SCHEDULE_CMD, "schedule command");
56
57 uv_thread_set_name_np(pthread_self(), "LIBUV_WORKER");
58 }
daemon/event_loop.h
+38 -23
@@ -5,31 +5,46 @@
5
6 enum event_loop_job {
7 UV_EVENT_JOB_NONE = 0,
8 - UV_EVENT_EXT_DECOMPRESSION,
9 - UV_EVENT_PAGE_LOOKUP,
10 - UV_EVENT_PAGE_POPULATION,
8 +
9 + // generic
10 + UV_EVENT_WORKER_INIT,
11 +
12 + // query related
13 + UV_EVENT_DBENGINE_QUERY,
14 + UV_EVENT_DBENGINE_EXTENT_CACHE_LOOKUP,
15 + UV_EVENT_DBENGINE_EXTENT_MMAP,
16 + UV_EVENT_DBENGINE_EXTENT_DECOMPRESSION,
17 + UV_EVENT_DBENGINE_EXTENT_PAGE_LOOKUP,
18 + UV_EVENT_DBENGINE_EXTENT_PAGE_POPULATION,
19 +
20 + // flushing related
21 + UV_EVENT_DBENGINE_FLUSH_MAIN_CACHE,
22 + UV_EVENT_DBENGINE_EXTENT_WRITE,
23 + UV_EVENT_DBENGINE_FLUSHED_TO_OPEN,
24 +
25 + // datafile full
26 + UV_EVENT_DBENGINE_JOURNAL_INDEX_WAIT,
27 + UV_EVENT_DBENGINE_JOURNAL_INDEX,
28 +
29 + // db rotation related
30 + UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT,
31 + UV_EVENT_DBENGINE_DATAFILE_DELETE,
32 + UV_EVENT_DBENGINE_FIND_ROTATED_METRICS, // find the metrics that are rotated
33 + UV_EVENT_DBENGINE_FIND_REMAINING_RETENTION, // find their remaining retention
34 + UV_EVENT_DBENGINE_POPULATE_MRG, // update mrg
35 +
36 + // other dbengine events
37 + UV_EVENT_DBENGINE_EVICT_MAIN_CACHE,
38 + UV_EVENT_DBENGINE_BUFFERS_CLEANUP,
39 + UV_EVENT_DBENGINE_QUIESCE,
40 + UV_EVENT_DBENGINE_SHUTDOWN,
41 +
42 + // metadata
43 UV_EVENT_METADATA_STORE,
12 - UV_EVENT_JOURNAL_INDEX_WAIT,
13 - UV_EVENT_JOURNAL_INDEX,
14 - UV_EVENT_SCHEDULE_CMD,
44 UV_EVENT_METADATA_CLEANUP,
16 - UV_EVENT_EXTENT_CACHE,
17 - UV_EVENT_EXTENT_MMAP,
18 - UV_EVENT_FLUSH_MAIN,
19 - UV_EVENT_EVICT_MAIN,
20 - UV_EVENT_ANALYZE_V2,
21 - UV_EVENT_RETENTION_V2,
22 - UV_EVENT_RETENTION_UPDATE,
23 - UV_EVENT_DATAFILE_ACQUIRE,
24 - UV_EVENT_DATAFILE_DELETE,
25 - UV_EVENT_FLUSHED_TO_OPEN,
26 - UV_EVENT_PREP_QUERY,
27 - UV_EVENT_WORKER_INIT,
28 - UV_EVENT_FLUSH_PAGES,
29 - UV_EVENT_BUFFERS_CLEANUP,
30 - UV_EVENT_QUIESCE,
31 - UV_EVENT_POPULATE_MRG,
32 - UV_EVENT_SHUTDOWN,
45 +
46 + // netdatacli
47 + UV_EVENT_SCHEDULE_CMD,
48 };
49
50 void register_libuv_worker_jobs();
daemon/global_statistics.c
+4 -1
@@ -1734,7 +1734,7 @@ static void dbengine2_statistics_charts(void) {
1734
1735 struct rrdeng_buffer_sizes buffers = rrdeng_get_buffer_sizes();
1736 size_t buffers_total_size = buffers.handles + buffers.xt_buf + buffers.xt_io + buffers.pdc + buffers.descriptors +
1737 - buffers.opcodes + buffers.wal + buffers.workers + buffers.epdl + buffers.deol + buffers.pd;
1737 + buffers.opcodes + buffers.wal + buffers.workers + buffers.epdl + buffers.deol + buffers.pd + buffers.pages;
1738
1739 #ifdef PDC_USE_JULYL
1740 buffers_total_size += buffers.julyl;
@@ -1798,6 +1798,7 @@ static void dbengine2_statistics_charts(void) {
1798 static RRDDIM *rd_pgc_buffers_epdl = NULL;
1799 static RRDDIM *rd_pgc_buffers_deol = NULL;
1800 static RRDDIM *rd_pgc_buffers_pd = NULL;
1801 + static RRDDIM *rd_pgc_buffers_pages = NULL;
1802 #ifdef PDC_USE_JULYL
1803 static RRDDIM *rd_pgc_buffers_julyl = NULL;
1804 #endif
@@ -1824,6 +1825,7 @@ static void dbengine2_statistics_charts(void) {
1825 rd_pgc_buffers_workers = rrddim_add(st_pgc_buffers, "workers", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1826 rd_pgc_buffers_pdc = rrddim_add(st_pgc_buffers, "pdc", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1827 rd_pgc_buffers_pd = rrddim_add(st_pgc_buffers, "pd", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1828 + rd_pgc_buffers_pages = rrddim_add(st_pgc_buffers, "pages", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1829 rd_pgc_buffers_xt_io = rrddim_add(st_pgc_buffers, "extent io", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1830 rd_pgc_buffers_xt_buf = rrddim_add(st_pgc_buffers, "extent buffers", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1831 rd_pgc_buffers_epdl = rrddim_add(st_pgc_buffers, "epdl", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
@@ -1841,6 +1843,7 @@ static void dbengine2_statistics_charts(void) {
1843 rrddim_set_by_pointer(st_pgc_buffers, rd_pgc_buffers_workers, (collected_number)buffers.workers);
1844 rrddim_set_by_pointer(st_pgc_buffers, rd_pgc_buffers_pdc, (collected_number)buffers.pdc);
1845 rrddim_set_by_pointer(st_pgc_buffers, rd_pgc_buffers_pd, (collected_number)buffers.pd);
1846 + rrddim_set_by_pointer(st_pgc_buffers, rd_pgc_buffers_pages, (collected_number)buffers.pages);
1847 rrddim_set_by_pointer(st_pgc_buffers, rd_pgc_buffers_xt_io, (collected_number)buffers.xt_io);
1848 rrddim_set_by_pointer(st_pgc_buffers, rd_pgc_buffers_xt_buf, (collected_number)buffers.xt_buf);
1849 rrddim_set_by_pointer(st_pgc_buffers, rd_pgc_buffers_epdl, (collected_number)buffers.epdl);
daemon/main.c
+19
@@ -428,6 +428,25 @@ void netdata_cleanup_and_exit(int ret) {
428
429 #ifdef ENABLE_DBENGINE
430 if(dbengine_enabled) {
431 + delta_shutdown_time("wait for dbengine collectors to finish");
432 +
433 + size_t running = 1;
434 + while(running) {
435 + running = 0;
436 + for (size_t tier = 0; tier < storage_tiers; tier++)
437 + running += rrdeng_collectors_running(multidb_ctx[tier]);
438 +
439 + if(running)
440 + sleep_usec(100 * USEC_PER_MS);
441 + }
442 +
443 + delta_shutdown_time("wait for dbengine main cache to finish flushing");
444 +
445 + while (pgc_hot_and_dirty_entries(main_cache)) {
446 + pgc_flush_all_hot_and_dirty_pages(main_cache, PGC_SECTION_ALL);
447 + sleep_usec(100 * USEC_PER_MS);
448 + }
449 +
450 delta_shutdown_time("stop dbengine tiers");
451 for (size_t tier = 0; tier < storage_tiers; tier++)
452 rrdeng_exit(multidb_ctx[tier]);
database/engine/cache.c
+20 -3
@@ -625,6 +625,9 @@ static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lo
625 return;
626 }
627
628 + __atomic_add_fetch(&cache->stats.hot2dirty_entries, 1, __ATOMIC_RELAXED);
629 + __atomic_add_fetch(&cache->stats.hot2dirty_size, page->assumed_size, __ATOMIC_RELAXED);
630 +
631 if(likely(flags & PGC_PAGE_HOT))
632 pgc_ll_del(cache, &cache->hot, page, true);
633
@@ -638,6 +641,9 @@ static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lo
641 // first add to linked list, the set the flag (required for move_page_last())
642 pgc_ll_add(cache, &cache->dirty, page, false);
643
644 + __atomic_sub_fetch(&cache->stats.hot2dirty_entries, 1, __ATOMIC_RELAXED);
645 + __atomic_sub_fetch(&cache->stats.hot2dirty_size, page->assumed_size, __ATOMIC_RELAXED);
646 +
647 page_transition_unlock(cache, page);
648 }
649
@@ -1592,12 +1598,12 @@ static bool flush_pages(PGC *cache, size_t max_flushes, Word_t section, bool wai
1598 internal_fatal(page_get_status_flags(tpg) != PGC_PAGE_DIRTY,
1599 "DBENGINE CACHE: page should be in the dirty list before saved");
1600
1595 - // remove it from the dirty list
1596 - pgc_ll_del(cache, &cache->dirty, tpg, true);
1597 -
1601 __atomic_add_fetch(&cache->stats.flushing_entries, 1, __ATOMIC_RELAXED);
1602 __atomic_add_fetch(&cache->stats.flushing_size, tpg->assumed_size, __ATOMIC_RELAXED);
1603
1604 + // remove it from the dirty list
1605 + pgc_ll_del(cache, &cache->dirty, tpg, true);
1606 +
1607 pages_removed_dirty_size += tpg->assumed_size;
1608 pages_removed_dirty++;
1609 }
@@ -1994,6 +2000,17 @@ struct pgc_statistics pgc_get_statistics(PGC *cache) {
2000 return cache->stats;
2001 }
2002
2003 +size_t pgc_hot_and_dirty_entries(PGC *cache) {
2004 + size_t entries = 0;
2005 +
2006 + entries += __atomic_load_n(&cache->hot.stats->entries, __ATOMIC_RELAXED);
2007 + entries += __atomic_load_n(&cache->dirty.stats->entries, __ATOMIC_RELAXED);
2008 + entries += __atomic_load_n(&cache->stats.flushing_entries, __ATOMIC_RELAXED);
2009 + entries += __atomic_load_n(&cache->stats.hot2dirty_entries, __ATOMIC_RELAXED);
2010 +
2011 + return entries;
2012 +}
2013 +
2014 void pgc_open_cache_to_journal_v2(PGC *cache, Word_t section, unsigned datafile_fileno, uint8_t type, migrate_to_v2_callback cb, void *data) {
2015 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.journal_v2_indexing_started, 1, __ATOMIC_RELAXED);
2016 __atomic_add_fetch(&cache->stats.workers_jv2_flush, 1, __ATOMIC_RELAXED);
database/engine/cache.h
+4
@@ -79,6 +79,9 @@ struct pgc_statistics {
79 size_t flushing_entries;
80 size_t flushing_size;
81
82 + size_t hot2dirty_entries;
83 + size_t hot2dirty_size;
84 +
85 PGC_CACHE_LINE_PADDING(4);
86
87 size_t acquires;
@@ -235,5 +238,6 @@ bool pgc_evict_pages(PGC *cache, size_t max_skip, size_t max_evict);
238 bool pgc_flush_pages(PGC *cache, size_t max_flushes);
239
240 struct pgc_statistics pgc_get_statistics(PGC *cache);
241 +size_t pgc_hot_and_dirty_entries(PGC *cache);
242
243 #endif // DBENGINE_CACHE_H
database/engine/datafile.c
+16 -14
@@ -536,8 +536,13 @@ int init_data_files(struct rrdengine_instance *ctx)
536 return ret;
537 }
538 }
539 - else if(ctx->loading.create_new_datafile_pair)
540 - create_new_datafile_pair(ctx);
539 + else {
540 + if (ctx->loading.create_new_datafile_pair)
541 + create_new_datafile_pair(ctx);
542 +
543 + while(rrdeng_ctx_exceeded_disk_quota(ctx))
544 + datafile_delete(ctx, ctx->datafiles.first, false, false);
545 + }
546
547 pgc_reset_hot_max(open_cache);
548 ctx->loading.create_new_datafile_pair = false;
@@ -548,22 +553,19 @@ void finalize_data_files(struct rrdengine_instance *ctx)
553 {
554 bool logged = false;
555
556 + logged = false;
557 + while(__atomic_load_n(&ctx->atomic.extents_currently_being_flushed, __ATOMIC_RELAXED)) {
558 + if(!logged) {
559 + info("Waiting for inflight flush to finish on tier %d...", ctx->config.tier);
560 + logged = true;
561 + }
562 + sleep_usec(100 * USEC_PER_MS);
563 + }
564 +
565 do {
566 struct rrdengine_datafile *datafile = ctx->datafiles.first;
567 struct rrdengine_journalfile *journalfile = datafile->journalfile;
568
555 - logged = false;
556 - if(datafile == ctx->datafiles.first->prev) {
557 - // this is the last file
558 - while(__atomic_load_n(&ctx->atomic.extents_currently_being_flushed, __ATOMIC_RELAXED)) {
559 - if(!logged) {
560 - info("Waiting for inflight flush to finish on tier %d to close last datafile %u...", ctx->config.tier, datafile->fileno);
561 - logged = true;
562 - }
563 - sleep_usec(100 * USEC_PER_MS);
564 - }
565 - }
566 -
569 logged = false;
570 size_t iterations = 100;
571 while(!datafile_acquire_for_deletion(datafile) && datafile != ctx->datafiles.first->prev && --iterations > 0) {
database/engine/metric.c
+13 -7
@@ -314,19 +314,25 @@ void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t
314 netdata_spinlock_unlock(&metric->spinlock);
315 }
316
317 -bool mrg_metric_set_first_time_s_if_zero(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
317 +bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
318 bool ret = false;
319
320 netdata_spinlock_lock(&metric->spinlock);
321 - if(!metric->first_time_s) {
321 + if(first_time_s > metric->first_time_s) {
322 metric->first_time_s = first_time_s;
323 + ret = true;
324 + }
325 + netdata_spinlock_unlock(&metric->spinlock);
326
324 -// if(unlikely(metric->latest_time_s_clean < metric->first_time_s))
325 -// metric->latest_time_s_clean = metric->first_time_s;
326 -//
327 -// if(unlikely(metric->latest_time_s_hot < metric->first_time_s))
328 -// metric->latest_time_s_hot = metric->first_time_s;
327 + return ret;
328 +}
329
330 +bool mrg_metric_set_first_time_s_if_zero(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
331 + bool ret = false;
332 +
333 + netdata_spinlock_lock(&metric->spinlock);
334 + if(!metric->first_time_s) {
335 + metric->first_time_s = first_time_s;
336 ret = true;
337 }
338 netdata_spinlock_unlock(&metric->spinlock);
database/engine/metric.h
+1
@@ -45,6 +45,7 @@ uuid_t *mrg_metric_uuid(MRG *mrg, METRIC *metric);
45 Word_t mrg_metric_section(MRG *mrg, METRIC *metric);
46
47 bool mrg_metric_set_first_time_s(MRG *mrg, METRIC *metric, time_t first_time_s);
48 +bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg, METRIC *metric, time_t first_time_s);
49 bool mrg_metric_set_first_time_s_if_zero(MRG *mrg, METRIC *metric, time_t first_time_s);
50 time_t mrg_metric_get_first_time_s(MRG *mrg, METRIC *metric);
51 void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s, time_t last_time_s, time_t update_every_s);
database/engine/pagecache.c
+58 -17
@@ -105,18 +105,21 @@ static inline struct page_details *pdc_find_page_for_time(
105 Pcvoid_t PArray,
106 time_t wanted_time_s,
107 size_t *gaps,
108 - PDC_PAGE_STATUS stop_at
108 + PDC_PAGE_STATUS mode,
109 + PDC_PAGE_STATUS skip_list
110 ) {
111 Word_t PIndexF = wanted_time_s, PIndexL = wanted_time_s;
112 Pvoid_t *PValueF, *PValueL;
113 struct page_details *pdF = NULL, *pdL = NULL;
114 bool firstF = true, firstL = true;
115
116 + PDC_PAGE_STATUS ignore_list = PDC_PAGE_QUERY_GLOBAL_SKIP_LIST | skip_list;
117 +
118 while ((PValueF = PDCJudyLFirstThenNext(PArray, &PIndexF, &firstF))) {
119 pdF = *PValueF;
120
121 PDC_PAGE_STATUS status = __atomic_load_n(&pdF->status, __ATOMIC_ACQUIRE);
119 - if (!(status & (PDC_PAGE_FAILED | PDC_PAGE_SKIP | PDC_PAGE_INVALID | PDC_PAGE_RELEASED | stop_at)))
122 + if (!(status & (ignore_list | mode)))
123 break;
124
125 pdF = NULL;
@@ -126,13 +129,14 @@ static inline struct page_details *pdc_find_page_for_time(
129 pdL = *PValueL;
130
131 PDC_PAGE_STATUS status = __atomic_load_n(&pdL->status, __ATOMIC_ACQUIRE);
129 - if(status & stop_at) {
130 - // don't go all the way back to the beginning - stop at the last processed
132 + if(status & mode) {
133 + // don't go all the way back to the beginning
134 + // stop at the last processed
135 pdL = NULL;
136 break;
137 }
138
135 - if (!(status & (PDC_PAGE_FAILED | PDC_PAGE_SKIP | PDC_PAGE_INVALID | PDC_PAGE_RELEASED)))
139 + if (!(status & ignore_list))
140 break;
141
142 pdL = NULL;
@@ -305,7 +309,14 @@ static size_t get_page_list_from_pgc(PGC *cache, METRIC *metric, struct rrdengin
309 pd->page_length = page_length;
310 pd->update_every_s = page_update_every_s;
311 pd->page = (open_cache_mode) ? NULL : page;
308 - pd->status |= ((pd->page) ? (PDC_PAGE_READY | PDC_PAGE_PRELOADED) : 0) | tags;
312 + pd->status |= tags;
313 +
314 + if((pd->page)) {
315 + pd->status |= PDC_PAGE_READY | PDC_PAGE_PRELOADED;
316 +
317 + if(pgc_page_data(page) == DBENGINE_EMPTY_PAGE)
318 + pd->status |= PDC_PAGE_EMPTY;
319 + }
320
321 if(open_cache_mode) {
322 struct rrdengine_datafile *datafile = pgc_page_data(page);
@@ -344,6 +355,22 @@ static size_t get_page_list_from_pgc(PGC *cache, METRIC *metric, struct rrdengin
355 return pages_found_in_cache;
356 }
357
358 +static void pgc_inject_gap(struct rrdengine_instance *ctx, METRIC *metric, time_t start_time_s, time_t end_time_s) {
359 + PGC_ENTRY page_entry = {
360 + .hot = false,
361 + .section = (Word_t)ctx,
362 + .metric_id = (Word_t)metric,
363 + .start_time_s = start_time_s,
364 + .end_time_s = end_time_s,
365 + .update_every_s = 0,
366 + .size = 0,
367 + .data = DBENGINE_EMPTY_PAGE,
368 + };
369 +
370 + PGC_PAGE *page = pgc_page_add_and_acquire(main_cache, page_entry, NULL);
371 + pgc_page_release(main_cache, page);
372 +}
373 +
374 static size_t list_has_time_gaps(
375 struct rrdengine_instance *ctx,
376 METRIC *metric,
@@ -354,7 +381,8 @@ static size_t list_has_time_gaps(
381 size_t *pages_found_pass4,
382 size_t *pages_pending,
383 size_t *pages_overlapping,
357 - time_t *optimal_end_time_s
384 + time_t *optimal_end_time_s,
385 + bool populate_gaps
386 ) {
387 // we will recalculate these, so zero them
388 *pages_pending = 0;
@@ -390,7 +418,7 @@ static size_t list_has_time_gaps(
418 size_t pages_pass2 = 0, pages_pass3 = 0;
419 while((pd = pdc_find_page_for_time(
420 JudyL_page_array, now_s, &gaps,
393 - PDC_PAGE_PREPROCESSED))) {
421 + PDC_PAGE_PREPROCESSED, 0))) {
422
423 pd->status |= PDC_PAGE_PREPROCESSED;
424 pages_pass2++;
@@ -398,6 +426,9 @@ static size_t list_has_time_gaps(
426 if(pd->update_every_s)
427 dt_s = pd->update_every_s;
428
429 + if(populate_gaps && pd->first_time_s > now_s)
430 + pgc_inject_gap(ctx, metric, now_s, pd->first_time_s);
431 +
432 now_s = pd->last_time_s + dt_s;
433 if(now_s > wanted_end_time_s) {
434 *optimal_end_time_s = pd->last_time_s;
@@ -405,6 +436,9 @@ static size_t list_has_time_gaps(
436 }
437 }
438
439 + if(populate_gaps && now_s < wanted_end_time_s)
440 + pgc_inject_gap(ctx, metric, now_s, wanted_end_time_s);
441 +
442 // ------------------------------------------------------------------------
443 // PASS 3: mark as skipped all the pages not useful
444
@@ -432,6 +466,10 @@ static size_t list_has_time_gaps(
466
467 pd->status &= ~PDC_PAGE_DISK_PENDING;
468 pd->status |= PDC_PAGE_READY | PDC_PAGE_PRELOADED | PDC_PAGE_PRELOADED_PASS4;
469 +
470 + if(pgc_page_data(pd->page) == DBENGINE_EMPTY_PAGE)
471 + pd->status |= PDC_PAGE_EMPTY;
472 +
473 }
474 else if(!(pd->status & PDC_PAGE_FAILED) && (pd->status & PDC_PAGE_DATAFILE_ACQUIRED)) {
475 (*pages_pending)++;
@@ -635,7 +673,7 @@ static Pvoid_t get_page_list(
673 if(pages_found_in_main_cache && !cache_gaps) {
674 query_gaps = list_has_time_gaps(ctx, metric, JudyL_page_array, wanted_start_time_s, wanted_end_time_s,
675 &pages_total, &pages_found_pass4, &pages_pending, &pages_overlapping,
638 - optimal_end_time_s);
676 + optimal_end_time_s, false);
677
678 if (pages_total && !query_gaps)
679 goto we_are_done;
@@ -657,7 +695,7 @@ static Pvoid_t get_page_list(
695 if(pages_found_in_open_cache) {
696 query_gaps = list_has_time_gaps(ctx, metric, JudyL_page_array, wanted_start_time_s, wanted_end_time_s,
697 &pages_total, &pages_found_pass4, &pages_pending, &pages_overlapping,
660 - optimal_end_time_s);
698 + optimal_end_time_s, false);
699
700 if (pages_total && !query_gaps)
701 goto we_are_done;
@@ -681,7 +719,7 @@ static Pvoid_t get_page_list(
719 pass4_ut = now_monotonic_usec();
720 query_gaps = list_has_time_gaps(ctx, metric, JudyL_page_array, wanted_start_time_s, wanted_end_time_s,
721 &pages_total, &pages_found_pass4, &pages_pending, &pages_overlapping,
684 - optimal_end_time_s);
722 + optimal_end_time_s, true);
723
724 we_are_done:
725
@@ -814,7 +852,7 @@ struct pgc_page *pg_cache_lookup_next(
852 preloaded = false;
853 struct page_details *pd = pdc_find_page_for_time(
854 pdc->page_list_JudyL, now_s, &gaps,
817 - PDC_PAGE_PROCESSED);
855 + PDC_PAGE_PROCESSED, PDC_PAGE_EMPTY);
856
857 if (!pd)
858 break;
@@ -839,14 +877,17 @@ struct pgc_page *pg_cache_lookup_next(
877 preloaded = pdc_page_status_check(pd, PDC_PAGE_PRELOADED);
878 waited = true;
879 }
880 + }
881
843 - if(!page || pdc_page_status_check(pd, PDC_PAGE_FAILED | PDC_PAGE_SKIP | PDC_PAGE_INVALID)) {
844 - page = NULL;
845 - continue;
846 - }
882 + if(page && pgc_page_data(page) == DBENGINE_EMPTY_PAGE)
883 + pdc_page_status_set(pd, PDC_PAGE_EMPTY);
884 +
885 + if(!page || pdc_page_status_check(pd, PDC_PAGE_QUERY_GLOBAL_SKIP_LIST | PDC_PAGE_EMPTY)) {
886 + page = NULL;
887 + continue;
888 }
889
849 - // we now have page
890 + // we now have page and is not empty
891
892 time_t page_start_time_s = pgc_page_start_time_s(page);
893 time_t page_end_time_s = pgc_page_end_time_s(page);
database/engine/pdc.c
+29 -81
@@ -790,43 +790,6 @@ void pdc_to_epdl_router(struct rrdengine_instance *ctx, PDC *pdc, execute_extent
790 pdc_release_and_destroy_if_unreferenced(pdc, true, true);
791 }
792
793 -static void fill_page_with_nulls(void *page, uint32_t page_length, uint8_t type) {
794 - switch(type) {
795 - case PAGE_METRICS: {
796 - storage_number n = pack_storage_number(NAN, SN_FLAG_NONE);
797 - storage_number *array = (storage_number *)page;
798 - size_t slots = page_length / sizeof(n);
799 - for(size_t i = 0; i < slots ; i++)
800 - array[i] = n;
801 - }
802 - break;
803 -
804 - case PAGE_TIER: {
805 - storage_number_tier1_t n = {
806 - .min_value = NAN,
807 - .max_value = NAN,
808 - .sum_value = NAN,
809 - .count = 1,
810 - .anomaly_count = 0,
811 - };
812 - storage_number_tier1_t *array = (storage_number_tier1_t *)page;
813 - size_t slots = page_length / sizeof(n);
814 - for(size_t i = 0; i < slots ; i++)
815 - array[i] = n;
816 - }
817 - break;
818 -
819 - default: {
820 - static bool logged = false;
821 - if(!logged) {
822 - error("DBENGINE: cannot fill page with nulls on unknown page type id %d", type);
823 - logged = true;
824 - }
825 - memset(page, 0, page_length);
826 - }
827 - }
828 -}
829 -
793 void collect_page_flags_to_buffer(BUFFER *wb, RRDENG_COLLECT_PAGE_FLAGS flags) {
794 if(flags & RRDENG_PAGE_PAST_COLLECTION)
795 buffer_strcat(wb, "PAST_COLLECTION ");
@@ -870,7 +833,6 @@ inline VALIDATED_PAGE_DESCRIPTOR validate_extent_page_descr(const struct rrdeng_
833 now_s,
834 overwrite_zero_update_every_s,
835 have_read_error,
873 - true,
836 "loaded", 0);
837 }
838
@@ -885,7 +847,6 @@ VALIDATED_PAGE_DESCRIPTOR validate_page(
847 time_t now_s, // can be zero, to disable future timestamp check
848 time_t overwrite_zero_update_every_s, // can be zero, if unknown
849 bool have_read_error,
888 - bool minimize_invalid_size,
850 const char *msg,
851 RRDENG_COLLECT_PAGE_FLAGS flags) {
852
@@ -988,21 +949,6 @@ VALIDATED_PAGE_DESCRIPTOR validate_page(
949 uuid_str, msg, vd.type,
950 vd.start_time_s, vd.end_time_s, now_s, vd.update_every_s, vd.page_length, vd.entries, wb?buffer_tostring(wb):""
951 );
991 -
992 - if(minimize_invalid_size) {
993 - // since the page is going to be loaded
994 - // let's minimize the memory the invalid page will occupy
995 -
996 - if(vd.start_time_s == vd.end_time_s) {
997 - vd.page_length = vd.point_size;
998 - vd.entries = 1;
999 - }
1000 - else {
1001 - vd.page_length = vd.point_size * 2;
1002 - vd.update_every_s = vd.end_time_s - vd.start_time_s;
1003 - vd.entries = 2;
1004 - }
1005 - }
952 }
953 else {
954 const char *err_valid = (vd.is_valid) ? "" : "found invalid, ";
@@ -1138,7 +1084,7 @@ static bool epdl_populate_pages_from_extent_data(
1084 }
1085
1086 if(worker)
1141 - worker_is_busy(UV_EVENT_EXT_DECOMPRESSION);
1087 + worker_is_busy(UV_EVENT_DBENGINE_EXTENT_DECOMPRESSION);
1088
1089 if (likely(!have_read_error && RRD_NO_COMPRESSION != header->compression_algorithm)) {
1090 // find the uncompressed extent size
@@ -1169,7 +1115,7 @@ static bool epdl_populate_pages_from_extent_data(
1115 }
1116
1117 if(worker)
1172 - worker_is_busy(UV_EVENT_PAGE_LOOKUP);
1118 + worker_is_busy(UV_EVENT_DBENGINE_EXTENT_PAGE_LOOKUP);
1119
1120 size_t stats_data_from_main_cache = 0;
1121 size_t stats_data_from_extent = 0;
@@ -1211,33 +1157,35 @@ static bool epdl_populate_pages_from_extent_data(
1157 have_read_error);
1158
1159 if(worker)
1214 - worker_is_busy(UV_EVENT_PAGE_POPULATION);
1160 + worker_is_busy(UV_EVENT_DBENGINE_EXTENT_PAGE_POPULATION);
1161
1216 - void *page_data = dbengine_page_alloc(vd.page_length);
1162 + void *page_data;
1163
1164 if (unlikely(!vd.is_valid)) {
1219 - fill_page_with_nulls(page_data, vd.page_length, vd.type);
1165 + page_data = DBENGINE_EMPTY_PAGE;
1166 stats_load_invalid_page++;
1167 }
1222 -
1223 - else if (RRD_NO_COMPRESSION == header->compression_algorithm) {
1224 - memcpy(page_data, data + payload_offset + page_offset, (size_t) vd.page_length);
1225 - stats_load_uncompressed++;
1226 - }
1227 -
1168 else {
1229 - if(unlikely(page_offset + vd.page_length > uncompressed_payload_length)) {
1230 - error_limit_static_global_var(erl, 10, 0);
1231 - error_limit(&erl,
1232 - "DBENGINE: page %u offset %u + page length %zu exceeds the uncompressed buffer size %u",
1233 - i, page_offset, vd.page_length, uncompressed_payload_length);
1234 -
1235 - fill_page_with_nulls(page_data, vd.page_length, vd.type);
1236 - stats_load_invalid_page++;
1169 + if (RRD_NO_COMPRESSION == header->compression_algorithm) {
1170 + page_data = dbengine_page_alloc(vd.page_length);
1171 + memcpy(page_data, data + payload_offset + page_offset, (size_t) vd.page_length);
1172 + stats_load_uncompressed++;
1173 }
1174 else {
1239 - memcpy(page_data, uncompressed_buf + page_offset, vd.page_length);
1240 - stats_load_compressed++;
1175 + if (unlikely(page_offset + vd.page_length > uncompressed_payload_length)) {
1176 + error_limit_static_global_var(erl, 10, 0);
1177 + error_limit(&erl,
1178 + "DBENGINE: page %u offset %u + page length %zu exceeds the uncompressed buffer size %u",
1179 + i, page_offset, vd.page_length, uncompressed_payload_length);
1180 +
1181 + page_data = DBENGINE_EMPTY_PAGE;
1182 + stats_load_invalid_page++;
1183 + }
1184 + else {
1185 + page_data = dbengine_page_alloc(vd.page_length);
1186 + memcpy(page_data, uncompressed_buf + page_offset, vd.page_length);
1187 + stats_load_compressed++;
1188 + }
1189 }
1190 }
1191
@@ -1248,7 +1196,7 @@ static bool epdl_populate_pages_from_extent_data(
1196 .start_time_s = vd.start_time_s,
1197 .end_time_s = vd.end_time_s,
1198 .update_every_s = vd.update_every_s,
1251 - .size = (size_t) vd.page_length,
1199 + .size = (size_t) ((page_data == DBENGINE_EMPTY_PAGE) ? 0 : vd.page_length),
1200 .data = page_data
1201 };
1202
@@ -1269,13 +1217,13 @@ static bool epdl_populate_pages_from_extent_data(
1217
1218 pd->page = page;
1219 pd->page_length = pgc_page_data_size(main_cache, page);
1272 - pdc_page_status_set(pd, PDC_PAGE_READY | tags);
1220 + pdc_page_status_set(pd, PDC_PAGE_READY | tags | ((page_data == DBENGINE_EMPTY_PAGE) ? PDC_PAGE_EMPTY : 0));
1221
1222 pd = pd->load.next;
1223 } while(pd);
1224
1225 if(worker)
1278 - worker_is_busy(UV_EVENT_PAGE_LOOKUP);
1226 + worker_is_busy(UV_EVENT_DBENGINE_EXTENT_PAGE_LOOKUP);
1227 }
1228
1229 if(stats_data_from_main_cache)
@@ -1332,7 +1280,7 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1280 }
1281
1282 if(worker)
1335 - worker_is_busy(UV_EVENT_EXTENT_CACHE);
1283 + worker_is_busy(UV_EVENT_DBENGINE_EXTENT_CACHE_LOOKUP);
1284
1285 bool extent_found_in_cache = false;
1286
@@ -1353,7 +1301,7 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1301 }
1302 else {
1303 if(worker)
1356 - worker_is_busy(UV_EVENT_EXTENT_MMAP);
1304 + worker_is_busy(UV_EVENT_DBENGINE_EXTENT_MMAP);
1305
1306 off_t map_start = ALIGN_BYTES_FLOOR(epdl->extent_offset);
1307 size_t length = ALIGN_BYTES_CEILING(epdl->extent_offset + epdl->extent_size) - map_start;
@@ -1369,7 +1317,7 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1317 fatal_assert(0 == ret);
1318
1319 if(worker)
1372 - worker_is_busy(UV_EVENT_EXTENT_CACHE);
1320 + worker_is_busy(UV_EVENT_DBENGINE_EXTENT_CACHE_LOOKUP);
1321
1322 bool added = false;
1323 extent_cache_page = pgc_page_add_and_acquire(extent_cache, (PGC_ENTRY) {
database/engine/rrdengine.c
+180 -28
@@ -754,15 +754,157 @@ static inline struct rrdeng_cmd rrdeng_deq_cmd(void) {
754
755 // ----------------------------------------------------------------------------
756
757 +#define MAX_PAGE_SIZES_TO_KEEP 3
758 +#define MIN_PAGES_PER_SIZE_TO_KEEP 100
759 +
760 +struct dbengine_page_size {
761 + SPINLOCK spinlock;
762 + size_t page_size; // read-only, no lock required to read it
763 +
764 + size_t demand;
765 + size_t supply;
766 +
767 + size_t hit;
768 + size_t miss;
769 +
770 + size_t used;
771 + size_t array_size;
772 + void **array;
773 +};
774 +
775 +struct {
776 + struct {
777 + size_t hit;
778 + size_t miss_wrong_size;
779 + size_t miss_short_supply;
780 + size_t cached_size;
781 + size_t struct_size;
782 + } atomic;
783 +
784 + struct dbengine_page_size slots[MAX_PAGE_SIZES_TO_KEEP];
785 +} dbengine_page_alloc_globals = {
786 + .atomic = {
787 + .struct_size = sizeof(dbengine_page_alloc_globals),
788 + }
789 +};
790 +
791 +__attribute__((constructor)) void initialize_sizes_to_slots(void) {
792 + uint8_t found[RRDENG_BLOCK_SIZE + 1];
793 + memset(found, 0, RRDENG_BLOCK_SIZE + 1);
794 +
795 + for(int i = 0; i < MAX_PAGE_SIZES_TO_KEEP ; i++) {
796 + struct dbengine_page_size *dps = &dbengine_page_alloc_globals.slots[i];
797 + memset(dps, 0, sizeof(struct dbengine_page_size));
798 + netdata_spinlock_init(&dps->spinlock);
799 + }
800 +
801 + for(int tier = 0; tier < MAX_PAGE_SIZES_TO_KEEP && tier < RRD_STORAGE_TIERS ; tier++) {
802 + size_t size = tier_page_size[tier];
803 +
804 + if(size <= RRDENG_BLOCK_SIZE && !found[size]) {
805 + struct dbengine_page_size *dps = &dbengine_page_alloc_globals.slots[tier];
806 + dps->page_size = size;
807 + found[size] = 1;
808 + }
809 + }
810 +}
811 +
812 +static inline struct dbengine_page_size *page_size_lookup(size_t size) {
813 + for(int i = 0; i < MAX_PAGE_SIZES_TO_KEEP ; i++) {
814 + if(size == dbengine_page_alloc_globals.slots[i].page_size)
815 + return &dbengine_page_alloc_globals.slots[i];
816 + }
817 + return NULL;
818 +}
819 +
820 +static void dbengine_page_alloc_cleanup1(void) {
821 + for(int i = 0; i < MAX_PAGE_SIZES_TO_KEEP ; i++) {
822 + void *page = NULL;
823 +
824 + struct dbengine_page_size *dps = &dbengine_page_alloc_globals.slots[i];
825 + netdata_spinlock_lock(&dps->spinlock);
826 + if(dps->used > MIN_PAGES_PER_SIZE_TO_KEEP) {
827 + dps->used--;
828 + internal_fatal(!dps->array[dps->used], "DBENGINE: slot should have a page but is empty");
829 + page = dps->array[dps->used];
830 + dps->array[dps->used] = NULL;
831 + __atomic_sub_fetch(&dbengine_page_alloc_globals.atomic.cached_size, dps->page_size, __ATOMIC_RELAXED);
832 + }
833 + netdata_spinlock_unlock(&dps->spinlock);
834 +
835 + if(page)
836 + freez(page);
837 + }
838 +}
839 +
840 void *dbengine_page_alloc(size_t size) {
758 - void *page = mallocz(size);
841 + void *page = NULL;
842 +
843 + struct dbengine_page_size *dps = page_size_lookup(size);
844 + if(dps) {
845 + netdata_spinlock_lock(&dps->spinlock);
846 + dps->demand++;
847 +
848 + if(dps->used > 0) {
849 + dps->hit++;
850 + dps->used--;
851 + internal_fatal(!dps->array[dps->used], "DBENGINE: slot should have a page but is empty");
852 + page = dps->array[dps->used];
853 + dps->array[dps->used] = NULL;
854 + __atomic_add_fetch(&dbengine_page_alloc_globals.atomic.hit, 1, __ATOMIC_RELAXED);
855 + __atomic_sub_fetch(&dbengine_page_alloc_globals.atomic.cached_size, dps->page_size, __ATOMIC_RELAXED);
856 + }
857 + else {
858 + dps->miss++;
859 + __atomic_add_fetch(&dbengine_page_alloc_globals.atomic.miss_short_supply, 1, __ATOMIC_RELAXED);
860 + }
861 +
862 + netdata_spinlock_unlock(&dps->spinlock);
863 + }
864 + else
865 + __atomic_add_fetch(&dbengine_page_alloc_globals.atomic.miss_wrong_size, 1, __ATOMIC_RELAXED);
866 +
867 + if(!page)
868 + page = mallocz(size);
869 +
870 return page;
871 }
872
873 void dbengine_page_free(void *page, size_t size __maybe_unused) {
763 - freez(page);
874 + if(unlikely(!page || page == DBENGINE_EMPTY_PAGE))
875 + return;
876 +
877 + struct dbengine_page_size *dps = page_size_lookup(size);
878 + if(dps) {
879 + netdata_spinlock_lock(&dps->spinlock);
880 + dps->supply++;
881 +
882 + if(dps->used == dps->array_size) {
883 + size_t new_array_size = dps->array_size ? dps->array_size * 2 : MIN_PAGES_PER_SIZE_TO_KEEP;
884 + dps->array = reallocz(dps->array, new_array_size * sizeof(void *));
885 +
886 + __atomic_add_fetch(&dbengine_page_alloc_globals.atomic.struct_size,
887 + (new_array_size - dps->array_size) * sizeof(void *), __ATOMIC_RELAXED);
888 +
889 + dps->array_size = new_array_size;
890 + }
891 +
892 + if(dps->used < dps->array_size) {
893 + dps->array[dps->used] = page;
894 + dps->used++;
895 + page = NULL;
896 + __atomic_add_fetch(&dbengine_page_alloc_globals.atomic.cached_size, dps->page_size, __ATOMIC_RELAXED);
897 + }
898 +
899 + netdata_spinlock_unlock(&dps->spinlock);
900 + }
901 +
902 + if(page)
903 + freez(page);
904 }
905
906 +// ----------------------------------------------------------------------------
907 +
908 void *dbengine_extent_alloc(size_t size) {
909 void *extent = mallocz(size);
910 return extent;
@@ -818,7 +960,7 @@ static void after_extent_flushed_to_open(struct rrdengine_instance *ctx __maybe_
960 }
961
962 static void *extent_flushed_to_open_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
821 - worker_is_busy(UV_EVENT_FLUSHED_TO_OPEN);
963 + worker_is_busy(UV_EVENT_DBENGINE_FLUSHED_TO_OPEN);
964
965 uv_fs_t *uv_fs_request = data;
966 struct extent_io_descriptor *xt_io_descr = uv_fs_request->data;
@@ -1099,7 +1241,7 @@ static void after_extent_write(struct rrdengine_instance *ctx __maybe_unused, vo
1241 }
1242
1243 static void *extent_write_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1102 - worker_is_busy(UV_EVENT_FLUSH_PAGES);
1244 + worker_is_busy(UV_EVENT_DBENGINE_EXTENT_WRITE);
1245 struct page_descr_with_data *base = data;
1246 struct extent_io_descriptor *xt_io_descr = datafile_extent_build(ctx, base, completion);
1247 return xt_io_descr;
@@ -1212,7 +1354,7 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1354 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.metrics_retention_started, 1, __ATOMIC_RELAXED);
1355
1356 if(worker)
1215 - worker_is_busy(UV_EVENT_ANALYZE_V2);
1357 + worker_is_busy(UV_EVENT_DBENGINE_FIND_ROTATED_METRICS);
1358
1359 struct rrdengine_journalfile *journalfile = datafile_to_delete->journalfile;
1360 struct journal_v2_header *j2_header = journalfile_v2_data_acquire(journalfile, NULL, 0, 0);
@@ -1233,8 +1375,8 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1375 if (!*PValue) {
1376 uuid_first_t_entry = mallocz(sizeof(*uuid_first_t_entry));
1377 uuid_first_t_entry->metric = metric;
1236 - uuid_first_t_entry->first_time_s = mrg_metric_get_first_time_s(main_mrg, metric);
1237 - uuid_first_t_entry->last_time_s = mrg_metric_get_latest_time_s(main_mrg, metric);
1378 + uuid_first_t_entry->first_time_s = LONG_MAX;
1379 + uuid_first_t_entry->last_time_s = 0;
1380 uuid_first_t_entry->uuid = mrg_metric_uuid(main_mrg, metric);
1381 *PValue = uuid_first_t_entry;
1382 count++;
@@ -1242,17 +1384,17 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1384 }
1385 journalfile_v2_data_release(journalfile);
1386
1245 - info("DBENGINE: recalculating retention for %u metrics", count);
1387 + info("DBENGINE: recalculating retention for %u metrics starting with datafile %u", count, first_datafile_remaining->fileno);
1388
1389 // Update the first time / last time for all metrics we plan to delete
1390
1391 if(worker)
1250 - worker_is_busy(UV_EVENT_RETENTION_V2);
1392 + worker_is_busy(UV_EVENT_DBENGINE_FIND_REMAINING_RETENTION);
1393
1394 find_uuid_first_time(ctx, first_datafile_remaining, metric_first_time_JudyL);
1395
1396 if(worker)
1255 - worker_is_busy(UV_EVENT_RETENTION_UPDATE);
1397 + worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);
1398
1399 info("DBENGINE: updating metric registry retention for %u metrics", count);
1400
@@ -1260,7 +1402,12 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1402 bool first_then_next = true;
1403 while ((PValue = JudyLFirstThenNext(metric_first_time_JudyL, &index, &first_then_next))) {
1404 uuid_first_t_entry = *PValue;
1263 - mrg_metric_set_first_time_s(main_mrg, uuid_first_t_entry->metric, uuid_first_t_entry->first_time_s);
1405 +
1406 + if (likely(uuid_first_t_entry->first_time_s != LONG_MAX && uuid_first_t_entry->last_time_s))
1407 + mrg_metric_set_first_time_s_if_bigger(main_mrg, uuid_first_t_entry->metric, uuid_first_t_entry->first_time_s);
1408 + else
1409 + mrg_metric_set_first_time_s(main_mrg, uuid_first_t_entry->metric, 0);
1410 +
1411 mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
1412 freez(uuid_first_t_entry);
1413 }
@@ -1271,18 +1418,18 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1418 worker_is_idle();
1419 }
1420
1274 -static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, bool worker) {
1421 +void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, bool update_retention, bool worker) {
1422 if(worker)
1276 - worker_is_busy(UV_EVENT_DATAFILE_ACQUIRE);
1423 + worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT);
1424
1425 bool datafile_got_for_deletion = datafile_acquire_for_deletion(datafile);
1426
1280 - if (ctx_is_available_for_queries(ctx))
1427 + if (update_retention)
1428 update_metrics_first_time_s(ctx, datafile, datafile->next, worker);
1429
1430 while (!datafile_got_for_deletion) {
1431 if(worker)
1285 - worker_is_busy(UV_EVENT_DATAFILE_ACQUIRE);
1432 + worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT);
1433
1434 datafile_got_for_deletion = datafile_acquire_for_deletion(datafile);
1435
@@ -1305,7 +1452,7 @@ static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_dat
1452 ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);
1453
1454 if(worker)
1308 - worker_is_busy(UV_EVENT_DATAFILE_DELETE);
1455 + worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE);
1456
1457 struct rrdengine_journalfile *journal_file;
1458 unsigned deleted_bytes, journal_file_bytes, datafile_bytes;
@@ -1341,15 +1488,16 @@ static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_dat
1488
1489 ctx_current_disk_space_decrease(ctx, deleted_bytes);
1490 info("DBENGINE: reclaimed %u bytes of disk space.", deleted_bytes);
1491 +}
1492 +
1493 +static void *database_rotate_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1494 + datafile_delete(ctx, ctx->datafiles.first, ctx_is_available_for_queries(ctx), true);
1495
1496 if (rrdeng_ctx_exceeded_disk_quota(ctx))
1497 rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1498
1499 rrdcontext_db_rotation();
1349 -}
1500
1351 -static void *database_rotate_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1352 - datafile_delete(ctx, ctx->datafiles.first, true);
1501 return data;
1502 }
1503
@@ -1358,7 +1506,7 @@ static void after_flush_all_hot_and_dirty_pages_of_section(struct rrdengine_inst
1506 }
1507
1508 static void *flush_all_hot_and_dirty_pages_of_section_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1361 - worker_is_busy(UV_EVENT_QUIESCE);
1509 + worker_is_busy(UV_EVENT_DBENGINE_QUIESCE);
1510 pgc_flush_all_hot_and_dirty_pages(main_cache, (Word_t)ctx);
1511 completion_mark_complete(&ctx->quiesce.completion);
1512 return data;
@@ -1369,7 +1517,7 @@ static void after_populate_mrg(struct rrdengine_instance *ctx __maybe_unused, vo
1517 }
1518
1519 static void *populate_mrg_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1372 - worker_is_busy(UV_EVENT_POPULATE_MRG);
1520 + worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);
1521
1522 do {
1523 struct rrdengine_datafile *datafile = NULL;
@@ -1409,7 +1557,7 @@ static void after_ctx_shutdown(struct rrdengine_instance *ctx __maybe_unused, vo
1557 }
1558
1559 static void *ctx_shutdown_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1412 - worker_is_busy(UV_EVENT_SHUTDOWN);
1560 + worker_is_busy(UV_EVENT_DBENGINE_SHUTDOWN);
1561
1562 completion_wait_for(&ctx->quiesce.completion);
1563 completion_destroy(&ctx->quiesce.completion);
@@ -1427,7 +1575,7 @@ static void *cache_flush_tp_worker(struct rrdengine_instance *ctx __maybe_unused
1575 if (!main_cache)
1576 return data;
1577
1430 - worker_is_busy(UV_EVENT_FLUSH_MAIN);
1578 + worker_is_busy(UV_EVENT_DBENGINE_FLUSH_MAIN_CACHE);
1579 pgc_flush_pages(main_cache, 0);
1580
1581 return data;
@@ -1437,7 +1585,7 @@ static void *cache_evict_tp_worker(struct rrdengine_instance *ctx __maybe_unused
1585 if (!main_cache)
1586 return data;
1587
1440 - worker_is_busy(UV_EVENT_EVICT_MAIN);
1588 + worker_is_busy(UV_EVENT_DBENGINE_EVICT_MAIN_CACHE);
1589 pgc_evict_pages(main_cache, 0, 0);
1590
1591 return data;
@@ -1448,7 +1596,7 @@ static void after_prep_query(struct rrdengine_instance *ctx __maybe_unused, void
1596 }
1597
1598 static void *query_prep_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *req __maybe_unused) {
1451 - worker_is_busy(UV_EVENT_PREP_QUERY);
1599 + worker_is_busy(UV_EVENT_DBENGINE_QUERY);
1600 PDC *pdc = data;
1601 rrdeng_prep_query(pdc);
1602 return data;
@@ -1516,7 +1664,7 @@ void pdc_route_synchronously(struct rrdengine_instance *ctx, struct page_details
1664 #define MAX_RETRIES_TO_START_INDEX (100)
1665 static void *journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1666 unsigned count = 0;
1519 - worker_is_busy(UV_EVENT_JOURNAL_INDEX_WAIT);
1667 + worker_is_busy(UV_EVENT_DBENGINE_JOURNAL_INDEX_WAIT);
1668
1669 while (__atomic_load_n(&ctx->atomic.now_deleting_files, __ATOMIC_RELAXED) && count++ < MAX_RETRIES_TO_START_INDEX)
1670 sleep_usec(100 * USEC_PER_MS);
@@ -1527,7 +1675,7 @@ static void *journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __mayb
1675 }
1676
1677 struct rrdengine_datafile *datafile = ctx->datafiles.first;
1530 - worker_is_busy(UV_EVENT_JOURNAL_INDEX);
1678 + worker_is_busy(UV_EVENT_DBENGINE_JOURNAL_INDEX);
1679 count = 0;
1680 while (datafile && datafile->fileno != ctx_last_fileno_get(ctx) && datafile->fileno != ctx_last_flush_fileno_get(ctx)) {
1681
@@ -1589,6 +1737,9 @@ struct rrdeng_buffer_sizes rrdeng_get_buffer_sizes(void) {
1737 .epdl = epdl_cache_size(),
1738 .deol = deol_cache_size(),
1739 .pd = pd_cache_size(),
1740 + .pages = __atomic_load_n(&dbengine_page_alloc_globals.atomic.cached_size, __ATOMIC_RELAXED) +
1741 + __atomic_load_n(&dbengine_page_alloc_globals.atomic.struct_size, __ATOMIC_RELAXED),
1742 +
1743 #ifdef PDC_USE_JULYL
1744 .julyl = julyl_cache_size(),
1745 #endif
@@ -1600,7 +1751,7 @@ static void after_cleanup(struct rrdengine_instance *ctx __maybe_unused, void *d
1751 }
1752
1753 static void *cleanup_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1603 - worker_is_busy(UV_EVENT_BUFFERS_CLEANUP);
1754 + worker_is_busy(UV_EVENT_DBENGINE_BUFFERS_CLEANUP);
1755
1756 rrdeng_cmd_cleanup1();
1757 work_request_cleanup1();
@@ -1613,6 +1764,7 @@ static void *cleanup_tp_worker(struct rrdengine_instance *ctx __maybe_unused, vo
1764 extent_buffer_cleanup1();
1765 epdl_cleanup1();
1766 deol_cleanup1();
1767 + dbengine_page_alloc_cleanup1();
1768
1769 {
1770 static time_t last_run_s = 0;
database/engine/rrdengine.h
+23 -17
@@ -71,36 +71,39 @@ typedef enum __attribute__ ((__packed__)) {
71 PDC_PAGE_FAILED = (1 << 1), // failed to be loaded (pd->page is null)
72 PDC_PAGE_SKIP = (1 << 2), // don't use this page, it is not good for us
73 PDC_PAGE_INVALID = (1 << 3), // don't use this page, it is invalid
74 + PDC_PAGE_EMPTY = (1 << 4), // the page is empty, does not have any data
75
76 // other statuses for tracking issues
76 - PDC_PAGE_PREPROCESSED = (1 << 4), // used during preprocessing
77 - PDC_PAGE_PROCESSED = (1 << 5), // processed by the query caller
78 - PDC_PAGE_RELEASED = (1 << 6), // already released
77 + PDC_PAGE_PREPROCESSED = (1 << 5), // used during preprocessing
78 + PDC_PAGE_PROCESSED = (1 << 6), // processed by the query caller
79 + PDC_PAGE_RELEASED = (1 << 7), // already released
80
81 // data found in cache (preloaded) or on disk?
81 - PDC_PAGE_PRELOADED = (1 << 7), // data found in memory
82 - PDC_PAGE_DISK_PENDING = (1 << 8), // data need to be loaded from disk
82 + PDC_PAGE_PRELOADED = (1 << 8), // data found in memory
83 + PDC_PAGE_DISK_PENDING = (1 << 9), // data need to be loaded from disk
84
85 // worker related statuses
85 - PDC_PAGE_FAILED_INVALID_EXTENT = (1 << 9),
86 - PDC_PAGE_FAILED_NOT_IN_EXTENT = (1 << 10),
87 - PDC_PAGE_FAILED_TO_MAP_EXTENT = (1 << 11),
88 - PDC_PAGE_FAILED_TO_ACQUIRE_DATAFILE= (1 << 12),
86 + PDC_PAGE_FAILED_INVALID_EXTENT = (1 << 10),
87 + PDC_PAGE_FAILED_NOT_IN_EXTENT = (1 << 11),
88 + PDC_PAGE_FAILED_TO_MAP_EXTENT = (1 << 12),
89 + PDC_PAGE_FAILED_TO_ACQUIRE_DATAFILE= (1 << 13),
90
90 - PDC_PAGE_EXTENT_FROM_CACHE = (1 << 13),
91 - PDC_PAGE_EXTENT_FROM_DISK = (1 << 14),
91 + PDC_PAGE_EXTENT_FROM_CACHE = (1 << 14),
92 + PDC_PAGE_EXTENT_FROM_DISK = (1 << 15),
93
93 - PDC_PAGE_CANCELLED = (1 << 15), // the query thread had left when we try to load the page
94 + PDC_PAGE_CANCELLED = (1 << 16), // the query thread had left when we try to load the page
95
95 - PDC_PAGE_SOURCE_MAIN_CACHE = (1 << 16),
96 - PDC_PAGE_SOURCE_OPEN_CACHE = (1 << 17),
97 - PDC_PAGE_SOURCE_JOURNAL_V2 = (1 << 18),
98 - PDC_PAGE_PRELOADED_PASS4 = (1 << 19),
96 + PDC_PAGE_SOURCE_MAIN_CACHE = (1 << 17),
97 + PDC_PAGE_SOURCE_OPEN_CACHE = (1 << 18),
98 + PDC_PAGE_SOURCE_JOURNAL_V2 = (1 << 19),
99 + PDC_PAGE_PRELOADED_PASS4 = (1 << 20),
100
101 // datafile acquired
102 PDC_PAGE_DATAFILE_ACQUIRED = (1 << 30),
103 } PDC_PAGE_STATUS;
104
105 +#define PDC_PAGE_QUERY_GLOBAL_SKIP_LIST (PDC_PAGE_FAILED | PDC_PAGE_SKIP | PDC_PAGE_INVALID | PDC_PAGE_RELEASED)
106 +
107 struct page_details {
108 struct {
109 struct rrdengine_datafile *ptr;
@@ -494,6 +497,8 @@ typedef struct validated_page_descriptor {
497 bool is_valid;
498 } VALIDATED_PAGE_DESCRIPTOR;
499
500 +#define DBENGINE_EMPTY_PAGE (void *)(-1)
501 +
502 #define page_entries_by_time(start_time_s, end_time_s, update_every_s) \
503 ((update_every_s) ? (((end_time_s) - ((start_time_s) - (update_every_s))) / (update_every_s)) : 1)
504
@@ -510,7 +515,6 @@ VALIDATED_PAGE_DESCRIPTOR validate_page(uuid_t *uuid,
515 time_t now_s,
516 time_t overwrite_zero_update_every_s,
517 bool have_read_error,
513 - bool minimize_invalid_size,
518 const char *msg,
519 RRDENG_COLLECT_PAGE_FLAGS flags);
520 VALIDATED_PAGE_DESCRIPTOR validate_extent_page_descr(const struct rrdeng_extent_page_descr *descr, time_t now_s, time_t overwrite_zero_update_every_s, bool have_read_error);
@@ -528,4 +532,6 @@ static inline time_t max_acceptable_collected_time(void) {
532 return now_realtime_sec() + 1;
533 }
534
535 +void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, bool update_retention, bool worker);
536 +
537 #endif /* NETDATA_RRDENGINE_H */
database/engine/rrdengineapi.c
+48 -28
@@ -190,7 +190,7 @@ static inline void check_and_fix_mrg_update_every(struct rrdeng_collect_handle *
190 (time_t)(handle->update_every_ut / USEC_PER_SEC), mrg_metric_get_update_every_s(main_mrg, handle->metric));
191
192 if(unlikely(!handle->update_every_ut))
193 - handle->update_every_ut = mrg_metric_get_update_every_s(main_mrg, handle->metric) * USEC_PER_SEC;
193 + handle->update_every_ut = (usec_t)mrg_metric_get_update_every_s(main_mrg, handle->metric) * USEC_PER_SEC;
194 else
195 mrg_metric_set_update_every(main_mrg, handle->metric, (time_t)(handle->update_every_ut / USEC_PER_SEC));
196 }
@@ -225,7 +225,6 @@ static inline bool check_completed_page_consistency(struct rrdeng_collect_handle
225 0, // do not check for future timestamps - we inherit the timestamps of the children
226 overwrite_zero_update_every_s,
227 false,
228 - false,
228 "collected",
229 handle->page_flags);
230
@@ -260,7 +259,7 @@ STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metri
259 handle->page = NULL;
260 handle->page_position = 0;
261 handle->page_entries_max = 0;
263 - handle->update_every_ut = update_every * USEC_PER_SEC;
262 + handle->update_every_ut = (usec_t)update_every * USEC_PER_SEC;
263 handle->options = is_1st_metric_writer ? RRDENG_1ST_METRIC_WRITER : 0;
264
265 __atomic_add_fetch(&ctx->atomic.collectors_running, 1, __ATOMIC_RELAXED);
@@ -271,7 +270,7 @@ STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metri
270 // if we don't set the page_end_time_ut during the first collection
271 // data collection may be able to go back in time and during the addition of new pages
272 // clean pages may be found matching ours!
274 - handle->page_end_time_ut = mrg_metric_get_latest_time_s(main_mrg, metric) * USEC_PER_SEC;
273 + handle->page_end_time_ut = (usec_t)mrg_metric_get_latest_time_s(main_mrg, metric) * USEC_PER_SEC;
274
275 mrg_metric_set_update_every(main_mrg, metric, update_every);
276
@@ -348,9 +347,13 @@ void rrdeng_store_metric_flush_current_page(STORAGE_COLLECT_HANDLE *collection_h
347 check_and_fix_mrg_update_every(handle);
348 }
349
351 -static void rrdeng_store_metric_create_new_page(struct rrdeng_collect_handle *handle, struct rrdengine_instance *ctx, usec_t point_in_time_ut, void *data, size_t data_size) {
350 +static void rrdeng_store_metric_create_new_page(struct rrdeng_collect_handle *handle,
351 + struct rrdengine_instance *ctx,
352 + usec_t point_in_time_ut,
353 + void *data,
354 + size_t data_size) {
355 time_t point_in_time_s = (time_t)(point_in_time_ut / USEC_PER_SEC);
353 - time_t update_every_s = (time_t)(handle->update_every_ut / USEC_PER_SEC);
356 + const time_t update_every_s = (time_t)(handle->update_every_ut / USEC_PER_SEC);
357
358 PGC_ENTRY page_entry = {
359 .section = (Word_t) ctx,
@@ -449,13 +452,13 @@ static void *rrdeng_alloc_new_metric_data(struct rrdeng_collect_handle *handle,
452 }
453
454 static void rrdeng_store_metric_append_point(STORAGE_COLLECT_HANDLE *collection_handle,
452 - usec_t point_in_time_ut,
453 - NETDATA_DOUBLE n,
454 - NETDATA_DOUBLE min_value,
455 - NETDATA_DOUBLE max_value,
456 - uint16_t count,
457 - uint16_t anomaly_count,
458 - SN_FLAGS flags)
455 + const usec_t point_in_time_ut,
456 + const NETDATA_DOUBLE n,
457 + const NETDATA_DOUBLE min_value,
458 + const NETDATA_DOUBLE max_value,
459 + const uint16_t count,
460 + const uint16_t anomaly_count,
461 + const SN_FLAGS flags)
462 {
463 struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle;
464 struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric);
@@ -584,18 +587,18 @@ static void store_metric_next_error_log(struct rrdeng_collect_handle *handle, us
587 }
588
589 void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle,
587 - usec_t point_in_time_ut,
588 - NETDATA_DOUBLE n,
589 - NETDATA_DOUBLE min_value,
590 - NETDATA_DOUBLE max_value,
591 - uint16_t count,
592 - uint16_t anomaly_count,
593 - SN_FLAGS flags)
590 + const usec_t point_in_time_ut,
591 + const NETDATA_DOUBLE n,
592 + const NETDATA_DOUBLE min_value,
593 + const NETDATA_DOUBLE max_value,
594 + const uint16_t count,
595 + const uint16_t anomaly_count,
596 + const SN_FLAGS flags)
597 {
598 struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle;
599
600 #ifdef NETDATA_INTERNAL_CHECKS
598 - if(unlikely(point_in_time_ut > max_acceptable_collected_time() * USEC_PER_SEC))
601 + if(unlikely(point_in_time_ut > (usec_t)max_acceptable_collected_time() * USEC_PER_SEC))
602 handle->page_flags |= RRDENG_PAGE_FUTURE_POINT;
603 #endif
604
@@ -638,13 +641,13 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle,
641 // loop to fill the gap
642 handle->page_flags |= RRDENG_PAGE_GAP;
643
641 - usec_t point_in_time_to_stop_ut = point_in_time_ut - handle->update_every_ut;
642 - for(usec_t next_point_in_time = handle->page_end_time_ut + handle->update_every_ut;
643 - next_point_in_time <= point_in_time_to_stop_ut ;
644 - next_point_in_time = handle->page_end_time_ut + handle->update_every_ut) {
644 + usec_t stop_ut = point_in_time_ut - handle->update_every_ut;
645 + for(usec_t this_ut = handle->page_end_time_ut + handle->update_every_ut;
646 + this_ut <= stop_ut ;
647 + this_ut = handle->page_end_time_ut + handle->update_every_ut) {
648 rrdeng_store_metric_append_point(
649 collection_handle,
647 - handle->page_end_time_ut + handle->update_every_ut,
650 + this_ut,
651 NAN, NAN, NAN,
652 1, 0,
653 SN_EMPTY_SLOT);
@@ -690,7 +693,7 @@ void rrdeng_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *col
693 check_and_fix_mrg_update_every(handle);
694
695 METRIC *metric = handle->metric;
693 - usec_t update_every_ut = update_every * USEC_PER_SEC;
696 + usec_t update_every_ut = (usec_t)update_every * USEC_PER_SEC;
697
698 if(update_every_ut == handle->update_every_ut)
699 return;
@@ -792,6 +795,8 @@ static bool rrdeng_load_page_next(struct storage_engine_query_handle *rrddim_han
795 if (unlikely(!handle->page))
796 return false;
797
798 + internal_fatal(pgc_page_data(handle->page) == DBENGINE_EMPTY_PAGE, "Empty page returned");
799 +
800 time_t page_start_time_s = pgc_page_start_time_s(handle->page);
801 time_t page_end_time_s = pgc_page_end_time_s(handle->page);
802 time_t page_update_every_s = pgc_page_update_every_s(handle->page);
@@ -1154,6 +1159,10 @@ int rrdeng_init(struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned p
1159 return UV_EIO;
1160 }
1161
1162 +size_t rrdeng_collectors_running(struct rrdengine_instance *ctx) {
1163 + return __atomic_load_n(&ctx->atomic.collectors_running, __ATOMIC_RELAXED);
1164 +}
1165 +
1166 /*
1167 * Returns 0 on success, 1 on error
1168 */
@@ -1163,10 +1172,21 @@ int rrdeng_exit(struct rrdengine_instance *ctx) {
1172
1173 // FIXME - ktsaou - properly cleanup ctx
1174 // 1. make sure all collectors are stopped
1166 - // 2. make new queries will not be accepted
1175 + // 2. make new queries will not be accepted (this is quiesce that has already run)
1176 // 3. flush this section of the main cache
1177 // 4. then wait for completion
1178
1179 + bool logged = false;
1180 + while(__atomic_load_n(&ctx->atomic.collectors_running, __ATOMIC_RELAXED) && !unittest_running) {
1181 + if(!logged) {
1182 + info("Waiting for collectors to finish on tier %d...", ctx->config.tier);
1183 + logged = true;
1184 + }
1185 + sleep_usec(100 * USEC_PER_MS);
1186 + }
1187 +
1188 + pgc_flush_all_hot_and_dirty_pages(main_cache, (Word_t)ctx);
1189 +
1190 struct completion completion = {};
1191 completion_init(&completion);
1192 rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_SHUTDOWN, NULL, &completion, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);
database/engine/rrdengineapi.h
+2
@@ -214,6 +214,7 @@ struct rrdeng_buffer_sizes {
214 size_t epdl;
215 size_t deol;
216 size_t pd;
217 + size_t pages;
218 #ifdef PDC_USE_JULYL
219 size_t julyl;
220 #endif
@@ -223,5 +224,6 @@ struct rrdeng_buffer_sizes rrdeng_get_buffer_sizes(void);
224 struct rrdeng_cache_efficiency_stats rrdeng_get_cache_efficiency_stats(void);
225
226 RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx);
227 +size_t rrdeng_collectors_running(struct rrdengine_instance *ctx);
228
229 #endif /* NETDATA_RRDENGINEAPI_H */
database/sqlite/sqlite_aclk_alert.c
+29 -11
@@ -270,20 +270,38 @@ void aclk_push_alert_event(struct aclk_database_worker_config *wc, struct aclk_d
270
271 sqlite3_stmt *res = NULL;
272
273 - buffer_sprintf(sql, "select aa.sequence_id, hl.unique_id, hl.alarm_id, hl.config_hash_id, hl.updated_by_id, hl.when_key, \
274 - hl.duration, hl.non_clear_duration, hl.flags, hl.exec_run_timestamp, hl.delay_up_to_timestamp, hl.name, \
275 - hl.chart, hl.family, hl.exec, hl.recipient, hl.source, hl.units, hl.info, hl.exec_code, hl.new_status, \
276 - hl.old_status, hl.delay, hl.new_value, hl.old_value, hl.last_repeat, hl.chart_context \
277 - from health_log_%s hl, aclk_alert_%s aa \
278 - where hl.unique_id = aa.alert_unique_id and aa.date_submitted is null \
279 - order by aa.sequence_id asc limit %d;", wc->uuid_str, wc->uuid_str, limit);
273 + buffer_sprintf(sql, "select aa.sequence_id, hl.unique_id, hl.alarm_id, hl.config_hash_id, hl.updated_by_id, hl.when_key, " \
274 + " hl.duration, hl.non_clear_duration, hl.flags, hl.exec_run_timestamp, hl.delay_up_to_timestamp, hl.name, " \
275 + " hl.chart, hl.family, hl.exec, hl.recipient, hl.source, hl.units, hl.info, hl.exec_code, hl.new_status, " \
276 + " hl.old_status, hl.delay, hl.new_value, hl.old_value, hl.last_repeat, hl.chart_context " \
277 + " from health_log_%s hl, aclk_alert_%s aa " \
278 + " where hl.unique_id = aa.alert_unique_id and aa.date_submitted is null " \
279 + " order by aa.sequence_id asc limit %d;", wc->uuid_str, wc->uuid_str, limit);
280
281 rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
282 if (rc != SQLITE_OK) {
283 - error_report("Failed to prepare statement when trying to send an alert update via ACLK");
284 - buffer_free(sql);
285 - freez(claim_id);
286 - return;
283 +
284 + // Try to create tables
285 + if (wc->host)
286 + sql_create_health_log_table(wc->host);
287 +
288 + BUFFER *sql_fix = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);
289 + buffer_sprintf(sql_fix, TABLE_ACLK_ALERT, wc->uuid_str);
290 + db_execute(buffer_tostring(sql_fix));
291 + buffer_flush(sql_fix);
292 + buffer_sprintf(sql_fix, INDEX_ACLK_ALERT, wc->uuid_str, wc->uuid_str);
293 + db_execute(buffer_tostring(sql_fix));
294 + buffer_free(sql_fix);
295 +
296 + // Try again
297 + rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
298 + if (rc != SQLITE_OK) {
299 + error_report("Failed to prepare statement when trying to send an alert update via ACLK");
300 +
301 + buffer_free(sql);
302 + freez(claim_id);
303 + return;
304 + }
305 }
306
307 char uuid_str[GUID_LEN + 1];
database/sqlite/sqlite_health.c
+12 -4
@@ -61,8 +61,12 @@ void sql_health_alarm_log_update(RRDHOST *host, ALARM_ENTRY *ae) {
61
62 rc = sqlite3_prepare_v2(db_meta, command, -1, &res, 0);
63 if (unlikely(rc != SQLITE_OK)) {
64 - error_report("HEALTH [%s]: Failed to prepare statement for SQL_UPDATE_HEALTH_LOG", rrdhost_hostname(host));
65 - return;
64 + sql_create_health_log_table(host);
65 + rc = sqlite3_prepare_v2(db_meta, command, -1, &res, 0);
66 + if (unlikely(rc != SQLITE_OK)) {
67 + error_report("HEALTH [%s]: Failed to prepare statement for SQL_INSERT_HEALTH_LOG", rrdhost_hostname(host));
68 + return;
69 + }
70 }
71
72 rc = sqlite3_bind_int64(res, 1, (sqlite3_int64) ae->updated_by_id);
@@ -134,8 +138,12 @@ void sql_health_alarm_log_insert(RRDHOST *host, ALARM_ENTRY *ae) {
138
139 rc = sqlite3_prepare_v2(db_meta, command, -1, &res, 0);
140 if (unlikely(rc != SQLITE_OK)) {
137 - error_report("HEALTH [%s]: Failed to prepare statement for SQL_INSERT_HEALTH_LOG", rrdhost_hostname(host));
138 - return;
141 + sql_create_health_log_table(host);
142 + rc = sqlite3_prepare_v2(db_meta, command, -1, &res, 0);
143 + if (unlikely(rc != SQLITE_OK)) {
144 + error_report("HEALTH [%s]: Failed to prepare statement for SQL_INSERT_HEALTH_LOG", rrdhost_hostname(host));
145 + return;
146 + }
147 }
148
149 rc = sqlite3_bind_text(res, 1, rrdhost_hostname(host), -1, SQLITE_STATIC);
streaming/replication.c
+41 -41
@@ -155,7 +155,7 @@ static struct replication_query *replication_query_prepare(
155 (unsigned long long) st->last_updated.tv_sec
156 );
157 #endif
158 - q->query.before = st->last_updated.tv_sec;
158 + q->query.before = MIN(st->last_updated.tv_sec, wall_clock_time);
159 }
160 }
161
@@ -209,9 +209,38 @@ static struct replication_query *replication_query_prepare(
209 return q;
210 }
211
212 -static void replication_query_finalize(struct replication_query *q, bool executed) {
212 +static void replication_send_chart_collection_state(BUFFER *wb, RRDSET *st) {
213 + RRDDIM *rd;
214 + rrddim_foreach_read(rd, st) {
215 + if(!rd->exposed) continue;
216 +
217 + buffer_sprintf(wb, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE " \"%s\" %llu %lld " NETDATA_DOUBLE_FORMAT " " NETDATA_DOUBLE_FORMAT "\n",
218 + rrddim_id(rd),
219 + (usec_t)rd->last_collected_time.tv_sec * USEC_PER_SEC + (usec_t)rd->last_collected_time.tv_usec,
220 + rd->last_collected_value,
221 + rd->last_calculated_value,
222 + rd->last_stored_value
223 + );
224 + }
225 + rrddim_foreach_done(rd);
226 +
227 + buffer_sprintf(wb, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE " %llu %llu\n",
228 + (usec_t)st->last_collected_time.tv_sec * USEC_PER_SEC + (usec_t)st->last_collected_time.tv_usec,
229 + (usec_t)st->last_updated.tv_sec * USEC_PER_SEC + (usec_t)st->last_updated.tv_usec
230 + );
231 +}
232 +
233 +static void replication_query_finalize(BUFFER *wb, struct replication_query *q, bool executed) {
234 size_t dimensions = q->dimensions;
235
236 + if(wb && q->query.enable_streaming)
237 + replication_send_chart_collection_state(wb, q->st);
238 +
239 + if(q->query.locked_data_collection) {
240 + netdata_spinlock_unlock(&q->st->data_collection_lock);
241 + q->query.locked_data_collection = false;
242 + }
243 +
244 // release all the dictionary items acquired
245 // finalize the queries
246 size_t queries = 0;
@@ -228,11 +257,6 @@ static void replication_query_finalize(struct replication_query *q, bool execute
257 queries++;
258 }
259
231 - if(q->query.locked_data_collection) {
232 - netdata_spinlock_unlock(&q->st->data_collection_lock);
233 - q->query.locked_data_collection = false;
234 - }
235 -
260 if(executed) {
261 netdata_spinlock_lock(&replication_queries.spinlock);
262 replication_queries.queries_started += queries;
@@ -311,8 +335,8 @@ static void replication_query_execute(BUFFER *wb, struct replication_query *q, s
335 continue;
336 }
337
314 - if(d->sp.end_time_s < now)
315 - // this dimension does not have any more data
338 + if(unlikely(d->sp.end_time_s < now || d->sp.end_time_s < d->sp.start_time_s))
339 + // this dimension does not provide any data
340 continue;
341
342 if(unlikely(!min_start_time))
@@ -325,13 +349,10 @@ static void replication_query_execute(BUFFER *wb, struct replication_query *q, s
349 min_end_time = MIN(min_end_time, d->sp.end_time_s);
350 }
351
328 - if(unlikely(min_end_time < now))
329 - break;
330 -
331 - if(likely(min_start_time <= now)) {
352 + if(likely(min_start_time <= now && min_end_time >= now)) {
353 // we have a valid point
354
334 - if (unlikely(min_end_time <= min_start_time))
355 + if (unlikely(min_end_time == min_start_time))
356 min_start_time = min_end_time - q->st->update_every;
357
358 #ifdef NETDATA_LOG_REPLICATION_REQUESTS
@@ -382,7 +403,11 @@ static void replication_query_execute(BUFFER *wb, struct replication_query *q, s
403
404 now = min_end_time + 1;
405 }
406 + else if(unlikely(min_end_time < now))
407 + // the query does not progress
408 + break;
409 else
410 + // we have gap - all points are in the future
411 now = min_start_time;
412 }
413
@@ -408,27 +433,6 @@ static void replication_query_execute(BUFFER *wb, struct replication_query *q, s
433 q->points_generated = points_generated;
434 }
435
411 -static void replication_send_chart_collection_state(BUFFER *wb, RRDSET *st) {
412 - RRDDIM *rd;
413 - rrddim_foreach_read(rd, st) {
414 - if(!rd->exposed) continue;
415 -
416 - buffer_sprintf(wb, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE " \"%s\" %llu %lld " NETDATA_DOUBLE_FORMAT " " NETDATA_DOUBLE_FORMAT "\n",
417 - rrddim_id(rd),
418 - (usec_t)rd->last_collected_time.tv_sec * USEC_PER_SEC + (usec_t)rd->last_collected_time.tv_usec,
419 - rd->last_collected_value,
420 - rd->last_calculated_value,
421 - rd->last_stored_value
422 - );
423 - }
424 - rrddim_foreach_done(rd);
425 -
426 - buffer_sprintf(wb, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE " %llu %llu\n",
427 - (usec_t)st->last_collected_time.tv_sec * USEC_PER_SEC + (usec_t)st->last_collected_time.tv_usec,
428 - (usec_t)st->last_updated.tv_sec * USEC_PER_SEC + (usec_t)st->last_updated.tv_usec
429 - );
430 -}
431 -
436 static struct replication_query *replication_response_prepare(RRDSET *st, bool requested_enable_streaming, time_t requested_after, time_t requested_before) {
437 time_t query_after = requested_after;
438 time_t query_before = requested_before;
@@ -475,7 +479,7 @@ static struct replication_query *replication_response_prepare(RRDSET *st, bool r
479 }
480
481 void replication_response_cancel_and_finalize(struct replication_query *q) {
478 - replication_query_finalize(q, false);
482 + replication_query_finalize(NULL, q, false);
483 }
484
485 static bool sender_is_still_connected_for_this_request(struct replication_request *rq);
@@ -502,13 +506,9 @@ bool replication_response_execute_and_finalize(struct replication_query *q, size
506 time_t before = q->query.before;
507 bool enable_streaming = q->query.enable_streaming;
508
505 - replication_query_finalize(q, q->query.execute);
509 + replication_query_finalize(wb, q, q->query.execute);
510 q = NULL; // IMPORTANT: q is invalid now
511
508 - // get again the world clock time
509 - if(enable_streaming)
510 - replication_send_chart_collection_state(wb, st);
511 -
512 // get a fresh retention to send to the parent
513 time_t wall_clock_time = now_realtime_sec();
514 time_t db_first_entry, db_last_entry;
web/api/queries/query.c
+13 -7
@@ -1018,15 +1018,16 @@ static void query_planer_activate_plan(QUERY_ENGINE_OPS *ops, size_t plan_id, ti
1018 ops->is_finished = qm->plan.array[plan_id].is_finished;
1019 ops->finalize = qm->plan.array[plan_id].finalize;
1020 ops->current_plan = plan_id;
1021 - ops->current_plan_expire_time = qm->plan.array[plan_id].before;
1021 +
1022 + if(plan_id + 1 < qm->plan.used && qm->plan.array[plan_id + 1].after < qm->plan.array[plan_id].before)
1023 + ops->current_plan_expire_time = qm->plan.array[plan_id + 1].after;
1024 + else
1025 + ops->current_plan_expire_time = qm->plan.array[plan_id].before;
1026 }
1027
1028 static void query_planer_next_plan(QUERY_ENGINE_OPS *ops, time_t now, time_t last_point_end_time) {
1029 QUERY_METRIC *qm = ops->qm;
1030
1027 - internal_fatal(now < ops->current_plan_expire_time && now < qm->plan.array[ops->current_plan].before,
1028 - "QUERY: switching query plan too early!");
1029 -
1031 size_t old_plan = ops->current_plan;
1032
1033 time_t next_plan_before_time;
@@ -1179,6 +1180,11 @@ static bool query_plan(QUERY_ENGINE_OPS *ops, time_t after_wanted, time_t before
1180 }
1181 #endif
1182
1183 + for(size_t p = 0; p < qm->plan.used ;p++) {
1184 + size_t tier = qm->plan.array[p].tier;
1185 + qm->plan.array[p].before += qm->tiers[tier].db_update_every_s - 1;
1186 + }
1187 +
1188 query_planer_initialize_plans(ops);
1189 query_planer_activate_plan(ops, 0, 0);
1190
@@ -1355,7 +1361,7 @@ static void rrd2rrdr_query_execute(RRDR *r, size_t dim_id_in_rrdr, QUERY_ENGINE_
1361 }
1362
1363 // check if the db is giving us zero duration points
1358 - if(unlikely(new_point.start_time == new_point.end_time)) {
1364 + if(unlikely(db_points_read_since_plan_switch > 1 && new_point.start_time == new_point.end_time)) {
1365 internal_error(true, "QUERY: '%s', dimension '%s' next_metric() returned point %zu start time %ld, end time %ld, that are both equal",
1366 qt->id, string2str(qm->dimension.id), new_point.id, new_point.start_time, new_point.end_time);
1367
@@ -1363,8 +1369,8 @@ static void rrd2rrdr_query_execute(RRDR *r, size_t dim_id_in_rrdr, QUERY_ENGINE_
1369 }
1370
1371 // check if the db is advancing the query
1366 - if(unlikely(new_point.end_time <= last1_point.end_time)) {
1367 - internal_error(db_points_read_since_plan_switch > 1,
1372 + if(unlikely(db_points_read_since_plan_switch > 1 && new_point.end_time <= last1_point.end_time)) {
1373 + internal_error(true,
1374 "QUERY: '%s', dimension '%s' next_metric() returned point %zu from %ld to %ld, before the last point %zu from %ld to %ld, now is %ld to %ld",
1375 qt->id, string2str(qm->dimension.id), new_point.id, new_point.start_time, new_point.end_time,
1376 last1_point.id, last1_point.start_time, last1_point.end_time, now_start_time, now_end_time);