@cryptotaxi247 / netdata-1 / commits / 3d0c2b9f4

Fast restart on busy parents (#19964)

Prepopulate UUIDs to MRG to improve startup times. - [x] Pre-populate UUIDs to MRG on start (from SQLite) - [x] Cleanup obsolete UUIDs from MRG once finished This improves startup times by 3x to 5x. --------- Co-authored-by: vkalintiris <vasilis@netdata.cloud> Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>

Costa Tsaousis committed Mar 28, 2025 at 19:31 UTC 3d0c2b9f410564d43b7db3580c5d49417ac0e46b
18 files changed +1177 -1020
CMakeLists.txt
+5 -2
@@ -1652,8 +1652,11 @@ if(ENABLE_DBENGINE)
1652 src/database/engine/page.h
1653 src/database/engine/cache.c
1654 src/database/engine/cache.h
1655 - src/database/engine/metric.c
1656 - src/database/engine/metric.h
1655 + src/database/engine/mrg.c
1656 + src/database/engine/mrg.h
1657 + src/database/engine/mrg-internals.h
1658 + src/database/engine/mrg-unittest.c
1659 + src/database/engine/mrg-load.c
1660 src/database/engine/pdc.c
1661 src/database/engine/pdc.h
1662 src/database/engine/dbengine-unittest.c
src/daemon/daemon-shutdown.c
+1 -1
@@ -310,7 +310,7 @@ static void netdata_cleanup_and_exit(EXIT_REASON reason, bool abnormal, bool exi
310 sqlite_close_databases();
311 watcher_step_complete(WATCHER_STEP_ID_CLOSE_SQL_DATABASES);
312 sqlite_library_shutdown();
313 -
313 +
314 // unlink the pid
315 if(pidfile && *pidfile && unlink(pidfile) != 0)
316 netdata_log_error("EXIT: cannot unlink pidfile '%s'.", pidfile);
src/daemon/main.c
+5
@@ -1125,6 +1125,11 @@ int netdata_main(int argc, char **argv) {
1125 webrtc_initialize();
1126 #endif
1127
1128 + // ----------------------------------------------------------------------------------------------------------------
1129 + delta_startup_time("mrg cleanup");
1130 +
1131 + mrg_metric_prepopulate_cleanup(main_mrg);
1132 +
1133 // ----------------------------------------------------------------------------------------------------------------
1134 delta_startup_time("done");
1135
src/database/engine/journalfile.c
+2 -3
@@ -707,10 +707,9 @@ static void journalfile_restore_extent_metadata(struct rrdengine_instance *ctx,
707
708 bool added;
709 metric = mrg_metric_add_and_acquire(main_mrg, entry, &added);
710 - if(added) {
711 - __atomic_add_fetch(&ctx->atomic.metrics, 1, __ATOMIC_RELAXED);
710 + if(added)
711 update_metric_time = false;
713 - }
712 +
713 if (vd.update_every_s) {
714 uint64_t samples = (vd.end_time_s - vd.start_time_s) / vd.update_every_s;
715 __atomic_add_fetch(&ctx->atomic.samples, samples, __ATOMIC_RELAXED);
src/database/engine/metric.c deleted
-993
@@ -1,993 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -#include "metric.h"
3 -#include "cache.h"
4 -#include "libnetdata/locks/locks.h"
5 -#include "rrddiskprotocol.h"
6 -
7 -struct metric {
8 - Word_t section; // never changes
9 - UUIDMAP_ID uuid; // never changes
10 -
11 - REFCOUNT refcount;
12 - uint8_t partition;
13 -
14 - uint32_t latest_update_every_s; // the latest data collection frequency
15 -
16 - time_t first_time_s; // the timestamp of the oldest point in the database
17 - time_t latest_time_s_clean; // the timestamp of the newest point in the database
18 - time_t latest_time_s_hot; // the timestamp of the latest point that has been collected (not yet stored)
19 -
20 -#ifdef NETDATA_INTERNAL_CHECKS
21 - pid_t writer;
22 -#endif
23 -
24 - // THIS IS allocated with malloc()
25 - // YOU HAVE TO INITIALIZE IT YOURSELF !
26 -};
27 -
28 -#define set_metric_field_with_condition(field, value, condition) ({ \
29 - typeof(field) _current = __atomic_load_n(&(field), __ATOMIC_RELAXED); \
30 - typeof(field) _wanted = value; \
31 - bool did_it = true; \
32 - \
33 - do { \
34 - if((condition) && (_current != _wanted)) { \
35 - ; \
36 - } \
37 - else { \
38 - did_it = false; \
39 - break; \
40 - } \
41 - } while(!__atomic_compare_exchange_n(&(field), &_current, _wanted, \
42 - false, __ATOMIC_RELAXED, __ATOMIC_RELAXED)); \
43 - \
44 - did_it; \
45 -})
46 -
47 -static struct aral_statistics mrg_aral_statistics;
48 -
49 -struct mrg {
50 - struct mrg_partition {
51 - ARAL *aral; // not protected by our spinlock - it has its own
52 -
53 - RW_SPINLOCK rw_spinlock;
54 - Pvoid_t uuid_judy; // JudyL: each UUID has a JudyL of sections (tiers)
55 -
56 - struct mrg_statistics stats;
57 - } index[UUIDMAP_PARTITIONS];
58 -};
59 -
60 -static inline void MRG_STATS_DUPLICATE_ADD(MRG *mrg, size_t partition) {
61 - mrg->index[partition].stats.additions_duplicate++;
62 -}
63 -
64 -static inline void MRG_STATS_ADDED_METRIC(MRG *mrg, size_t partition) {
65 - mrg->index[partition].stats.entries++;
66 - mrg->index[partition].stats.additions++;
67 - mrg->index[partition].stats.size += sizeof(METRIC);
68 -}
69 -
70 -static inline void MRG_STATS_DELETED_METRIC(MRG *mrg, size_t partition) {
71 - mrg->index[partition].stats.entries--;
72 - mrg->index[partition].stats.size -= sizeof(METRIC);
73 - mrg->index[partition].stats.deletions++;
74 -}
75 -
76 -static inline void MRG_STATS_SEARCH_HIT(MRG *mrg, size_t partition) {
77 - __atomic_add_fetch(&mrg->index[partition].stats.search_hits, 1, __ATOMIC_RELAXED);
78 -}
79 -
80 -static inline void MRG_STATS_SEARCH_MISS(MRG *mrg, size_t partition) {
81 - __atomic_add_fetch(&mrg->index[partition].stats.search_misses, 1, __ATOMIC_RELAXED);
82 -}
83 -
84 -static inline void MRG_STATS_DELETE_MISS(MRG *mrg, size_t partition) {
85 - mrg->index[partition].stats.delete_misses++;
86 -}
87 -
88 -#define mrg_index_read_lock(mrg, partition) rw_spinlock_read_lock(&(mrg)->index[partition].rw_spinlock)
89 -#define mrg_index_read_unlock(mrg, partition) rw_spinlock_read_unlock(&(mrg)->index[partition].rw_spinlock)
90 -#define mrg_index_write_lock(mrg, partition) rw_spinlock_write_lock(&(mrg)->index[partition].rw_spinlock)
91 -#define mrg_index_write_unlock(mrg, partition) rw_spinlock_write_unlock(&(mrg)->index[partition].rw_spinlock)
92 -
93 -static inline void mrg_stats_judy_mem(MRG *mrg, size_t partition, int64_t judy_mem) {
94 - __atomic_add_fetch(&mrg->index[partition].stats.size, judy_mem, __ATOMIC_RELAXED);
95 -}
96 -
97 -static ALWAYS_INLINE time_t mrg_metric_get_first_time_s_smart(MRG *mrg __maybe_unused, METRIC *metric) {
98 - time_t first_time_s = __atomic_load_n(&metric->first_time_s, __ATOMIC_RELAXED);
99 -
100 - if(first_time_s <= 0) {
101 - first_time_s = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
102 - if(first_time_s <= 0)
103 - first_time_s = __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED);
104 -
105 - if(first_time_s <= 0)
106 - first_time_s = 0;
107 - else
108 - __atomic_store_n(&metric->first_time_s, first_time_s, __ATOMIC_RELAXED);
109 - }
110 -
111 - return first_time_s;
112 -}
113 -
114 -static inline void metric_log(MRG *mrg __maybe_unused, METRIC *metric, const char *msg) {
115 - struct rrdengine_instance *ctx = (struct rrdengine_instance *)metric->section;
116 -
117 - nd_uuid_t uuid;
118 - uuidmap_uuid(metric->uuid, uuid);
119 - char uuid_txt[UUID_STR_LEN];
120 - uuid_unparse_lower(uuid, uuid_txt);
121 - nd_log(NDLS_DAEMON, NDLP_ERR,
122 - "METRIC: %s on %s at tier %d, refcount %d, partition %u, "
123 - "retention [%ld - %ld (hot), %ld (clean)], update every %"PRIu32
124 -#ifdef NETDATA_INTERNAL_CHECKS
125 - ", writer pid %d "
126 -#endif
127 - " --- PLEASE OPEN A GITHUB ISSUE TO REPORT THIS LOG LINE TO NETDATA --- ",
128 - msg,
129 - uuid_txt,
130 - ctx->config.tier,
131 - metric->refcount,
132 - metric->partition,
133 - metric->first_time_s,
134 - metric->latest_time_s_hot,
135 - metric->latest_time_s_clean,
136 - metric->latest_update_every_s
137 -#ifdef NETDATA_INTERNAL_CHECKS
138 - , (int)metric->writer
139 -#endif
140 - );
141 -}
142 -
143 -static ALWAYS_INLINE bool acquired_metric_has_retention(MRG *mrg, METRIC *metric) {
144 - time_t first, last;
145 - mrg_metric_get_retention(mrg, metric, &first, &last, NULL);
146 - bool rc = (first != 0 && last != 0 && first <= last);
147 -
148 - if(!rc && __atomic_load_n(&mrg->index[metric->partition].stats.writers, __ATOMIC_RELAXED) > 0)
149 - rc = true;
150 -
151 - return rc;
152 -}
153 -
154 -static ALWAYS_INLINE void acquired_for_deletion_metric_delete(MRG *mrg, METRIC *metric) {
155 - JudyAllocThreadPulseReset();
156 -
157 - size_t partition = metric->partition;
158 -
159 - mrg_index_write_lock(mrg, partition);
160 -
161 - Pvoid_t *sections_judy_pptr = JudyLGet(mrg->index[partition].uuid_judy, metric->uuid, PJE0);
162 - if(unlikely(sections_judy_pptr == PJERR))
163 - fatal("METRIC: corrupted JudyL");
164 -
165 - if(unlikely(!sections_judy_pptr || !*sections_judy_pptr)) {
166 - MRG_STATS_DELETE_MISS(mrg, partition);
167 - mrg_index_write_unlock(mrg, partition);
168 - return;
169 - }
170 -
171 - int rc = JudyLDel(sections_judy_pptr, metric->section, PJE0);
172 - if(unlikely(!rc)) {
173 - MRG_STATS_DELETE_MISS(mrg, partition);
174 - mrg_index_write_unlock(mrg, partition);
175 - mrg_stats_judy_mem(mrg, partition, JudyAllocThreadPulseGetAndReset());
176 - return;
177 - }
178 -
179 - if(!*sections_judy_pptr) {
180 - rc = JudyLDel(&mrg->index[partition].uuid_judy, metric->uuid, PJE0);
181 -
182 - if(unlikely(!rc))
183 - fatal("DBENGINE METRIC: cannot delete UUID from JudyL");
184 - }
185 -
186 - MRG_STATS_DELETED_METRIC(mrg, partition);
187 -
188 - mrg_index_write_unlock(mrg, partition);
189 - uuidmap_free(metric->uuid);
190 - aral_freez(mrg->index[partition].aral, metric);
191 - mrg_stats_judy_mem(mrg, partition, JudyAllocThreadPulseGetAndReset());
192 -}
193 -
194 -static ALWAYS_INLINE bool metric_acquire(MRG *mrg, METRIC *metric) {
195 - REFCOUNT rc = refcount_acquire_advanced(&metric->refcount);
196 - if(!REFCOUNT_ACQUIRED(rc))
197 - return false;
198 -
199 - size_t partition = metric->partition;
200 -
201 - if(rc == 1)
202 - __atomic_add_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
203 -
204 - __atomic_add_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
205 -
206 - return true;
207 -}
208 -
209 -static ALWAYS_INLINE bool metric_release(MRG *mrg, METRIC *metric) {
210 - size_t partition = metric->partition;
211 -
212 - REFCOUNT refcount = refcount_release(&metric->refcount);
213 -
214 - if(!refcount && !acquired_metric_has_retention(mrg, metric) && refcount_acquire_for_deletion(&metric->refcount))
215 - refcount = REFCOUNT_DELETED;
216 -
217 - if(refcount == 0 || refcount == REFCOUNT_DELETED) {
218 - __atomic_sub_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
219 -
220 - if(refcount == REFCOUNT_DELETED)
221 - acquired_for_deletion_metric_delete(mrg, metric);
222 - }
223 -
224 - __atomic_sub_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
225 -
226 - return refcount == REFCOUNT_DELETED;
227 -}
228 -
229 -static ALWAYS_INLINE METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
230 - JudyAllocThreadPulseReset();
231 -
232 - UUIDMAP_ID id = uuidmap_create(*entry->uuid);
233 -
234 - size_t partition = uuid_to_uuidmap_partition(*entry->uuid);
235 -
236 - METRIC *allocation = aral_mallocz(mrg->index[partition].aral);
237 - Pvoid_t *PValue;
238 -
239 - while(1) {
240 - mrg_index_write_lock(mrg, partition);
241 -
242 - Pvoid_t *sections_judy_pptr = JudyLIns(&mrg->index[partition].uuid_judy, id, PJE0);
243 - if (unlikely(!sections_judy_pptr || sections_judy_pptr == PJERR))
244 - fatal("DBENGINE METRIC: corrupted UUIDs JudyL array");
245 -
246 - PValue = JudyLIns(sections_judy_pptr, entry->section, PJE0);
247 - if (unlikely(!PValue || PValue == PJERR))
248 - fatal("DBENGINE METRIC: corrupted section JudyL array");
249 -
250 - if (unlikely(*PValue != NULL)) {
251 - METRIC *metric = *PValue;
252 -
253 - if(!metric_acquire(mrg, metric)) {
254 - mrg_index_write_unlock(mrg, partition);
255 - continue;
256 - }
257 -
258 - MRG_STATS_DUPLICATE_ADD(mrg, partition);
259 - mrg_index_write_unlock(mrg, partition);
260 -
261 - if (ret)
262 - *ret = false;
263 -
264 - uuidmap_free(id);
265 - aral_freez(mrg->index[partition].aral, allocation);
266 -
267 - mrg_stats_judy_mem(mrg, partition, JudyAllocThreadPulseGetAndReset());
268 - return metric;
269 - }
270 -
271 - break;
272 - }
273 -
274 - METRIC *metric = allocation;
275 - metric->uuid = id;
276 - metric->section = entry->section;
277 - metric->first_time_s = MAX(0, entry->first_time_s);
278 - metric->latest_time_s_clean = MAX(0, entry->last_time_s);
279 - metric->latest_time_s_hot = 0;
280 - metric->latest_update_every_s = entry->latest_update_every_s;
281 -#ifdef NETDATA_INTERNAL_CHECKS
282 - metric->writer = 0;
283 -#endif
284 - metric->refcount = 1;
285 - metric->partition = partition;
286 - *PValue = metric;
287 -
288 - __atomic_add_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
289 - __atomic_add_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
290 -
291 - MRG_STATS_ADDED_METRIC(mrg, partition);
292 -
293 - mrg_index_write_unlock(mrg, partition);
294 -
295 - if(ret)
296 - *ret = true;
297 -
298 - mrg_stats_judy_mem(mrg, partition, JudyAllocThreadPulseGetAndReset());
299 - return metric;
300 -}
301 -
302 -static ALWAYS_INLINE METRIC *metric_get_and_acquire_by_id(MRG *mrg, UUIDMAP_ID id, Word_t section) {
303 - size_t partition = uuidmap_id_to_partition(id);
304 -
305 - while(1) {
306 - mrg_index_read_lock(mrg, partition);
307 -
308 - Pvoid_t *sections_judy_pptr = JudyLGet(mrg->index[partition].uuid_judy, id, PJE0);
309 - if (unlikely(!sections_judy_pptr)) {
310 - mrg_index_read_unlock(mrg, partition);
311 - MRG_STATS_SEARCH_MISS(mrg, partition);
312 - return NULL;
313 - }
314 -
315 - Pvoid_t *PValue = JudyLGet(*sections_judy_pptr, section, PJE0);
316 - if (unlikely(!PValue)) {
317 - mrg_index_read_unlock(mrg, partition);
318 - MRG_STATS_SEARCH_MISS(mrg, partition);
319 - return NULL;
320 - }
321 -
322 - METRIC *metric = *PValue;
323 -
324 - if(metric && !metric_acquire(mrg, metric))
325 - metric = NULL;
326 -
327 - mrg_index_read_unlock(mrg, partition);
328 -
329 - if(metric) {
330 - MRG_STATS_SEARCH_HIT(mrg, partition);
331 - return metric;
332 - }
333 - }
334 -}
335 -
336 -// ----------------------------------------------------------------------------
337 -// public API
338 -
339 -inline MRG *mrg_create(void) {
340 - MRG *mrg = callocz(1, sizeof(MRG));
341 -
342 - for(size_t i = 0; i < _countof(mrg->index) ; i++) {
343 - rw_spinlock_init(&mrg->index[i].rw_spinlock);
344 -
345 - char buf[ARAL_MAX_NAME + 1];
346 - snprintfz(buf, ARAL_MAX_NAME, "mrg[%zu]", i);
347 -
348 - mrg->index[i].aral = aral_create(buf, sizeof(METRIC), 0, 16384, &mrg_aral_statistics, NULL, NULL,
349 - false, false, true);
350 - }
351 - pulse_aral_register_statistics(&mrg_aral_statistics, "mrg");
352 -
353 - return mrg;
354 -}
355 -
356 -struct aral_statistics *mrg_aral_stats(void) {
357 - return &mrg_aral_statistics;
358 -}
359 -
360 -size_t mrg_destroy(MRG *mrg) {
361 - if (!mrg)
362 - return 0;
363 -
364 - size_t referenced = 0;
365 -
366 - // Traverse all partitions
367 - for (size_t partition = 0; partition < UUIDMAP_PARTITIONS; partition++) {
368 - // Lock the partition to prevent new entries while we're cleaning up
369 - mrg_index_write_lock(mrg, partition);
370 -
371 - Word_t uuid_index = 0;
372 - Pvoid_t *uuid_pvalue;
373 -
374 - // Traverse all UUIDs in this partition
375 - for (uuid_pvalue = JudyLFirst(mrg->index[partition].uuid_judy, &uuid_index, PJE0);
376 - uuid_pvalue != NULL && uuid_pvalue != PJERR;
377 - uuid_pvalue = JudyLNext(mrg->index[partition].uuid_judy, &uuid_index, PJE0)) {
378 -
379 - if (!(*uuid_pvalue))
380 - continue;
381 -
382 - // Get the sections judy for this UUID
383 - Pvoid_t sections_judy = *uuid_pvalue;
384 - Word_t section_index = 0;
385 - Pvoid_t *section_pvalue;
386 -
387 - // Traverse all sections for this UUID
388 - for (section_pvalue = JudyLFirst(sections_judy, &section_index, PJE0);
389 - section_pvalue != NULL && section_pvalue != PJERR;
390 - section_pvalue = JudyLNext(sections_judy, &section_index, PJE0)) {
391 -
392 - if (!(*section_pvalue))
393 - continue;
394 -
395 - METRIC *metric = *section_pvalue;
396 -
397 - // Try to acquire metric for deletion
398 - if (!refcount_acquire_for_deletion(&metric->refcount))
399 - referenced++;
400 -
401 - uuidmap_free(metric->uuid);
402 - aral_freez(mrg->index[partition].aral, metric);
403 - MRG_STATS_DELETED_METRIC(mrg, partition);
404 - }
405 -
406 - JudyLFreeArray(&sections_judy, PJE0);
407 - }
408 -
409 - JudyLFreeArray(&mrg->index[partition].uuid_judy, PJE0);
410 -
411 - // Unlock the partition
412 - mrg_index_write_unlock(mrg, partition);
413 -
414 - // Destroy the aral for this partition
415 - aral_destroy(mrg->index[partition].aral);
416 - }
417 -
418 - // Unregister the aral statistics
419 - pulse_aral_unregister_statistics(&mrg_aral_statistics);
420 -
421 - // Free the MRG structure
422 - freez(mrg);
423 -
424 - return referenced;
425 -}
426 -
427 -ALWAYS_INLINE METRIC *mrg_metric_add_and_acquire(MRG *mrg, MRG_ENTRY entry, bool *ret) {
428 -// internal_fatal(entry.latest_time_s > max_acceptable_collected_time(),
429 -// "DBENGINE METRIC: metric latest time is in the future");
430 -
431 - return metric_add_and_acquire(mrg, &entry, ret);
432 -}
433 -
434 -ALWAYS_INLINE METRIC *mrg_metric_get_and_acquire_by_uuid(MRG *mrg, nd_uuid_t *uuid, Word_t section) {
435 - UUIDMAP_ID id = uuidmap_create(*uuid);
436 - METRIC *metric = metric_get_and_acquire_by_id(mrg, id, section);
437 - uuidmap_free(id);
438 - return metric;
439 -}
440 -
441 -ALWAYS_INLINE METRIC *mrg_metric_get_and_acquire_by_id(MRG *mrg, UUIDMAP_ID id, Word_t section) {
442 - return metric_get_and_acquire_by_id(mrg, id, section);
443 -}
444 -
445 -ALWAYS_INLINE bool mrg_metric_release_and_delete(MRG *mrg, METRIC *metric) {
446 - return metric_release(mrg, metric);
447 -}
448 -
449 -ALWAYS_INLINE METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric) {
450 - metric_acquire(mrg, metric);
451 - return metric;
452 -}
453 -
454 -ALWAYS_INLINE void mrg_metric_release(MRG *mrg, METRIC *metric) {
455 - metric_release(mrg, metric);
456 -}
457 -
458 -ALWAYS_INLINE Word_t mrg_metric_id(MRG *mrg __maybe_unused, METRIC *metric) {
459 - return (Word_t)metric;
460 -}
461 -
462 -ALWAYS_INLINE nd_uuid_t *mrg_metric_uuid(MRG *mrg __maybe_unused, METRIC *metric) {
463 - return uuidmap_uuid_ptr(metric->uuid);
464 -}
465 -
466 -ALWAYS_INLINE UUIDMAP_ID mrg_metric_uuidmap_id_dup(MRG *mrg __maybe_unused, METRIC *metric) {
467 - return uuidmap_dup(metric->uuid);
468 -}
469 -
470 -ALWAYS_INLINE Word_t mrg_metric_section(MRG *mrg __maybe_unused, METRIC *metric) {
471 - return metric->section;
472 -}
473 -
474 -ALWAYS_INLINE bool mrg_metric_set_first_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
475 - internal_fatal(first_time_s < 0, "DBENGINE METRIC: timestamp is negative");
476 -
477 - if(first_time_s == LONG_MAX)
478 - first_time_s = 0;
479 -
480 - if(unlikely(first_time_s < 0))
481 - return false;
482 -
483 - __atomic_store_n(&metric->first_time_s, first_time_s, __ATOMIC_RELAXED);
484 -
485 - return true;
486 -}
487 -
488 -ALWAYS_INLINE void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s, time_t last_time_s, uint32_t update_every_s) {
489 - internal_fatal(first_time_s < 0 || last_time_s < 0,
490 - "DBENGINE METRIC: timestamp is negative");
491 - internal_fatal(first_time_s > max_acceptable_collected_time(),
492 - "DBENGINE METRIC: metric first time is in the future");
493 - internal_fatal(last_time_s > max_acceptable_collected_time(),
494 - "DBENGINE METRIC: metric last time is in the future");
495 -
496 - if(first_time_s > 0 && first_time_s != LONG_MAX)
497 - set_metric_field_with_condition(metric->first_time_s, first_time_s, _current <= 0 || (_wanted != 0 && _wanted != LONG_MAX && _wanted < _current));
498 -
499 - if(last_time_s > 0) {
500 - if(set_metric_field_with_condition(metric->latest_time_s_clean, last_time_s, _current <= 0 || _wanted > _current) &&
501 - update_every_s > 0)
502 - // set the latest update every too
503 - set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, true);
504 - }
505 - else if(update_every_s > 0)
506 - // set it only if it is invalid
507 - set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, _current <= 0);
508 -}
509 -
510 -ALWAYS_INLINE bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
511 - internal_fatal(first_time_s < 0, "DBENGINE METRIC: timestamp is negative");
512 - return set_metric_field_with_condition(metric->first_time_s, first_time_s, _wanted != 0 && _wanted != LONG_MAX && _wanted > _current);
513 -}
514 -
515 -ALWAYS_INLINE time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
516 - return mrg_metric_get_first_time_s_smart(mrg, metric);
517 -}
518 -
519 -void mrg_metric_clear_retention(MRG *mrg __maybe_unused, METRIC *metric) {
520 - __atomic_store_n(&metric->first_time_s, 0, __ATOMIC_RELAXED);
521 - __atomic_store_n(&metric->latest_time_s_clean, 0, __ATOMIC_RELAXED);
522 - __atomic_store_n(&metric->latest_time_s_hot, 0, __ATOMIC_RELAXED);
523 -}
524 -
525 -ALWAYS_INLINE_HOT void mrg_metric_get_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t *first_time_s, time_t *last_time_s, uint32_t *update_every_s) {
526 - time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
527 - time_t hot = __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED);
528 -
529 - *last_time_s = MAX(clean, hot);
530 - *first_time_s = mrg_metric_get_first_time_s_smart(mrg, metric);
531 - if (update_every_s)
532 - *update_every_s = __atomic_load_n(&metric->latest_update_every_s, __ATOMIC_RELAXED);
533 -}
534 -
535 -ALWAYS_INLINE bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
536 - internal_fatal(latest_time_s < 0, "DBENGINE METRIC: timestamp is negative");
537 -
538 -// internal_fatal(latest_time_s > max_acceptable_collected_time(),
539 -// "DBENGINE METRIC: metric latest time is in the future");
540 -
541 -// internal_fatal(metric->latest_time_s_clean > latest_time_s,
542 -// "DBENGINE METRIC: metric new clean latest time is older than the previous one");
543 -
544 - if(latest_time_s > 0) {
545 - if(set_metric_field_with_condition(metric->latest_time_s_clean, latest_time_s, true)) {
546 - set_metric_field_with_condition(metric->first_time_s, latest_time_s, _current <= 0 || _wanted < _current);
547 -
548 - return true;
549 - }
550 - }
551 -
552 - return false;
553 -}
554 -
555 -// returns true when metric still has retention
556 -ALWAYS_INLINE bool mrg_metric_has_zero_disk_retention(MRG *mrg __maybe_unused, METRIC *metric) {
557 - Word_t section = mrg_metric_section(mrg, metric);
558 - bool do_again = false;
559 - size_t countdown = 5;
560 -
561 - do {
562 - time_t min_first_time_s = LONG_MAX;
563 - time_t max_end_time_s = 0;
564 - PGC_PAGE *page;
565 - PGC_SEARCH method = PGC_SEARCH_FIRST;
566 - time_t page_first_time_s = 0;
567 - time_t page_end_time_s = 0;
568 - while ((page = pgc_page_get_and_acquire(main_cache, section, (Word_t)metric, page_first_time_s, method))) {
569 - method = PGC_SEARCH_NEXT;
570 -
571 - bool is_hot = pgc_is_page_hot(page);
572 - bool is_dirty = pgc_is_page_dirty(page);
573 - page_first_time_s = pgc_page_start_time_s(page);
574 - page_end_time_s = pgc_page_end_time_s(page);
575 -
576 - if ((is_hot || is_dirty) && page_first_time_s > 0 && page_first_time_s < min_first_time_s)
577 - min_first_time_s = page_first_time_s;
578 -
579 - if (is_dirty && page_end_time_s > max_end_time_s)
580 - max_end_time_s = page_end_time_s;
581 -
582 - pgc_page_release(main_cache, page);
583 - }
584 -
585 - if (min_first_time_s == LONG_MAX)
586 - min_first_time_s = 0;
587 -
588 - if (--countdown && !min_first_time_s && __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED))
589 - do_again = true;
590 - else {
591 - internal_error(!countdown, "METRIC: giving up on updating the retention of metric without disk retention");
592 -
593 - do_again = false;
594 - set_metric_field_with_condition(metric->first_time_s, min_first_time_s, true);
595 - set_metric_field_with_condition(metric->latest_time_s_clean, max_end_time_s, true);
596 - }
597 - } while(do_again);
598 -
599 - time_t first, last;
600 - mrg_metric_get_retention(mrg, metric, &first, &last, NULL);
601 - return (first && last && first < last);
602 -}
603 -
604 -ALWAYS_INLINE bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
605 - internal_fatal(latest_time_s < 0, "DBENGINE METRIC: timestamp is negative");
606 -
607 -// internal_fatal(latest_time_s > max_acceptable_collected_time(),
608 -// "DBENGINE METRIC: metric latest time is in the future");
609 -
610 - if(likely(latest_time_s > 0)) {
611 - __atomic_store_n(&metric->latest_time_s_hot, latest_time_s, __ATOMIC_RELAXED);
612 - return true;
613 - }
614 -
615 - return false;
616 -}
617 -
618 -ALWAYS_INLINE time_t mrg_metric_get_latest_clean_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
619 - time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
620 - return clean;
621 -}
622 -
623 -ALWAYS_INLINE time_t mrg_metric_get_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
624 - time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
625 - time_t hot = __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED);
626 -
627 - return MAX(clean, hot);
628 -}
629 -
630 -ALWAYS_INLINE bool mrg_metric_set_update_every(MRG *mrg __maybe_unused, METRIC *metric, uint32_t update_every_s) {
631 - if(likely(update_every_s > 0))
632 - return set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, true);
633 -
634 - return false;
635 -}
636 -
637 -ALWAYS_INLINE_HOT bool mrg_metric_set_update_every_s_if_zero(MRG *mrg __maybe_unused, METRIC *metric, uint32_t update_every_s) {
638 - if(likely(update_every_s > 0))
639 - return set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, _current <= 0);
640 -
641 - return false;
642 -}
643 -
644 -ALWAYS_INLINE uint32_t mrg_metric_get_update_every_s(MRG *mrg __maybe_unused, METRIC *metric) {
645 - return __atomic_load_n(&metric->latest_update_every_s, __ATOMIC_RELAXED);
646 -}
647 -
648 -#ifdef NETDATA_INTERNAL_CHECKS
649 -ALWAYS_INLINE bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) {
650 - pid_t expected = __atomic_load_n(&metric->writer, __ATOMIC_RELAXED);
651 - pid_t wanted = gettid_cached();
652 - bool done = true;
653 -
654 - do {
655 - if(expected != 0) {
656 - done = false;
657 - break;
658 - }
659 - } while(!__atomic_compare_exchange_n(&metric->writer, &expected, wanted, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
660 -
661 - if(done)
662 - __atomic_add_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
663 - else
664 - __atomic_add_fetch(&mrg->index[metric->partition].stats.writers_conflicts, 1, __ATOMIC_RELAXED);
665 -
666 - return done;
667 -}
668 -
669 -ALWAYS_INLINE bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) {
670 - // this function can be called from a different thread than the one than the writer
671 -
672 - pid_t expected = __atomic_load_n(&metric->writer, __ATOMIC_RELAXED);
673 - pid_t wanted = 0;
674 - bool done = true;
675 -
676 - do {
677 - if(!expected) {
678 - done = false;
679 - break;
680 - }
681 - } while(!__atomic_compare_exchange_n(&metric->writer, &expected, wanted, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
682 -
683 - if(done)
684 - __atomic_sub_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
685 -
686 - return done;
687 -}
688 -#endif
689 -
690 -inline void mrg_update_metric_retention_and_granularity_by_uuid(
691 - MRG *mrg, Word_t section, nd_uuid_t *uuid,
692 - time_t first_time_s, time_t last_time_s,
693 - uint32_t update_every_s, time_t now_s)
694 -{
695 - if(unlikely(last_time_s > now_s)) {
696 - nd_log_limit_static_global_var(erl, 1, 0);
697 - nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
698 - "DBENGINE JV2: wrong last time on-disk (%ld - %ld, now %ld), "
699 - "fixing last time to now",
700 - first_time_s, last_time_s, now_s);
701 - last_time_s = now_s;
702 - }
703 -
704 - if (unlikely(first_time_s > last_time_s)) {
705 - nd_log_limit_static_global_var(erl, 1, 0);
706 - nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
707 - "DBENGINE JV2: wrong first time on-disk (%ld - %ld, now %ld), "
708 - "fixing first time to last time",
709 - first_time_s, last_time_s, now_s);
710 -
711 - first_time_s = last_time_s;
712 - }
713 -
714 - if (unlikely(first_time_s == 0 || last_time_s == 0)) {
715 - nd_log_limit_static_global_var(erl, 1, 0);
716 - nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
717 - "DBENGINE JV2: zero on-disk timestamps (%ld - %ld, now %ld), "
718 - "using them as-is",
719 - first_time_s, last_time_s, now_s);
720 - }
721 -
722 - bool added = false;
723 - METRIC *metric = mrg_metric_get_and_acquire_by_uuid(mrg, uuid, section);
724 - if (!metric) {
725 - MRG_ENTRY entry = {
726 - .uuid = uuid,
727 - .section = section,
728 - .first_time_s = first_time_s,
729 - .last_time_s = last_time_s,
730 - .latest_update_every_s = update_every_s
731 - };
732 - metric = mrg_metric_add_and_acquire(mrg, entry, &added);
733 - }
734 -
735 - struct rrdengine_instance *ctx = (struct rrdengine_instance *) section;
736 - if (likely(!added)) {
737 - uint64_t old_samples = 0;
738 -
739 - if (update_every_s && metric->latest_update_every_s && metric->latest_time_s_clean)
740 - old_samples = (metric->latest_time_s_clean - metric->first_time_s) / metric->latest_update_every_s;
741 -
742 - mrg_metric_expand_retention(mrg, metric, first_time_s, last_time_s, update_every_s);
743 -
744 - uint64_t new_samples = 0;
745 - if (update_every_s && metric->latest_update_every_s && metric->latest_time_s_clean)
746 - new_samples = (metric->latest_time_s_clean - metric->first_time_s) / metric->latest_update_every_s;
747 -
748 - __atomic_add_fetch(&ctx->atomic.samples, new_samples - old_samples, __ATOMIC_RELAXED);
749 - }
750 - else {
751 - // Newly added
752 - if (update_every_s) {
753 - uint64_t samples = (last_time_s - first_time_s) / update_every_s;
754 - __atomic_add_fetch(&ctx->atomic.samples, samples, __ATOMIC_RELAXED);
755 - }
756 - __atomic_add_fetch(&ctx->atomic.metrics, 1, __ATOMIC_RELAXED);
757 - }
758 -
759 - mrg_metric_release(mrg, metric);
760 -}
761 -
762 -inline void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
763 - memset(s, 0, sizeof(struct mrg_statistics));
764 -
765 - for(size_t i = 0; i < _countof(mrg->index) ;i++) {
766 - s->entries += __atomic_load_n(&mrg->index[i].stats.entries, __ATOMIC_RELAXED);
767 - s->entries_acquired += __atomic_load_n(&mrg->index[i].stats.entries_acquired, __ATOMIC_RELAXED);
768 - s->size += __atomic_load_n(&mrg->index[i].stats.size, __ATOMIC_RELAXED);
769 - s->current_references += __atomic_load_n(&mrg->index[i].stats.current_references, __ATOMIC_RELAXED);
770 - s->additions += __atomic_load_n(&mrg->index[i].stats.additions, __ATOMIC_RELAXED);
771 - s->additions_duplicate += __atomic_load_n(&mrg->index[i].stats.additions_duplicate, __ATOMIC_RELAXED);
772 - s->deletions += __atomic_load_n(&mrg->index[i].stats.deletions, __ATOMIC_RELAXED);
773 - s->delete_having_retention_or_referenced += __atomic_load_n(&mrg->index[i].stats.delete_having_retention_or_referenced, __ATOMIC_RELAXED);
774 - s->delete_misses += __atomic_load_n(&mrg->index[i].stats.delete_misses, __ATOMIC_RELAXED);
775 - s->search_hits += __atomic_load_n(&mrg->index[i].stats.search_hits, __ATOMIC_RELAXED);
776 - s->search_misses += __atomic_load_n(&mrg->index[i].stats.search_misses, __ATOMIC_RELAXED);
777 - s->writers += __atomic_load_n(&mrg->index[i].stats.writers, __ATOMIC_RELAXED);
778 - s->writers_conflicts += __atomic_load_n(&mrg->index[i].stats.writers_conflicts, __ATOMIC_RELAXED);
779 - }
780 -
781 - s->size += sizeof(MRG);
782 -}
783 -
784 -// ----------------------------------------------------------------------------
785 -// unit test
786 -
787 -struct mrg_stress_entry {
788 - nd_uuid_t uuid;
789 - time_t after;
790 - time_t before;
791 -};
792 -
793 -struct mrg_stress {
794 - MRG *mrg;
795 - bool stop;
796 - size_t entries;
797 - struct mrg_stress_entry *array;
798 - size_t updates;
799 -};
800 -
801 -static void *mrg_stress(void *ptr) {
802 - struct mrg_stress *t = ptr;
803 - MRG *mrg = t->mrg;
804 -
805 - ssize_t start = 0;
806 - ssize_t end = (ssize_t)t->entries;
807 - ssize_t step = 1;
808 -
809 - if(gettid_cached() % 2) {
810 - start = (ssize_t)t->entries - 1;
811 - end = -1;
812 - step = -1;
813 - }
814 -
815 - while(!__atomic_load_n(&t->stop, __ATOMIC_RELAXED) && !nd_thread_signaled_to_cancel()) {
816 - for (ssize_t i = start; i != end; i += step) {
817 - struct mrg_stress_entry *e = &t->array[i];
818 -
819 - time_t after = __atomic_sub_fetch(&e->after, 1, __ATOMIC_RELAXED);
820 - time_t before = __atomic_add_fetch(&e->before, 1, __ATOMIC_RELAXED);
821 -
822 - mrg_update_metric_retention_and_granularity_by_uuid(
823 - mrg, 0x01,
824 - &e->uuid,
825 - after,
826 - before,
827 - 1,
828 - before);
829 -
830 - __atomic_add_fetch(&t->updates, 1, __ATOMIC_RELAXED);
831 - }
832 - }
833 -
834 - return ptr;
835 -}
836 -
837 -int mrg_unittest(void) {
838 - MRG *mrg = mrg_create();
839 - METRIC *m1_t0, *m2_t0, *m3_t0, *m4_t0;
840 - METRIC *m1_t1, *m2_t1, *m3_t1, *m4_t1;
841 - bool ret;
842 -
843 - nd_uuid_t test_uuid;
844 - uuid_generate(test_uuid);
845 - MRG_ENTRY entry = {
846 - .uuid = &test_uuid,
847 - .section = 0,
848 - .first_time_s = 2,
849 - .last_time_s = 3,
850 - .latest_update_every_s = 4,
851 - };
852 - m1_t0 = mrg_metric_add_and_acquire(mrg, entry, &ret);
853 - if(!ret)
854 - fatal("DBENGINE METRIC: failed to add metric");
855 -
856 - // add the same metric again
857 - m2_t0 = mrg_metric_add_and_acquire(mrg, entry, &ret);
858 - if(m2_t0 != m1_t0)
859 - fatal("DBENGINE METRIC: adding the same metric twice, does not return the same pointer");
860 - if(ret)
861 - fatal("DBENGINE METRIC: managed to add the same metric twice");
862 -
863 - m3_t0 = mrg_metric_get_and_acquire_by_uuid(mrg, entry.uuid, entry.section);
864 - if(m3_t0 != m1_t0)
865 - fatal("DBENGINE METRIC: cannot find the metric added");
866 -
867 - // add the same metric again
868 - m4_t0 = mrg_metric_add_and_acquire(mrg, entry, &ret);
869 - if(m4_t0 != m1_t0)
870 - fatal("DBENGINE METRIC: adding the same metric twice, does not return the same pointer");
871 - if(ret)
872 - fatal("DBENGINE METRIC: managed to add the same metric twice");
873 -
874 - // add the same metric in another section
875 - entry.section = 1;
876 - m1_t1 = mrg_metric_add_and_acquire(mrg, entry, &ret);
877 - if(!ret)
878 - fatal("DBENGINE METRIC: failed to add metric in section %zu", (size_t)entry.section);
879 -
880 - // add the same metric again
881 - m2_t1 = mrg_metric_add_and_acquire(mrg, entry, &ret);
882 - if(m2_t1 != m1_t1)
883 - fatal("DBENGINE METRIC: adding the same metric twice (section %zu), does not return the same pointer", (size_t)entry.section);
884 - if(ret)
885 - fatal("DBENGINE METRIC: managed to add the same metric twice in (section 0)");
886 -
887 - m3_t1 = mrg_metric_get_and_acquire_by_uuid(mrg, entry.uuid, entry.section);
888 - if(m3_t1 != m1_t1)
889 - fatal("DBENGINE METRIC: cannot find the metric added (section %zu)", (size_t)entry.section);
890 -
891 - // delete the first metric
892 - mrg_metric_release(mrg, m2_t0);
893 - mrg_metric_release(mrg, m3_t0);
894 - mrg_metric_release(mrg, m4_t0);
895 - mrg_metric_set_first_time_s(mrg, m1_t0, 0);
896 - mrg_metric_set_clean_latest_time_s(mrg, m1_t0, 0);
897 - mrg_metric_set_hot_latest_time_s(mrg, m1_t0, 0);
898 - if(!mrg_metric_release_and_delete(mrg, m1_t0))
899 - fatal("DBENGINE METRIC: cannot delete the first metric");
900 -
901 - m4_t1 = mrg_metric_get_and_acquire_by_uuid(mrg, entry.uuid, entry.section);
902 - if(m4_t1 != m1_t1)
903 - fatal("DBENGINE METRIC: cannot find the metric added (section %zu), after deleting the first one", (size_t)entry.section);
904 -
905 - // delete the second metric
906 - mrg_metric_release(mrg, m2_t1);
907 - mrg_metric_release(mrg, m3_t1);
908 - mrg_metric_release(mrg, m4_t1);
909 - mrg_metric_set_first_time_s(mrg, m1_t1, 0);
910 - mrg_metric_set_clean_latest_time_s(mrg, m1_t1, 0);
911 - mrg_metric_set_hot_latest_time_s(mrg, m1_t1, 0);
912 - if(!mrg_metric_release_and_delete(mrg, m1_t1))
913 - fatal("DBENGINE METRIC: cannot delete the second metric");
914 -
915 - struct mrg_statistics s;
916 - mrg_get_statistics(mrg, &s);
917 - if(s.entries != 0)
918 - fatal("DBENGINE METRIC: invalid entries counter");
919 -
920 - size_t entries = 1000000;
921 - size_t threads = _countof(mrg->index) / 3 + 1;
922 - size_t tiers = 3;
923 - size_t run_for_secs = 5;
924 - netdata_log_info("preparing stress test of %zu entries...", entries);
925 - struct mrg_stress t = {
926 - .mrg = mrg,
927 - .entries = entries,
928 - .array = callocz(entries, sizeof(struct mrg_stress_entry)),
929 - };
930 -
931 - time_t now = max_acceptable_collected_time();
932 - for(size_t i = 0; i < entries ;i++) {
933 - uuid_generate_random(t.array[i].uuid);
934 - t.array[i].after = now / 3;
935 - t.array[i].before = now / 2;
936 - }
937 - netdata_log_info("stress test is populating MRG with 3 tiers...");
938 - for(size_t i = 0; i < entries ;i++) {
939 - struct mrg_stress_entry *e = &t.array[i];
940 - for(size_t tier = 1; tier <= tiers ;tier++) {
941 - mrg_update_metric_retention_and_granularity_by_uuid(
942 - mrg, tier,
943 - &e->uuid,
944 - e->after,
945 - e->before,
946 - 1,
947 - e->before);
948 - }
949 - }
950 - netdata_log_info("stress test ready to run...");
951 -
952 - usec_t started_ut = now_monotonic_usec();
953 -
954 - ND_THREAD *th[threads];
955 - for(size_t i = 0; i < threads ; i++) {
956 - char buf[15 + 1];
957 - snprintfz(buf, sizeof(buf) - 1, "TH[%zu]", i);
958 - th[i] = nd_thread_create(buf, NETDATA_THREAD_OPTION_JOINABLE | NETDATA_THREAD_OPTION_DONT_LOG, mrg_stress, &t);
959 - }
960 -
961 - sleep_usec(run_for_secs * USEC_PER_SEC);
962 - __atomic_store_n(&t.stop, true, __ATOMIC_RELAXED);
963 -
964 - for(size_t i = 0; i < threads ; i++)
965 - nd_thread_signal_cancel(th[i]);
966 -
967 - for(size_t i = 0; i < threads ; i++)
968 - nd_thread_join(th[i]);
969 -
970 - usec_t ended_ut = now_monotonic_usec();
971 -
972 - struct mrg_statistics stats;
973 - mrg_get_statistics(mrg, &stats);
974 -
975 - netdata_log_info("DBENGINE METRIC: did %zu additions, %zu duplicate additions, "
976 - "%zu deletions, %zu wrong deletions, "
977 - "%zu successful searches, %zu wrong searches, "
978 - "in %"PRIu64" usecs",
979 - stats.additions, stats.additions_duplicate,
980 - stats.deletions, stats.delete_misses,
981 - stats.search_hits, stats.search_misses,
982 - ended_ut - started_ut);
983 -
984 - netdata_log_info("DBENGINE METRIC: updates performance: %0.2fk/sec total, %0.2fk/sec/thread",
985 - (double)t.updates / (double)((ended_ut - started_ut) / USEC_PER_SEC) / 1000.0,
986 - (double)t.updates / (double)((ended_ut - started_ut) / USEC_PER_SEC) / 1000.0 / threads);
987 -
988 - mrg_destroy(mrg);
989 -
990 - netdata_log_info("DBENGINE METRIC: all tests passed!");
991 -
992 - return 0;
993 -}
src/database/engine/mrg-internals.h new
+352
@@ -0,0 +1,352 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#ifndef NETDATA_MRG_INTERNALS_H
4 +#define NETDATA_MRG_INTERNALS_H
5 +
6 +#include "mrg.h"
7 +#include "cache.h"
8 +#include "libnetdata/locks/locks.h"
9 +#include "rrddiskprotocol.h"
10 +
11 +struct metric {
12 + Word_t section; // never changes
13 + UUIDMAP_ID uuid; // never changes
14 +
15 + REFCOUNT refcount;
16 + uint8_t partition;
17 +
18 + uint32_t latest_update_every_s; // the latest data collection frequency
19 +
20 + time_t first_time_s; // the timestamp of the oldest point in the database
21 + time_t latest_time_s_clean; // the timestamp of the newest point in the database
22 + time_t latest_time_s_hot; // the timestamp of the latest point that has been collected (not yet stored)
23 +
24 +#ifdef NETDATA_INTERNAL_CHECKS
25 + pid_t writer;
26 +#endif
27 +
28 + // THIS IS allocated with malloc()
29 + // YOU HAVE TO INITIALIZE IT YOURSELF!
30 +};
31 +
32 +#define set_metric_field_with_condition(field, value, condition) ({ \
33 + typeof(field) _current = __atomic_load_n(&(field), __ATOMIC_RELAXED); \
34 + typeof(field) _wanted = value; \
35 + bool did_it = true; \
36 + \
37 + do { \
38 + if((condition) && (_current != _wanted)) { \
39 + ; \
40 + } \
41 + else { \
42 + did_it = false; \
43 + break; \
44 + } \
45 + } while(!__atomic_compare_exchange_n(&(field), &_current, _wanted, \
46 + false, __ATOMIC_RELAXED, __ATOMIC_RELAXED)); \
47 + \
48 + did_it; \
49 +})
50 +
51 +extern struct aral_statistics mrg_aral_statistics;
52 +
53 +struct mrg {
54 + struct mrg_partition {
55 + ARAL *aral; // not protected by our spinlock - it has its own
56 +
57 + RW_SPINLOCK rw_spinlock;
58 + Pvoid_t uuid_judy; // JudyL: each UUID has a JudyL of sections (tiers)
59 +
60 + struct mrg_statistics stats;
61 + } index[UUIDMAP_PARTITIONS];
62 +};
63 +
64 +static inline void MRG_STATS_DUPLICATE_ADD(MRG *mrg, size_t partition) {
65 + mrg->index[partition].stats.additions_duplicate++;
66 +}
67 +
68 +static inline void MRG_STATS_ADDED_METRIC(MRG *mrg, size_t partition, Word_t section) {
69 + mrg->index[partition].stats.entries++;
70 + mrg->index[partition].stats.additions++;
71 + mrg->index[partition].stats.size += sizeof(METRIC);
72 + struct rrdengine_instance *ctx = (struct rrdengine_instance *) section;
73 + __atomic_add_fetch(&ctx->atomic.metrics, 1, __ATOMIC_RELAXED);
74 +}
75 +
76 +static inline void MRG_STATS_DELETED_METRIC(MRG *mrg, size_t partition, Word_t section) {
77 + mrg->index[partition].stats.entries--;
78 + mrg->index[partition].stats.size -= sizeof(METRIC);
79 + mrg->index[partition].stats.deletions++;
80 + struct rrdengine_instance *ctx = (struct rrdengine_instance *) section;
81 + __atomic_sub_fetch(&ctx->atomic.metrics, 1, __ATOMIC_RELAXED);
82 +}
83 +
84 +static inline void MRG_STATS_SEARCH_HIT(MRG *mrg, size_t partition) {
85 + __atomic_add_fetch(&mrg->index[partition].stats.search_hits, 1, __ATOMIC_RELAXED);
86 +}
87 +
88 +static inline void MRG_STATS_SEARCH_MISS(MRG *mrg, size_t partition) {
89 + __atomic_add_fetch(&mrg->index[partition].stats.search_misses, 1, __ATOMIC_RELAXED);
90 +}
91 +
92 +static inline void MRG_STATS_DELETE_MISS(MRG *mrg, size_t partition) {
93 + mrg->index[partition].stats.delete_misses++;
94 +}
95 +
96 +#define mrg_index_read_lock(mrg, partition) rw_spinlock_read_lock(&(mrg)->index[partition].rw_spinlock)
97 +#define mrg_index_read_unlock(mrg, partition) rw_spinlock_read_unlock(&(mrg)->index[partition].rw_spinlock)
98 +#define mrg_index_write_lock(mrg, partition) rw_spinlock_write_lock(&(mrg)->index[partition].rw_spinlock)
99 +#define mrg_index_write_unlock(mrg, partition) rw_spinlock_write_unlock(&(mrg)->index[partition].rw_spinlock)
100 +
101 +static inline void mrg_stats_judy_mem(MRG *mrg, size_t partition, int64_t judy_mem) {
102 + __atomic_add_fetch(&mrg->index[partition].stats.size, judy_mem, __ATOMIC_RELAXED);
103 +}
104 +
105 +static inline void metric_log(MRG *mrg __maybe_unused, METRIC *metric, const char *msg) {
106 + struct rrdengine_instance *ctx = (struct rrdengine_instance *)metric->section;
107 +
108 + nd_uuid_t uuid;
109 + uuidmap_uuid(metric->uuid, uuid);
110 + char uuid_txt[UUID_STR_LEN];
111 + uuid_unparse_lower(uuid, uuid_txt);
112 + nd_log(NDLS_DAEMON, NDLP_ERR,
113 + "METRIC: %s on %s at tier %d, refcount %d, partition %u, "
114 + "retention [%ld - %ld (hot), %ld (clean)], update every %"PRIu32
115 +#ifdef NETDATA_INTERNAL_CHECKS
116 + ", writer pid %d "
117 +#endif
118 + " --- PLEASE OPEN A GITHUB ISSUE TO REPORT THIS LOG LINE TO NETDATA --- ",
119 + msg,
120 + uuid_txt,
121 + ctx->config.tier,
122 + metric->refcount,
123 + metric->partition,
124 + metric->first_time_s,
125 + metric->latest_time_s_hot,
126 + metric->latest_time_s_clean,
127 + metric->latest_update_every_s
128 +#ifdef NETDATA_INTERNAL_CHECKS
129 + , (int)metric->writer
130 +#endif
131 + );
132 +}
133 +
134 +
135 +ALWAYS_INLINE
136 +static time_t mrg_metric_get_first_time_s_smart(MRG *mrg __maybe_unused, METRIC *metric) {
137 + time_t first_time_s = __atomic_load_n(&metric->first_time_s, __ATOMIC_RELAXED);
138 +
139 + if(first_time_s <= 0) {
140 + first_time_s = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
141 + if(first_time_s <= 0)
142 + first_time_s = __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED);
143 +
144 + if(first_time_s <= 0)
145 + first_time_s = 0;
146 + else
147 + __atomic_store_n(&metric->first_time_s, first_time_s, __ATOMIC_RELAXED);
148 + }
149 +
150 + return first_time_s;
151 +}
152 +
153 +ALWAYS_INLINE
154 +static bool acquired_metric_has_retention(MRG *mrg, METRIC *metric) {
155 + time_t first, last;
156 + mrg_metric_get_retention(mrg, metric, &first, &last, NULL);
157 + bool rc = (first != 0 && last != 0 && first <= last);
158 +
159 + if(!rc && __atomic_load_n(&mrg->index[metric->partition].stats.writers, __ATOMIC_RELAXED) > 0)
160 + rc = true;
161 +
162 + return rc;
163 +}
164 +
165 +ALWAYS_INLINE
166 +static void acquired_for_deletion_metric_delete(MRG *mrg, METRIC *metric) {
167 + JudyAllocThreadPulseReset();
168 +
169 + size_t partition = metric->partition;
170 +
171 + mrg_index_write_lock(mrg, partition);
172 +
173 + Pvoid_t *sections_judy_pptr = JudyLGet(mrg->index[partition].uuid_judy, metric->uuid, PJE0);
174 + if(unlikely(sections_judy_pptr == PJERR))
175 + fatal("METRIC: corrupted JudyL");
176 +
177 + if(unlikely(!sections_judy_pptr || !*sections_judy_pptr)) {
178 + MRG_STATS_DELETE_MISS(mrg, partition);
179 + mrg_index_write_unlock(mrg, partition);
180 + return;
181 + }
182 +
183 + int rc = JudyLDel(sections_judy_pptr, metric->section, PJE0);
184 + if(unlikely(!rc)) {
185 + MRG_STATS_DELETE_MISS(mrg, partition);
186 + mrg_index_write_unlock(mrg, partition);
187 + mrg_stats_judy_mem(mrg, partition, JudyAllocThreadPulseGetAndReset());
188 + return;
189 + }
190 +
191 + if(!*sections_judy_pptr) {
192 + rc = JudyLDel(&mrg->index[partition].uuid_judy, metric->uuid, PJE0);
193 +
194 + if(unlikely(!rc))
195 + fatal("DBENGINE METRIC: cannot delete UUID from JudyL");
196 + }
197 +
198 + MRG_STATS_DELETED_METRIC(mrg, partition, metric->section);
199 +
200 + mrg_index_write_unlock(mrg, partition);
201 + uuidmap_free(metric->uuid);
202 + aral_freez(mrg->index[partition].aral, metric);
203 + mrg_stats_judy_mem(mrg, partition, JudyAllocThreadPulseGetAndReset());
204 +}
205 +
206 +ALWAYS_INLINE
207 +static bool metric_acquire(MRG *mrg, METRIC *metric) {
208 + REFCOUNT rc = refcount_acquire_advanced(&metric->refcount);
209 + if(!REFCOUNT_ACQUIRED(rc))
210 + return false;
211 +
212 + size_t partition = metric->partition;
213 +
214 + if(rc == 1)
215 + __atomic_add_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
216 +
217 + __atomic_add_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
218 +
219 + return true;
220 +}
221 +
222 +ALWAYS_INLINE
223 +static bool metric_release(MRG *mrg, METRIC *metric) {
224 + size_t partition = metric->partition;
225 +
226 + REFCOUNT refcount = refcount_release(&metric->refcount);
227 +
228 + if(!refcount && !acquired_metric_has_retention(mrg, metric) && refcount_acquire_for_deletion(&metric->refcount))
229 + refcount = REFCOUNT_DELETED;
230 +
231 + if(refcount == 0 || refcount == REFCOUNT_DELETED) {
232 + __atomic_sub_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
233 +
234 + if(refcount == REFCOUNT_DELETED)
235 + acquired_for_deletion_metric_delete(mrg, metric);
236 + }
237 +
238 + __atomic_sub_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
239 +
240 + return refcount == REFCOUNT_DELETED;
241 +}
242 +
243 +ALWAYS_INLINE
244 +static METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
245 + JudyAllocThreadPulseReset();
246 +
247 + UUIDMAP_ID id = uuidmap_create(*entry->uuid);
248 +
249 + size_t partition = uuid_to_uuidmap_partition(*entry->uuid);
250 +
251 + METRIC *allocation = aral_mallocz(mrg->index[partition].aral);
252 + Pvoid_t *PValue;
253 +
254 + while(1) {
255 + mrg_index_write_lock(mrg, partition);
256 +
257 + Pvoid_t *sections_judy_pptr = JudyLIns(&mrg->index[partition].uuid_judy, id, PJE0);
258 + if (unlikely(!sections_judy_pptr || sections_judy_pptr == PJERR))
259 + fatal("DBENGINE METRIC: corrupted UUIDs JudyL array");
260 +
261 + PValue = JudyLIns(sections_judy_pptr, entry->section, PJE0);
262 + if (unlikely(!PValue || PValue == PJERR))
263 + fatal("DBENGINE METRIC: corrupted section JudyL array");
264 +
265 + if (unlikely(*PValue != NULL)) {
266 + METRIC *metric = *PValue;
267 +
268 + if(!metric_acquire(mrg, metric)) {
269 + mrg_index_write_unlock(mrg, partition);
270 + continue;
271 + }
272 +
273 + MRG_STATS_DUPLICATE_ADD(mrg, partition);
274 + mrg_index_write_unlock(mrg, partition);
275 +
276 + if (ret)
277 + *ret = false;
278 +
279 + uuidmap_free(id);
280 + aral_freez(mrg->index[partition].aral, allocation);
281 +
282 + mrg_stats_judy_mem(mrg, partition, JudyAllocThreadPulseGetAndReset());
283 + return metric;
284 + }
285 +
286 + break;
287 + }
288 +
289 + METRIC *metric = allocation;
290 + metric->uuid = id;
291 + metric->section = entry->section;
292 + metric->first_time_s = MAX(0, entry->first_time_s);
293 + metric->latest_time_s_clean = MAX(0, entry->last_time_s);
294 + metric->latest_time_s_hot = 0;
295 + metric->latest_update_every_s = entry->latest_update_every_s;
296 +#ifdef NETDATA_INTERNAL_CHECKS
297 + metric->writer = 0;
298 +#endif
299 + metric->refcount = 1;
300 + metric->partition = partition;
301 + *PValue = metric;
302 +
303 + __atomic_add_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
304 + __atomic_add_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
305 +
306 + MRG_STATS_ADDED_METRIC(mrg, partition, metric->section);
307 +
308 + mrg_index_write_unlock(mrg, partition);
309 +
310 + if(ret)
311 + *ret = true;
312 +
313 + mrg_stats_judy_mem(mrg, partition, JudyAllocThreadPulseGetAndReset());
314 + return metric;
315 +}
316 +
317 +ALWAYS_INLINE
318 +static METRIC *metric_get_and_acquire_by_id(MRG *mrg, UUIDMAP_ID id, Word_t section) {
319 + size_t partition = uuidmap_id_to_partition(id);
320 +
321 + while(1) {
322 + mrg_index_read_lock(mrg, partition);
323 +
324 + Pvoid_t *sections_judy_pptr = JudyLGet(mrg->index[partition].uuid_judy, id, PJE0);
325 + if (unlikely(!sections_judy_pptr)) {
326 + mrg_index_read_unlock(mrg, partition);
327 + MRG_STATS_SEARCH_MISS(mrg, partition);
328 + return NULL;
329 + }
330 +
331 + Pvoid_t *PValue = JudyLGet(*sections_judy_pptr, section, PJE0);
332 + if (unlikely(!PValue)) {
333 + mrg_index_read_unlock(mrg, partition);
334 + MRG_STATS_SEARCH_MISS(mrg, partition);
335 + return NULL;
336 + }
337 +
338 + METRIC *metric = *PValue;
339 +
340 + if(metric && !metric_acquire(mrg, metric))
341 + metric = NULL;
342 +
343 + mrg_index_read_unlock(mrg, partition);
344 +
345 + if(metric) {
346 + MRG_STATS_SEARCH_HIT(mrg, partition);
347 + return metric;
348 + }
349 + }
350 +}
351 +
352 +#endif //NETDATA_MRG_INTERNALS_H
src/database/engine/mrg-load.c new
+49
@@ -0,0 +1,49 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#include "mrg-internals.h"
4 +
5 +DEFINE_JUDYL_TYPED(METRIC, METRIC *);
6 +METRIC_JudyLSet acquired_metrics = { 0 };
7 +size_t acquired_metrics_counter = 0;
8 +size_t acquired_metrics_deleted = 0;
9 +
10 +ALWAYS_INLINE
11 +static void mrg_metric_prepopulate(MRG *mrg, Word_t section, nd_uuid_t *uuid) {
12 + MRG_ENTRY entry = {
13 + .uuid = uuid,
14 + .section = section,
15 + .first_time_s = 0,
16 + .last_time_s = 0,
17 + .latest_update_every_s = 0,
18 + };
19 + bool added = false;
20 + METRIC *metric = metric_add_and_acquire(mrg, &entry, &added);
21 + if(likely(added)) {
22 + METRIC_SET(&acquired_metrics, acquired_metrics_counter++, metric);
23 + return;
24 + }
25 + mrg_metric_release(mrg, metric);
26 +}
27 +
28 +static void mrg_release_cb(Word_t idx __maybe_unused, METRIC *m, void *data) {
29 + MRG *mrg = data;
30 + if(mrg_metric_release(mrg, m))
31 + acquired_metrics_deleted++;
32 +}
33 +
34 +void mrg_metric_prepopulate_cleanup(MRG *mrg) {
35 + acquired_metrics_deleted = 0;
36 + METRIC_FREE(&acquired_metrics, mrg_release_cb, mrg);
37 +
38 + if(acquired_metrics_counter || acquired_metrics_deleted)
39 + nd_log(NDLS_DAEMON, NDLP_INFO, "MRG DUMP: Prepopulated %zu metrics, released %zu, deleted %zu",
40 + acquired_metrics_counter, acquired_metrics_counter - acquired_metrics_deleted, acquired_metrics_deleted);
41 +
42 + acquired_metrics_counter = 0;
43 +}
44 +
45 +// Main function to load metrics from the database
46 +bool mrg_load(MRG *mrg) {
47 + size_t processed_metrics = populate_metrics_from_database(mrg, (void (*)(void *, Word_t, nd_uuid_t *))mrg_metric_prepopulate);
48 + return processed_metrics > 0;
49 +}
src/database/engine/mrg-unittest.c new
+211
@@ -0,0 +1,211 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#include "mrg-internals.h"
4 +
5 +struct mrg_stress_entry {
6 + nd_uuid_t uuid;
7 + time_t after;
8 + time_t before;
9 +};
10 +
11 +struct mrg_stress {
12 + MRG *mrg;
13 + bool stop;
14 + size_t entries;
15 + struct mrg_stress_entry *array;
16 + size_t updates;
17 +};
18 +
19 +static void *mrg_stress(void *ptr) {
20 + struct mrg_stress *t = ptr;
21 + MRG *mrg = t->mrg;
22 +
23 + ssize_t start = 0;
24 + ssize_t end = (ssize_t)t->entries;
25 + ssize_t step = 1;
26 +
27 + if(gettid_cached() % 2) {
28 + start = (ssize_t)t->entries - 1;
29 + end = -1;
30 + step = -1;
31 + }
32 +
33 + while(!__atomic_load_n(&t->stop, __ATOMIC_RELAXED) && !nd_thread_signaled_to_cancel()) {
34 + for (ssize_t i = start; i != end; i += step) {
35 + struct mrg_stress_entry *e = &t->array[i];
36 +
37 + time_t after = __atomic_sub_fetch(&e->after, 1, __ATOMIC_RELAXED);
38 + time_t before = __atomic_add_fetch(&e->before, 1, __ATOMIC_RELAXED);
39 +
40 + mrg_update_metric_retention_and_granularity_by_uuid(
41 + mrg, 0x01,
42 + &e->uuid,
43 + after,
44 + before,
45 + 1,
46 + before);
47 +
48 + __atomic_add_fetch(&t->updates, 1, __ATOMIC_RELAXED);
49 + }
50 + }
51 +
52 + return ptr;
53 +}
54 +
55 +int mrg_unittest(void) {
56 + MRG *mrg = mrg_create();
57 + METRIC *m1_t0, *m2_t0, *m3_t0, *m4_t0;
58 + METRIC *m1_t1, *m2_t1, *m3_t1, *m4_t1;
59 + bool ret;
60 +
61 + nd_uuid_t test_uuid;
62 + uuid_generate(test_uuid);
63 + MRG_ENTRY entry = {
64 + .uuid = &test_uuid,
65 + .section = 0,
66 + .first_time_s = 2,
67 + .last_time_s = 3,
68 + .latest_update_every_s = 4,
69 + };
70 + m1_t0 = mrg_metric_add_and_acquire(mrg, entry, &ret);
71 + if(!ret)
72 + fatal("DBENGINE METRIC: failed to add metric");
73 +
74 + // add the same metric again
75 + m2_t0 = mrg_metric_add_and_acquire(mrg, entry, &ret);
76 + if(m2_t0 != m1_t0)
77 + fatal("DBENGINE METRIC: adding the same metric twice, does not return the same pointer");
78 + if(ret)
79 + fatal("DBENGINE METRIC: managed to add the same metric twice");
80 +
81 + m3_t0 = mrg_metric_get_and_acquire_by_uuid(mrg, entry.uuid, entry.section);
82 + if(m3_t0 != m1_t0)
83 + fatal("DBENGINE METRIC: cannot find the metric added");
84 +
85 + // add the same metric again
86 + m4_t0 = mrg_metric_add_and_acquire(mrg, entry, &ret);
87 + if(m4_t0 != m1_t0)
88 + fatal("DBENGINE METRIC: adding the same metric twice, does not return the same pointer");
89 + if(ret)
90 + fatal("DBENGINE METRIC: managed to add the same metric twice");
91 +
92 + // add the same metric in another section
93 + entry.section = 1;
94 + m1_t1 = mrg_metric_add_and_acquire(mrg, entry, &ret);
95 + if(!ret)
96 + fatal("DBENGINE METRIC: failed to add metric in section %zu", (size_t)entry.section);
97 +
98 + // add the same metric again
99 + m2_t1 = mrg_metric_add_and_acquire(mrg, entry, &ret);
100 + if(m2_t1 != m1_t1)
101 + fatal("DBENGINE METRIC: adding the same metric twice (section %zu), does not return the same pointer", (size_t)entry.section);
102 + if(ret)
103 + fatal("DBENGINE METRIC: managed to add the same metric twice in (section 0)");
104 +
105 + m3_t1 = mrg_metric_get_and_acquire_by_uuid(mrg, entry.uuid, entry.section);
106 + if(m3_t1 != m1_t1)
107 + fatal("DBENGINE METRIC: cannot find the metric added (section %zu)", (size_t)entry.section);
108 +
109 + // delete the first metric
110 + mrg_metric_release(mrg, m2_t0);
111 + mrg_metric_release(mrg, m3_t0);
112 + mrg_metric_release(mrg, m4_t0);
113 + mrg_metric_set_first_time_s(mrg, m1_t0, 0);
114 + mrg_metric_set_clean_latest_time_s(mrg, m1_t0, 0);
115 + mrg_metric_set_hot_latest_time_s(mrg, m1_t0, 0);
116 + if(!mrg_metric_release_and_delete(mrg, m1_t0))
117 + fatal("DBENGINE METRIC: cannot delete the first metric");
118 +
119 + m4_t1 = mrg_metric_get_and_acquire_by_uuid(mrg, entry.uuid, entry.section);
120 + if(m4_t1 != m1_t1)
121 + fatal("DBENGINE METRIC: cannot find the metric added (section %zu), after deleting the first one", (size_t)entry.section);
122 +
123 + // delete the second metric
124 + mrg_metric_release(mrg, m2_t1);
125 + mrg_metric_release(mrg, m3_t1);
126 + mrg_metric_release(mrg, m4_t1);
127 + mrg_metric_set_first_time_s(mrg, m1_t1, 0);
128 + mrg_metric_set_clean_latest_time_s(mrg, m1_t1, 0);
129 + mrg_metric_set_hot_latest_time_s(mrg, m1_t1, 0);
130 + if(!mrg_metric_release_and_delete(mrg, m1_t1))
131 + fatal("DBENGINE METRIC: cannot delete the second metric");
132 +
133 + struct mrg_statistics s;
134 + mrg_get_statistics(mrg, &s);
135 + if(s.entries != 0)
136 + fatal("DBENGINE METRIC: invalid entries counter");
137 +
138 + size_t entries = 1000000;
139 + size_t threads = _countof(mrg->index) / 3 + 1;
140 + size_t tiers = 3;
141 + size_t run_for_secs = 5;
142 + netdata_log_info("preparing stress test of %zu entries...", entries);
143 + struct mrg_stress t = {
144 + .mrg = mrg,
145 + .entries = entries,
146 + .array = callocz(entries, sizeof(struct mrg_stress_entry)),
147 + };
148 +
149 + time_t now = max_acceptable_collected_time();
150 + for(size_t i = 0; i < entries ;i++) {
151 + uuid_generate_random(t.array[i].uuid);
152 + t.array[i].after = now / 3;
153 + t.array[i].before = now / 2;
154 + }
155 + netdata_log_info("stress test is populating MRG with 3 tiers...");
156 + for(size_t i = 0; i < entries ;i++) {
157 + struct mrg_stress_entry *e = &t.array[i];
158 + for(size_t tier = 1; tier <= tiers ;tier++) {
159 + mrg_update_metric_retention_and_granularity_by_uuid(
160 + mrg, tier,
161 + &e->uuid,
162 + e->after,
163 + e->before,
164 + 1,
165 + e->before);
166 + }
167 + }
168 + netdata_log_info("stress test ready to run...");
169 +
170 + usec_t started_ut = now_monotonic_usec();
171 +
172 + ND_THREAD *th[threads];
173 + for(size_t i = 0; i < threads ; i++) {
174 + char buf[15 + 1];
175 + snprintfz(buf, sizeof(buf) - 1, "TH[%zu]", i);
176 + th[i] = nd_thread_create(buf, NETDATA_THREAD_OPTION_JOINABLE | NETDATA_THREAD_OPTION_DONT_LOG, mrg_stress, &t);
177 + }
178 +
179 + sleep_usec(run_for_secs * USEC_PER_SEC);
180 + __atomic_store_n(&t.stop, true, __ATOMIC_RELAXED);
181 +
182 + for(size_t i = 0; i < threads ; i++)
183 + nd_thread_signal_cancel(th[i]);
184 +
185 + for(size_t i = 0; i < threads ; i++)
186 + nd_thread_join(th[i]);
187 +
188 + usec_t ended_ut = now_monotonic_usec();
189 +
190 + struct mrg_statistics stats;
191 + mrg_get_statistics(mrg, &stats);
192 +
193 + netdata_log_info("DBENGINE METRIC: did %zu additions, %zu duplicate additions, "
194 + "%zu deletions, %zu wrong deletions, "
195 + "%zu successful searches, %zu wrong searches, "
196 + "in %"PRIu64" usecs",
197 + stats.additions, stats.additions_duplicate,
198 + stats.deletions, stats.delete_misses,
199 + stats.search_hits, stats.search_misses,
200 + ended_ut - started_ut);
201 +
202 + netdata_log_info("DBENGINE METRIC: updates performance: %0.2fk/sec total, %0.2fk/sec/thread",
203 + (double)t.updates / (double)((ended_ut - started_ut) / USEC_PER_SEC) / 1000.0,
204 + (double)t.updates / (double)((ended_ut - started_ut) / USEC_PER_SEC) / 1000.0 / threads);
205 +
206 + mrg_destroy(mrg);
207 +
208 + netdata_log_info("DBENGINE METRIC: all tests passed!");
209 +
210 + return 0;
211 +}
src/database/engine/mrg.c new
+475
@@ -0,0 +1,475 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +#include "mrg-internals.h"
4 +
5 +struct aral_statistics mrg_aral_statistics;
6 +
7 +// ----------------------------------------------------------------------------
8 +// public API
9 +
10 +inline MRG *mrg_create(void) {
11 + MRG *mrg = callocz(1, sizeof(MRG));
12 +
13 + for(size_t i = 0; i < _countof(mrg->index) ; i++) {
14 + rw_spinlock_init(&mrg->index[i].rw_spinlock);
15 +
16 + char buf[ARAL_MAX_NAME + 1];
17 + snprintfz(buf, ARAL_MAX_NAME, "mrg[%zu]", i);
18 +
19 + mrg->index[i].aral = aral_create(buf, sizeof(METRIC), 0, 16384, &mrg_aral_statistics, NULL, NULL,
20 + false, false, true);
21 + }
22 + pulse_aral_register_statistics(&mrg_aral_statistics, "mrg");
23 +
24 + mrg_load(mrg);
25 + return mrg;
26 +}
27 +
28 +struct aral_statistics *mrg_aral_stats(void) {
29 + return &mrg_aral_statistics;
30 +}
31 +
32 +size_t mrg_destroy(MRG *mrg) {
33 + if (!mrg)
34 + return 0;
35 +
36 + size_t referenced = 0;
37 +
38 + // Traverse all partitions
39 + for (size_t partition = 0; partition < UUIDMAP_PARTITIONS; partition++) {
40 + // Lock the partition to prevent new entries while we're cleaning up
41 + mrg_index_write_lock(mrg, partition);
42 +
43 + Word_t uuid_index = 0;
44 + Pvoid_t *uuid_pvalue;
45 +
46 + // Traverse all UUIDs in this partition
47 + for (uuid_pvalue = JudyLFirst(mrg->index[partition].uuid_judy, &uuid_index, PJE0);
48 + uuid_pvalue != NULL && uuid_pvalue != PJERR;
49 + uuid_pvalue = JudyLNext(mrg->index[partition].uuid_judy, &uuid_index, PJE0)) {
50 +
51 + if (!(*uuid_pvalue))
52 + continue;
53 +
54 + // Get the sections judy for this UUID
55 + Pvoid_t sections_judy = *uuid_pvalue;
56 + Word_t section_index = 0;
57 + Pvoid_t *section_pvalue;
58 +
59 + // Traverse all sections for this UUID
60 + for (section_pvalue = JudyLFirst(sections_judy, &section_index, PJE0);
61 + section_pvalue != NULL && section_pvalue != PJERR;
62 + section_pvalue = JudyLNext(sections_judy, &section_index, PJE0)) {
63 +
64 + if (!(*section_pvalue))
65 + continue;
66 +
67 + METRIC *metric = *section_pvalue;
68 +
69 + // Try to acquire metric for deletion
70 + if (!refcount_acquire_for_deletion(&metric->refcount))
71 + referenced++;
72 +
73 + uuidmap_free(metric->uuid);
74 + MRG_STATS_DELETED_METRIC(mrg, partition, metric->section);
75 + aral_freez(mrg->index[partition].aral, metric);
76 + }
77 +
78 + JudyLFreeArray(&sections_judy, PJE0);
79 + }
80 +
81 + JudyLFreeArray(&mrg->index[partition].uuid_judy, PJE0);
82 +
83 + // Unlock the partition
84 + mrg_index_write_unlock(mrg, partition);
85 +
86 + // Destroy the aral for this partition
87 + aral_destroy(mrg->index[partition].aral);
88 + }
89 +
90 + // Unregister the aral statistics
91 + pulse_aral_unregister_statistics(&mrg_aral_statistics);
92 +
93 + // Free the MRG structure
94 + freez(mrg);
95 +
96 + return referenced;
97 +}
98 +
99 +ALWAYS_INLINE
100 +METRIC *mrg_metric_add_and_acquire(MRG *mrg, MRG_ENTRY entry, bool *ret) {
101 +// internal_fatal(entry.latest_time_s > max_acceptable_collected_time(),
102 +// "DBENGINE METRIC: metric latest time is in the future");
103 +
104 + return metric_add_and_acquire(mrg, &entry, ret);
105 +}
106 +
107 +ALWAYS_INLINE
108 +METRIC *mrg_metric_get_and_acquire_by_uuid(MRG *mrg, nd_uuid_t *uuid, Word_t section) {
109 + UUIDMAP_ID id = uuidmap_create(*uuid);
110 + METRIC *metric = metric_get_and_acquire_by_id(mrg, id, section);
111 + uuidmap_free(id);
112 + return metric;
113 +}
114 +
115 +ALWAYS_INLINE
116 +METRIC *mrg_metric_get_and_acquire_by_id(MRG *mrg, UUIDMAP_ID id, Word_t section) {
117 + return metric_get_and_acquire_by_id(mrg, id, section);
118 +}
119 +
120 +ALWAYS_INLINE
121 +bool mrg_metric_release_and_delete(MRG *mrg, METRIC *metric) {
122 + return metric_release(mrg, metric);
123 +}
124 +
125 +ALWAYS_INLINE
126 +METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric) {
127 + metric_acquire(mrg, metric);
128 + return metric;
129 +}
130 +
131 +ALWAYS_INLINE
132 +bool mrg_metric_release(MRG *mrg, METRIC *metric) {
133 + return metric_release(mrg, metric);
134 +}
135 +
136 +ALWAYS_INLINE
137 +Word_t mrg_metric_id(MRG *mrg __maybe_unused, METRIC *metric) {
138 + return (Word_t)metric;
139 +}
140 +
141 +ALWAYS_INLINE
142 +nd_uuid_t *mrg_metric_uuid(MRG *mrg __maybe_unused, METRIC *metric) {
143 + return uuidmap_uuid_ptr(metric->uuid);
144 +}
145 +
146 +ALWAYS_INLINE
147 +UUIDMAP_ID mrg_metric_uuidmap_id_dup(MRG *mrg __maybe_unused, METRIC *metric) {
148 + return uuidmap_dup(metric->uuid);
149 +}
150 +
151 +ALWAYS_INLINE
152 +Word_t mrg_metric_section(MRG *mrg __maybe_unused, METRIC *metric) {
153 + return metric->section;
154 +}
155 +
156 +ALWAYS_INLINE
157 +bool mrg_metric_set_first_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
158 + internal_fatal(first_time_s < 0, "DBENGINE METRIC: timestamp is negative");
159 +
160 + if(first_time_s == LONG_MAX)
161 + first_time_s = 0;
162 +
163 + if(unlikely(first_time_s < 0))
164 + return false;
165 +
166 + __atomic_store_n(&metric->first_time_s, first_time_s, __ATOMIC_RELAXED);
167 +
168 + return true;
169 +}
170 +
171 +ALWAYS_INLINE
172 +void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s, time_t last_time_s, uint32_t update_every_s) {
173 + internal_fatal(first_time_s < 0 || last_time_s < 0,
174 + "DBENGINE METRIC: timestamp is negative");
175 + internal_fatal(first_time_s > max_acceptable_collected_time(),
176 + "DBENGINE METRIC: metric first time is in the future");
177 + internal_fatal(last_time_s > max_acceptable_collected_time(),
178 + "DBENGINE METRIC: metric last time is in the future");
179 +
180 + if(first_time_s > 0 && first_time_s != LONG_MAX)
181 + set_metric_field_with_condition(metric->first_time_s, first_time_s, _current <= 0 || (_wanted != 0 && _wanted != LONG_MAX && _wanted < _current));
182 +
183 + if(last_time_s > 0) {
184 + if(set_metric_field_with_condition(metric->latest_time_s_clean, last_time_s, _current <= 0 || _wanted > _current) &&
185 + update_every_s > 0)
186 + // set the latest update every too
187 + set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, true);
188 + }
189 + else if(update_every_s > 0)
190 + // set it only if it is invalid
191 + set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, _current <= 0);
192 +}
193 +
194 +ALWAYS_INLINE
195 +bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
196 + internal_fatal(first_time_s < 0, "DBENGINE METRIC: timestamp is negative");
197 + return set_metric_field_with_condition(metric->first_time_s, first_time_s, _wanted != 0 && _wanted != LONG_MAX && _wanted > _current);
198 +}
199 +
200 +ALWAYS_INLINE
201 +time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
202 + return mrg_metric_get_first_time_s_smart(mrg, metric);
203 +}
204 +
205 +void mrg_metric_clear_retention(MRG *mrg __maybe_unused, METRIC *metric) {
206 + __atomic_store_n(&metric->first_time_s, 0, __ATOMIC_RELAXED);
207 + __atomic_store_n(&metric->latest_time_s_clean, 0, __ATOMIC_RELAXED);
208 + __atomic_store_n(&metric->latest_time_s_hot, 0, __ATOMIC_RELAXED);
209 +}
210 +
211 +ALWAYS_INLINE_HOT
212 +void mrg_metric_get_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t *first_time_s, time_t *last_time_s, uint32_t *update_every_s) {
213 + time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
214 + time_t hot = __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED);
215 +
216 + *last_time_s = MAX(clean, hot);
217 + *first_time_s = mrg_metric_get_first_time_s_smart(mrg, metric);
218 + if (update_every_s)
219 + *update_every_s = __atomic_load_n(&metric->latest_update_every_s, __ATOMIC_RELAXED);
220 +}
221 +
222 +ALWAYS_INLINE
223 +bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
224 + internal_fatal(latest_time_s < 0, "DBENGINE METRIC: timestamp is negative");
225 +
226 +// internal_fatal(latest_time_s > max_acceptable_collected_time(),
227 +// "DBENGINE METRIC: metric latest time is in the future");
228 +
229 +// internal_fatal(metric->latest_time_s_clean > latest_time_s,
230 +// "DBENGINE METRIC: metric new clean latest time is older than the previous one");
231 +
232 + if(latest_time_s > 0) {
233 + if(set_metric_field_with_condition(metric->latest_time_s_clean, latest_time_s, true)) {
234 + set_metric_field_with_condition(metric->first_time_s, latest_time_s, _current <= 0 || _wanted < _current);
235 +
236 + return true;
237 + }
238 + }
239 +
240 + return false;
241 +}
242 +
243 +// returns true when metric still has retention
244 +ALWAYS_INLINE
245 +bool mrg_metric_has_zero_disk_retention(MRG *mrg __maybe_unused, METRIC *metric) {
246 + Word_t section = mrg_metric_section(mrg, metric);
247 + bool do_again = false;
248 + size_t countdown = 5;
249 +
250 + do {
251 + time_t min_first_time_s = LONG_MAX;
252 + time_t max_end_time_s = 0;
253 + PGC_PAGE *page;
254 + PGC_SEARCH method = PGC_SEARCH_FIRST;
255 + time_t page_first_time_s = 0;
256 + while ((page = pgc_page_get_and_acquire(main_cache, section, (Word_t)metric, page_first_time_s, method))) {
257 + method = PGC_SEARCH_NEXT;
258 +
259 + bool is_hot = pgc_is_page_hot(page);
260 + bool is_dirty = pgc_is_page_dirty(page);
261 + page_first_time_s = pgc_page_start_time_s(page);
262 + time_t page_end_time_s = pgc_page_end_time_s(page);
263 +
264 + if ((is_hot || is_dirty) && page_first_time_s > 0 && page_first_time_s < min_first_time_s)
265 + min_first_time_s = page_first_time_s;
266 +
267 + if (is_dirty && page_end_time_s > max_end_time_s)
268 + max_end_time_s = page_end_time_s;
269 +
270 + pgc_page_release(main_cache, page);
271 + }
272 +
273 + if (min_first_time_s == LONG_MAX)
274 + min_first_time_s = 0;
275 +
276 + if (--countdown && !min_first_time_s && __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED))
277 + do_again = true;
278 + else {
279 + internal_error(!countdown, "METRIC: giving up on updating the retention of metric without disk retention");
280 +
281 + do_again = false;
282 + set_metric_field_with_condition(metric->first_time_s, min_first_time_s, true);
283 + set_metric_field_with_condition(metric->latest_time_s_clean, max_end_time_s, true);
284 + }
285 + } while(do_again);
286 +
287 + time_t first, last;
288 + mrg_metric_get_retention(mrg, metric, &first, &last, NULL);
289 + return (first && last && first < last);
290 +}
291 +
292 +ALWAYS_INLINE_HOT
293 +bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
294 + internal_fatal(latest_time_s < 0, "DBENGINE METRIC: timestamp is negative");
295 +
296 +// internal_fatal(latest_time_s > max_acceptable_collected_time(),
297 +// "DBENGINE METRIC: metric latest time is in the future");
298 +
299 + if(likely(latest_time_s > 0)) {
300 + __atomic_store_n(&metric->latest_time_s_hot, latest_time_s, __ATOMIC_RELAXED);
301 + return true;
302 + }
303 +
304 + return false;
305 +}
306 +
307 +ALWAYS_INLINE
308 +time_t mrg_metric_get_latest_clean_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
309 + time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
310 + return clean;
311 +}
312 +
313 +ALWAYS_INLINE_HOT
314 +time_t mrg_metric_get_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
315 + time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
316 + time_t hot = __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED);
317 +
318 + return MAX(clean, hot);
319 +}
320 +
321 +ALWAYS_INLINE
322 +bool mrg_metric_set_update_every(MRG *mrg __maybe_unused, METRIC *metric, uint32_t update_every_s) {
323 + if(likely(update_every_s > 0))
324 + return set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, true);
325 +
326 + return false;
327 +}
328 +
329 +ALWAYS_INLINE_HOT
330 +bool mrg_metric_set_update_every_s_if_zero(MRG *mrg __maybe_unused, METRIC *metric, uint32_t update_every_s) {
331 + if(likely(update_every_s > 0))
332 + return set_metric_field_with_condition(metric->latest_update_every_s, update_every_s, _current <= 0);
333 +
334 + return false;
335 +}
336 +
337 +ALWAYS_INLINE
338 +uint32_t mrg_metric_get_update_every_s(MRG *mrg __maybe_unused, METRIC *metric) {
339 + return __atomic_load_n(&metric->latest_update_every_s, __ATOMIC_RELAXED);
340 +}
341 +
342 +#ifdef NETDATA_INTERNAL_CHECKS
343 +ALWAYS_INLINE bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) {
344 + pid_t expected = __atomic_load_n(&metric->writer, __ATOMIC_RELAXED);
345 + pid_t wanted = gettid_cached();
346 + bool done = true;
347 +
348 + do {
349 + if(expected != 0) {
350 + done = false;
351 + break;
352 + }
353 + } while(!__atomic_compare_exchange_n(&metric->writer, &expected, wanted, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
354 +
355 + if(done)
356 + __atomic_add_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
357 + else
358 + __atomic_add_fetch(&mrg->index[metric->partition].stats.writers_conflicts, 1, __ATOMIC_RELAXED);
359 +
360 + return done;
361 +}
362 +
363 +ALWAYS_INLINE bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) {
364 + // this function can be called from a different thread than the one than the writer
365 +
366 + pid_t expected = __atomic_load_n(&metric->writer, __ATOMIC_RELAXED);
367 + pid_t wanted = 0;
368 + bool done = true;
369 +
370 + do {
371 + if(!expected) {
372 + done = false;
373 + break;
374 + }
375 + } while(!__atomic_compare_exchange_n(&metric->writer, &expected, wanted, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
376 +
377 + if(done)
378 + __atomic_sub_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
379 +
380 + return done;
381 +}
382 +#endif
383 +
384 +inline void mrg_update_metric_retention_and_granularity_by_uuid(
385 + MRG *mrg, Word_t section, nd_uuid_t *uuid,
386 + time_t first_time_s, time_t last_time_s,
387 + uint32_t update_every_s, time_t now_s)
388 +{
389 + if(unlikely(last_time_s > now_s)) {
390 + nd_log_limit_static_global_var(erl, 1, 0);
391 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
392 + "DBENGINE JV2: wrong last time on-disk (%ld - %ld, now %ld), "
393 + "fixing last time to now",
394 + first_time_s, last_time_s, now_s);
395 + last_time_s = now_s;
396 + }
397 +
398 + if (unlikely(first_time_s > last_time_s)) {
399 + nd_log_limit_static_global_var(erl, 1, 0);
400 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
401 + "DBENGINE JV2: wrong first time on-disk (%ld - %ld, now %ld), "
402 + "fixing first time to last time",
403 + first_time_s, last_time_s, now_s);
404 +
405 + first_time_s = last_time_s;
406 + }
407 +
408 + if (unlikely(first_time_s == 0 || last_time_s == 0)) {
409 + nd_log_limit_static_global_var(erl, 1, 0);
410 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
411 + "DBENGINE JV2: zero on-disk timestamps (%ld - %ld, now %ld), "
412 + "using them as-is",
413 + first_time_s, last_time_s, now_s);
414 + }
415 +
416 + bool added = false;
417 + METRIC *metric = mrg_metric_get_and_acquire_by_uuid(mrg, uuid, section);
418 + if (!metric) {
419 + MRG_ENTRY entry = {
420 + .uuid = uuid,
421 + .section = section,
422 + .first_time_s = first_time_s,
423 + .last_time_s = last_time_s,
424 + .latest_update_every_s = update_every_s,
425 + };
426 + metric = mrg_metric_add_and_acquire(mrg, entry, &added);
427 + }
428 +
429 + struct rrdengine_instance *ctx = (struct rrdengine_instance *) section;
430 + if (likely(!added)) {
431 + uint64_t old_samples = 0;
432 +
433 + if (update_every_s && metric->latest_update_every_s && metric->latest_time_s_clean)
434 + old_samples = (metric->latest_time_s_clean - metric->first_time_s) / metric->latest_update_every_s;
435 +
436 + mrg_metric_expand_retention(mrg, metric, first_time_s, last_time_s, update_every_s);
437 +
438 + uint64_t new_samples = 0;
439 + if (update_every_s && metric->latest_update_every_s && metric->latest_time_s_clean)
440 + new_samples = (metric->latest_time_s_clean - metric->first_time_s) / metric->latest_update_every_s;
441 +
442 + __atomic_add_fetch(&ctx->atomic.samples, new_samples - old_samples, __ATOMIC_RELAXED);
443 + }
444 + else {
445 + // Newly added
446 + if (update_every_s) {
447 + uint64_t samples = (last_time_s - first_time_s) / update_every_s;
448 + __atomic_add_fetch(&ctx->atomic.samples, samples, __ATOMIC_RELAXED);
449 + }
450 + }
451 +
452 + mrg_metric_release(mrg, metric);
453 +}
454 +
455 +inline void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
456 + memset(s, 0, sizeof(struct mrg_statistics));
457 +
458 + for(size_t i = 0; i < _countof(mrg->index) ;i++) {
459 + s->entries += __atomic_load_n(&mrg->index[i].stats.entries, __ATOMIC_RELAXED);
460 + s->entries_acquired += __atomic_load_n(&mrg->index[i].stats.entries_acquired, __ATOMIC_RELAXED);
461 + s->size += __atomic_load_n(&mrg->index[i].stats.size, __ATOMIC_RELAXED);
462 + s->current_references += __atomic_load_n(&mrg->index[i].stats.current_references, __ATOMIC_RELAXED);
463 + s->additions += __atomic_load_n(&mrg->index[i].stats.additions, __ATOMIC_RELAXED);
464 + s->additions_duplicate += __atomic_load_n(&mrg->index[i].stats.additions_duplicate, __ATOMIC_RELAXED);
465 + s->deletions += __atomic_load_n(&mrg->index[i].stats.deletions, __ATOMIC_RELAXED);
466 + s->delete_having_retention_or_referenced += __atomic_load_n(&mrg->index[i].stats.delete_having_retention_or_referenced, __ATOMIC_RELAXED);
467 + s->delete_misses += __atomic_load_n(&mrg->index[i].stats.delete_misses, __ATOMIC_RELAXED);
468 + s->search_hits += __atomic_load_n(&mrg->index[i].stats.search_hits, __ATOMIC_RELAXED);
469 + s->search_misses += __atomic_load_n(&mrg->index[i].stats.search_misses, __ATOMIC_RELAXED);
470 + s->writers += __atomic_load_n(&mrg->index[i].stats.writers, __ATOMIC_RELAXED);
471 + s->writers_conflicts += __atomic_load_n(&mrg->index[i].stats.writers_conflicts, __ATOMIC_RELAXED);
472 + }
473 +
474 + s->size += sizeof(MRG);
475 +}
src/database/engine/mrg.h renamed
+5 -1
@@ -46,7 +46,7 @@ MRG *mrg_create(void);
46 size_t mrg_destroy(MRG *mrg);
47
48 METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric);
49 -void mrg_metric_release(MRG *mrg, METRIC *metric);
49 +bool mrg_metric_release(MRG *mrg, METRIC *metric);
50
51 METRIC *mrg_metric_add_and_acquire(MRG *mrg, MRG_ENTRY entry, bool *ret);
52 METRIC *mrg_metric_get_and_acquire_by_id(MRG *mrg, UUIDMAP_ID id, Word_t section);
@@ -89,4 +89,8 @@ void mrg_update_metric_retention_and_granularity_by_uuid(
89 time_t first_time_s, time_t last_time_s,
90 uint32_t update_every_s, time_t now_s);
91
92 +bool mrg_save(MRG *mrg);
93 +bool mrg_load(MRG *mrg);
94 +void mrg_metric_prepopulate_cleanup(MRG *mrg);
95 +
96 #endif // DBENGINE_METRIC_H
src/database/engine/rrdengine.h
+1 -1
@@ -15,7 +15,7 @@
15 #include "journalfile.h"
16 #include "rrdengineapi.h"
17 #include "pagecache.h"
18 -#include "metric.h"
18 +#include "mrg.h"
19 #include "cache.h"
20 #include "pdc.h"
21 #include "page.h"
src/database/engine/rrdengineapi.c
-2
@@ -166,8 +166,6 @@ static METRIC *rrdeng_metric_create(STORAGE_INSTANCE *si, nd_uuid_t *uuid) {
166
167 bool added;
168 METRIC *metric = mrg_metric_add_and_acquire(main_mrg, entry, &added);
169 - if (added)
170 - __atomic_add_fetch(&ctx->atomic.metrics, 1, __ATOMIC_RELAXED);
169 return metric;
170 }
171
src/database/sqlite/sqlite_metadata.c
+44
@@ -2061,6 +2061,50 @@ static void after_metadata_hosts(uv_work_t *req, int status __maybe_unused)
2061 freez(data);
2062 }
2063
2064 +#define GET_UUID_LIST "SELECT dim_id FROM dimension"
2065 +size_t populate_metrics_from_database(void *mrg, void (*populate_cb)(void *mrg, Word_t section, nd_uuid_t *uuid))
2066 +{
2067 + sqlite3_stmt *res = NULL;
2068 + sqlite3 *local_meta_db = NULL;
2069 +
2070 + char sqlite_database[FILENAME_MAX + 1];
2071 + snprintfz(sqlite_database, sizeof(sqlite_database) - 1, "%s/netdata-meta.db", netdata_configured_cache_dir);
2072 + int rc = sqlite3_open_v2(sqlite_database, &local_meta_db, SQLITE_OPEN_READONLY | SQLITE_OPEN_NOMUTEX, NULL);
2073 + if (rc != SQLITE_OK) {
2074 + sqlite3_close(local_meta_db);
2075 + local_meta_db = NULL;
2076 + }
2077 +
2078 + if (local_meta_db)
2079 + db_execute(local_meta_db, "PRAGMA cache_size=10000");
2080 +
2081 + if (!PREPARE_STATEMENT(local_meta_db ? local_meta_db : db_meta, GET_UUID_LIST, &res)) {
2082 + sqlite3_close(local_meta_db);
2083 + return 0;
2084 + }
2085 +
2086 + size_t count = 0;
2087 +
2088 + usec_t started_ut = now_monotonic_usec();
2089 + while (sqlite3_step(res) == SQLITE_ROW) {
2090 + nd_uuid_t *uuid = (nd_uuid_t *)sqlite3_column_blob(res, 0);
2091 +
2092 + for (size_t tier = 0; tier < nd_profile.storage_tiers ; tier++) {
2093 + if (unlikely(!multidb_ctx[tier]))
2094 + continue;
2095 +
2096 + populate_cb(mrg, (Word_t)multidb_ctx[tier], uuid);
2097 + }
2098 + count++;
2099 + }
2100 +
2101 + SQLITE_FINALIZE(res);
2102 + sqlite3_close(local_meta_db);
2103 + COMPUTE_DURATION(report_duration, "us", started_ut, now_monotonic_usec());
2104 + nd_log_daemon(NDLP_INFO, "MRG: Loaded %zu metrics from database in %s", count, report_duration);
2105 + return count;
2106 +}
2107 +
2108 static void metadata_scan_host(RRDHOST *host, BUFFER *work_buffer, bool shutting_down)
2109 {
2110 static bool skip_models = false;
src/database/sqlite/sqlite_metadata.h
+1
@@ -62,6 +62,7 @@ void metadata_sync_shutdown_background_wait(void);
62 void metadata_queue_ctx_host_cleanup(nd_uuid_t *host_uuid, const char *context);
63 void store_host_info_and_metadata(RRDHOST *host, BUFFER *work_buffer);
64 void metadata_execute_store_statement(sqlite3_stmt *stmt);
65 +size_t populate_metrics_from_database(void *mrg, void (*populate_cb)(void *mrg, Word_t section, nd_uuid_t *uuid));
66
67 // UNIT TEST
68 int metadata_unittest(void);
src/exporting/prometheus/prometheus.c
+3 -3
@@ -4,7 +4,7 @@
4
5 DEFINE_JUDYL_TYPED(PROM_CONTEXT_OPTIONS, PROMETHEUS_OUTPUT_OPTIONS);
6
7 -static void PROM_CONTEXT_OPTIONS_free_cb(Word_t index, PROMETHEUS_OUTPUT_OPTIONS options __maybe_unused) {
7 +static void PROM_CONTEXT_OPTIONS_free_cb(Word_t index, PROMETHEUS_OUTPUT_OPTIONS options __maybe_unused, void *data __maybe_unused) {
8 STRING *context_id = (STRING *)index;
9 string_freez(context_id);
10 }
@@ -985,7 +985,7 @@ void rrd_stats_api_v1_charts_allmetrics_prometheus_single_host(
985 rrd_stats_api_v1_charts_allmetrics_prometheus(
986 prometheus_exporter_instance, host, filter_string, wb, prefix, exporting_options, 0, output_options, &context_options);
987
988 - PROM_CONTEXT_OPTIONS_FREE(&context_options, PROM_CONTEXT_OPTIONS_free_cb);
988 + PROM_CONTEXT_OPTIONS_FREE(&context_options, PROM_CONTEXT_OPTIONS_free_cb, NULL);
989 }
990
991 /**
@@ -1030,5 +1030,5 @@ void rrd_stats_api_v1_charts_allmetrics_prometheus_all_hosts(
1030 }
1031 dfe_done(host);
1032
1033 - PROM_CONTEXT_OPTIONS_FREE(&context_options, PROM_CONTEXT_OPTIONS_free_cb);
1033 + PROM_CONTEXT_OPTIONS_FREE(&context_options, PROM_CONTEXT_OPTIONS_free_cb, NULL);
1034 }
src/libnetdata/libjudy/judyl-typed.h
+19 -10
@@ -13,54 +13,63 @@
13 Pvoid_t judyl; \
14 } NAME##_JudyLSet; \
15 \
16 - static inline void NAME##_INIT(NAME##_JudyLSet *set) { \
16 + ALWAYS_INLINE \
17 + static void NAME##_INIT(NAME##_JudyLSet *set) { \
18 set->judyl = NULL; \
19 } \
20 \
20 - static inline bool NAME##_SET(NAME##_JudyLSet *set, Word_t index, TYPE value) { \
21 + ALWAYS_INLINE \
22 + static bool NAME##_SET(NAME##_JudyLSet *set, Word_t index, TYPE value) { \
23 Pvoid_t *pValue = JudyLIns(&set->judyl, index, PJE0); \
24 if (pValue == PJERR) return false; \
25 *pValue = (void *)PACK_MACRO(value); \
26 return true; \
27 } \
28 \
27 - static inline TYPE NAME##_GET(NAME##_JudyLSet *set, Word_t index) { \
29 + ALWAYS_INLINE \
30 + static TYPE NAME##_GET(NAME##_JudyLSet *set, Word_t index) { \
31 Pvoid_t *pValue = JudyLGet(set->judyl, index, PJE0); \
32 return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
33 } \
34 \
32 - static inline bool NAME##_DEL(NAME##_JudyLSet *set, Word_t index) { \
35 + ALWAYS_INLINE \
36 + static bool NAME##_DEL(NAME##_JudyLSet *set, Word_t index) { \
37 return JudyLDel(&set->judyl, index, PJE0) == 1; \
38 } \
39 \
36 - static inline TYPE NAME##_FIRST(NAME##_JudyLSet *set, Word_t *index) { \
40 + ALWAYS_INLINE \
41 + static TYPE NAME##_FIRST(NAME##_JudyLSet *set, Word_t *index) { \
42 Pvoid_t *pValue = JudyLFirst(set->judyl, index, PJE0); \
43 return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
44 } \
45 \
41 - static inline TYPE NAME##_NEXT(NAME##_JudyLSet *set, Word_t *index) { \
46 + ALWAYS_INLINE \
47 + static TYPE NAME##_NEXT(NAME##_JudyLSet *set, Word_t *index) { \
48 Pvoid_t *pValue = JudyLNext(set->judyl, index, PJE0); \
49 return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
50 } \
51 \
46 - static inline TYPE NAME##_LAST(NAME##_JudyLSet *set, Word_t *index) { \
52 + ALWAYS_INLINE \
53 + static TYPE NAME##_LAST(NAME##_JudyLSet *set, Word_t *index) { \
54 Pvoid_t *pValue = JudyLLast(set->judyl, index, PJE0); \
55 return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
56 } \
57 \
51 - static inline TYPE NAME##_PREV(NAME##_JudyLSet *set, Word_t *index) { \
58 + ALWAYS_INLINE \
59 + static TYPE NAME##_PREV(NAME##_JudyLSet *set, Word_t *index) { \
60 Pvoid_t *pValue = JudyLPrev(set->judyl, index, PJE0); \
61 return (pValue != NULL) ? (TYPE)UNPACK_MACRO(*pValue) : (TYPE){0}; \
62 } \
63 \
56 - static inline void NAME##_FREE(NAME##_JudyLSet *set, void (*callback)(Word_t, TYPE)) { \
64 + ALWAYS_INLINE \
65 + static void NAME##_FREE(NAME##_JudyLSet *set, void (*callback)(Word_t, TYPE, void *), void *data) { \
66 Word_t index = 0; \
67 Pvoid_t *pValue; \
68 if (callback) { \
69 for (pValue = JudyLFirst(set->judyl, &index, PJE0); \
70 pValue != NULL; \
71 pValue = JudyLNext(set->judyl, &index, PJE0)) { \
63 - callback(index, (TYPE)UNPACK_MACRO(*pValue)); \
72 + callback(index, (TYPE)UNPACK_MACRO(*pValue), data); \
73 } \
74 } \
75 JudyLFreeArray(&set->judyl, PJE0); \
src/libnetdata/socket/nd-poll.c
+3 -3
@@ -238,7 +238,7 @@ int nd_poll_wait(nd_poll_t *ndpl, int timeout_ms, nd_poll_result_t *result) {
238 } while(true);
239 }
240
241 -static void nd_poll_free_callback(Word_t fd __maybe_unused, struct fd_info *fdi) {
241 +static void nd_poll_free_callback(Word_t fd __maybe_unused, struct fd_info *fdi, void *data __maybe_unused) {
242 freez(fdi);
243 }
244
@@ -246,7 +246,7 @@ static void nd_poll_free_callback(Word_t fd __maybe_unused, struct fd_info *fdi)
246 void nd_poll_destroy(nd_poll_t *ndpl) {
247 if (ndpl) {
248 close(ndpl->epoll_fd);
249 - POINTERS_FREE(&ndpl->pointers, nd_poll_free_callback);
249 + POINTERS_FREE(&ndpl->pointers, nd_poll_free_callback, NULL);
250 freez(ndpl);
251 }
252 }
@@ -440,7 +440,7 @@ int nd_poll_wait(nd_poll_t *ndpl, int timeout_ms, nd_poll_result_t *result) {
440 void nd_poll_destroy(nd_poll_t *ndpl) {
441 if (ndpl) {
442 free(ndpl->fds);
443 - POINTERS_FREE(&ndpl->pointers, NULL);
443 + POINTERS_FREE(&ndpl->pointers, NULL, NULL);
444 freez(ndpl);
445 }
446 }
src/streaming/stream-thread.c
+1 -1
@@ -625,7 +625,7 @@ void *stream_thread(void *ptr) {
625 // cleanup receiver and dispatcher
626 stream_sender_cleanup(sth);
627 stream_receiver_cleanup(sth);
628 - META_FREE(&sth->run.meta, NULL);
628 + META_FREE(&sth->run.meta, NULL, NULL);
629
630 // cleanup the thread structures
631 spinlock_lock(&sth->messages.spinlock);