@cryptotaxi247 / netdata-1 / commits / eb6942a2d

Improve jv2 load (#19995)

* Use one uv worker to per tier to populate journal v2 files Each worker will share a pool of netdata_conf_cpus() threads * Improve MRG population logging and handle empty datafile scenario

Stelios Fragkakis committed Apr 7, 2025 at 15:38 UTC eb6942a2d2435ee76c144b87b2f0e646dd3d611d

2 files changed +175 -21

src/database/engine/rrdengine.c

+172 -9

@@ -1329,20 +1329,87 @@ static void *flush_dirty_pages_of_section_tp_worker(struct rrdengine_instance *c

1329 return data;

1330 }

1331

1332 +struct mrg_load_thread {

1333 + int max_threads;

1334 + uv_thread_t thread;

1335 + uv_sem_t *sem;

1336 + int tier;

1337 + struct rrdengine_datafile *datafile;

1338 + bool busy;

1339 + bool finished;

1340 +};

1341 +

1342 +size_t max_running_threads = 0;

1343 +size_t running_threads = 0;

1344 +

1345 +void journalfile_v2_populate_retention_to_mrg_worker(void *arg)

1346 +{

1347 + struct mrg_load_thread *mlt = arg;

1348 + uv_sem_wait(mlt->sem);

1349 +

1350 + struct rrdengine_instance *ctx = mlt->datafile->ctx;

1351 +

1352 + size_t current_threads = __atomic_add_fetch(&running_threads, 1, __ATOMIC_RELAXED);

1353 + size_t prev_max;

1354 + do {

1355 + prev_max = __atomic_load_n(&max_running_threads, __ATOMIC_RELAXED);

1356 + if (current_threads <= prev_max) {

1357 + break;

1358 + }

1359 + } while (!__atomic_compare_exchange_n(

1360 + &max_running_threads, &prev_max, current_threads, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));

1361 +

1362 + journalfile_v2_populate_retention_to_mrg(ctx, mlt->datafile->journalfile);

1363 +

1364 + __atomic_sub_fetch(&running_threads, 1, __ATOMIC_RELAXED);

1365 + uv_sem_post(mlt->sem);

1366 +

1367 + // Signal completion - this needs to be last

1368 + __atomic_store_n(&mlt->finished, true, __ATOMIC_RELEASE);

1369 +}

1370

1371 static void after_populate_mrg(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) {

1334 - ;

1372 + if (completion)

1373 + completion_mark_complete(completion);

1374 }

1375

1337 -static void *populate_mrg_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {

1376 +static void *populate_mrg_tp_worker(

1377 + struct rrdengine_instance *ctx,

1378 + void *data,

1379 + struct completion *completion __maybe_unused,

1380 + uv_work_t *uv_work_req __maybe_unused)

1381 +{

1382 worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);

1383

1384 + struct mrg_load_thread *mlt = data;

1385 + size_t max_threads = mlt->max_threads;

1386 + int tier = ctx->config.tier;

1387 +

1388 + size_t thread_index = 0;

1389 + int rc;

1390 +

1391 + uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1392 + size_t total_datafiles = 0;

1393 + size_t populated_datafiles = 0;

1394 + for (struct rrdengine_datafile *df = ctx->datafiles.first; df; df = df->next) {

1395 + total_datafiles++;

1396 + if (df->populate_mrg.populated)

1397 + populated_datafiles++;

1398 + }

1399 + uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1400 +

1401 + if (total_datafiles == 0) {

1402 + nd_log_daemon(NDLP_WARNING, "DBENGINE: No datafiles to populate MRG");

1403 + worker_is_idle();

1404 + return data;

1405 + }

1406 +

1407 do {

1408 struct rrdengine_datafile *datafile = NULL;

1409

1343 - // find a datafile to work

1410 + // find a datafile to work on

1411 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1345 - for(datafile = ctx->datafiles.first; datafile ; datafile = datafile->next) {

1412 + for(datafile = ctx->datafiles.first; datafile; datafile = datafile->next) {

1413 if(!spinlock_trylock(&datafile->populate_mrg.spinlock))

1414 continue;

1415

@@ -1359,14 +1426,99 @@ static void *populate_mrg_tp_worker(struct rrdengine_instance *ctx __maybe_unuse

1426 if(!datafile)

1427 break;

1428

1362 - journalfile_v2_populate_retention_to_mrg(ctx, datafile->journalfile);

1363 - datafile->populate_mrg.populated = true;

1364 - spinlock_unlock(&datafile->populate_mrg.spinlock);

1429 + // Datafile populate mrg spinlock is acquired

1430 + // Find an available thread slot or join finished threads

1431 + bool thread_slot_found = false;

1432 +

1433 + while (!thread_slot_found) {

1434 + // First, check for any finished threads to clean up

1435 + for (size_t index = 0; index < max_threads; index++) {

1436 + if (__atomic_load_n(&mlt[index].finished, __ATOMIC_RELAXED) &&

1437 + __atomic_load_n(&mlt[index].tier, __ATOMIC_ACQUIRE) == tier) {

1438 +

1439 + rc = uv_thread_join(&(mlt[index].thread));

1440 + if (rc)

1441 + nd_log_daemon(NDLP_WARNING, "Failed to join thread, rc = %d", rc);

1442 +

1443 + __atomic_store_n(&mlt[index].busy, false, __ATOMIC_RELEASE);

1444 + __atomic_store_n(&mlt[index].finished, false, __ATOMIC_RELEASE);

1445 + mlt[index].datafile->populate_mrg.populated = true;

1446 + populated_datafiles++;

1447 + spinlock_unlock(&mlt[index].datafile->populate_mrg.spinlock);

1448 +

1449 + // We've cleaned up a thread slot, but we'll still look for a free one

1450 + }

1451 + }

1452 +

1453 + // Look for a free thread slot

1454 + for (size_t index = 0; index < max_threads; index++) {

1455 + bool expected = false;

1456 + if (__atomic_compare_exchange_n(&(mlt[index].busy), &expected, true, false,

1457 + __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {

1458 + thread_index = index;

1459 + thread_slot_found = true;

1460 + break;

1461 + }

1462 + }

1463 +

1464 + if (!thread_slot_found) {

1465 + // If we couldn't find a free slot after cleanup, wait a bit and try again

1466 + sleep_usec(10 * USEC_PER_MS);

1467 + }

1468 + }

1469

1470 + // We have a thread slot (thread_index) and a datafile to process

1471 + __atomic_store_n(&mlt[thread_index].tier, tier, __ATOMIC_RELAXED);

1472 + mlt[thread_index].datafile = datafile;

1473 +

1474 + rc = uv_thread_create(&mlt[thread_index].thread,

1475 + journalfile_v2_populate_retention_to_mrg_worker,

1476 + &mlt[thread_index]);

1477 +

1478 + if (rc) {

1479 + nd_log_daemon(NDLP_WARNING, "Failed to create thread, rc = %d", rc);

1480 + __atomic_store_n(&mlt[thread_index].busy, false, __ATOMIC_RELEASE);

1481 + spinlock_unlock(&datafile->populate_mrg.spinlock);

1482 + }

1483 + nd_log_limit_static_thread_var(erl, 10, 0);

1484 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_INFO, "DBENGINE: Tier %d MRG population completed: %.2f%% (%zu/%zu)", tier, (populated_datafiles * 100.0) / total_datafiles,

1485 + populated_datafiles, total_datafiles);

1486 } while(1);

1487

1368 - completion_mark_complete(completion);

1488 + // We've processed all datafiles. Now wait for all our threads to complete

1489 + bool threads_still_running;

1490 + do {

1491 + threads_still_running = false;

1492 +

1493 + for (size_t index = 0; index < max_threads; index++) {

1494 + if (__atomic_load_n(&mlt[index].busy, __ATOMIC_ACQUIRE) &&

1495 + __atomic_load_n(&mlt[index].tier, __ATOMIC_ACQUIRE) == tier) {

1496 +

1497 + if (__atomic_load_n(&mlt[index].finished, __ATOMIC_RELAXED)) {

1498 + // Thread is finished, join it

1499 + rc = uv_thread_join(&(mlt[index].thread));

1500 + if (rc)

1501 + nd_log_daemon(NDLP_WARNING, "Failed to join thread, rc = %d", rc);

1502 +

1503 + __atomic_store_n(&mlt[index].busy, false, __ATOMIC_RELEASE);

1504 + __atomic_store_n(&mlt[index].finished, false, __ATOMIC_RELEASE);

1505 + mlt[index].datafile->populate_mrg.populated = true;

1506 + spinlock_unlock(&mlt[index].datafile->populate_mrg.spinlock);

1507 + } else {

1508 + // Thread is still running

1509 + threads_still_running = true;

1510 + }

1511 + }

1512 + }

1513 +

1514 + if (threads_still_running) {

1515 + // Wait a bit before checking again

1516 + sleep_usec(10 * USEC_PER_MS);

1517 + }

1518 +

1519 + } while (threads_still_running);

1520

1521 + worker_is_idle();

1522 return data;

1523 }

1524

@@ -1934,6 +2086,17 @@ void dbengine_event_loop(void* arg) {

2086 fatal_assert(0 == uv_timer_start(&main->retention_timer, retention_timer_cb, TIMER_PERIOD_MS * 60, TIMER_PERIOD_MS * 60));

2087

2088 bool shutdown = false;

2089 + size_t cpus = netdata_conf_cpus();

2090 + uv_sem_t sem;

2091 + uv_sem_init(&sem, (unsigned int) cpus);

2092 + struct mrg_load_thread *mlt = callocz(cpus, sizeof(*mlt));

2093 + for (size_t i = 0; i < cpus; i++) {

2094 + mlt[i].sem = &sem;

2095 + mlt[i].max_threads = cpus;

2096 + mlt[i].busy = false;

2097 + mlt[i].finished = false;

2098 + }

2099 +

2100 while (likely(!shutdown)) {

2101 worker_is_idle();

2102 uv_run(&main->loop, UV_RUN_DEFAULT);

@@ -2038,7 +2201,7 @@ void dbengine_event_loop(void* arg) {

2201 case RRDENG_OPCODE_CTX_POPULATE_MRG: {

2202 struct rrdengine_instance *ctx = cmd.ctx;

2203 struct completion *completion = cmd.completion;

2041 - work_dispatch(ctx, NULL, completion, opcode, populate_mrg_tp_worker, after_populate_mrg);

2204 + work_dispatch(ctx, mlt, completion, opcode, populate_mrg_tp_worker, after_populate_mrg);

2205 break;

2206 }

2207

src/database/engine/rrdengineapi.c

+3 -12

@@ -1113,20 +1113,11 @@ static void rrdeng_populate_mrg(struct rrdengine_instance *ctx) {

1113 datafiles++;

1114 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1115

1116 - ssize_t cpus = (ssize_t)netdata_conf_cpus() / (ssize_t)nd_profile.storage_tiers;

1117 - if(cpus > (ssize_t)datafiles)

1118 - cpus = (ssize_t)datafiles;

1119 -

1120 - if(cpus > (ssize_t)libuv_worker_threads)

1121 - cpus = (ssize_t)libuv_worker_threads;

1122 -

1123 - if(cpus >= (ssize_t)netdata_conf_cpus() / 2)

1124 - cpus = (ssize_t)(netdata_conf_cpus() / 2 - 1);

1125 -

1116 + ssize_t cpus = (ssize_t)netdata_conf_cpus();

1117 if(cpus < 1)

1118 cpus = 1;

1119

1129 - netdata_log_info("DBENGINE: populating retention to MRG from %zu journal files of tier %d, using %zd threads...", datafiles, ctx->config.tier, cpus);

1120 + 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);

1121

1122 if(datafiles > 2) {

1123 struct rrdengine_datafile *datafile;

@@ -1147,7 +1138,7 @@ static void rrdeng_populate_mrg(struct rrdengine_instance *ctx) {

1138 }

1139 }

1140

1150 - ctx->loading.populate_mrg.size = cpus;

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++) {