Use spinlock for reference counting. (#16901)
vkalintiris committed
Feb 1, 2024 at 16:36 UTC
2e459de7ecb985162b48c65ebfa742e5d745af8c
1 file changed
+20
-16
src/database/engine/metric.c
+20
-16
@@ -1,6 +1,7 @@
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
typedef int32_t REFCOUNT;
@@ -16,6 +17,8 @@ struct metric {
17
uint32_t latest_update_every_s; // the latest data collection frequency
18
pid_t writer;
19
uint8_t partition;
20
+
21
+ SPINLOCK refcount_spinlock;
22
REFCOUNT refcount;
23
24
// THIS IS allocated with malloc()
@@ -131,17 +134,17 @@ static inline time_t mrg_metric_get_first_time_s_smart(MRG *mrg __maybe_unused,
134
}
135
136
static inline REFCOUNT metric_acquire(MRG *mrg __maybe_unused, METRIC *metric) {
134
- size_t partition = metric->partition;
135
- REFCOUNT expected = __atomic_load_n(&metric->refcount, __ATOMIC_RELAXED);
136
- REFCOUNT refcount;
137
+ spinlock_lock(&metric->refcount_spinlock);
138
138
- do {
139
- if(expected < 0)
140
- fatal("METRIC: refcount is %d (negative) during acquire", metric->refcount);
139
+ if (metric->refcount >= 0)
140
+ metric->refcount += 1;
141
+ else
142
+ fatal("METRIC: refcount is %d (negative) during acquire", metric->refcount);
143
142
- refcount = expected + 1;
143
- } while(!__atomic_compare_exchange_n(&metric->refcount, &expected, refcount, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
144
+ REFCOUNT refcount = metric->refcount;
145
+ spinlock_unlock(&metric->refcount_spinlock);
146
147
+ size_t partition = metric->partition;
148
if(refcount == 1)
149
__atomic_add_fetch(&mrg->index[partition].stats.entries_referenced, 1, __ATOMIC_RELAXED);
150
@@ -151,17 +154,17 @@ static inline REFCOUNT metric_acquire(MRG *mrg __maybe_unused, METRIC *metric) {
154
}
155
156
static inline bool metric_release_and_can_be_deleted(MRG *mrg __maybe_unused, METRIC *metric) {
154
- size_t partition = metric->partition;
155
- REFCOUNT expected = __atomic_load_n(&metric->refcount, __ATOMIC_RELAXED);
156
- REFCOUNT refcount;
157
+ spinlock_lock(&metric->refcount_spinlock);
158
158
- do {
159
- if(expected <= 0)
160
- fatal("METRIC: refcount is %d (zero or negative) during release", metric->refcount);
159
+ if (metric->refcount <= 0)
160
+ fatal("METRIC: refcount is %d (zero or negative) during release", metric->refcount);
161
162
- refcount = expected - 1;
163
- } while(!__atomic_compare_exchange_n(&metric->refcount, &expected, refcount, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
162
+ metric->refcount -= 1;
163
+ REFCOUNT refcount = metric->refcount;
164
165
+ spinlock_unlock(&metric->refcount_spinlock);
166
+
167
+ size_t partition = metric->partition;
168
if(unlikely(!refcount))
169
__atomic_sub_fetch(&mrg->index[partition].stats.entries_referenced, 1, __ATOMIC_RELAXED);
170
@@ -223,6 +226,7 @@ static inline METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *r
226
metric->writer = 0;
227
metric->refcount = 0;
228
metric->partition = partition;
229
+ spinlock_init(&metric->refcount_spinlock);
230
metric_acquire(mrg, metric);
231
*PValue = metric;
232