@cryptotaxi247 / netdata-1 / commits / 68658fc1e

DBENGINE v2 - improvements 2 (#14257)

* allow extents to be merged for as long as possible * do not block the event loop while recalculating retention due to datafile rotation * buffers are incrementally cleaned up, every second, by just 1 entry * fix order of commands * remove newline * measure cancelled extent read requests * count all cancelled extent requests * do not double count failed pages * fixed cancelled name * Fix error and warnings when compiling with --disable-dbengine * when the timeframe is outside retention and whole query should fail * do not mark as failed pages that have been loaded but have been skipped * added chart to show cache memory calculation variables * LONG_MAX for 32-bit compatibility * fix cache size calculation on 32-bit * fix cache size calculation on 32-bit - use unsinged long long * fix compilation warnings on 32-bits * fix another compilation warning on 32-bits * fix compilation warnings on older 32-bit compilers * fix compilation warnings on older 32-bit compilers - more of them * disable ML threads joining Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>

Costa Tsaousis committed Jan 13, 2023 at 19:52 UTC 68658fc1e0a0902343bb165b7ed12b6fba1f2202
24 files changed +412 -199
aclk/aclk.c
+1 -1
@@ -459,7 +459,7 @@ static int aclk_block_till_recon_allowed() {
459 next_connection_attempt = now_realtime_sec() + (recon_delay / MSEC_PER_SEC);
460 last_backoff_value = (float)recon_delay / MSEC_PER_SEC;
461
462 - info("Wait before attempting to reconnect in %.3f seconds\n", recon_delay / (float)MSEC_PER_SEC);
462 + info("Wait before attempting to reconnect in %.3f seconds", recon_delay / (float)MSEC_PER_SEC);
463 // we want to wake up from time to time to check netdata_exit
464 while (recon_delay)
465 {
daemon/global_statistics.c
+66 -3
@@ -1023,6 +1023,15 @@ struct dbengine2_cache_pointers {
1023 RRDDIM *rd_pgc_memory_evicting;
1024 RRDDIM *rd_pgc_memory_flushing;
1025
1026 + RRDSET *st_pgc_tm;
1027 + RRDDIM *rd_pgc_tm_current;
1028 + RRDDIM *rd_pgc_tm_wanted;
1029 + RRDDIM *rd_pgc_tm_hot_max;
1030 + RRDDIM *rd_pgc_tm_dirty_max;
1031 + RRDDIM *rd_pgc_tm_hot;
1032 + RRDDIM *rd_pgc_tm_dirty;
1033 + RRDDIM *rd_pgc_tm_referenced;
1034 +
1035 RRDSET *st_pgc_pages;
1036 RRDDIM *rd_pgc_pages_clean;
1037 RRDDIM *rd_pgc_pages_hot;
@@ -1168,7 +1177,6 @@ static void dbengine2_cache_statistics_charts(struct dbengine2_cache_pointers *p
1177 }
1178
1179 {
1171 -
1180 if (unlikely(!ptrs->st_pgc_memory)) {
1181 BUFFER *id = buffer_create(100);
1182 buffer_sprintf(id, "dbengine_%s_cache_memory", name);
@@ -1222,6 +1230,56 @@ static void dbengine2_cache_statistics_charts(struct dbengine2_cache_pointers *p
1230 rrdset_done(ptrs->st_pgc_memory);
1231 }
1232
1233 + {
1234 + if (unlikely(!ptrs->st_pgc_tm)) {
1235 + BUFFER *id = buffer_create(100);
1236 + buffer_sprintf(id, "dbengine_%s_target_memory", name);
1237 +
1238 + BUFFER *family = buffer_create(100);
1239 + buffer_sprintf(family, "dbengine %s cache", name);
1240 +
1241 + BUFFER *title = buffer_create(100);
1242 + buffer_sprintf(title, "Netdata %s Target Cache Memory", name);
1243 +
1244 + ptrs->st_pgc_tm = rrdset_create_localhost(
1245 + "netdata",
1246 + buffer_tostring(id),
1247 + NULL,
1248 + buffer_tostring(family),
1249 + NULL,
1250 + buffer_tostring(title),
1251 + "bytes",
1252 + "netdata",
1253 + "stats",
1254 + priority,
1255 + localhost->rrd_update_every,
1256 + RRDSET_TYPE_LINE);
1257 +
1258 + ptrs->rd_pgc_tm_current = rrddim_add(ptrs->st_pgc_tm, "current", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1259 + ptrs->rd_pgc_tm_wanted = rrddim_add(ptrs->st_pgc_tm, "wanted", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1260 + ptrs->rd_pgc_tm_referenced = rrddim_add(ptrs->st_pgc_tm, "referenced", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1261 + ptrs->rd_pgc_tm_hot_max = rrddim_add(ptrs->st_pgc_tm, "hot max", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1262 + ptrs->rd_pgc_tm_dirty_max = rrddim_add(ptrs->st_pgc_tm, "dirty max", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1263 + ptrs->rd_pgc_tm_hot = rrddim_add(ptrs->st_pgc_tm, "hot", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1264 + ptrs->rd_pgc_tm_dirty = rrddim_add(ptrs->st_pgc_tm, "dirty", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
1265 +
1266 + buffer_free(id);
1267 + buffer_free(family);
1268 + buffer_free(title);
1269 + priority++;
1270 + }
1271 +
1272 + rrddim_set_by_pointer(ptrs->st_pgc_tm, ptrs->rd_pgc_tm_current, (collected_number)pgc_stats->current_cache_size);
1273 + rrddim_set_by_pointer(ptrs->st_pgc_tm, ptrs->rd_pgc_tm_wanted, (collected_number)pgc_stats->wanted_cache_size);
1274 + rrddim_set_by_pointer(ptrs->st_pgc_tm, ptrs->rd_pgc_tm_referenced, (collected_number)pgc_stats->referenced_size);
1275 + rrddim_set_by_pointer(ptrs->st_pgc_tm, ptrs->rd_pgc_tm_hot_max, (collected_number)pgc_stats->queues.hot.max_size);
1276 + rrddim_set_by_pointer(ptrs->st_pgc_tm, ptrs->rd_pgc_tm_dirty_max, (collected_number)pgc_stats->queues.dirty.max_size);
1277 + rrddim_set_by_pointer(ptrs->st_pgc_tm, ptrs->rd_pgc_tm_hot, (collected_number)pgc_stats->queues.hot.size);
1278 + rrddim_set_by_pointer(ptrs->st_pgc_tm, ptrs->rd_pgc_tm_dirty, (collected_number)pgc_stats->queues.dirty.size);
1279 +
1280 + rrdset_done(ptrs->st_pgc_tm);
1281 + }
1282 +
1283 {
1284 if (unlikely(!ptrs->st_pgc_pages)) {
1285 BUFFER *id = buffer_create(100);
@@ -1972,6 +2030,7 @@ static void dbengine2_statistics_charts(void) {
2030 static RRDDIM *rd_unavailable = NULL;
2031 static RRDDIM *rd_unroutable = NULL;
2032 static RRDDIM *rd_not_found = NULL;
2033 + static RRDDIM *rd_cancelled = NULL;
2034 static RRDDIM *rd_invalid_extent = NULL;
2035 static RRDDIM *rd_extent_merged = NULL;
2036
@@ -1996,9 +2055,10 @@ static void dbengine2_statistics_charts(void) {
2055 rd_mmap_failed = rrddim_add(st_query_pages_from_disk, "fail cant mmap", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2056 rd_unavailable = rrddim_add(st_query_pages_from_disk, "fail unavailable", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2057 rd_unroutable = rrddim_add(st_query_pages_from_disk, "fail unroutable", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
1999 - rd_not_found = rrddim_add(st_query_pages_from_disk, "fail uuid not found", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2058 + rd_not_found = rrddim_add(st_query_pages_from_disk, "fail not found", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2059 rd_invalid_extent = rrddim_add(st_query_pages_from_disk, "fail invalid extent", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2060 rd_extent_merged = rrddim_add(st_query_pages_from_disk, "extent merged", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
2061 + rd_cancelled = rrddim_add(st_query_pages_from_disk, "cancelled", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2062 }
2063 priority++;
2064
@@ -2008,7 +2068,8 @@ static void dbengine2_statistics_charts(void) {
2068 rrddim_set_by_pointer(st_query_pages_from_disk, rd_mmap_failed, (collected_number)cache_efficiency_stats.pages_load_fail_cant_mmap_extent);
2069 rrddim_set_by_pointer(st_query_pages_from_disk, rd_unavailable, (collected_number)cache_efficiency_stats.pages_load_fail_datafile_not_available);
2070 rrddim_set_by_pointer(st_query_pages_from_disk, rd_unroutable, (collected_number)cache_efficiency_stats.pages_load_fail_unroutable);
2011 - rrddim_set_by_pointer(st_query_pages_from_disk, rd_not_found, (collected_number)cache_efficiency_stats.pages_load_fail_uuid_not_found);
2071 + rrddim_set_by_pointer(st_query_pages_from_disk, rd_not_found, (collected_number)cache_efficiency_stats.pages_load_fail_not_found);
2072 + rrddim_set_by_pointer(st_query_pages_from_disk, rd_cancelled, (collected_number)cache_efficiency_stats.pages_load_fail_cancelled);
2073 rrddim_set_by_pointer(st_query_pages_from_disk, rd_invalid_extent, (collected_number)cache_efficiency_stats.pages_load_fail_invalid_extent);
2074 rrddim_set_by_pointer(st_query_pages_from_disk, rd_extent_merged, (collected_number)cache_efficiency_stats.pages_load_extent_merged);
2075
@@ -3778,10 +3839,12 @@ void *global_statistics_main(void *ptr)
3839 worker_is_busy(WORKER_JOB_REGISTRY);
3840 registry_statistics();
3841
3842 +#ifdef ENABLE_DBENGINE
3843 if(dbengine_enabled) {
3844 worker_is_busy(WORKER_JOB_DBENGINE);
3845 dbengine2_statistics_charts();
3846 }
3847 +#endif
3848
3849 worker_is_busy(WORKER_JOB_HEARTBEAT);
3850 update_heartbeat_charts();
database/engine/cache.c
+1 -1
@@ -310,7 +310,7 @@ static inline size_t cache_usage_per1000(PGC *cache, size_t *size_to_evict) {
310
311 current_cache_size = __atomic_load_n(&cache->stats.size, __ATOMIC_RELAXED);
312
313 - per1000 = current_cache_size * 1000 / wanted_cache_size;
313 + per1000 = (size_t)((unsigned long long)current_cache_size * 1000UL / (unsigned long long)wanted_cache_size);
314
315 __atomic_store_n(&cache->usage.per1000, per1000, __ATOMIC_RELAXED);
316 __atomic_store_n(&cache->stats.wanted_cache_size, wanted_cache_size, __ATOMIC_RELAXED);
database/engine/datafile.h
+9 -7
@@ -24,7 +24,14 @@ struct rrdengine_instance;
24 #define MAX_DATAFILES (65536) /* Supports up to 64TiB for now */
25 #define TARGET_DATAFILES (50)
26
27 -#define DATAFILE_IDEAL_IO_SIZE (1048576U)
27 +typedef enum __attribute__ ((__packed__)) {
28 + DATAFILE_ACQUIRE_OPEN_CACHE = 0,
29 + DATAFILE_ACQUIRE_PAGE_DETAILS,
30 + DATAFILE_ACQUIRE_RETENTION,
31 +
32 + // terminator
33 + DATAFILE_ACQUIRE_MAX,
34 +} DATAFILE_ACQUIRE_REASONS;
35
36 /* only one event loop is supported for now */
37 struct rrdengine_datafile {
@@ -47,7 +54,7 @@ struct rrdengine_datafile {
54 struct {
55 SPINLOCK spinlock;
56 unsigned lockers;
50 - unsigned lockers_by_reason[2];
57 + unsigned lockers_by_reason[DATAFILE_ACQUIRE_MAX];
58 bool available;
59 time_t time_to_evict;
60 } users;
@@ -58,11 +65,6 @@ struct rrdengine_datafile {
65 } extent_queries;
66 };
67
61 -typedef enum __attribute__ ((__packed__)) {
62 - DATAFILE_ACQUIRE_OPEN_CACHE = 0,
63 - DATAFILE_ACQUIRE_PAGE_DETAILS = 1,
64 -} DATAFILE_ACQUIRE_REASONS;
65 -
68 void datafile_acquire_dup(struct rrdengine_datafile *df);
69 bool datafile_acquire(struct rrdengine_datafile *df, DATAFILE_ACQUIRE_REASONS reason);
70 void datafile_release(struct rrdengine_datafile *df, DATAFILE_ACQUIRE_REASONS reason);
database/engine/journalfile.c
+5 -5
@@ -652,7 +652,7 @@ static int check_journal_v2_file(void *data_start, size_t file_size, uint32_t or
652 }
653
654 metric++;
655 - if (((uint8_t *) metric - (uint8_t *) data_start) > (uint32_t) file_size) {
655 + if ((uint32_t)((uint8_t *) metric - (uint8_t *) data_start) > (uint32_t) file_size) {
656 info("DBENGINE: verification failed EOF reached -- total entries %u, verified %u", entries, verified);
657 return 1;
658 }
@@ -919,13 +919,13 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
919 struct rrdengine_instance *ctx = (struct rrdengine_instance *) section;
920 struct rrdengine_journalfile *journalfile = (struct rrdengine_journalfile *) user_data;
921 struct rrdengine_datafile *datafile = journalfile->datafile;
922 - time_t min_time_s = LLONG_MAX;
922 + time_t min_time_s = LONG_MAX;
923 time_t max_time_s = 0;
924 struct jv2_metrics_info *metric_info;
925
926 generate_journalfilepath_v2(datafile, path, sizeof(path));
927
928 - info("DBENGINE: indexing file '%s': extents %lu, metrics %lu, pages %lu",
928 + info("DBENGINE: indexing file '%s': extents %zu, metrics %zu, pages %zu",
929 path,
930 number_of_extents,
931 number_of_metrics,
@@ -1062,7 +1062,7 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1062 // Calculate start of the pages start for next descriptor
1063 pages_offset += (metric_info->number_of_pages * (sizeof(struct journal_page_list)) + sizeof(struct journal_page_header) + sizeof(struct journal_v2_block_trailer));
1064 // Verify we are at the right location
1065 - if (pages_offset != (next_page_address - data_start)) {
1065 + if (pages_offset != (uint32_t)(next_page_address - data_start)) {
1066 // make sure checks fail so that we abort
1067 data = data_start;
1068 break;
@@ -1092,7 +1092,7 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1092
1093 internal_error(true, "DBENGINE: FILE COMPLETED --------> %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1094
1095 - info("DBENGINE: migrated journal file '%s', file size %lu", path, total_file_size);
1095 + info("DBENGINE: migrated journal file '%s', file size %zu", path, total_file_size);
1096
1097 SET_JOURNAL_DATA(journalfile, data_start);
1098 SET_JOURNAL_DATA_SIZE(journalfile, total_file_size);
database/engine/metric.c
+4 -4
@@ -331,11 +331,11 @@ time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
331 bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
332 netdata_spinlock_lock(&metric->timestamps_lock);
333
334 - internal_fatal(latest_time_s > now_realtime_sec() + 1,
335 - "DBENGINE METRIC: metric latest time is in the future");
334 +// internal_fatal(latest_time_s > now_realtime_sec() + 1,
335 +// "DBENGINE METRIC: metric latest time is in the future");
336
337 - internal_fatal(metric->latest_time_s_clean > latest_time_s,
338 - "DBENGINE METRIC: metric new clean latest time is older than the previous one");
337 +// internal_fatal(metric->latest_time_s_clean > latest_time_s,
338 +// "DBENGINE METRIC: metric new clean latest time is older than the previous one");
339
340 metric->latest_time_s_clean = latest_time_s;
341
database/engine/pagecache.c
+2 -1
@@ -440,6 +440,7 @@ static size_t list_has_time_gaps(
440 (*pages_pending)++;
441
442 if (pd->status & PDC_PAGE_DISK_PENDING) {
443 + internal_fatal(pd->status & PDC_PAGE_SKIP, "page is disk pending and skipped");
444 internal_fatal(!pd->datafile.ptr, "datafile is NULL");
445 internal_fatal(!pd->datafile.extent.bytes, "datafile.extent.bytes zero");
446 internal_fatal(!pd->datafile.extent.pos, "datafile.extent.pos is zero");
@@ -633,7 +634,7 @@ static Pvoid_t get_page_list(
634 pass1_ut = now_monotonic_usec();
635 size_t pages_pass1 = get_page_list_from_pgc(main_cache, metric, ctx, wanted_start_time_s, wanted_end_time_s,
636 &JudyL_page_array, &cache_gaps,
636 - false, PDC_PAGE_PRELOADED_PASS1 | PDC_PAGE_SOURCE_MAIN_CACHE);
637 + false, PDC_PAGE_SOURCE_MAIN_CACHE);
638 query_gaps += cache_gaps;
639 pages_found_in_main_cache += pages_pass1;
640 pages_total += pages_pass1;
database/engine/pdc.c
+129 -60
@@ -12,6 +12,7 @@ struct extent_page_details_list {
12 struct rrdengine_datafile *datafile;
13
14 struct rrdeng_cmd *cmd;
15 + bool head_to_datafile_extent_queries_pending_for_extent;
16
17 struct {
18 struct extent_page_details_list *prev;
@@ -59,18 +60,24 @@ static struct {
60 },
61 };
62
62 -void pdc_cleanup(void) {
63 - netdata_spinlock_lock(&pdc_globals.protected.spinlock);
63 +void pdc_cleanup1(void) {
64 + PDC *item = NULL;
65 +
66 + if(!netdata_spinlock_trylock(&pdc_globals.protected.spinlock))
67 + return;
68
65 - while(pdc_globals.protected.available_items && pdc_globals.protected.available > (size_t)libuv_worker_threads) {
66 - PDC *item = pdc_globals.protected.available_items;
69 + if(pdc_globals.protected.available_items && pdc_globals.protected.available > (size_t)libuv_worker_threads) {
70 + item = pdc_globals.protected.available_items;
71 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(pdc_globals.protected.available_items, item, cache.prev, cache.next);
68 - freez(item);
72 pdc_globals.protected.available--;
70 - __atomic_sub_fetch(&pdc_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
73 }
74
75 netdata_spinlock_unlock(&pdc_globals.protected.spinlock);
76 +
77 + if(item) {
78 + freez(item);
79 + __atomic_sub_fetch(&pdc_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
80 + }
81 }
82
83 PDC *pdc_get(void) {
@@ -132,18 +139,24 @@ static struct {
139 },
140 };
141
135 -void page_details_cleanup(void) {
136 - netdata_spinlock_lock(&page_details_globals.protected.spinlock);
142 +void page_details_cleanup1(void) {
143 + struct page_details *item = NULL;
144
138 - while(page_details_globals.protected.available_items && page_details_globals.protected.available > (size_t)libuv_worker_threads * 2) {
139 - struct page_details *item = page_details_globals.protected.available_items;
145 + if(!netdata_spinlock_trylock(&page_details_globals.protected.spinlock))
146 + return;
147 +
148 + if(page_details_globals.protected.available_items && page_details_globals.protected.available > (size_t)libuv_worker_threads * 2) {
149 + item = page_details_globals.protected.available_items;
150 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(page_details_globals.protected.available_items, item, cache.prev, cache.next);
141 - freez(item);
151 page_details_globals.protected.available--;
143 - __atomic_sub_fetch(&page_details_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
152 }
153
154 netdata_spinlock_unlock(&page_details_globals.protected.spinlock);
155 +
156 + if(item) {
157 + freez(item);
158 + __atomic_sub_fetch(&page_details_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
159 + }
160 }
161
162 struct page_details *page_details_get(void) {
@@ -205,18 +218,24 @@ static struct {
218 },
219 };
220
208 -void epdl_cleanup(void) {
209 - netdata_spinlock_lock(&epdl_globals.protected.spinlock);
221 +void epdl_cleanup1(void) {
222 + EPDL *item = NULL;
223
211 - while(epdl_globals.protected.available_items && epdl_globals.protected.available > 100) {
212 - EPDL *item = epdl_globals.protected.available_items;
224 + if(!netdata_spinlock_trylock(&epdl_globals.protected.spinlock))
225 + return;
226 +
227 + if(epdl_globals.protected.available_items && epdl_globals.protected.available > 100) {
228 + item = epdl_globals.protected.available_items;
229 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(epdl_globals.protected.available_items, item, cache.prev, cache.next);
214 - freez(item);
230 epdl_globals.protected.available--;
216 - __atomic_sub_fetch(&epdl_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
231 }
232
233 netdata_spinlock_unlock(&epdl_globals.protected.spinlock);
234 +
235 + if(item) {
236 + freez(item);
237 + __atomic_sub_fetch(&epdl_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
238 + }
239 }
240
241 static EPDL *epdl_get(void) {
@@ -278,18 +297,24 @@ static struct {
297 },
298 };
299
281 -void deol_cleanup(void) {
282 - netdata_spinlock_lock(&deol_globals.protected.spinlock);
300 +void deol_cleanup1(void) {
301 + DEOL *item = NULL;
302
284 - while(deol_globals.protected.available_items && deol_globals.protected.available > 100) {
285 - DEOL *item = deol_globals.protected.available_items;
303 + if(!netdata_spinlock_trylock(&deol_globals.protected.spinlock))
304 + return;
305 +
306 + if(deol_globals.protected.available_items && deol_globals.protected.available > 100) {
307 + item = deol_globals.protected.available_items;
308 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(deol_globals.protected.available_items, item, cache.prev, cache.next);
287 - freez(item);
309 deol_globals.protected.available--;
289 - __atomic_sub_fetch(&deol_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
310 }
311
312 netdata_spinlock_unlock(&deol_globals.protected.spinlock);
313 +
314 + if(item) {
315 + freez(item);
316 + __atomic_sub_fetch(&deol_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
317 + }
318 }
319
320 static DEOL *deol_get(void) {
@@ -366,20 +391,26 @@ void extent_buffer_init(void) {
391 extent_buffer_globals.max_size = max_size;
392 }
393
369 -void extent_buffer_cleanup(void) {
370 - netdata_spinlock_lock(&extent_buffer_globals.protected.spinlock);
394 +void extent_buffer_cleanup1(void) {
395 + struct extent_buffer *item = NULL;
396
372 - while(extent_buffer_globals.protected.available_items && extent_buffer_globals.protected.available > 1) {
373 - struct extent_buffer *item = extent_buffer_globals.protected.available_items;
374 - size_t bytes = sizeof(struct extent_buffer) + item->bytes;
397 + if(!netdata_spinlock_trylock(&extent_buffer_globals.protected.spinlock))
398 + return;
399 +
400 + if(extent_buffer_globals.protected.available_items && extent_buffer_globals.protected.available > 1) {
401 + item = extent_buffer_globals.protected.available_items;
402 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(extent_buffer_globals.protected.available_items, item, cache.prev, cache.next);
376 - freez(item);
403 extent_buffer_globals.protected.available--;
378 - __atomic_sub_fetch(&extent_buffer_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
379 - __atomic_sub_fetch(&extent_buffer_globals.atomics.allocated_bytes, bytes, __ATOMIC_RELAXED);
404 }
405
406 netdata_spinlock_unlock(&extent_buffer_globals.protected.spinlock);
407 +
408 + if(item) {
409 + size_t bytes = sizeof(struct extent_buffer) + item->bytes;
410 + freez(item);
411 + __atomic_sub_fetch(&extent_buffer_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
412 + __atomic_sub_fetch(&extent_buffer_globals.atomics.allocated_bytes, bytes, __ATOMIC_RELAXED);
413 + }
414 }
415
416 struct extent_buffer *extent_buffer_get(size_t size) {
@@ -462,7 +493,7 @@ static void epdl_mark_all_not_loaded_pages_as_failed(EPDL *epdl, PDC_PAGE_STATUS
493 while ((PValue = PDCJudyLFirstThenNext(*pd_by_start_time_s_JudyL, &start_time_index, &start_time_first))) {
494 struct page_details *pd = *PValue;
495
465 - if(!pd->page) {
496 + if(!pd->page && !pdc_page_status_check(pd, PDC_PAGE_FAILED|PDC_PAGE_READY)) {
497 pdc_page_status_set(pd, PDC_PAGE_FAILED | tags);
498 pages_matched++;
499 }
@@ -522,7 +553,7 @@ static void pdc_destroy(PDC *pdc) {
553 struct page_details *pd;
554 Word_t time_index = 0;
555 bool first_then_next = true;
525 - size_t unroutable = 0;
556 + size_t unroutable = 0, cancelled = 0;
557 while((PValue = PDCJudyLFirstThenNext(pdc->page_list_JudyL, &time_index, &first_then_next))) {
558 pd = *PValue;
559
@@ -536,10 +567,12 @@ static void pdc_destroy(PDC *pdc) {
567
568 internal_fatal(pd->datafile.ptr, "DBENGINE: page details has a datafile.ptr that is not released.");
569
539 - if(!pd->page && !(status & (PDC_PAGE_READY | PDC_PAGE_FAILED | PDC_PAGE_RELEASED | PDC_PAGE_SKIP | PDC_PAGE_INVALID))) {
570 + if(!pd->page && !(status & (PDC_PAGE_READY | PDC_PAGE_FAILED | PDC_PAGE_RELEASED | PDC_PAGE_SKIP | PDC_PAGE_INVALID | PDC_PAGE_CANCELLED))) {
571 // pdc_page_status_set(pd, PDC_PAGE_FAILED);
572 unroutable++;
573 }
574 + else if(!pd->page && (status & PDC_PAGE_CANCELLED))
575 + cancelled++;
576
577 if(pd->page && !(status & PDC_PAGE_RELEASED)) {
578 pgc_page_release(main_cache, pd->page);
@@ -557,6 +590,9 @@ static void pdc_destroy(PDC *pdc) {
590
591 if(unroutable)
592 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_fail_unroutable, unroutable, __ATOMIC_RELAXED);
593 +
594 + if(cancelled)
595 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_fail_cancelled, cancelled, __ATOMIC_RELAXED);
596 }
597
598 void pdc_acquire(PDC *pdc) {
@@ -619,10 +655,13 @@ static bool epdl_pending_add(EPDL *epdl) {
655
656 EPDL *base = *PValue;
657
622 - if(!base)
658 + if(!base) {
659 added_new = true;
660 + epdl->head_to_datafile_extent_queries_pending_for_extent = true;
661 + }
662 else {
663 added_new = false;
664 + epdl->head_to_datafile_extent_queries_pending_for_extent = false;
665 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_extent_merged, 1, __ATOMIC_RELAXED);
666
667 if(base->pdc->priority > epdl->pdc->priority)
@@ -639,9 +678,12 @@ static bool epdl_pending_add(EPDL *epdl) {
678
679 static void epdl_pending_del(EPDL *epdl) {
680 netdata_spinlock_lock(&epdl->datafile->extent_queries.spinlock);
642 - int rc = JudyLDel(&epdl->datafile->extent_queries.pending_epdl_by_extent_offset_judyL, epdl->extent_offset, PJE0);
643 - (void)rc;
644 - internal_fatal(!rc, "DBENGINE: epdl not found in pending list");
681 + if(epdl->head_to_datafile_extent_queries_pending_for_extent) {
682 + epdl->head_to_datafile_extent_queries_pending_for_extent = false;
683 + int rc = JudyLDel(&epdl->datafile->extent_queries.pending_epdl_by_extent_offset_judyL, epdl->extent_offset, PJE0);
684 + (void) rc;
685 + internal_fatal(!rc, "DBENGINE: epdl not found in pending list");
686 + }
687 netdata_spinlock_unlock(&epdl->datafile->extent_queries.spinlock);
688 }
689
@@ -858,22 +900,32 @@ inline VALIDATED_PAGE_DESCRIPTOR validate_extent_page_descr(const struct rrdeng_
900 return vd;
901 }
902
861 -static struct page_details *epdl_to_pd_load_list(EPDL *epdl, Word_t metric_id, time_t start_time_s) {
903 +static 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) {
904 +
905 + if(unlikely(epdl->head_to_datafile_extent_queries_pending_for_extent))
906 + // stop appending more pages to this epdl
907 + epdl_pending_del(epdl);
908 +
909 struct page_details *pd_list = NULL;
910
911 for(EPDL *ep = epdl; ep ;ep = ep->query.next) {
912 Pvoid_t *pd_by_start_time_s_judyL = PDCJudyLGet(ep->page_details_by_metric_id_JudyL, metric_id, PJE0);
913 internal_fatal(pd_by_start_time_s_judyL == PJERR, "DBENGINE: corrupted extent metrics JudyL");
914
868 - if (pd_by_start_time_s_judyL && *pd_by_start_time_s_judyL) {
915 + if (unlikely(pd_by_start_time_s_judyL && *pd_by_start_time_s_judyL)) {
916 Pvoid_t *pd_pptr = PDCJudyLGet(*pd_by_start_time_s_judyL, start_time_s, PJE0);
917 internal_fatal(pd_pptr == PJERR, "DBENGINE: corrupted metric page details JudyHS");
918
872 - if (pd_pptr && *pd_pptr) {
919 + if(likely(pd_pptr && *pd_pptr)) {
920 struct page_details *pd = *pd_pptr;
921 internal_fatal(metric_id != pd->metric_id, "DBENGINE: metric ids do not match");
922
876 - DOUBLE_LINKED_LIST_APPEND_UNSAFE(pd_list, pd, load.prev, load.next);
923 + if(likely(!pd->page)) {
924 + if (unlikely(__atomic_load_n(&ep->pdc->workers_should_stop, __ATOMIC_RELAXED)))
925 + pdc_page_status_set(pd, PDC_PAGE_FAILED | PDC_PAGE_CANCELLED);
926 + else
927 + DOUBLE_LINKED_LIST_APPEND_UNSAFE(pd_list, pd, load.prev, load.next);
928 + }
929 }
930 }
931 }
@@ -904,6 +956,14 @@ static bool epdl_populate_pages_from_extent_data(
956 bool can_use_data = true;
957 if(data_length < sizeof(*header) + sizeof(header->descr[0]) + sizeof(*trailer)) {
958 can_use_data = false;
959 +
960 + // added to satisfy the requirements of older compilers (prevent warnings)
961 + payload_length = 0;
962 + payload_offset = 0;
963 + trailer_offset = 0;
964 + count = 0;
965 + header = NULL;
966 + trailer = NULL;
967 }
968 else {
969 header = data;
@@ -1006,7 +1066,7 @@ static bool epdl_populate_pages_from_extent_data(
1066 }
1067 mrg_metric_release(main_mrg, metric);
1068
1009 - struct page_details *pd_list = epdl_to_pd_load_list(epdl, metric_id, start_time_s);
1069 + struct page_details *pd_list = epdl_get_pd_load_link_list_from_metric_start_time(epdl, metric_id, start_time_s);
1070 if(likely(!pd_list))
1071 continue;
1072
@@ -1114,7 +1174,8 @@ static bool epdl_populate_pages_from_extent_data(
1174 }
1175
1176 void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *epdl, bool worker) {
1117 - epdl_pending_del(epdl);
1177 + size_t *statistics_counter = NULL;
1178 + PDC_PAGE_STATUS not_loaded_pages_tag = 0, loaded_pages_tag = 0;
1179
1180 bool should_stop = __atomic_load_n(&epdl->pdc->workers_should_stop, __ATOMIC_RELAXED);
1181 for(EPDL *ep = epdl->query.next; ep ;ep = ep->query.next) {
@@ -1123,20 +1184,21 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1184 internal_fatal(ep->extent_size != epdl->extent_size, "DBENGINE: extent sizes do not match");
1185 internal_fatal(ep->file != epdl->file, "DBENGINE: files do not match");
1186
1126 - PDC *pdc = ep->pdc;
1127 - if(!__atomic_load_n(&pdc->workers_should_stop, __ATOMIC_RELAXED)) {
1187 + if(!__atomic_load_n(&ep->pdc->workers_should_stop, __ATOMIC_RELAXED)) {
1188 should_stop = false;
1189 break;
1190 }
1191 }
1192
1133 - if(should_stop)
1193 + if(unlikely(should_stop)) {
1194 + statistics_counter = &rrdeng_cache_efficiency_stats.pages_load_fail_cancelled;
1195 + not_loaded_pages_tag = PDC_PAGE_CANCELLED;
1196 goto cleanup;
1197 + }
1198
1199 if(worker)
1200 worker_is_busy(UV_EVENT_EXTENT_CACHE);
1201
1139 - PDC_PAGE_STATUS not_loaded_pages_tag = 0, loaded_pages_tag = 0;
1202 bool extent_found_in_cache = false;
1203
1204 void *extent_compressed_data = NULL;
@@ -1150,8 +1212,8 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1212 internal_fatal(epdl->extent_size != pgc_page_data_size(extent_cache, extent_cache_page),
1213 "DBENGINE: cache size does not match the expected size");
1214
1153 - loaded_pages_tag |= PDC_PAGE_LOADED_FROM_EXTENT_CACHE;
1154 - not_loaded_pages_tag |= PDC_PAGE_LOADED_FROM_EXTENT_CACHE;
1215 + loaded_pages_tag |= PDC_PAGE_EXTENT_FROM_CACHE;
1216 + not_loaded_pages_tag |= PDC_PAGE_EXTENT_FROM_CACHE;
1217 extent_found_in_cache = true;
1218 }
1219 else {
@@ -1194,8 +1256,8 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1256
1257 extent_compressed_data = pgc_page_data(extent_cache_page);
1258
1197 - loaded_pages_tag |= PDC_PAGE_LOADED_FROM_DISK;
1198 - not_loaded_pages_tag |= PDC_PAGE_LOADED_FROM_DISK;
1259 + loaded_pages_tag |= PDC_PAGE_EXTENT_FROM_DISK;
1260 + not_loaded_pages_tag |= PDC_PAGE_EXTENT_FROM_DISK;
1261 }
1262 }
1263
@@ -1208,26 +1270,33 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1270 if(extent_used) {
1271 // since the extent was used, all the pages that are not
1272 // loaded from this extent, were not found in the extent
1211 - not_loaded_pages_tag |= PDC_PAGE_FAILED_UUID_NOT_IN_EXTENT;
1273 + not_loaded_pages_tag |= PDC_PAGE_FAILED_NOT_IN_EXTENT;
1274 + statistics_counter = &rrdeng_cache_efficiency_stats.pages_load_fail_not_found;
1275 }
1213 - else
1276 + else {
1277 not_loaded_pages_tag |= PDC_PAGE_FAILED_INVALID_EXTENT;
1278 + statistics_counter = &rrdeng_cache_efficiency_stats.pages_load_fail_invalid_extent;
1279 + }
1280 }
1216 - else
1281 + else {
1282 not_loaded_pages_tag |= PDC_PAGE_FAILED_TO_MAP_EXTENT;
1218 -
1283 + statistics_counter = &rrdeng_cache_efficiency_stats.pages_load_fail_cant_mmap_extent;
1284 + }
1285
1286 if(extent_cache_page)
1287 pgc_page_release(extent_cache, extent_cache_page);
1288
1289 +cleanup:
1290 + // remove it from the datafile extent_queries
1291 + // this can be called multiple times safely
1292 + epdl_pending_del(epdl);
1293 +
1294 // mark all pending pages as failed
1295 for(EPDL *ep = epdl; ep ;ep = ep->query.next) {
1296 epdl_mark_all_not_loaded_pages_as_failed(
1226 - ep, not_loaded_pages_tag,
1227 - &rrdeng_cache_efficiency_stats.pages_load_fail_cant_mmap_extent);
1297 + ep, not_loaded_pages_tag, statistics_counter);
1298 }
1299
1230 -cleanup:
1300 for(EPDL *ep = epdl, *next = NULL; ep ; ep = next) {
1301 next = ep->query.next;
1302
database/engine/pdc.h
+5 -5
@@ -40,11 +40,11 @@ size_t epdl_cache_size(void);
40 size_t deol_cache_size(void);
41 size_t extent_buffer_cache_size(void);
42
43 -void pdc_cleanup(void);
44 -void page_details_cleanup(void);
45 -void epdl_cleanup(void);
46 -void deol_cleanup(void);
47 -void extent_buffer_cleanup(void);
43 +void pdc_cleanup1(void);
44 +void page_details_cleanup1(void);
45 +void epdl_cleanup1(void);
46 +void deol_cleanup1(void);
47 +void extent_buffer_cleanup1(void);
48
49 void epdl_cmd_dequeued(void *epdl_ptr);
50 void epdl_cmd_queued(void *epdl_ptr, struct rrdeng_cmd *cmd);
database/engine/rrdengine.c
+114 -63
@@ -23,8 +23,6 @@ struct rrdeng_main {
23 uv_timer_t timer;
24 pid_t tid;
25
26 - time_t last_buffers_cleanup_s;
27 -
26 size_t flushes_running;
27 size_t evictions_running;
28 } rrdeng_main = {
@@ -32,7 +30,6 @@ struct rrdeng_main {
30 .loop = {},
31 .async = {},
32 .timer = {},
35 - .last_buffers_cleanup_s = 0,
33 .flushes_running = 0,
34 .evictions_running = 0,
35 };
@@ -117,16 +114,23 @@ static inline bool work_request_full(void) {
114 return __atomic_load_n(&work_request_globals.atomics.dispatched, __ATOMIC_RELAXED) >= (size_t)(libuv_worker_threads - RESERVED_LIBUV_WORKER_THREADS);
115 }
116
120 -static void work_request_cleanup(void) {
121 - netdata_spinlock_lock(&work_request_globals.protected.spinlock);
122 - while(work_request_globals.protected.available_items && work_request_globals.protected.available > (size_t)libuv_worker_threads) {
123 - struct rrdeng_work *item = work_request_globals.protected.available_items;
117 +static void work_request_cleanup1(void) {
118 + struct rrdeng_work *item = NULL;
119 +
120 + if(!netdata_spinlock_trylock(&work_request_globals.protected.spinlock))
121 + return;
122 +
123 + if(work_request_globals.protected.available_items && work_request_globals.protected.available > (size_t)libuv_worker_threads) {
124 + item = work_request_globals.protected.available_items;
125 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(work_request_globals.protected.available_items, item, cache.prev, cache.next);
125 - freez(item);
126 work_request_globals.protected.available--;
127 - __atomic_sub_fetch(&work_request_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
127 }
128 netdata_spinlock_unlock(&work_request_globals.protected.spinlock);
129 +
130 + if(item) {
131 + freez(item);
132 + __atomic_sub_fetch(&work_request_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
133 + }
134 }
135
136 static inline void work_done(struct rrdeng_work *work_request) {
@@ -232,18 +236,24 @@ static struct {
236 },
237 };
238
235 -static void page_descriptor_cleanup(void) {
236 - netdata_spinlock_lock(&page_descriptor_globals.protected.spinlock);
239 +static void page_descriptor_cleanup1(void) {
240 + struct page_descr_with_data *item = NULL;
241 +
242 + if(!netdata_spinlock_trylock(&page_descriptor_globals.protected.spinlock))
243 + return;
244
238 - while(page_descriptor_globals.protected.available_items && page_descriptor_globals.protected.available > MAX_PAGES_PER_EXTENT) {
239 - struct page_descr_with_data *item = page_descriptor_globals.protected.available_items;
245 + if(page_descriptor_globals.protected.available_items && page_descriptor_globals.protected.available > MAX_PAGES_PER_EXTENT) {
246 + item = page_descriptor_globals.protected.available_items;
247 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(page_descriptor_globals.protected.available_items, item, cache.prev, cache.next);
241 - freez(item);
248 page_descriptor_globals.protected.available--;
243 - __atomic_sub_fetch(&page_descriptor_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
249 }
250
251 netdata_spinlock_unlock(&page_descriptor_globals.protected.spinlock);
252 +
253 + if(item) {
254 + freez(item);
255 + __atomic_sub_fetch(&page_descriptor_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
256 + }
257 }
258
259 struct page_descr_with_data *page_descriptor_get(void) {
@@ -302,16 +312,23 @@ static struct {
312 },
313 };
314
305 -static void extent_io_descriptor_cleanup(void) {
306 - netdata_spinlock_lock(&extent_io_descriptor_globals.protected.spinlock);
307 - while(extent_io_descriptor_globals.protected.available_items && extent_io_descriptor_globals.protected.available > (size_t)libuv_worker_threads) {
308 - struct extent_io_descriptor *item = extent_io_descriptor_globals.protected.available_items;
315 +static void extent_io_descriptor_cleanup1(void) {
316 + struct extent_io_descriptor *item = NULL;
317 +
318 + if(!netdata_spinlock_trylock(&extent_io_descriptor_globals.protected.spinlock))
319 + return;
320 +
321 + if(extent_io_descriptor_globals.protected.available_items && extent_io_descriptor_globals.protected.available > (size_t)libuv_worker_threads) {
322 + item = extent_io_descriptor_globals.protected.available_items;
323 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(extent_io_descriptor_globals.protected.available_items, item, cache.prev, cache.next);
310 - freez(item);
324 extent_io_descriptor_globals.protected.available--;
312 - __atomic_sub_fetch(&extent_io_descriptor_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
325 }
326 netdata_spinlock_unlock(&extent_io_descriptor_globals.protected.spinlock);
327 +
328 + if(item) {
329 + freez(item);
330 + __atomic_sub_fetch(&extent_io_descriptor_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
331 + }
332 }
333
334 static struct extent_io_descriptor *extent_io_descriptor_get(void) {
@@ -369,18 +386,24 @@ static struct {
386 },
387 };
388
372 -static void rrdeng_query_handle_cleanup(void) {
373 - netdata_spinlock_lock(&rrdeng_query_handle_globals.protected.spinlock);
389 +static void rrdeng_query_handle_cleanup1(void) {
390 + struct rrdeng_query_handle *item = NULL;
391
375 - while(rrdeng_query_handle_globals.protected.available_items && rrdeng_query_handle_globals.protected.available > 10) {
376 - struct rrdeng_query_handle *item = rrdeng_query_handle_globals.protected.available_items;
392 + if(!netdata_spinlock_trylock(&rrdeng_query_handle_globals.protected.spinlock))
393 + return;
394 +
395 + if(rrdeng_query_handle_globals.protected.available_items && rrdeng_query_handle_globals.protected.available > 10) {
396 + item = rrdeng_query_handle_globals.protected.available_items;
397 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(rrdeng_query_handle_globals.protected.available_items, item, cache.prev, cache.next);
378 - freez(item);
398 rrdeng_query_handle_globals.protected.available--;
380 - __atomic_sub_fetch(&rrdeng_query_handle_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
399 }
400
401 netdata_spinlock_unlock(&rrdeng_query_handle_globals.protected.spinlock);
402 +
403 + if(item) {
404 + freez(item);
405 + __atomic_sub_fetch(&rrdeng_query_handle_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
406 + }
407 }
408
409 struct rrdeng_query_handle *rrdeng_query_handle_get(void) {
@@ -438,19 +461,25 @@ static struct {
461 },
462 };
463
441 -static void wal_cleanup(void) {
442 - netdata_spinlock_lock(&wal_globals.protected.spinlock);
464 +static void wal_cleanup1(void) {
465 + WAL *wal = NULL;
466
444 - while(wal_globals.protected.available_items && wal_globals.protected.available > storage_tiers) {
445 - WAL *wal = wal_globals.protected.available_items;
467 + if(!netdata_spinlock_trylock(&wal_globals.protected.spinlock))
468 + return;
469 +
470 + if(wal_globals.protected.available_items && wal_globals.protected.available > storage_tiers) {
471 + wal = wal_globals.protected.available_items;
472 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(wal_globals.protected.available_items, wal, cache.prev, cache.next);
447 - posix_memfree(wal->buf);
448 - freez(wal);
473 wal_globals.protected.available--;
450 - __atomic_sub_fetch(&wal_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
474 }
475
476 netdata_spinlock_unlock(&wal_globals.protected.spinlock);
477 +
478 + if(wal) {
479 + posix_memfree(wal->buf);
480 + freez(wal);
481 + __atomic_sub_fetch(&wal_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
482 + }
483 }
484
485 WAL *wal_get(struct rrdengine_instance *ctx, unsigned size) {
@@ -557,16 +586,23 @@ static struct {
586 },
587 };
588
560 -static void rrdeng_cmd_cleanup(void) {
561 - netdata_spinlock_lock(&rrdeng_cmd_globals.cache.spinlock);
562 - while(rrdeng_cmd_globals.cache.available_items && rrdeng_cmd_globals.cache.available > 100) {
563 - struct rrdeng_cmd *item = rrdeng_cmd_globals.cache.available_items;
589 +static void rrdeng_cmd_cleanup1(void) {
590 + struct rrdeng_cmd *item = NULL;
591 +
592 + if(!netdata_spinlock_trylock(&rrdeng_cmd_globals.cache.spinlock))
593 + return;
594 +
595 + if(rrdeng_cmd_globals.cache.available_items && rrdeng_cmd_globals.cache.available > 100) {
596 + item = rrdeng_cmd_globals.cache.available_items;
597 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(rrdeng_cmd_globals.cache.available_items, item, cache.prev, cache.next);
565 - freez(item);
598 rrdeng_cmd_globals.cache.available--;
567 - __atomic_sub_fetch(&rrdeng_cmd_globals.cache.atomics.allocated, 1, __ATOMIC_RELAXED);
599 }
600 netdata_spinlock_unlock(&rrdeng_cmd_globals.cache.spinlock);
601 +
602 + if(item) {
603 + freez(item);
604 + __atomic_sub_fetch(&rrdeng_cmd_globals.cache.atomics.allocated, 1, __ATOMIC_RELAXED);
605 + }
606 }
607
608 void rrdeng_enqueue_epdl_cmd(struct rrdeng_cmd *cmd) {
@@ -994,18 +1030,38 @@ static int journal_metric_uuid_compare(const void *key, const void *metric)
1030 return uuid_compare(*(uuid_t *) key, ((struct journal_metric_list *) metric)->uuid);
1031 }
1032
997 -void find_uuid_first_time(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, Pvoid_t metric_first_time_JudyL)
998 -{
1033 +struct rrdengine_datafile *datafile_release_and_acquire_next_for_retention(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile) {
1034 +
1035 + uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1036 +
1037 + struct rrdengine_datafile *next_datafile = datafile->next;
1038 +
1039 + while(next_datafile && !datafile_acquire(next_datafile, DATAFILE_ACQUIRE_RETENTION))
1040 + next_datafile = next_datafile->next;
1041 +
1042 + uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1043 +
1044 + datafile_release(datafile, DATAFILE_ACQUIRE_RETENTION);
1045 +
1046 + return next_datafile;
1047 +}
1048 +
1049 +void find_uuid_first_time(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, Pvoid_t metric_first_time_JudyL) {
1050 + // acquire the datafile to work with it
1051 + uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1052 + while(datafile && !datafile_acquire(datafile, DATAFILE_ACQUIRE_RETENTION))
1053 + datafile = datafile->next;
1054 + uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1055 +
1056 if (unlikely(!datafile))
1057 return;
1058
1059 unsigned v2_count = 0;
1060 unsigned journalfile_count = 0;
1004 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1061 while (datafile) {
1062 struct journal_v2_header *journal_header = (struct journal_v2_header *) GET_JOURNAL_DATA(datafile->journalfile);
1063 if (!journal_header || !datafile->users.available) {
1008 - datafile = datafile->next;
1064 + datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
1065 continue;
1066 }
1067
@@ -1031,9 +1087,8 @@ void find_uuid_first_time(struct rrdengine_instance *ctx, struct rrdengine_dataf
1087 v2_count++;
1088 }
1089 journalfile_count++;
1034 - datafile = datafile->next;
1090 + datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
1091 }
1036 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1092
1093 // Let's scan the open cache for almost exact match
1094 bool first_then_next = true;
@@ -1388,25 +1443,21 @@ void timer_cb(uv_timer_t* handle) {
1443 rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_INIT, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1444 rrdeng_enq_cmd(NULL, RRDENG_OPCODE_EVICT_INIT, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1445
1391 - time_t now_s = now_monotonic_sec();
1392 - if(now_s - rrdeng_main.last_buffers_cleanup_s > 600) {
1393 - rrdeng_main.last_buffers_cleanup_s = now_s;
1394 -
1395 - work_request_cleanup();
1396 - page_descriptor_cleanup();
1397 - extent_io_descriptor_cleanup();
1398 - rrdeng_cmd_cleanup();
1399 - pdc_cleanup();
1400 - page_details_cleanup();
1401 - rrdeng_query_handle_cleanup();
1402 - wal_cleanup();
1403 - extent_buffer_cleanup();
1404 - epdl_cleanup();
1405 - deol_cleanup();
1446 + rrdeng_cmd_cleanup1();
1447 + work_request_cleanup1();
1448 + page_descriptor_cleanup1();
1449 + extent_io_descriptor_cleanup1();
1450 + pdc_cleanup1();
1451 + page_details_cleanup1();
1452 + rrdeng_query_handle_cleanup1();
1453 + wal_cleanup1();
1454 + extent_buffer_cleanup1();
1455 + epdl_cleanup1();
1456 + deol_cleanup1();
1457 +
1458 #ifdef PDC_USE_JULYL
1407 - julyl_cleanup();
1459 + julyl_cleanup1();
1460 #endif
1409 - }
1461
1462 worker_is_idle();
1463 }
database/engine/rrdengine.h
+8 -9
@@ -88,20 +88,19 @@ typedef enum __attribute__ ((__packed__)) {
88
89 // worker related statuses
90 PDC_PAGE_FAILED_INVALID_EXTENT = (1 << 9),
91 - PDC_PAGE_FAILED_UUID_NOT_IN_EXTENT = (1 << 10),
91 + PDC_PAGE_FAILED_NOT_IN_EXTENT = (1 << 10),
92 PDC_PAGE_FAILED_TO_MAP_EXTENT = (1 << 11),
93 PDC_PAGE_FAILED_TO_ACQUIRE_DATAFILE= (1 << 12),
94
95 - PDC_PAGE_LOADED_FROM_EXTENT_CACHE = (1 << 13),
96 - PDC_PAGE_LOADED_FROM_DISK = (1 << 14),
95 + PDC_PAGE_EXTENT_FROM_CACHE = (1 << 13),
96 + PDC_PAGE_EXTENT_FROM_DISK = (1 << 14),
97
98 - PDC_PAGE_PRELOADED_PASS1 = (1 << 15),
99 - PDC_PAGE_PRELOADED_PASS4 = (1 << 16),
100 - PDC_PAGE_PRELOADED_WORKER = (1 << 17),
98 + PDC_PAGE_CANCELLED = (1 << 15), // the query thread had left when we try to load the page
99
102 - PDC_PAGE_SOURCE_MAIN_CACHE = (1 << 19),
103 - PDC_PAGE_SOURCE_OPEN_CACHE = (1 << 19),
104 - PDC_PAGE_SOURCE_JOURNAL_V2 = (1 << 20),
100 + PDC_PAGE_SOURCE_MAIN_CACHE = (1 << 16),
101 + PDC_PAGE_SOURCE_OPEN_CACHE = (1 << 17),
102 + PDC_PAGE_SOURCE_JOURNAL_V2 = (1 << 18),
103 + PDC_PAGE_PRELOADED_PASS4 = (1 << 19),
104
105 // datafile acquired
106 PDC_PAGE_DATAFILE_ACQUIRED = (1 << 30),
database/engine/rrdengineapi.c
+1 -2
@@ -727,9 +727,8 @@ void rrdeng_load_metric_finalize(struct storage_engine_query_handle *rrddim_hand
727 if (handle->page)
728 pgc_page_release(main_cache, handle->page);
729
730 - if(!pdc_release_and_destroy_if_unreferenced(handle->pdc, false, false)) {
730 + if(!pdc_release_and_destroy_if_unreferenced(handle->pdc, false, false))
731 __atomic_store_n(&handle->pdc->workers_should_stop, true, __ATOMIC_RELAXED);
732 - }
732
733 unregister_query_handle(handle);
734 rrdeng_query_handle_release(handle);
database/engine/rrdengineapi.h
+2 -1
@@ -164,8 +164,9 @@ struct rrdeng_cache_efficiency_stats {
164 size_t pages_load_fail_cant_mmap_extent;
165 size_t pages_load_fail_datafile_not_available;
166 size_t pages_load_fail_unroutable;
167 - size_t pages_load_fail_uuid_not_found;
167 + size_t pages_load_fail_not_found;
168 size_t pages_load_fail_invalid_extent;
169 + size_t pages_load_fail_cancelled;
170
171 // timings for query preparation
172 size_t prep_time_to_route;
database/ram/rrddim_mem.c
+1 -1
@@ -136,7 +136,7 @@ STORAGE_COLLECT_HANDLE *rrddim_collect_init(STORAGE_METRIC_HANDLE *db_metric_han
136 RRDDIM *rd = mh->rd;
137
138 update_metric_handle_from_rrddim(mh, rd);
139 - internal_fatal(mh->update_every_s != update_every, "RRDDIM: update requested does not match the dimension");
139 + internal_fatal((uint32_t)mh->update_every_s != update_every, "RRDDIM: update requested does not match the dimension");
140
141 struct mem_collect_handle *ch = callocz(1, sizeof(struct mem_collect_handle));
142 ch->rd = rd;
database/rrdhost.c
+1 -1
@@ -31,7 +31,7 @@ netdata_rwlock_t rrd_rwlock = NETDATA_RWLOCK_INITIALIZER;
31 time_t rrdset_free_obsolete_time_s = 3600;
32 time_t rrdhost_free_orphan_time_s = 3600;
33
34 -bool is_storage_engine_shared(STORAGE_INSTANCE *engine) {
34 +bool is_storage_engine_shared(STORAGE_INSTANCE *engine __maybe_unused) {
35 #ifdef ENABLE_DBENGINE
36 for(size_t tier = 0; tier < storage_tiers ;tier++) {
37 if (engine == (STORAGE_INSTANCE *)multidb_ctx[tier])
database/sqlite/sqlite_metadata.c
+1 -1
@@ -610,7 +610,7 @@ bind_fail:
610 return 1;
611 }
612
613 -static bool dimension_can_be_deleted(uuid_t *dim_uuid)
613 +static bool dimension_can_be_deleted(uuid_t *dim_uuid __maybe_unused)
614 {
615 #ifdef ENABLE_DBENGINE
616 if(dbengine_enabled) {
libnetdata/july/july.c
+13 -7
@@ -56,20 +56,26 @@ static struct {
56 },
57 };
58
59 -void julyl_cleanup(void) {
60 - netdata_spinlock_lock(&julyl_globals.protected.spinlock);
59 +void julyl_cleanup1(void) {
60 + struct JulyL *item = NULL;
61 +
62 + if(!netdata_spinlock_trylock(&julyl_globals.protected.spinlock))
63 + return;
64
62 - while(julyl_globals.protected.available_items && julyl_globals.protected.available > 10) {
63 - struct JulyL *item = julyl_globals.protected.available_items;
65 + if(julyl_globals.protected.available_items && julyl_globals.protected.available > 10) {
66 + item = julyl_globals.protected.available_items;
67 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(julyl_globals.protected.available_items, item, cache.prev, cache.next);
68 + julyl_globals.protected.available--;
69 + }
70 +
71 + netdata_spinlock_unlock(&julyl_globals.protected.spinlock);
72 +
73 + if(item) {
74 size_t bytes = item->bytes;
75 freez(item);
67 - julyl_globals.protected.available--;
76 __atomic_sub_fetch(&julyl_globals.atomics.bytes, bytes, __ATOMIC_RELAXED);
77 __atomic_sub_fetch(&julyl_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
78 }
71 -
72 - netdata_spinlock_unlock(&julyl_globals.protected.spinlock);
79 }
80
81 struct JulyL *julyl_get(void) {
libnetdata/july/july.h
+1 -1
@@ -33,7 +33,7 @@ static inline PPvoid_t JulyLLastThenPrev(Pcvoid_t PArray, Word_t * PIndex, bool
33 return JulyLPrev(PArray, PIndex, PJE0);
34 }
35
36 -void julyl_cleanup(void);
36 +void julyl_cleanup1(void);
37 size_t julyl_cache_size(void);
38 size_t julyl_bytes_moved(void);
39
ml/Dimension.cc
+1 -1
@@ -221,7 +221,7 @@ void Dimension::scheduleForTraining(time_t CurrT) {
221 break;
222 }
223 case TrainingStatus::Trained: {
224 - bool NeedsTraining = LastTrainingTime + (Cfg.TrainEvery * updateEvery()) < CurrT;
224 + bool NeedsTraining = (time_t)(LastTrainingTime + (Cfg.TrainEvery * updateEvery())) < CurrT;
225
226 if (NeedsTraining) {
227 Host *H = reinterpret_cast<Host *>(RD->rrdset->rrdhost->ml_host);
ml/Host.cc
+13 -4
@@ -338,10 +338,10 @@ void Host::startAnomalyDetectionThreads() {
338 char Tag[NETDATA_THREAD_TAG_MAX + 1];
339
340 snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "TRAIN[%s]", rrdhost_hostname(RH));
341 - netdata_thread_create(&TrainingThread, Tag, NETDATA_THREAD_OPTION_JOINABLE, train_main, static_cast<void *>(this));
341 + netdata_thread_create(&TrainingThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, train_main, static_cast<void *>(this));
342
343 snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "DETECT[%s]", rrdhost_hostname(RH));
344 - netdata_thread_create(&DetectionThread, Tag, NETDATA_THREAD_OPTION_JOINABLE, detect_main, static_cast<void *>(this));
344 + netdata_thread_create(&DetectionThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, detect_main, static_cast<void *>(this));
345 }
346
347 void Host::stopAnomalyDetectionThreads(bool join) {
@@ -362,7 +362,16 @@ void Host::stopAnomalyDetectionThreads(bool join) {
362 if(join && !ThreadsJoined) {
363 ThreadsJoined = true;
364 ThreadsRunning = false;
365 - netdata_thread_join(TrainingThread, nullptr);
366 - netdata_thread_join(DetectionThread, nullptr);
365 +
366 + // these fail on alpine linux and our CI hangs forever
367 + // failing to compile static builds
368 +
369 + // commenting them, until we find a solution
370 +
371 + // to enable again:
372 + // NETDATA_THREAD_OPTION_DEFAULT needs to become NETDATA_THREAD_OPTION_JOINABLE
373 +
374 + //netdata_thread_join(TrainingThread, nullptr);
375 + //netdata_thread_join(DetectionThread, nullptr);
376 }
377 }
streaming/compression.c
+1 -1
@@ -244,7 +244,7 @@ static size_t lz4_decompressor_decompress(struct decompressor_state *state, cons
244 , state->stream->size
245 , state->stream->write_at
246 , decompressed_size
247 - , state->stream->write_at + decompressed_size - state->stream->size
247 + , (size_t)(state->stream->write_at + decompressed_size - state->stream->size)
248 );
249
250 state->stream->write_at += decompressed_size;
streaming/rrdpush.c
+1 -1
@@ -974,7 +974,7 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
974 */
975
976 {
977 - time_t age;
977 + time_t age = 0;
978 bool receiver_stale = false;
979 bool receiver_working = false;
980
web/api/queries/query.c
+31 -18
@@ -719,7 +719,11 @@ static size_t query_metric_first_working_tier(QUERY_METRIC *qm) {
719 }
720
721 static long query_plan_points_coverage_weight(time_t db_first_time_s, time_t db_last_time_s, time_t db_update_every_s, time_t after_wanted, time_t before_wanted, size_t points_wanted, size_t tier __maybe_unused) {
722 - if(db_first_time_s == 0 || db_last_time_s == 0 || db_update_every_s == 0)
722 + if(db_first_time_s == 0 ||
723 + db_last_time_s == 0 ||
724 + db_update_every_s == 0 ||
725 + db_first_time_s > before_wanted ||
726 + db_last_time_s < after_wanted)
727 return -LONG_MAX;
728
729 time_t common_first_t = MAX(db_first_time_s, after_wanted);
@@ -774,13 +778,15 @@ static size_t query_metric_best_tier_for_timeframe(QUERY_METRIC *qm, time_t afte
778 !first_time_s ||
779 !last_time_s ||
780 !update_every_s ||
777 - first_time_s > max_last_time_s ||
778 - last_time_s < min_first_time_s
781 + first_time_s > before_wanted ||
782 + last_time_s < after_wanted
783 ) {
784 qm->tiers[tier].weight = -LONG_MAX;
785 continue;
786 }
787
788 + internal_fatal(first_time_s > before_wanted || last_time_s < after_wanted, "QUERY: invalid db durations");
789 +
790 qm->tiers[tier].weight = query_plan_points_coverage_weight(
791 min_first_time_s, max_last_time_s, update_every_s,
792 after_wanted, before_wanted, points_wanted, tier);
@@ -1003,6 +1009,8 @@ static void query_planer_activate_plan(QUERY_ENGINE_OPS *ops, size_t plan_id, ti
1009 internal_fatal(!qm->plan.array[plan_id].initialized, "QUERY: plan has not been initialized");
1010 internal_fatal(qm->plan.array[plan_id].finalized, "QUERY: plan has been finalized");
1011
1012 + internal_fatal(qm->plan.array[plan_id].after > qm->plan.array[plan_id].before, "QUERY: flipped after/before");
1013 +
1014 ops->tier = qm->plan.array[plan_id].tier;
1015 ops->tier_ptr = &qm->tiers[ops->tier];
1016 ops->handle = &qm->plan.array[plan_id].handle;
@@ -1071,12 +1079,16 @@ static bool query_plan(QUERY_ENGINE_OPS *ops, time_t after_wanted, time_t before
1079
1080 if(!query_metric_is_valid_tier(qm, selected_tier))
1081 return false;
1082 +
1083 + if(qm->tiers[selected_tier].db_first_time_s > before_wanted ||
1084 + qm->tiers[selected_tier].db_last_time_s < after_wanted)
1085 + return false;
1086 }
1087
1088 qm->plan.used = 1;
1089 qm->plan.array[0].tier = selected_tier;
1078 - qm->plan.array[0].after = qm->tiers[selected_tier].db_first_time_s;
1079 - qm->plan.array[0].before = qm->tiers[selected_tier].db_last_time_s;
1090 + qm->plan.array[0].after = (qm->tiers[selected_tier].db_first_time_s < after_wanted) ? after_wanted : qm->tiers[selected_tier].db_first_time_s;
1091 + qm->plan.array[0].before = (qm->tiers[selected_tier].db_last_time_s > before_wanted) ? before_wanted : qm->tiers[selected_tier].db_last_time_s;
1092
1093 if(!(ops->r->internal.query_options & RRDR_OPTION_SELECTED_TIER)) {
1094 // the selected tier
@@ -1091,14 +1103,14 @@ static bool query_plan(QUERY_ENGINE_OPS *ops, time_t after_wanted, time_t before
1103 continue;
1104
1105 // find the first time of this tier
1094 - time_t first_time_s = qm->tiers[tr].db_first_time_s;
1106 + time_t tier_first_time_s = qm->tiers[tr].db_first_time_s;
1107
1108 // can it help?
1097 - if (first_time_s < selected_tier_first_time_s) {
1109 + if (tier_first_time_s < selected_tier_first_time_s) {
1110 // it can help us add detail at the beginning of the query
1111 QUERY_PLAN_ENTRY t = {
1112 .tier = tr,
1101 - .after = (first_time_s < after_wanted) ? after_wanted : first_time_s,
1113 + .after = (tier_first_time_s < after_wanted) ? after_wanted : tier_first_time_s,
1114 .before = selected_tier_first_time_s,
1115 .initialized = false,
1116 .finalized = false,
@@ -1124,17 +1136,17 @@ static bool query_plan(QUERY_ENGINE_OPS *ops, time_t after_wanted, time_t before
1136 continue;
1137
1138 // find the last time of this tier
1127 - time_t last_time_s = qm->tiers[tr].db_last_time_s;
1139 + time_t tier_last_time_s = qm->tiers[tr].db_last_time_s;
1140
1141 //buffer_sprintf(wb, ": EVAL BEFORE tier %d, %ld", tier, last_time_s);
1142
1143 // can it help?
1132 - if (last_time_s > selected_tier_last_time_s) {
1144 + if (tier_last_time_s > selected_tier_last_time_s) {
1145 // it can help us add detail at the end of the query
1146 QUERY_PLAN_ENTRY t = {
1147 .tier = tr,
1148 .after = selected_tier_last_time_s,
1137 - .before = (last_time_s > before_wanted) ? before_wanted : last_time_s,
1149 + .before = (tier_last_time_s > before_wanted) ? before_wanted : tier_last_time_s,
1150 .initialized = false,
1151 .finalized = false,
1152 };
@@ -1156,16 +1168,17 @@ static bool query_plan(QUERY_ENGINE_OPS *ops, time_t after_wanted, time_t before
1168 if(qm->plan.used > 1)
1169 qsort(&qm->plan.array, qm->plan.used, sizeof(QUERY_PLAN_ENTRY), compare_query_plan_entries_on_start_time);
1170
1159 - // make sure it has the whole timeframe we need
1160 - if(qm->plan.array[0].after < after_wanted)
1161 - qm->plan.array[0].after = after_wanted;
1162 -
1163 - if(qm->plan.array[qm->plan.used - 1].before > before_wanted)
1164 - qm->plan.array[qm->plan.used - 1].before = before_wanted;
1165 -
1171 if(!query_metric_is_valid_tier(qm, qm->plan.array[0].tier))
1172 return false;
1173
1174 +#ifdef NETDATA_INTERNAL_CHECKS
1175 + for(size_t p = 0; p < qm->plan.used ;p++) {
1176 + internal_fatal(qm->plan.array[p].after > qm->plan.array[p].before, "QUERY: flipped after/before");
1177 + internal_fatal(qm->plan.array[p].after < after_wanted, "QUERY: too small plan first time");
1178 + internal_fatal(qm->plan.array[p].before > before_wanted, "QUERY: too big plan last time");
1179 + }
1180 +#endif
1181 +
1182 query_planer_initialize_plans(ops);
1183 query_planer_activate_plan(ops, 0, 0);
1184
web/api/web_api_v1.c
+1 -1
@@ -1510,7 +1510,7 @@ int web_client_api_request_v1_functions(RRDHOST *host, struct web_client *w, cha
1510 }
1511
1512 #ifndef ENABLE_DBENGINE
1513 -int web_client_api_request_v1_dbengine_stats(RRDHOST *host, struct web_client *w, char *url) {
1513 +int web_client_api_request_v1_dbengine_stats(RRDHOST *host __maybe_unused, struct web_client *w __maybe_unused, char *url __maybe_unused) {
1514 return HTTP_RESP_NOT_FOUND;
1515 }
1516 #else