improve database indexing and rotation handling in event loop (#20459)
Stelios Fragkakis committed
Jun 10, 2025 at 23:43 UTC
8b0721ab929e3b61dc21113f2eb46216d28d03a7
2 files changed
+13
-9
src/database/engine/rrdengine.c
+11
-7
@@ -165,10 +165,18 @@ static inline enum LIBUV_WORKERS_STATUS work_request_full(void) {
165
return LIBUV_WORKERS_RELAXED;
166
}
167
168
+// This needs to be called from event loop thread only (callback)
169
static inline void check_and_schedule_db_rotation(struct rrdengine_instance *ctx)
170
{
171
internal_fatal(rrdeng_main.tid != gettid_cached(), "check_and_schedule_db_rotation() can only be run from the event loop thread");
172
173
+ if (__atomic_load_n(&ctx->atomic.needs_indexing, __ATOMIC_RELAXED)) {
174
+ if (ctx->datafiles.pending_index == false) {
175
+ ctx->datafiles.pending_index = true;
176
+ rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_INDEX, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
177
+ }
178
+ }
179
+
180
if (ctx->datafiles.pending_rotate) {
181
nd_log_daemon(NDLP_DEBUG, "DBENGINE: tier %d is already pending rotation", ctx->config.tier);
182
return;
@@ -670,8 +678,7 @@ extent_flush_to_open(struct rrdengine_instance *ctx, struct extent_io_descriptor
678
spinlock_unlock(&datafile->writers.spinlock);
679
680
if(datafile->fileno != ctx_last_fileno_get(ctx) && still_running)
673
- // we just finished a flushing on a datafile that is not the active one
674
- rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_INDEX, datafile, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
681
+ __atomic_store_n(&ctx->atomic.needs_indexing, true, __ATOMIC_RELAXED);
682
683
worker_is_idle();
684
}
@@ -716,8 +723,7 @@ static struct rrdengine_datafile *get_datafile_to_write_extent(struct rrdengine_
723
uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
724
725
if(datafile_is_full(ctx, datafile) && create_new_datafile_pair(ctx, true) == 0)
719
- rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_INDEX, datafile, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL,
720
- NULL);
726
+ __atomic_store_n(&ctx->atomic.needs_indexing, true, __ATOMIC_RELAXED);
727
728
netdata_mutex_unlock(&mutex);
729
@@ -935,9 +941,6 @@ done:
941
static void after_database_rotate(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
942
__atomic_store_n(&ctx->atomic.now_deleting_files, false, __ATOMIC_RELAXED);
943
938
- if (__atomic_load_n(&ctx->atomic.needs_indexing, __ATOMIC_RELAXED))
939
- rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_INDEX, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
940
-
944
check_and_schedule_db_rotation(ctx);
945
}
946
@@ -2308,6 +2311,7 @@ void *dbengine_event_loop(void* arg) {
2311
case RRDENG_OPCODE_JOURNAL_INDEX: {
2312
struct rrdengine_instance *ctx = cmd.ctx;
2313
struct rrdengine_datafile *datafile = cmd.data;
2314
+ // We no longer have an indexing command pending
2315
ctx->datafiles.pending_index = false;
2316
if (NOT_INDEXING_FILES(ctx) && ctx_is_available_for_queries(ctx)) {
2317
__atomic_store_n(&ctx->atomic.migration_to_v2_running, true, __ATOMIC_RELAXED);
src/database/engine/rrdengine.h
+2
-2
@@ -387,8 +387,8 @@ struct rrdengine_instance {
387
struct {
388
uv_rwlock_t rwlock; // the linked list of datafiles is protected by this lock
389
bool disk_time; // true: delete for disk quota, false: delete for retention
390
- bool pending_rotate;
391
- bool pending_index;
390
+ bool pending_rotate; // Change from event loop
391
+ bool pending_index; // Change from event loop
392
struct rrdengine_datafile *first; // oldest - the newest with ->first->prev
393
} datafiles;
394