Simplify MRG loading mechanism logic (#20562)
Stelios Fragkakis committed
Jun 24, 2025 at 16:51 UTC
43810529dfb808c9b883acc7d2624bc140393f93
2 files changed
+14
-41
src/database/engine/rrdengine.h
+1
-5
@@ -425,11 +425,7 @@ struct rrdengine_instance {
425
} quiesce;
426
427
struct {
428
- struct {
429
- size_t size;
430
- struct completion *array;
431
- } populate_mrg;
432
-
428
+ struct completion load_mrg;
429
bool create_new_datafile_pair;
430
} loading;
431
src/database/engine/rrdengineapi.c
+13
-36
@@ -1106,7 +1106,8 @@ void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long
1106
fatal_assert(RRDENG_NR_STATS == 38);
1107
}
1108
1109
-static void rrdeng_populate_mrg(struct rrdengine_instance *ctx) {
1109
+static void rrdeng_populate_mrg(struct rrdengine_instance *ctx)
1110
+{
1111
uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1112
size_t datafiles = 0;
1113
for(struct rrdengine_datafile *df = ctx->datafiles.first; df ;df = df->next)
@@ -1119,44 +1120,20 @@ static void rrdeng_populate_mrg(struct rrdengine_instance *ctx) {
1120
1121
netdata_log_info("DBENGINE: populating retention to MRG from %zu journal files of tier %d, using a shared pool of %zd threads...", datafiles, ctx->config.tier, cpus);
1122
1122
- if(datafiles > 2) {
1123
- struct rrdengine_datafile *datafile;
1124
-
1125
- datafile = ctx->datafiles.first->prev;
1126
- if(!(datafile->journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE))
1127
- datafile = datafile->prev;
1128
-
1129
- if(datafile->journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE) {
1130
- journalfile_v2_populate_retention_to_mrg(ctx, datafile->journalfile);
1131
- datafile->populate_mrg.populated = true;
1132
- }
1133
-
1134
- datafile = ctx->datafiles.first;
1135
- if(datafile->journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE) {
1136
- journalfile_v2_populate_retention_to_mrg(ctx, datafile->journalfile);
1137
- datafile->populate_mrg.populated = true;
1138
- }
1139
- }
1140
-
1141
- ctx->loading.populate_mrg.size = 1;
1142
- ctx->loading.populate_mrg.array = callocz(ctx->loading.populate_mrg.size, sizeof(struct completion));
1143
-
1144
- for (size_t i = 0; i < ctx->loading.populate_mrg.size; i++) {
1145
- completion_init(&ctx->loading.populate_mrg.array[i]);
1146
- rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_POPULATE_MRG, NULL, &ctx->loading.populate_mrg.array[i],
1147
- STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1148
- }
1123
+ completion_init(&ctx->loading.load_mrg);
1124
+ rrdeng_enq_cmd(
1125
+ ctx,
1126
+ RRDENG_OPCODE_CTX_POPULATE_MRG,
1127
+ NULL,
1128
+ &ctx->loading.load_mrg,
1129
+ STORAGE_PRIORITY_INTERNAL_DBENGINE,
1130
+ NULL,
1131
+ NULL);
1132
}
1133
1134
void rrdeng_readiness_wait(struct rrdengine_instance *ctx) {
1152
- for (size_t i = 0; i < ctx->loading.populate_mrg.size; i++) {
1153
- completion_wait_for(&ctx->loading.populate_mrg.array[i]);
1154
- completion_destroy(&ctx->loading.populate_mrg.array[i]);
1155
- }
1156
-
1157
- freez(ctx->loading.populate_mrg.array);
1158
- ctx->loading.populate_mrg.array = NULL;
1159
- ctx->loading.populate_mrg.size = 0;
1135
+ completion_wait_for(&ctx->loading.load_mrg);
1136
+ completion_destroy(&ctx->loading.load_mrg);
1137
1138
if(__atomic_load_n(&ctx->atomic.first_time_s, __ATOMIC_RELAXED) == LONG_MAX)
1139
__atomic_store_n(&ctx->atomic.first_time_s, now_realtime_sec(), __ATOMIC_RELAXED);