16
time_t latest_time_s_hot; // latest time of the currently collected page
17
uint32_t latest_update_every_s; //
18
pid_t writer;
19
+ uint8_t partition;
20
METRIC_FLAGS flags;
21
REFCOUNT refcount;
22
SPINLOCK spinlock; // protects all variable members
31
ARAL *aral[MRG_PARTITIONS];
32
33
struct pgc_index {
33
- netdata_rwlock_t rwlock;
34
+ MRG_CACHE_LINE_PADDING(0);
35
+
36
+ RW_SPINLOCK rw_spinlock;
37
+
38
+ MRG_CACHE_LINE_PADDING(1);
39
+
40
Pvoid_t uuid_judy; // each UUID has a JudyL of sections (tiers)
35
- } index[MRG_PARTITIONS];
41
37
- struct mrg_statistics stats;
42
+ MRG_CACHE_LINE_PADDING(2);
43
+
44
+ struct mrg_statistics stats;
45
+
46
+ MRG_CACHE_LINE_PADDING(3);
47
+ } index[MRG_PARTITIONS];
48
49
+#ifdef NETDATA_INTERNAL_CHECKS
50
size_t entries_per_partition[MRG_PARTITIONS];
51
+#endif
52
};
53
42
-static inline void MRG_STATS_DUPLICATE_ADD(MRG *mrg) {
43
- __atomic_add_fetch(&mrg->stats.additions_duplicate, 1, __ATOMIC_RELAXED);
54
+static inline void MRG_STATS_DUPLICATE_ADD(MRG *mrg, size_t partition) {
55
+ mrg->index[partition].stats.additions_duplicate++;
56
}
57
58
static inline void MRG_STATS_ADDED_METRIC(MRG *mrg, size_t partition) {
47
- __atomic_add_fetch(&mrg->stats.entries, 1, __ATOMIC_RELAXED);
48
- __atomic_add_fetch(&mrg->stats.additions, 1, __ATOMIC_RELAXED);
49
- __atomic_add_fetch(&mrg->stats.size, sizeof(METRIC), __ATOMIC_RELAXED);
59
+ mrg->index[partition].stats.entries++;
60
+ mrg->index[partition].stats.additions++;
61
+ mrg->index[partition].stats.size += sizeof(METRIC);
62
63
+#ifdef NETDATA_INTERNAL_CHECKS
64
__atomic_add_fetch(&mrg->entries_per_partition[partition], 1, __ATOMIC_RELAXED);
65
+#endif
66
}
67
68
static inline void MRG_STATS_DELETED_METRIC(MRG *mrg, size_t partition) {
55
- __atomic_sub_fetch(&mrg->stats.entries, 1, __ATOMIC_RELAXED);
56
- __atomic_sub_fetch(&mrg->stats.size, sizeof(METRIC), __ATOMIC_RELAXED);
57
- __atomic_add_fetch(&mrg->stats.deletions, 1, __ATOMIC_RELAXED);
69
+ mrg->index[partition].stats.entries--;
70
+ mrg->index[partition].stats.size -= sizeof(METRIC);
71
+ mrg->index[partition].stats.deletions++;
72
73
+#ifdef NETDATA_INTERNAL_CHECKS
74
__atomic_sub_fetch(&mrg->entries_per_partition[partition], 1, __ATOMIC_RELAXED);
75
+#endif
76
}
77
62
-static inline void MRG_STATS_SEARCH_HIT(MRG *mrg) {
63
- __atomic_add_fetch(&mrg->stats.search_hits, 1, __ATOMIC_RELAXED);
78
+static inline void MRG_STATS_SEARCH_HIT(MRG *mrg, size_t partition) {
79
+ __atomic_add_fetch(&mrg->index[partition].stats.search_hits, 1, __ATOMIC_RELAXED);
80
}
81
66
-static inline void MRG_STATS_SEARCH_MISS(MRG *mrg) {
67
- __atomic_add_fetch(&mrg->stats.search_misses, 1, __ATOMIC_RELAXED);
82
+static inline void MRG_STATS_SEARCH_MISS(MRG *mrg, size_t partition) {
83
+ __atomic_add_fetch(&mrg->index[partition].stats.search_misses, 1, __ATOMIC_RELAXED);
84
}
85
70
-static inline void MRG_STATS_DELETE_MISS(MRG *mrg) {
71
- __atomic_add_fetch(&mrg->stats.delete_misses, 1, __ATOMIC_RELAXED);
86
+static inline void MRG_STATS_DELETE_MISS(MRG *mrg, size_t partition) {
87
+ mrg->index[partition].stats.delete_misses++;
88
}
89
90
static inline void mrg_index_read_lock(MRG *mrg, size_t partition) {
75
- netdata_rwlock_rdlock(&mrg->index[partition].rwlock);
91
+ rw_spinlock_read_lock(&mrg->index[partition].rw_spinlock);
92
}
93
static inline void mrg_index_read_unlock(MRG *mrg, size_t partition) {
78
- netdata_rwlock_unlock(&mrg->index[partition].rwlock);
94
+ rw_spinlock_read_unlock(&mrg->index[partition].rw_spinlock);
95
}
96
static inline void mrg_index_write_lock(MRG *mrg, size_t partition) {
81
- netdata_rwlock_wrlock(&mrg->index[partition].rwlock);
97
+ rw_spinlock_write_lock(&mrg->index[partition].rw_spinlock);
98
}
99
static inline void mrg_index_write_unlock(MRG *mrg, size_t partition) {
84
- netdata_rwlock_unlock(&mrg->index[partition].rwlock);
100
+ rw_spinlock_write_unlock(&mrg->index[partition].rw_spinlock);
101
}
102
87
-static inline void mrg_stats_size_judyl_change(MRG *mrg, size_t mem_before_judyl, size_t mem_after_judyl) {
103
+static inline void mrg_stats_size_judyl_change(MRG *mrg, size_t mem_before_judyl, size_t mem_after_judyl, size_t partition) {
104
if(mem_after_judyl > mem_before_judyl)
89
- __atomic_add_fetch(&mrg->stats.size, mem_after_judyl - mem_before_judyl, __ATOMIC_RELAXED);
105
+ __atomic_add_fetch(&mrg->index[partition].stats.size, mem_after_judyl - mem_before_judyl, __ATOMIC_RELAXED);
106
else if(mem_after_judyl < mem_before_judyl)
91
- __atomic_sub_fetch(&mrg->stats.size, mem_before_judyl - mem_after_judyl, __ATOMIC_RELAXED);
107
+ __atomic_sub_fetch(&mrg->index[partition].stats.size, mem_before_judyl - mem_after_judyl, __ATOMIC_RELAXED);
108
}
109
94
-static inline void mrg_stats_size_judyhs_added_uuid(MRG *mrg) {
95
- __atomic_add_fetch(&mrg->stats.size, JUDYHS_INDEX_SIZE_ESTIMATE(sizeof(uuid_t)), __ATOMIC_RELAXED);
110
+static inline void mrg_stats_size_judyhs_added_uuid(MRG *mrg, size_t partition) {
111
+ __atomic_add_fetch(&mrg->index[partition].stats.size, JUDYHS_INDEX_SIZE_ESTIMATE(sizeof(uuid_t)), __ATOMIC_RELAXED);
112
}
113
98
-static inline void mrg_stats_size_judyhs_removed_uuid(MRG *mrg) {
99
- __atomic_sub_fetch(&mrg->stats.size, JUDYHS_INDEX_SIZE_ESTIMATE(sizeof(uuid_t)), __ATOMIC_RELAXED);
114
+static inline void mrg_stats_size_judyhs_removed_uuid(MRG *mrg, size_t partition) {
115
+ __atomic_sub_fetch(&mrg->index[partition].stats.size, JUDYHS_INDEX_SIZE_ESTIMATE(sizeof(uuid_t)), __ATOMIC_RELAXED);
116
}
117
118
static inline size_t uuid_partition(MRG *mrg __maybe_unused, uuid_t *uuid) {
121
}
122
123
static inline bool metric_has_retention_unsafe(MRG *mrg __maybe_unused, METRIC *metric) {
124
+ size_t partition = metric->partition;
125
+
126
bool has_retention = (metric->first_time_s > 0 || metric->latest_time_s_clean > 0 || metric->latest_time_s_hot > 0);
127
128
if(has_retention && !(metric->flags & METRIC_FLAG_HAS_RETENTION)) {
129
metric->flags |= METRIC_FLAG_HAS_RETENTION;
112
- __atomic_add_fetch(&mrg->stats.entries_with_retention, 1, __ATOMIC_RELAXED);
130
+ __atomic_add_fetch(&mrg->index[partition].stats.entries_with_retention, 1, __ATOMIC_RELAXED);
131
}
132
else if(!has_retention && (metric->flags & METRIC_FLAG_HAS_RETENTION)) {
133
metric->flags &= ~METRIC_FLAG_HAS_RETENTION;
116
- __atomic_sub_fetch(&mrg->stats.entries_with_retention, 1, __ATOMIC_RELAXED);
134
+ __atomic_sub_fetch(&mrg->index[partition].stats.entries_with_retention, 1, __ATOMIC_RELAXED);
135
}
136
137
return has_retention;
138
}
139
140
static inline REFCOUNT metric_acquire(MRG *mrg __maybe_unused, METRIC *metric, bool having_spinlock) {
141
+ size_t partition = metric->partition;
142
REFCOUNT refcount;
143
144
if(!having_spinlock)
126
- netdata_spinlock_lock(&metric->spinlock);
145
+ spinlock_lock(&metric->spinlock);
146
147
if(unlikely(metric->refcount < 0))
148
fatal("METRIC: refcount is %d (negative) during acquire", metric->refcount);
153
metric_has_retention_unsafe(mrg, metric);
154
155
if(!having_spinlock)
137
- netdata_spinlock_unlock(&metric->spinlock);
156
+ spinlock_unlock(&metric->spinlock);
157
158
if(refcount == 1)
140
- __atomic_add_fetch(&mrg->stats.entries_referenced, 1, __ATOMIC_RELAXED);
159
+ __atomic_add_fetch(&mrg->index[partition].stats.entries_referenced, 1, __ATOMIC_RELAXED);
160
142
- __atomic_add_fetch(&mrg->stats.current_references, 1, __ATOMIC_RELAXED);
161
+ __atomic_add_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
162
163
return refcount;
164
}
165
166
static inline bool metric_release_and_can_be_deleted(MRG *mrg __maybe_unused, METRIC *metric) {
167
bool ret = true;
168
+ size_t partition = metric->partition;
169
REFCOUNT refcount;
170
151
- netdata_spinlock_lock(&metric->spinlock);
171
+ spinlock_lock(&metric->spinlock);
172
173
if(unlikely(metric->refcount <= 0))
174
fatal("METRIC: refcount is %d (zero or negative) during release", metric->refcount);
178
if(likely(metric_has_retention_unsafe(mrg, metric) || refcount != 0))
179
ret = false;
180
161
- netdata_spinlock_unlock(&metric->spinlock);
181
+ spinlock_unlock(&metric->spinlock);
182
183
if(unlikely(!refcount))
164
- __atomic_sub_fetch(&mrg->stats.entries_referenced, 1, __ATOMIC_RELAXED);
184
+ __atomic_sub_fetch(&mrg->index[partition].stats.entries_referenced, 1, __ATOMIC_RELAXED);
185
166
- __atomic_sub_fetch(&mrg->stats.current_references, 1, __ATOMIC_RELAXED);
186
+ __atomic_sub_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
187
188
return ret;
189
}
202
fatal("DBENGINE METRIC: corrupted UUIDs JudyHS array");
203
204
if(unlikely(!*sections_judy_pptr))
185
- mrg_stats_size_judyhs_added_uuid(mrg);
205
+ mrg_stats_size_judyhs_added_uuid(mrg, partition);
206
207
mem_before_judyl = JudyLMemUsed(*sections_judy_pptr);
208
Pvoid_t *PValue = JudyLIns(sections_judy_pptr, entry->section, PJE0);
209
mem_after_judyl = JudyLMemUsed(*sections_judy_pptr);
190
- mrg_stats_size_judyl_change(mrg, mem_before_judyl, mem_after_judyl);
210
+ mrg_stats_size_judyl_change(mrg, mem_before_judyl, mem_after_judyl, partition);
211
212
if(unlikely(!PValue || PValue == PJERR))
213
fatal("DBENGINE METRIC: corrupted section JudyL array");
216
METRIC *metric = *PValue;
217
218
metric_acquire(mrg, metric, false);
219
+
220
+ MRG_STATS_DUPLICATE_ADD(mrg, partition);
221
+
222
mrg_index_write_unlock(mrg, partition);
223
224
if(ret)
226
227
aral_freez(mrg->aral[partition], allocation);
228
206
- MRG_STATS_DUPLICATE_ADD(mrg);
229
return metric;
230
}
231
239
metric->writer = 0;
240
metric->refcount = 0;
241
metric->flags = 0;
220
- netdata_spinlock_init(&metric->spinlock);
242
+ metric->partition = partition;
243
+ spinlock_init(&metric->spinlock);
244
metric_acquire(mrg, metric, true); // no spinlock use required here
245
*PValue = metric;
246
247
+ MRG_STATS_ADDED_METRIC(mrg, partition);
248
+
249
mrg_index_write_unlock(mrg, partition);
250
251
if(ret)
252
*ret = true;
253
229
- MRG_STATS_ADDED_METRIC(mrg, partition);
230
-
254
return metric;
255
}
256
262
Pvoid_t *sections_judy_pptr = JudyHSGet(mrg->index[partition].uuid_judy, uuid, sizeof(uuid_t));
263
if(unlikely(!sections_judy_pptr)) {
264
mrg_index_read_unlock(mrg, partition);
242
- MRG_STATS_SEARCH_MISS(mrg);
265
+ MRG_STATS_SEARCH_MISS(mrg, partition);
266
return NULL;
267
}
268
269
Pvoid_t *PValue = JudyLGet(*sections_judy_pptr, section, PJE0);
270
if(unlikely(!PValue)) {
271
mrg_index_read_unlock(mrg, partition);
249
- MRG_STATS_SEARCH_MISS(mrg);
272
+ MRG_STATS_SEARCH_MISS(mrg, partition);
273
return NULL;
274
}
275
279
280
mrg_index_read_unlock(mrg, partition);
281
259
- MRG_STATS_SEARCH_HIT(mrg);
282
+ MRG_STATS_SEARCH_HIT(mrg, partition);
283
return metric;
284
}
285
291
mrg_index_write_lock(mrg, partition);
292
293
if(!metric_release_and_can_be_deleted(mrg, metric)) {
294
+ mrg->index[partition].stats.delete_having_retention_or_referenced++;
295
mrg_index_write_unlock(mrg, partition);
272
- __atomic_add_fetch(&mrg->stats.delete_having_retention_or_referenced, 1, __ATOMIC_RELAXED);
296
return false;
297
}
298
299
Pvoid_t *sections_judy_pptr = JudyHSGet(mrg->index[partition].uuid_judy, &metric->uuid, sizeof(uuid_t));
300
if(unlikely(!sections_judy_pptr || !*sections_judy_pptr)) {
301
+ MRG_STATS_DELETE_MISS(mrg, partition);
302
mrg_index_write_unlock(mrg, partition);
279
- MRG_STATS_DELETE_MISS(mrg);
303
return false;
304
}
305
306
mem_before_judyl = JudyLMemUsed(*sections_judy_pptr);
307
int rc = JudyLDel(sections_judy_pptr, metric->section, PJE0);
308
mem_after_judyl = JudyLMemUsed(*sections_judy_pptr);
286
- mrg_stats_size_judyl_change(mrg, mem_before_judyl, mem_after_judyl);
309
+ mrg_stats_size_judyl_change(mrg, mem_before_judyl, mem_after_judyl, partition);
310
311
if(unlikely(!rc)) {
312
+ MRG_STATS_DELETE_MISS(mrg, partition);
313
mrg_index_write_unlock(mrg, partition);
290
- MRG_STATS_DELETE_MISS(mrg);
314
return false;
315
}
316
318
rc = JudyHSDel(&mrg->index[partition].uuid_judy, &metric->uuid, sizeof(uuid_t), PJE0);
319
if(unlikely(!rc))
320
fatal("DBENGINE METRIC: cannot delete UUID from JudyHS");
298
- mrg_stats_size_judyhs_removed_uuid(mrg);
321
+ mrg_stats_size_judyhs_removed_uuid(mrg, partition);
322
}
323
324
+ MRG_STATS_DELETED_METRIC(mrg, partition);
325
+
326
mrg_index_write_unlock(mrg, partition);
327
328
aral_freez(mrg->aral[partition], metric);
329
305
- MRG_STATS_DELETED_METRIC(mrg, partition);
306
-
330
return true;
331
}
332
337
MRG *mrg = callocz(1, sizeof(MRG));
338
339
for(size_t i = 0; i < MRG_PARTITIONS ; i++) {
317
- netdata_rwlock_init(&mrg->index[i].rwlock);
340
+ rw_spinlock_init(&mrg->index[i].rw_spinlock);
341
342
char buf[ARAL_MAX_NAME + 1];
343
snprintfz(buf, ARAL_MAX_NAME, "mrg[%zu]", i);
351
false);
352
}
353
331
- mrg->stats.size = sizeof(MRG);
332
-
354
return mrg;
355
}
356
414
if(unlikely(first_time_s < 0))
415
return false;
416
396
- netdata_spinlock_lock(&metric->spinlock);
417
+ spinlock_lock(&metric->spinlock);
418
metric->first_time_s = first_time_s;
419
metric_has_retention_unsafe(mrg, metric);
399
- netdata_spinlock_unlock(&metric->spinlock);
420
+ spinlock_unlock(&metric->spinlock);
421
422
return true;
423
}
442
if(unlikely(!first_time_s && !last_time_s && !update_every_s))
443
return;
444
424
- netdata_spinlock_lock(&metric->spinlock);
445
+ spinlock_lock(&metric->spinlock);
446
447
if(unlikely(first_time_s && (!metric->first_time_s || first_time_s < metric->first_time_s)))
448
metric->first_time_s = first_time_s;
457
metric->latest_update_every_s = (uint32_t) update_every_s;
458
459
metric_has_retention_unsafe(mrg, metric);
439
- netdata_spinlock_unlock(&metric->spinlock);
460
+ spinlock_unlock(&metric->spinlock);
461
}
462
463
bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
465
466
bool ret = false;
467
447
- netdata_spinlock_lock(&metric->spinlock);
468
+ spinlock_lock(&metric->spinlock);
469
if(first_time_s > metric->first_time_s) {
470
metric->first_time_s = first_time_s;
471
ret = true;
472
}
473
metric_has_retention_unsafe(mrg, metric);
453
- netdata_spinlock_unlock(&metric->spinlock);
474
+ spinlock_unlock(&metric->spinlock);
475
476
return ret;
477
}
479
time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
480
time_t first_time_s;
481
461
- netdata_spinlock_lock(&metric->spinlock);
482
+ spinlock_lock(&metric->spinlock);
483
484
if(unlikely(!metric->first_time_s)) {
485
if(metric->latest_time_s_clean)
491
492
first_time_s = metric->first_time_s;
493
473
- netdata_spinlock_unlock(&metric->spinlock);
494
+ spinlock_unlock(&metric->spinlock);
495
496
return first_time_s;
497
}
498
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) {
479
- netdata_spinlock_lock(&metric->spinlock);
500
+ spinlock_lock(&metric->spinlock);
501
502
if(unlikely(!metric->first_time_s)) {
503
if(metric->latest_time_s_clean)
511
*last_time_s = MAX(metric->latest_time_s_clean, metric->latest_time_s_hot);
512
*update_every_s = metric->latest_update_every_s;
513
493
- netdata_spinlock_unlock(&metric->spinlock);
514
+ spinlock_unlock(&metric->spinlock);
515
}
516
517
bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
520
if(unlikely(latest_time_s < 0))
521
return false;
522
502
- netdata_spinlock_lock(&metric->spinlock);
523
+ spinlock_lock(&metric->spinlock);
524
525
// internal_fatal(latest_time_s > max_acceptable_collected_time(),
526
// "DBENGINE METRIC: metric latest time is in the future");
534
metric->first_time_s = latest_time_s;
535
536
metric_has_retention_unsafe(mrg, metric);
516
- netdata_spinlock_unlock(&metric->spinlock);
537
+ spinlock_unlock(&metric->spinlock);
538
return true;
539
}
540
572
if (min_first_time_s == LONG_MAX)
573
min_first_time_s = 0;
574
554
- netdata_spinlock_lock(&metric->spinlock);
575
+ spinlock_lock(&metric->spinlock);
576
if (--countdown && !min_first_time_s && metric->latest_time_s_hot)
577
do_again = true;
578
else {
584
585
ret = metric_has_retention_unsafe(mrg, metric);
586
}
566
- netdata_spinlock_unlock(&metric->spinlock);
587
+ spinlock_unlock(&metric->spinlock);
588
} while(do_again);
589
590
return ret;
599
if(unlikely(latest_time_s < 0))
600
return false;
601
581
- netdata_spinlock_lock(&metric->spinlock);
602
+ spinlock_lock(&metric->spinlock);
603
metric->latest_time_s_hot = latest_time_s;
604
605
if(unlikely(!metric->first_time_s))
606
metric->first_time_s = latest_time_s;
607
608
metric_has_retention_unsafe(mrg, metric);
588
- netdata_spinlock_unlock(&metric->spinlock);
609
+ spinlock_unlock(&metric->spinlock);
610
return true;
611
}
612
613
time_t mrg_metric_get_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
614
time_t max;
594
- netdata_spinlock_lock(&metric->spinlock);
615
+ spinlock_lock(&metric->spinlock);
616
max = MAX(metric->latest_time_s_clean, metric->latest_time_s_hot);
596
- netdata_spinlock_unlock(&metric->spinlock);
617
+ spinlock_unlock(&metric->spinlock);
618
return max;
619
}
620
624
if(update_every_s <= 0)
625
return false;
626
606
- netdata_spinlock_lock(&metric->spinlock);
627
+ spinlock_lock(&metric->spinlock);
628
metric->latest_update_every_s = (uint32_t) update_every_s;
608
- netdata_spinlock_unlock(&metric->spinlock);
629
+ spinlock_unlock(&metric->spinlock);
630
631
return true;
632
}
637
if(update_every_s <= 0)
638
return false;
639
619
- netdata_spinlock_lock(&metric->spinlock);
640
+ spinlock_lock(&metric->spinlock);
641
if(!metric->latest_update_every_s)
642
metric->latest_update_every_s = (uint32_t) update_every_s;
622
- netdata_spinlock_unlock(&metric->spinlock);
643
+ spinlock_unlock(&metric->spinlock);
644
645
return true;
646
}
648
time_t mrg_metric_get_update_every_s(MRG *mrg __maybe_unused, METRIC *metric) {
649
time_t update_every_s;
650
630
- netdata_spinlock_lock(&metric->spinlock);
651
+ spinlock_lock(&metric->spinlock);
652
update_every_s = metric->latest_update_every_s;
632
- netdata_spinlock_unlock(&metric->spinlock);
653
+ spinlock_unlock(&metric->spinlock);
654
655
return update_every_s;
656
}
657
658
bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) {
659
bool done = false;
639
- netdata_spinlock_lock(&metric->spinlock);
660
+ spinlock_lock(&metric->spinlock);
661
if(!metric->writer) {
662
metric->writer = gettid();
642
- __atomic_add_fetch(&mrg->stats.writers, 1, __ATOMIC_RELAXED);
663
+ __atomic_add_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
664
done = true;
665
}
666
else
646
- __atomic_add_fetch(&mrg->stats.writers_conflicts, 1, __ATOMIC_RELAXED);
647
- netdata_spinlock_unlock(&metric->spinlock);
667
+ __atomic_add_fetch(&mrg->index[metric->partition].stats.writers_conflicts, 1, __ATOMIC_RELAXED);
668
+ spinlock_unlock(&metric->spinlock);
669
return done;
670
}
671
672
bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) {
673
bool done = false;
653
- netdata_spinlock_lock(&metric->spinlock);
674
+ spinlock_lock(&metric->spinlock);
675
if(metric->writer) {
676
metric->writer = 0;
656
- __atomic_sub_fetch(&mrg->stats.writers, 1, __ATOMIC_RELAXED);
677
+ __atomic_sub_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
678
done = true;
679
}
659
- netdata_spinlock_unlock(&metric->spinlock);
680
+ spinlock_unlock(&metric->spinlock);
681
return done;
682
}
683
663
-struct mrg_statistics mrg_get_statistics(MRG *mrg) {
664
- // FIXME - use atomics
665
- return mrg->stats;
684
+void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
685
+ memset(s, 0, sizeof(struct mrg_statistics));
686
+
687
+ for(int i = 0; i < MRG_PARTITIONS ;i++) {
688
+ s->entries += __atomic_load_n(&mrg->index[i].stats.entries, __ATOMIC_RELAXED);
689
+ s->entries_referenced += __atomic_load_n(&mrg->index[i].stats.entries_referenced, __ATOMIC_RELAXED);
690
+ s->entries_with_retention += __atomic_load_n(&mrg->index[i].stats.entries_with_retention, __ATOMIC_RELAXED);
691
+ s->size += __atomic_load_n(&mrg->index[i].stats.size, __ATOMIC_RELAXED);
692
+ s->current_references += __atomic_load_n(&mrg->index[i].stats.current_references, __ATOMIC_RELAXED);
693
+ s->additions += __atomic_load_n(&mrg->index[i].stats.additions, __ATOMIC_RELAXED);
694
+ s->additions_duplicate += __atomic_load_n(&mrg->index[i].stats.additions_duplicate, __ATOMIC_RELAXED);
695
+ s->deletions += __atomic_load_n(&mrg->index[i].stats.deletions, __ATOMIC_RELAXED);
696
+ s->delete_having_retention_or_referenced += __atomic_load_n(&mrg->index[i].stats.delete_having_retention_or_referenced, __ATOMIC_RELAXED);
697
+ s->delete_misses += __atomic_load_n(&mrg->index[i].stats.delete_misses, __ATOMIC_RELAXED);
698
+ s->search_hits += __atomic_load_n(&mrg->index[i].stats.search_hits, __ATOMIC_RELAXED);
699
+ s->search_misses += __atomic_load_n(&mrg->index[i].stats.search_misses, __ATOMIC_RELAXED);
700
+ s->writers += __atomic_load_n(&mrg->index[i].stats.writers, __ATOMIC_RELAXED);
701
+ s->writers_conflicts += __atomic_load_n(&mrg->index[i].stats.writers_conflicts, __ATOMIC_RELAXED);
702
+ }
703
+
704
+ s->size += sizeof(MRG);
705
}
706
707
// ----------------------------------------------------------------------------
889
if(!mrg_metric_release_and_delete(mrg, m1_t1))
890
fatal("DBENGINE METRIC: cannot delete the second metric");
891
853
- if(mrg->stats.entries != 0)
892
+ struct mrg_statistics s;
893
+ mrg_get_statistics(mrg, &s);
894
+ if(s.entries != 0)
895
fatal("DBENGINE METRIC: invalid entries counter");
896
897
#ifdef MRG_STRESS_TEST