Avoid duplicate keys in labels (#16014)
* Avoid duplicate keys in labels * Properly delete old label * Simplify memory statistics logging
Stelios Fragkakis committed
Oct 2, 2023 at 18:07 UTC
d34dbf844fdf26ddbf4dedfa08d6a1d08974e499
1 file changed
+80
-27
database/rrdlabels.c
+80
-27
@@ -671,6 +671,25 @@ void rrdlabels_destroy(RRDLABELS *labels)
671
freez(labels);
672
}
673
674
+static RRDLABEL *rrdlabels_find_label_with_key_unsafe(RRDLABELS *labels, RRDLABEL *label)
675
+{
676
+ if (unlikely(!labels))
677
+ return NULL;
678
+
679
+ Pvoid_t *PValue;
680
+ Word_t Index = 0;
681
+ bool first_then_next = true;
682
+ RRDLABEL *found = NULL;
683
+ while ((PValue = JudyLFirstThenNext(labels->JudyL, &Index, &first_then_next))) {
684
+ RRDLABEL *lb = (RRDLABEL *)Index;
685
+ if (lb->index.key == label->index.key && lb->index.value != label->index.value) {
686
+ found = (RRDLABEL *)Index;
687
+ break;
688
+ }
689
+ }
690
+ return found;
691
+}
692
+
693
// ----------------------------------------------------------------------------
694
// rrdlabels_add()
695
@@ -680,6 +699,8 @@ static void labels_add_already_sanitized(RRDLABELS *labels, const char *key, con
699
700
spinlock_lock(&labels->spinlock);
701
702
+ RRDLABEL *old_key = rrdlabels_find_label_with_key_unsafe(labels, label);
703
+
704
size_t mem_before_judyl = JudyLMemUsed(labels->JudyL);
705
706
Pvoid_t *PValue = JudyLIns(&labels->JudyL, (Word_t) label, PJE0);
@@ -687,8 +708,19 @@ static void labels_add_already_sanitized(RRDLABELS *labels, const char *key, con
708
fatal("RRDLABELS: corrupted labels JudyL array");
709
710
if (!*PValue) {
690
- *((RRDLABEL_SRC *)PValue) = (ls & ~(RRDLABEL_FLAG_NEW | RRDLABEL_FLAG_OLD));
711
+ RRDLABEL_SRC new_ls;
712
+ if (old_key)
713
+ new_ls = ((ls & ~(RRDLABEL_FLAG_NEW | RRDLABEL_FLAG_OLD)) | RRDLABEL_FLAG_OLD);
714
+ else
715
+ new_ls = ((ls & ~(RRDLABEL_FLAG_NEW | RRDLABEL_FLAG_OLD)) | RRDLABEL_FLAG_NEW);
716
+ *((RRDLABEL_SRC *)PValue) = new_ls;
717
+
718
labels->version++;
719
+
720
+ if (old_key) {
721
+ (void)JudyLDel(&labels->JudyL, (Word_t) old_key, PJE0);
722
+ delete_label((RRDLABEL *)old_key);
723
+ }
724
size_t mem_after_judyl = JudyLMemUsed(labels->JudyL);
725
STATS_PLUS_MEMORY(&dictionary_stats_category_rrdlabels, 0, mem_after_judyl - mem_before_judyl, 0);
726
}
@@ -876,12 +908,54 @@ void rrdlabels_get_value_to_buffer_or_unset(RRDLABELS *labels, BUFFER *wb, const
908
string_freez(this_key);
909
}
910
879
-void rrdlabels_unmark_all(RRDLABELS *labels __maybe_unused) {
911
+static void rrdlabels_unmark_all_unsafe(RRDLABELS *labels)
912
+{
913
+ Pvoid_t *PValue;
914
+ Word_t Index = 0;
915
+ bool first_then_next = true;
916
+ while ((PValue = JudyLFirstThenNext(labels->JudyL, &Index, &first_then_next)))
917
+ *((RRDLABEL_SRC *)PValue) &= ~(RRDLABEL_FLAG_OLD | RRDLABEL_FLAG_NEW);
918
}
919
882
-void rrdlabels_remove_all_unmarked(RRDLABELS *labels __maybe_unused) {
920
+void rrdlabels_unmark_all(RRDLABELS *labels)
921
+{
922
+ spinlock_lock(&labels->spinlock);
923
+
924
+ rrdlabels_unmark_all_unsafe(labels);
925
+
926
+ spinlock_unlock(&labels->spinlock);
927
}
928
929
+static void rrdlabels_remove_all_unmarked_unsafe(RRDLABELS *labels)
930
+{
931
+ Pvoid_t *PValue;
932
+ Word_t Index = 0;
933
+ bool first_then_next = true;
934
+
935
+ while ((PValue = JudyLFirstThenNext(labels->JudyL, &Index, &first_then_next))) {
936
+ if (!((*((RRDLABEL_SRC *)PValue)) & (RRDLABEL_FLAG_OLD | RRDLABEL_FLAG_NEW | RRDLABEL_FLAG_PERMANENT))) {
937
+
938
+ size_t mem_before_judyl = JudyLMemUsed(labels->JudyL);
939
+ (void)JudyLDel(&labels->JudyL, Index, PJE0);
940
+ size_t mem_after_judyl = JudyLMemUsed(labels->JudyL);
941
+
942
+ STATS_MINUS_MEMORY(&dictionary_stats_category_rrdlabels, 0, mem_before_judyl - mem_after_judyl, 0);
943
+
944
+ delete_label((RRDLABEL *)Index);
945
+ if (labels->JudyL != (Pvoid_t) NULL) {
946
+ Index = 0;
947
+ first_then_next = true;
948
+ }
949
+ }
950
+ }
951
+}
952
+
953
+void rrdlabels_remove_all_unmarked(RRDLABELS *labels)
954
+{
955
+ spinlock_lock(&labels->spinlock);
956
+ rrdlabels_remove_all_unmarked_unsafe(labels);
957
+ spinlock_unlock(&labels->spinlock);
958
+}
959
960
// ----------------------------------------------------------------------------
961
// rrdlabels_walkthrough_read()
@@ -916,13 +990,10 @@ void rrdlabels_migrate_to_these(RRDLABELS *dst, RRDLABELS *src) {
990
spinlock_lock(&dst->spinlock);
991
spinlock_lock(&src->spinlock);
992
993
+ rrdlabels_unmark_all_unsafe(dst);
994
+
995
RRDLABEL *label;
996
Pvoid_t *PValue;
921
- Word_t Index = 0;
922
- bool first_then_next = true;
923
- while ((PValue = JudyLFirstThenNext(dst->JudyL, &Index, &first_then_next))) {
924
- *((RRDLABEL_SRC *)PValue) &= ~(RRDLABEL_FLAG_OLD | RRDLABEL_FLAG_NEW);
925
- }
997
998
RRDLABEL_SRC ls;
999
lfe_start_nolock(src, label, ls)
@@ -945,25 +1016,7 @@ void rrdlabels_migrate_to_these(RRDLABELS *dst, RRDLABELS *src) {
1016
}
1017
lfe_done_nolock();
1018
948
- Index = 0;
949
- first_then_next = true;
950
- while ((PValue = JudyLFirstThenNext(dst->JudyL, &Index, &first_then_next))) {
951
- if (!((*((RRDLABEL_SRC *)PValue)) & (RRDLABEL_FLAG_OLD | RRDLABEL_FLAG_NEW | RRDLABEL_FLAG_PERMANENT))) {
952
-
953
- size_t mem_before_judyl = JudyLMemUsed(dst->JudyL);
954
- (void)JudyLDel(&dst->JudyL, Index, PJE0);
955
- size_t mem_after_judyl = JudyLMemUsed(dst->JudyL);
956
-
957
- STATS_MINUS_MEMORY(&dictionary_stats_category_rrdlabels, 0, mem_before_judyl - mem_after_judyl, 0);
958
-
959
- delete_label((RRDLABEL *)Index);
960
- if (dst->JudyL != (Pvoid_t) NULL) {
961
- Index = 0;
962
- first_then_next = true;
963
- }
964
- }
965
- }
966
-
1019
+ rrdlabels_remove_all_unmarked_unsafe(dst);
1020
dst->version = src->version;
1021
1022
spinlock_unlock(&src->spinlock);