Relax jnfv2 caching (#15224)
* readers should be able to recursively acquire the lock, even when there is a writer waiting * dont madvise dontneed and random * dont validate extents and metrics on jnfv2 * dont validate crc * Delay journal metric check * added MRG stress test --------- Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>
Costa Tsaousis committed
Jun 26, 2023 at 12:49 UTC
f90d56f18d29c2835bc278f6a22e840230b9ca86
4 files changed
+201
-208
database/engine/journalfile.c
+21
-58
@@ -1,57 +1,6 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
#include "rrdengine.h"
3
4
-
5
-// DBENGINE2: Helper
6
-
7
-static void update_metric_retention_and_granularity_by_uuid(
8
- struct rrdengine_instance *ctx, uuid_t *uuid,
9
- time_t first_time_s, time_t last_time_s,
10
- time_t update_every_s, time_t now_s)
11
-{
12
- if(unlikely(last_time_s > now_s)) {
13
- error_limit_static_global_var(erl, 1, 0);
14
- error_limit(&erl, "DBENGINE JV2: wrong last time on-disk (%ld - %ld, now %ld), "
15
- "fixing last time to now",
16
- first_time_s, last_time_s, now_s);
17
- last_time_s = now_s;
18
- }
19
-
20
- if (unlikely(first_time_s > last_time_s)) {
21
- error_limit_static_global_var(erl, 1, 0);
22
- error_limit(&erl, "DBENGINE JV2: wrong first time on-disk (%ld - %ld, now %ld), "
23
- "fixing first time to last time",
24
- first_time_s, last_time_s, now_s);
25
-
26
- first_time_s = last_time_s;
27
- }
28
-
29
- if (unlikely(first_time_s == 0 || last_time_s == 0)) {
30
- error_limit_static_global_var(erl, 1, 0);
31
- error_limit(&erl, "DBENGINE JV2: zero on-disk timestamps (%ld - %ld, now %ld), "
32
- "using them as-is",
33
- first_time_s, last_time_s, now_s);
34
- }
35
-
36
- bool added = false;
37
- METRIC *metric = mrg_metric_get_and_acquire(main_mrg, uuid, (Word_t) ctx);
38
- if (!metric) {
39
- MRG_ENTRY entry = {
40
- .section = (Word_t) ctx,
41
- .first_time_s = first_time_s,
42
- .last_time_s = last_time_s,
43
- .latest_update_every_s = (uint32_t) update_every_s
44
- };
45
- uuid_copy(entry.uuid, *uuid);
46
- metric = mrg_metric_add_and_acquire(main_mrg, entry, &added);
47
- }
48
-
49
- if (likely(!added))
50
- mrg_metric_expand_retention(main_mrg, metric, first_time_s, last_time_s, update_every_s);
51
-
52
- mrg_metric_release(main_mrg, metric);
53
-}
54
-
4
static void after_extent_write_journalfile_v1_io(uv_fs_t* req)
5
{
6
worker_is_busy(RRDENG_FLUSH_TRANSACTION_BUFFER_CB);
@@ -265,8 +214,9 @@ static struct journal_v2_header *journalfile_v2_mounted_data_get(struct rrdengin
214
215
madvise_dontfork(journalfile->mmap.data, journalfile->mmap.size);
216
madvise_dontdump(journalfile->mmap.data, journalfile->mmap.size);
268
- madvise_random(journalfile->mmap.data, journalfile->mmap.size);
269
- madvise_dontneed(journalfile->mmap.data, journalfile->mmap.size);
217
+// madvise_willneed(journalfile->mmap.data, journalfile->v2.size_of_directory);
218
+// madvise_random(journalfile->mmap.data, journalfile->mmap.size);
219
+// madvise_dontneed(journalfile->mmap.data, journalfile->mmap.size);
220
221
spinlock_lock(&journalfile->v2.spinlock);
222
journalfile->v2.flags |= JOURNALFILE_FLAG_IS_AVAILABLE | JOURNALFILE_FLAG_IS_MOUNTED;
@@ -459,6 +409,7 @@ void journalfile_v2_data_set(struct rrdengine_journalfile *journalfile, int fd,
409
struct journal_v2_header *j2_header = journalfile->mmap.data;
410
journalfile->v2.first_time_s = (time_t)(j2_header->start_time_ut / USEC_PER_SEC);
411
journalfile->v2.last_time_s = (time_t)(j2_header->end_time_ut / USEC_PER_SEC);
412
+ // journalfile->v2.size_of_directory = j2_header->metric_offset + j2_header->metric_count * sizeof(struct journal_metric_list);
413
414
journalfile_v2_mounted_data_unmount(journalfile, true, true);
415
@@ -957,12 +908,12 @@ static int journalfile_v2_validate(void *data_start, size_t journal_v2_file_size
908
rc = journalfile_check_v2_extent_list(data_start, journal_v2_file_size);
909
if (rc) return 1;
910
960
- rc = journalfile_check_v2_metric_list(data_start, journal_v2_file_size);
961
- if (rc) return 1;
962
-
911
if (!db_engine_journal_check)
912
return 0;
913
914
+ rc = journalfile_check_v2_metric_list(data_start, journal_v2_file_size);
915
+ if (rc) return 1;
916
+
917
// Verify complete UUID chain
918
919
struct journal_metric_list *metric = (void *) (data_start + j2_header->metric_offset);
@@ -1027,6 +978,15 @@ void journalfile_v2_populate_retention_to_mrg(struct rrdengine_instance *ctx, st
978
uint8_t *data_start = (uint8_t *)j2_header;
979
uint32_t entries = j2_header->metric_count;
980
981
+ if (journalfile->v2.flags & JOURNALFILE_FLAG_METRIC_CRC_CHECK) {
982
+ journalfile->v2.flags &= ~JOURNALFILE_FLAG_METRIC_CRC_CHECK;
983
+ if (journalfile_check_v2_metric_list(data_start, j2_header->journal_v2_file_size)) {
984
+ journalfile->v2.flags &= ~JOURNALFILE_FLAG_IS_AVAILABLE;
985
+ // needs rebuild
986
+ return;
987
+ }
988
+ }
989
+
990
struct journal_metric_list *metric = (struct journal_metric_list *) (data_start + j2_header->metric_offset);
991
time_t header_start_time_s = (time_t) (j2_header->start_time_ut / USEC_PER_SEC);
992
time_t now_s = max_acceptable_collected_time();
@@ -1034,8 +994,8 @@ void journalfile_v2_populate_retention_to_mrg(struct rrdengine_instance *ctx, st
994
time_t start_time_s = header_start_time_s + metric->delta_start_s;
995
time_t end_time_s = header_start_time_s + metric->delta_end_s;
996
1037
- update_metric_retention_and_granularity_by_uuid(
1038
- ctx, &metric->uuid, start_time_s, end_time_s, (time_t) metric->update_every_s, now_s);
997
+ mrg_update_metric_retention_and_granularity_by_uuid(
998
+ main_mrg, (Word_t)ctx, &metric->uuid, start_time_s, end_time_s, (time_t) metric->update_every_s, now_s);
999
1000
metric++;
1001
}
@@ -1138,6 +1098,9 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
1098
);
1099
1100
// Initialize the journal file to be able to access the data
1101
+
1102
+ if (!db_engine_journal_check)
1103
+ journalfile->v2.flags |= JOURNALFILE_FLAG_METRIC_CRC_CHECK;
1104
journalfile_v2_data_set(journalfile, fd, data_start, journal_v2_file_size);
1105
1106
ctx_current_disk_space_increase(ctx, journal_v2_file_size);
database/engine/journalfile.h
+2
@@ -21,6 +21,7 @@ typedef enum __attribute__ ((__packed__)) {
21
JOURNALFILE_FLAG_IS_AVAILABLE = (1 << 0),
22
JOURNALFILE_FLAG_IS_MOUNTED = (1 << 1),
23
JOURNALFILE_FLAG_MOUNTED_FOR_RETENTION = (1 << 2),
24
+ JOURNALFILE_FLAG_METRIC_CRC_CHECK = (1 << 3),
25
} JOURNALFILE_FLAGS;
26
27
/* only one event loop is supported for now */
@@ -39,6 +40,7 @@ struct rrdengine_journalfile {
40
time_t first_time_s;
41
time_t last_time_s;
42
time_t not_needed_since_s;
43
+ // uint32_t size_of_directory;
44
} v2;
45
46
struct {
database/engine/metric.c
+172
-150
@@ -188,7 +188,7 @@ static inline bool metric_release_and_can_be_deleted(MRG *mrg __maybe_unused, ME
188
return ret;
189
}
190
191
-static METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
191
+static inline METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
192
size_t partition = uuid_partition(mrg, &entry->uuid);
193
194
METRIC *allocation = aral_mallocz(mrg->aral[partition]);
@@ -230,6 +230,7 @@ static METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
230
}
231
232
METRIC *metric = allocation;
233
+ // memcpy(metric->uuid, entry->uuid, sizeof(uuid_t));
234
uuid_copy(metric->uuid, entry->uuid);
235
metric->section = entry->section;
236
metric->first_time_s = MAX(0, entry->first_time_s);
@@ -254,7 +255,7 @@ static METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
255
return metric;
256
}
257
257
-static METRIC *metric_get_and_acquire(MRG *mrg, uuid_t *uuid, Word_t section) {
258
+static inline METRIC *metric_get_and_acquire(MRG *mrg, uuid_t *uuid, Word_t section) {
259
size_t partition = uuid_partition(mrg, uuid);
260
261
mrg_index_read_lock(mrg, partition);
@@ -283,8 +284,8 @@ static METRIC *metric_get_and_acquire(MRG *mrg, uuid_t *uuid, Word_t section) {
284
return metric;
285
}
286
286
-static bool acquired_metric_del(MRG *mrg, METRIC *metric) {
287
- size_t partition = uuid_partition(mrg, &metric->uuid);
287
+static inline bool acquired_metric_del(MRG *mrg, METRIC *metric) {
288
+ size_t partition = metric->partition;
289
290
size_t mem_before_judyl, mem_after_judyl;
291
@@ -333,7 +334,7 @@ static bool acquired_metric_del(MRG *mrg, METRIC *metric) {
334
// ----------------------------------------------------------------------------
335
// public API
336
336
-MRG *mrg_create(void) {
337
+inline MRG *mrg_create(void) {
338
MRG *mrg = callocz(1, sizeof(MRG));
339
340
for(size_t i = 0; i < MRG_PARTITIONS ; i++) {
@@ -354,15 +355,15 @@ MRG *mrg_create(void) {
355
return mrg;
356
}
357
357
-size_t mrg_aral_structures(void) {
358
+inline size_t mrg_aral_structures(void) {
359
return aral_structures_from_stats(&mrg_aral_statistics);
360
}
361
361
-size_t mrg_aral_overhead(void) {
362
+inline size_t mrg_aral_overhead(void) {
363
return aral_overhead_from_stats(&mrg_aral_statistics);
364
}
365
365
-void mrg_destroy(MRG *mrg __maybe_unused) {
366
+inline void mrg_destroy(MRG *mrg __maybe_unused) {
367
// no destruction possible
368
// we can't traverse the metrics list
369
@@ -372,43 +373,43 @@ void mrg_destroy(MRG *mrg __maybe_unused) {
373
;
374
}
375
375
-METRIC *mrg_metric_add_and_acquire(MRG *mrg, MRG_ENTRY entry, bool *ret) {
376
+inline METRIC *mrg_metric_add_and_acquire(MRG *mrg, MRG_ENTRY entry, bool *ret) {
377
// internal_fatal(entry.latest_time_s > max_acceptable_collected_time(),
378
// "DBENGINE METRIC: metric latest time is in the future");
379
380
return metric_add_and_acquire(mrg, &entry, ret);
381
}
382
382
-METRIC *mrg_metric_get_and_acquire(MRG *mrg, uuid_t *uuid, Word_t section) {
383
+inline METRIC *mrg_metric_get_and_acquire(MRG *mrg, uuid_t *uuid, Word_t section) {
384
return metric_get_and_acquire(mrg, uuid, section);
385
}
386
386
-bool mrg_metric_release_and_delete(MRG *mrg, METRIC *metric) {
387
+inline bool mrg_metric_release_and_delete(MRG *mrg, METRIC *metric) {
388
return acquired_metric_del(mrg, metric);
389
}
390
390
-METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric) {
391
+inline METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric) {
392
metric_acquire(mrg, metric, false);
393
return metric;
394
}
395
395
-bool mrg_metric_release(MRG *mrg, METRIC *metric) {
396
+inline bool mrg_metric_release(MRG *mrg, METRIC *metric) {
397
return metric_release_and_can_be_deleted(mrg, metric);
398
}
399
399
-Word_t mrg_metric_id(MRG *mrg __maybe_unused, METRIC *metric) {
400
+inline Word_t mrg_metric_id(MRG *mrg __maybe_unused, METRIC *metric) {
401
return (Word_t)metric;
402
}
403
403
-uuid_t *mrg_metric_uuid(MRG *mrg __maybe_unused, METRIC *metric) {
404
+inline uuid_t *mrg_metric_uuid(MRG *mrg __maybe_unused, METRIC *metric) {
405
return &metric->uuid;
406
}
407
407
-Word_t mrg_metric_section(MRG *mrg __maybe_unused, METRIC *metric) {
408
+inline Word_t mrg_metric_section(MRG *mrg __maybe_unused, METRIC *metric) {
409
return metric->section;
410
}
411
411
-bool mrg_metric_set_first_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
412
+inline bool mrg_metric_set_first_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
413
internal_fatal(first_time_s < 0, "DBENGINE METRIC: timestamp is negative");
414
415
if(unlikely(first_time_s < 0))
@@ -422,7 +423,7 @@ bool mrg_metric_set_first_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t
423
return true;
424
}
425
425
-void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s, time_t last_time_s, time_t update_every_s) {
426
+inline void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s, time_t last_time_s, time_t update_every_s) {
427
internal_fatal(first_time_s < 0 || last_time_s < 0 || update_every_s < 0,
428
"DBENGINE METRIC: timestamp is negative");
429
internal_fatal(first_time_s > max_acceptable_collected_time(),
@@ -460,7 +461,7 @@ void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t
461
spinlock_unlock(&metric->spinlock);
462
}
463
463
-bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
464
+inline bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
465
internal_fatal(first_time_s < 0, "DBENGINE METRIC: timestamp is negative");
466
467
bool ret = false;
@@ -476,7 +477,7 @@ bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metr
477
return ret;
478
}
479
479
-time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
480
+inline time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
481
time_t first_time_s;
482
483
spinlock_lock(&metric->spinlock);
@@ -496,7 +497,7 @@ time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
497
return first_time_s;
498
}
499
499
-void mrg_metric_get_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t *first_time_s, time_t *last_time_s, time_t *update_every_s) {
500
+inline void mrg_metric_get_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t *first_time_s, time_t *last_time_s, time_t *update_every_s) {
501
spinlock_lock(&metric->spinlock);
502
503
if(unlikely(!metric->first_time_s)) {
@@ -514,7 +515,7 @@ void mrg_metric_get_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t *f
515
spinlock_unlock(&metric->spinlock);
516
}
517
517
-bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
518
+inline bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
519
internal_fatal(latest_time_s < 0, "DBENGINE METRIC: timestamp is negative");
520
521
if(unlikely(latest_time_s < 0))
@@ -539,7 +540,7 @@ bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric,
540
}
541
542
// returns true when metric still has retention
542
-bool mrg_metric_zero_disk_retention(MRG *mrg __maybe_unused, METRIC *metric) {
543
+inline bool mrg_metric_zero_disk_retention(MRG *mrg __maybe_unused, METRIC *metric) {
544
Word_t section = mrg_metric_section(mrg, metric);
545
bool do_again = false;
546
size_t countdown = 5;
@@ -590,7 +591,7 @@ bool mrg_metric_zero_disk_retention(MRG *mrg __maybe_unused, METRIC *metric) {
591
return ret;
592
}
593
593
-bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
594
+inline bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
595
internal_fatal(latest_time_s < 0, "DBENGINE METRIC: timestamp is negative");
596
597
// internal_fatal(latest_time_s > max_acceptable_collected_time(),
@@ -610,7 +611,7 @@ bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, t
611
return true;
612
}
613
613
-time_t mrg_metric_get_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
614
+inline time_t mrg_metric_get_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
615
time_t max;
616
spinlock_lock(&metric->spinlock);
617
max = MAX(metric->latest_time_s_clean, metric->latest_time_s_hot);
@@ -618,7 +619,7 @@ time_t mrg_metric_get_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
619
return max;
620
}
621
621
-bool mrg_metric_set_update_every(MRG *mrg __maybe_unused, METRIC *metric, time_t update_every_s) {
622
+inline bool mrg_metric_set_update_every(MRG *mrg __maybe_unused, METRIC *metric, time_t update_every_s) {
623
internal_fatal(update_every_s < 0, "DBENGINE METRIC: timestamp is negative");
624
625
if(update_every_s <= 0)
@@ -631,7 +632,7 @@ bool mrg_metric_set_update_every(MRG *mrg __maybe_unused, METRIC *metric, time_t
632
return true;
633
}
634
634
-bool mrg_metric_set_update_every_s_if_zero(MRG *mrg __maybe_unused, METRIC *metric, time_t update_every_s) {
635
+inline bool mrg_metric_set_update_every_s_if_zero(MRG *mrg __maybe_unused, METRIC *metric, time_t update_every_s) {
636
internal_fatal(update_every_s < 0, "DBENGINE METRIC: timestamp is negative");
637
638
if(update_every_s <= 0)
@@ -645,7 +646,7 @@ bool mrg_metric_set_update_every_s_if_zero(MRG *mrg __maybe_unused, METRIC *metr
646
return true;
647
}
648
648
-time_t mrg_metric_get_update_every_s(MRG *mrg __maybe_unused, METRIC *metric) {
649
+inline time_t mrg_metric_get_update_every_s(MRG *mrg __maybe_unused, METRIC *metric) {
650
time_t update_every_s;
651
652
spinlock_lock(&metric->spinlock);
@@ -655,7 +656,7 @@ time_t mrg_metric_get_update_every_s(MRG *mrg __maybe_unused, METRIC *metric) {
656
return update_every_s;
657
}
658
658
-bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) {
659
+inline bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) {
660
bool done = false;
661
spinlock_lock(&metric->spinlock);
662
if(!metric->writer) {
@@ -669,7 +670,7 @@ bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) {
670
return done;
671
}
672
672
-bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) {
673
+inline bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) {
674
bool done = false;
675
spinlock_lock(&metric->spinlock);
676
if(metric->writer) {
@@ -681,7 +682,56 @@ bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) {
682
return done;
683
}
684
684
-void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
685
+inline void mrg_update_metric_retention_and_granularity_by_uuid(
686
+ MRG *mrg, Word_t section, uuid_t *uuid,
687
+ time_t first_time_s, time_t last_time_s,
688
+ time_t update_every_s, time_t now_s)
689
+{
690
+ if(unlikely(last_time_s > now_s)) {
691
+ error_limit_static_global_var(erl, 1, 0);
692
+ error_limit(&erl, "DBENGINE JV2: wrong last time on-disk (%ld - %ld, now %ld), "
693
+ "fixing last time to now",
694
+ first_time_s, last_time_s, now_s);
695
+ last_time_s = now_s;
696
+ }
697
+
698
+ if (unlikely(first_time_s > last_time_s)) {
699
+ error_limit_static_global_var(erl, 1, 0);
700
+ error_limit(&erl, "DBENGINE JV2: wrong first time on-disk (%ld - %ld, now %ld), "
701
+ "fixing first time to last time",
702
+ first_time_s, last_time_s, now_s);
703
+
704
+ first_time_s = last_time_s;
705
+ }
706
+
707
+ if (unlikely(first_time_s == 0 || last_time_s == 0)) {
708
+ error_limit_static_global_var(erl, 1, 0);
709
+ error_limit(&erl, "DBENGINE JV2: zero on-disk timestamps (%ld - %ld, now %ld), "
710
+ "using them as-is",
711
+ first_time_s, last_time_s, now_s);
712
+ }
713
+
714
+ bool added = false;
715
+ METRIC *metric = mrg_metric_get_and_acquire(mrg, uuid, section);
716
+ if (!metric) {
717
+ MRG_ENTRY entry = {
718
+ .section = section,
719
+ .first_time_s = first_time_s,
720
+ .last_time_s = last_time_s,
721
+ .latest_update_every_s = (uint32_t) update_every_s
722
+ };
723
+ // memcpy(entry.uuid, *uuid, sizeof(uuid_t));
724
+ uuid_copy(entry.uuid, *uuid);
725
+ metric = mrg_metric_add_and_acquire(mrg, entry, &added);
726
+ }
727
+
728
+ if (likely(!added))
729
+ mrg_metric_expand_retention(mrg, metric, first_time_s, last_time_s, update_every_s);
730
+
731
+ mrg_metric_release(mrg, metric);
732
+}
733
+
734
+inline void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
735
memset(s, 0, sizeof(struct mrg_statistics));
736
737
for(int i = 0; i < MRG_PARTITIONS ;i++) {
@@ -707,111 +757,55 @@ void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
757
// ----------------------------------------------------------------------------
758
// unit test
759
710
-#ifdef MRG_STRESS_TEST
711
-
712
-static void mrg_stress(MRG *mrg, size_t entries, size_t sections) {
713
- bool ret;
760
+struct mrg_stress_entry {
761
+ uuid_t uuid;
762
+ time_t after;
763
+ time_t before;
764
+};
765
715
- info("DBENGINE METRIC: stress testing %zu entries on %zu sections...", entries, sections);
766
+struct mrg_stress {
767
+ MRG *mrg;
768
+ bool stop;
769
+ size_t entries;
770
+ struct mrg_stress_entry *array;
771
+ size_t updates;
772
+};
773
717
- METRIC *array[entries][sections];
718
- for(size_t i = 0; i < entries ; i++) {
719
- MRG_ENTRY e = {
720
- .first_time_s = (time_t)(i + 1),
721
- .latest_time_s = (time_t)(i + 2),
722
- .latest_update_every_s = (time_t)(i + 3),
723
- };
724
- uuid_generate_random(e.uuid);
774
+static void *mrg_stress(void *ptr) {
775
+ struct mrg_stress *t = ptr;
776
+ MRG *mrg = t->mrg;
777
726
- for(size_t section = 0; section < sections ;section++) {
727
- e.section = section;
728
- array[i][section] = mrg_metric_add_and_acquire(mrg, e, &ret);
729
- if(!ret)
730
- fatal("DBENGINE METRIC: failed to add metric %zu, section %zu", i, section);
778
+ ssize_t start = 0;
779
+ ssize_t end = (ssize_t)t->entries;
780
+ ssize_t step = 1;
781
732
- if(mrg_metric_add_and_acquire(mrg, e, &ret) != array[i][section])
733
- fatal("DBENGINE METRIC: adding the same metric twice, returns a different metric");
782
+ if(gettid() % 2) {
783
+ start = (ssize_t)t->entries - 1;
784
+ end = -1;
785
+ step = -1;
786
+ }
787
735
- if(ret)
736
- fatal("DBENGINE METRIC: adding the same metric twice, returns success");
788
+ while(!__atomic_load_n(&t->stop, __ATOMIC_RELAXED)) {
789
+ for (ssize_t i = start; i != end; i += step) {
790
+ struct mrg_stress_entry *e = &t->array[i];
791
738
- if(mrg_metric_get_and_acquire(mrg, &e.uuid, e.section) != array[i][section])
739
- fatal("DBENGINE METRIC: cannot get back the same metric");
792
+ time_t after = __atomic_sub_fetch(&e->after, 1, __ATOMIC_RELAXED);
793
+ time_t before = __atomic_add_fetch(&e->before, 1, __ATOMIC_RELAXED);
794
741
- if(uuid_compare(*mrg_metric_uuid(mrg, array[i][section]), e.uuid) != 0)
742
- fatal("DBENGINE METRIC: uuids do not match");
743
- }
744
- }
795
+ mrg_update_metric_retention_and_granularity_by_uuid(
796
+ mrg, 0x01,
797
+ &e->uuid,
798
+ after,
799
+ before,
800
+ 1,
801
+ before);
802
746
- for(size_t i = 0; i < entries ; i++) {
747
- for (size_t section = 0; section < sections; section++) {
748
- uuid_t uuid;
749
- uuid_generate_random(uuid);
750
-
751
- if(mrg_metric_get_and_acquire(mrg, &uuid, section))
752
- fatal("DBENGINE METRIC: found non-existing uuid");
753
-
754
- if(mrg_metric_id(mrg, array[i][section]) != (Word_t)array[i][section])
755
- fatal("DBENGINE METRIC: metric id does not match");
756
-
757
- if(mrg_metric_get_first_time_s(mrg, array[i][section]) != (time_t)(i + 1))
758
- fatal("DBENGINE METRIC: wrong first time returned");
759
- if(mrg_metric_get_latest_time_s(mrg, array[i][section]) != (time_t)(i + 2))
760
- fatal("DBENGINE METRIC: wrong latest time returned");
761
- if(mrg_metric_get_update_every_s(mrg, array[i][section]) != (time_t)(i + 3))
762
- fatal("DBENGINE METRIC: wrong latest time returned");
763
-
764
- if(!mrg_metric_set_first_time_s(mrg, array[i][section], (time_t)((i + 1) * 2)))
765
- fatal("DBENGINE METRIC: cannot set first time");
766
- if(!mrg_metric_set_clean_latest_time_s(mrg, array[i][section], (time_t) ((i + 1) * 3)))
767
- fatal("DBENGINE METRIC: cannot set latest time");
768
- if(!mrg_metric_set_update_every(mrg, array[i][section], (time_t)((i + 1) * 4)))
769
- fatal("DBENGINE METRIC: cannot set update every");
770
-
771
- if(mrg_metric_get_first_time_s(mrg, array[i][section]) != (time_t)((i + 1) * 2))
772
- fatal("DBENGINE METRIC: wrong first time returned");
773
- if(mrg_metric_get_latest_time_s(mrg, array[i][section]) != (time_t)((i + 1) * 3))
774
- fatal("DBENGINE METRIC: wrong latest time returned");
775
- if(mrg_metric_get_update_every_s(mrg, array[i][section]) != (time_t)((i + 1) * 4))
776
- fatal("DBENGINE METRIC: wrong latest time returned");
803
+ __atomic_add_fetch(&t->updates, 1, __ATOMIC_RELAXED);
804
}
805
}
806
780
- for(size_t i = 0; i < entries ; i++) {
781
- for (size_t section = 0; section < sections; section++) {
782
- if(!mrg_metric_release_and_delete(mrg, array[i][section]))
783
- fatal("DBENGINE METRIC: failed to delete metric");
784
- }
785
- }
786
-}
787
-
788
-static void *mrg_stress_test_thread1(void *ptr) {
789
- MRG *mrg = ptr;
790
-
791
- for(int i = 0; i < 5 ; i++)
792
- mrg_stress(mrg, 10000, 5);
793
-
794
- return ptr;
795
-}
796
-
797
-static void *mrg_stress_test_thread2(void *ptr) {
798
- MRG *mrg = ptr;
799
-
800
- for(int i = 0; i < 10 ; i++)
801
- mrg_stress(mrg, 500, 50);
802
-
803
- return ptr;
804
-}
805
-
806
-static void *mrg_stress_test_thread3(void *ptr) {
807
- MRG *mrg = ptr;
808
-
809
- for(int i = 0; i < 50 ; i++)
810
- mrg_stress(mrg, 5000, 1);
811
-
807
return ptr;
808
}
814
-#endif
809
810
int mrg_unittest(void) {
811
MRG *mrg = mrg_create();
@@ -894,47 +888,75 @@ int mrg_unittest(void) {
888
if(s.entries != 0)
889
fatal("DBENGINE METRIC: invalid entries counter");
890
897
-#ifdef MRG_STRESS_TEST
898
- usec_t started_ut = now_monotonic_usec();
899
- pthread_t thread1;
900
- netdata_thread_create(&thread1, "TH1",
901
- NETDATA_THREAD_OPTION_JOINABLE | NETDATA_THREAD_OPTION_DONT_LOG,
902
- mrg_stress_test_thread1, mrg);
891
+ size_t entries = 1000000;
892
+ size_t threads = MRG_PARTITIONS / 3 + 1;
893
+ size_t tiers = 3;
894
+ size_t run_for_secs = 5;
895
+ info("preparing stress test of %zu entries...", entries);
896
+ struct mrg_stress t = {
897
+ .mrg = mrg,
898
+ .entries = entries,
899
+ .array = callocz(entries, sizeof(struct mrg_stress_entry)),
900
+ };
901
+
902
+ time_t now = max_acceptable_collected_time();
903
+ for(size_t i = 0; i < entries ;i++) {
904
+ uuid_generate_random(t.array[i].uuid);
905
+ t.array[i].after = now / 3;
906
+ t.array[i].before = now / 2;
907
+ }
908
+ info("stress test is populating MRG with 3 tiers...");
909
+ for(size_t i = 0; i < entries ;i++) {
910
+ struct mrg_stress_entry *e = &t.array[i];
911
+ for(size_t tier = 1; tier <= tiers ;tier++) {
912
+ mrg_update_metric_retention_and_granularity_by_uuid(
913
+ mrg, tier,
914
+ &e->uuid,
915
+ e->after,
916
+ e->before,
917
+ 1,
918
+ e->before);
919
+ }
920
+ }
921
+ info("stress test ready to run...");
922
904
- pthread_t thread2;
905
- netdata_thread_create(&thread2, "TH2",
906
- NETDATA_THREAD_OPTION_JOINABLE | NETDATA_THREAD_OPTION_DONT_LOG,
907
- mrg_stress_test_thread2, mrg);
923
+ usec_t started_ut = now_monotonic_usec();
924
909
- pthread_t thread3;
910
- netdata_thread_create(&thread3, "TH3",
911
- NETDATA_THREAD_OPTION_JOINABLE | NETDATA_THREAD_OPTION_DONT_LOG,
912
- mrg_stress_test_thread3, mrg);
925
+ pthread_t th[threads];
926
+ for(size_t i = 0; i < threads ; i++) {
927
+ char buf[15 + 1];
928
+ snprintfz(buf, 15, "TH[%zu]", i);
929
+ netdata_thread_create(&th[i], buf,
930
+ NETDATA_THREAD_OPTION_JOINABLE | NETDATA_THREAD_OPTION_DONT_LOG,
931
+ mrg_stress, &t);
932
+ }
933
934
+ sleep_usec(run_for_secs * USEC_PER_SEC);
935
+ __atomic_store_n(&t.stop, true, __ATOMIC_RELAXED);
936
915
- sleep_usec(5 * USEC_PER_SEC);
937
+ for(size_t i = 0; i < threads ; i++)
938
+ netdata_thread_cancel(th[i]);
939
917
- netdata_thread_cancel(thread1);
918
- netdata_thread_cancel(thread2);
919
- netdata_thread_cancel(thread3);
940
+ for(size_t i = 0; i < threads ; i++)
941
+ netdata_thread_join(th[i], NULL);
942
921
- netdata_thread_join(thread1, NULL);
922
- netdata_thread_join(thread2, NULL);
923
- netdata_thread_join(thread3, NULL);
943
usec_t ended_ut = now_monotonic_usec();
944
945
+ struct mrg_statistics stats;
946
+ mrg_get_statistics(mrg, &stats);
947
+
948
info("DBENGINE METRIC: did %zu additions, %zu duplicate additions, "
949
"%zu deletions, %zu wrong deletions, "
950
"%zu successful searches, %zu wrong searches, "
929
- "%zu successful pointer validations, %zu wrong pointer validations "
951
"in %llu usecs",
931
- mrg->stats.additions, mrg->stats.additions_duplicate,
932
- mrg->stats.deletions, mrg->stats.delete_misses,
933
- mrg->stats.search_hits, mrg->stats.search_misses,
934
- mrg->stats.pointer_validation_hits, mrg->stats.pointer_validation_misses,
952
+ stats.additions, stats.additions_duplicate,
953
+ stats.deletions, stats.delete_misses,
954
+ stats.search_hits, stats.search_misses,
955
ended_ut - started_ut);
956
937
-#endif
957
+ info("DBENGINE METRIC: updates performance: %0.2fk/sec total, %0.2fk/sec/thread",
958
+ (double)t.updates / (double)((ended_ut - started_ut) / USEC_PER_SEC) / 1000.0,
959
+ (double)t.updates / (double)((ended_ut - started_ut) / USEC_PER_SEC) / 1000.0 / threads);
960
961
mrg_destroy(mrg);
962
database/engine/metric.h
+6
@@ -87,4 +87,10 @@ void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s);
87
size_t mrg_aral_structures(void);
88
size_t mrg_aral_overhead(void);
89
90
+
91
+void mrg_update_metric_retention_and_granularity_by_uuid(
92
+ MRG *mrg, Word_t section, uuid_t *uuid,
93
+ time_t first_time_s, time_t last_time_s,
94
+ time_t update_every_s, time_t now_s);
95
+
96
#endif // DBENGINE_METRIC_H