@cryptotaxi247 / netdata-1 / commits / 71c7e4420

Improve agent startup (#21260)

* Improve agent startup * Config worker for tier jv2 mrg load * Cleanup unused variables * Mark unused variable * Remove unused variable

Stelios Fragkakis committed Nov 5, 2025 at 15:33 UTC 71c7e44206ce757a930b17f018a533189d0d3f23

8 files changed +79 -128

src/daemon/libuv_workers.c

@@ -45,6 +45,7 @@ static void register_libuv_worker_jobs_internal(void) {

45 worker_register_job_name(UV_EVENT_DBENGINE_FLUSH_DIRTY, "dbengine flush dirty");

46 worker_register_job_name(UV_EVENT_DBENGINE_QUIESCE, "dbengine quiesce");

47 worker_register_job_name(UV_EVENT_DBENGINE_SHUTDOWN, "dbengine shutdown");

48 + worker_register_job_name(UV_EVENT_DBENGINE_MRG_LOAD, "jv2 mrg load");

50 // metadata

51 worker_register_job_name(UV_EVENT_HOST_CONTEXT_LOAD, "metadata load host context");

src/daemon/libuv_workers.h

@@ -42,6 +42,7 @@ enum event_loop_job {

42 UV_EVENT_DBENGINE_BUFFERS_CLEANUP,

43 UV_EVENT_DBENGINE_FLUSH_DIRTY,

44 UV_EVENT_DBENGINE_QUIESCE,

45 + UV_EVENT_DBENGINE_MRG_LOAD,

46 UV_EVENT_DBENGINE_SHUTDOWN,

48 // metadata

src/database/engine/journalfile.c

+12 -1

@@ -1008,6 +1008,8 @@ void journalfile_v2_populate_retention_to_mrg(struct rrdengine_instance *ctx, st

1008 time_t global_first_time_s;

1009 bool failed = false;

1010 uint32_t entries;

1011 + // Calculate number of samples here and update once the file is loaded

1012 + uint64_t journal_samples = 0;

1013 PROTECTED_ACCESS_SETUP(data_start, journalfile->mmap.size, path_v2, "mrg-load");

1014 if(no_signal_received) {

1015 entries = j2_header->metric_count;

@@ -1020,7 +1022,14 @@ void journalfile_v2_populate_retention_to_mrg(struct rrdengine_instance *ctx, st

1022 time_t end_time_s = header_start_time_s + metric->delta_end_s;

1023

1024 mrg_update_metric_retention_and_granularity_by_uuid(

1023 - main_mrg, (Word_t)ctx, &metric->uuid, start_time_s, end_time_s, metric->update_every_s, now_s);

1025 + main_mrg,

1026 + (Word_t)ctx,

1027 + &metric->uuid,

1028 + start_time_s,

1029 + end_time_s,

1030 + metric->update_every_s,

1031 + now_s,

1032 + &journal_samples);

1033

1034 metric++;

1035 }

@@ -1032,6 +1041,8 @@ void journalfile_v2_populate_retention_to_mrg(struct rrdengine_instance *ctx, st

1041 if (unlikely(failed))

1042 return;

1043

1044 + __atomic_add_fetch(&ctx->atomic.samples, journal_samples, __ATOMIC_RELAXED);

1045 +

1046 usec_t ended_ut = now_monotonic_usec();

1047

1048 nd_log_daemon(NDLP_DEBUG, "DBENGINE: journal v2 of tier %d, datafile %u populated, size: %0.2f MiB, metrics: %0.2f k, %0.2f ms"

src/database/engine/mrg-unittest.c

+2 -7

@@ -38,12 +38,7 @@ static void mrg_stress(void *ptr) {

38 time_t before = __atomic_add_fetch(&e->before, 1, __ATOMIC_RELAXED);

40 mrg_update_metric_retention_and_granularity_by_uuid(

41 - mrg, 0x01,

42 - &e->uuid,

43 - after,

44 - before,

45 - 1,

46 - before);

41 + mrg, 0x01, &e->uuid, after, before, 1, before, NULL);

43 __atomic_add_fetch(&t->updates, 1, __ATOMIC_RELAXED);

44 }

@@ -160,7 +155,7 @@ int mrg_unittest(void) {

155 e->after,

156 e->before,

157 1,

163 - e->before);

158 + e->before, NULL);

159 }

160 }

161 netdata_log_info("stress test ready to run...");

src/database/engine/mrg.c

+12 -6

@@ -382,9 +382,14 @@ ALWAYS_INLINE bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) {

382 #endif

383

384 inline void mrg_update_metric_retention_and_granularity_by_uuid(

385 - MRG *mrg, Word_t section, nd_uuid_t *uuid,

386 - time_t first_time_s, time_t last_time_s,

387 - uint32_t update_every_s, time_t now_s)

385 + MRG *mrg,

386 + Word_t section,

387 + nd_uuid_t(*uuid),

388 + time_t first_time_s,

389 + time_t last_time_s,

390 + uint32_t update_every_s,

391 + time_t now_s,

392 + uint64_t *journal_samples)

393 {

394 if(unlikely(last_time_s > now_s)) {

395 nd_log_limit_static_global_var(erl, 1, 0);

@@ -426,7 +431,6 @@ inline void mrg_update_metric_retention_and_granularity_by_uuid(

431 metric = mrg_metric_add_and_acquire(mrg, entry, &added);

432 }

433

429 - struct rrdengine_instance *ctx = (struct rrdengine_instance *) section;

434 if (likely(!added)) {

435 uint64_t old_samples = 0;

436

@@ -439,13 +443,15 @@ inline void mrg_update_metric_retention_and_granularity_by_uuid(

443 if (update_every_s && metric->latest_update_every_s && metric->latest_time_s_clean)

444 new_samples = (metric->latest_time_s_clean - metric->first_time_s) / metric->latest_update_every_s;

445

442 - __atomic_add_fetch(&ctx->atomic.samples, new_samples - old_samples, __ATOMIC_RELAXED);

446 + if (journal_samples)

447 + *journal_samples += (new_samples - old_samples);

448 }

449 else {

450 // Newly added

451 if (update_every_s) {

452 uint64_t samples = (last_time_s - first_time_s) / update_every_s;

448 - __atomic_add_fetch(&ctx->atomic.samples, samples, __ATOMIC_RELAXED);

453 + if (journal_samples)

454 + *journal_samples += samples;

455 }

456 }

457

src/database/engine/mrg.h

+8 -3

@@ -85,9 +85,14 @@ void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s);

85 struct aral_statistics *mrg_aral_stats(void);

87 void mrg_update_metric_retention_and_granularity_by_uuid(

88 - MRG *mrg, Word_t section, nd_uuid_t *uuid,

89 - time_t first_time_s, time_t last_time_s,

90 - uint32_t update_every_s, time_t now_s);

88 + MRG *mrg,

89 + Word_t section,

90 + nd_uuid_t(*uuid),

91 + time_t first_time_s,

92 + time_t last_time_s,

93 + uint32_t update_every_s,

94 + time_t now_s,

95 + uint64_t *journal_samples);

97 bool mrg_save(MRG *mrg);

98 bool mrg_load(MRG *mrg);

src/database/engine/rrdengine.c

+42 -111

@@ -1586,44 +1586,42 @@ static void *flush_dirty_pages_of_section_tp_worker(struct rrdengine_instance *c

1586 }

1587

1588 struct mrg_load_thread {

1589 - int max_threads;

1590 - ND_THREAD *thread;

1589 uv_sem_t *sem;

1592 - int tier;

1590 struct rrdengine_datafile *datafile;

1594 - bool busy;

1595 - bool finished;

1591 + size_t *total;

1592 + size_t *populated_datafiles;

1593 };

1594

1598 -size_t max_running_threads = 0;

1599 -size_t running_threads = 0;

1600 -

1595 void journalfile_v2_populate_retention_to_mrg_worker(void *arg)

1596 {

1597 struct mrg_load_thread *mlt = arg;

1604 - uv_sem_wait(mlt->sem);

1605 -

1598 struct rrdengine_instance *ctx = mlt->datafile->ctx;

1599

1608 - size_t current_threads = __atomic_add_fetch(&running_threads, 1, __ATOMIC_RELAXED);

1609 - size_t prev_max;

1610 - do {

1611 - prev_max = __atomic_load_n(&max_running_threads, __ATOMIC_RELAXED);

1612 - if (current_threads <= prev_max) {

1613 - break;

1614 - }

1615 - } while (!__atomic_compare_exchange_n(

1616 - &max_running_threads, &prev_max, current_threads, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));

1617 -

1600 journalfile_v2_populate_retention_to_mrg(ctx, mlt->datafile->journalfile);

1601

1620 - __atomic_sub_fetch(&running_threads, 1, __ATOMIC_RELAXED);

1602 uv_sem_post(mlt->sem);

1603 +}

1604 +

1605 +static void *tier_mrg_load(

1606 + struct rrdengine_instance *ctx __maybe_unused,

1607 + void *data,

1608 + struct completion *completion __maybe_unused,

1609 + uv_work_t *req __maybe_unused)

1610 +{

1611 + worker_is_busy(UV_EVENT_DBENGINE_MRG_LOAD);

1612 + struct mrg_load_thread *mlt = data;

1613 + journalfile_v2_populate_retention_to_mrg_worker(mlt);

1614 + mlt->datafile->populate_mrg.populated = true;

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

1616

1623 - // Signal completion - this needs to be last

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

1617 + __atomic_add_fetch(mlt->populated_datafiles, 1, __ATOMIC_RELAXED);

1618 + __atomic_sub_fetch(mlt->total, 1, __ATOMIC_RELEASE);

1619 + freez(mlt);

1620 + worker_is_idle();

1621 + return NULL;

1622 }

1623

1624 +

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

1626 if (completion)

1627 completion_mark_complete(completion);

@@ -1638,12 +1636,8 @@ static void *populate_mrg_tp_worker(

1636 worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);

1637

1638 struct mrg_load_thread *mlt = data;

1641 - size_t max_threads = mlt->max_threads;

1639 int tier = ctx->config.tier;

1640

1644 - size_t thread_index = 0;

1645 - int rc;

1646 -

1641 netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

1642

1643 size_t total_datafiles = 0;

@@ -1662,6 +1656,7 @@ static void *populate_mrg_tp_worker(

1656 return data;

1657 }

1658

1659 + size_t total = 0;

1660 do {

1661 struct rrdengine_datafile *datafile = NULL;

1662

@@ -1689,96 +1684,29 @@ static void *populate_mrg_tp_worker(

1684 if(!datafile)

1685 break;

1686

1692 - // Datafile populate mrg spinlock is acquired

1693 - // Find an available thread slot or join finished threads

1694 - bool thread_slot_found = false;

1695 -

1696 - while (!thread_slot_found) {

1697 - // First, check for any finished threads to clean up

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

1699 - if (__atomic_load_n(&mlt[index].finished, __ATOMIC_RELAXED) &&

1700 - __atomic_load_n(&mlt[index].tier, __ATOMIC_ACQUIRE) == tier) {

1701 -

1702 - rc = nd_thread_join(mlt[index].thread);

1703 - if (rc)

1704 - nd_log_daemon(NDLP_WARNING, "Failed to join thread, rc = %d", rc);

1705 -

1706 - __atomic_store_n(&mlt[index].busy, false, __ATOMIC_RELEASE);

1707 - __atomic_store_n(&mlt[index].finished, false, __ATOMIC_RELEASE);

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

1709 - populated_datafiles++;

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

1711 -

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

1713 - }

1714 - }

1715 -

1716 - // Look for a free thread slot

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

1718 - bool expected = false;

1719 - if (__atomic_compare_exchange_n(&(mlt[index].busy), &expected, true, false,

1720 - __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {

1721 - thread_index = index;

1722 - thread_slot_found = true;

1723 - break;

1724 - }

1725 - }

1726 -

1727 - if (!thread_slot_found) {

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

1729 - sleep_usec(10 * USEC_PER_MS);

1730 - }

1731 - }

1732 -

1733 - // We have a thread slot (thread_index) and a datafile to process

1734 - __atomic_store_n(&mlt[thread_index].tier, tier, __ATOMIC_RELAXED);

1735 - mlt[thread_index].datafile = datafile;

1736 -

1737 - mlt[thread_index].thread = nd_thread_create("MRGLOAD", NETDATA_THREAD_OPTION_DEFAULT, journalfile_v2_populate_retention_to_mrg_worker,

1738 - &mlt[thread_index]);

1739 -

1740 - if (!mlt[thread_index].thread) {

1741 - nd_log_daemon(NDLP_WARNING, "Failed to create thread for MRG population");

1742 - __atomic_store_n(&mlt[thread_index].busy, false, __ATOMIC_RELEASE);

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

1744 - }

1687 + uv_sem_wait(mlt->sem);

1688 + struct mrg_load_thread *local_mlt = callocz(1, sizeof(struct mrg_load_thread));

1689 + local_mlt->datafile = datafile;

1690 + local_mlt->sem = mlt->sem;

1691 + local_mlt->total = &total;

1692 + local_mlt->populated_datafiles = &populated_datafiles;

1693 + __atomic_add_fetch(local_mlt->total, 1, __ATOMIC_RELAXED);

1694 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_MRG_LOAD, local_mlt, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1695 nd_log_limit_static_thread_var(erl, 10, 0);

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

1697 populated_datafiles, total_datafiles);

1698 } while(1);

1699

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

1751 - bool threads_still_running;

1701 + size_t pending;

1702 do {

1753 - threads_still_running = false;

1754 -

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

1756 - if (__atomic_load_n(&mlt[index].busy, __ATOMIC_ACQUIRE) &&

1757 - __atomic_load_n(&mlt[index].tier, __ATOMIC_ACQUIRE) == tier) {

1758 -

1759 - if (__atomic_load_n(&mlt[index].finished, __ATOMIC_RELAXED)) {

1760 - // Thread is finished, join it

1761 - rc = nd_thread_join((mlt[index].thread));

1762 - if (rc)

1763 - nd_log_daemon(NDLP_WARNING, "Failed to join thread, rc = %d", rc);

1764 -

1765 - __atomic_store_n(&mlt[index].busy, false, __ATOMIC_RELEASE);

1766 - __atomic_store_n(&mlt[index].finished, false, __ATOMIC_RELEASE);

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

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

1769 - } else {

1770 - // Thread is still running

1771 - threads_still_running = true;

1772 - }

1773 - }

1774 - }

1775 -

1776 - if (threads_still_running) {

1777 - // Wait a bit before checking again

1703 + pending = __atomic_load_n(&total, __ATOMIC_ACQUIRE);

1704 + if (pending) {

1705 + nd_log_limit_static_thread_var(erl, 10, 0);

1706 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_INFO, "DBENGINE: Tier %d Waiting for %zu threads", tier, total);

1707 sleep_usec(10 * USEC_PER_MS);

1708 }

1780 -

1781 - } while (threads_still_running);

1709 + } while (pending > 0);

1710

1711 worker_is_idle();

1712 return data;

@@ -2392,6 +2320,7 @@ void dbengine_event_loop(void* arg) {

2320 worker_register_job_name(RRDENG_OPCODE_CTX_QUIESCE, "ctx quiesce");

2321 worker_register_job_name(RRDENG_OPCODE_SHUTDOWN_EVLOOP, "dbengine shutdown");

2322 worker_register_job_name(RRDENG_OPCODE_PARALLEL_WEIGHT, "parallel weight");

2323 + worker_register_job_name(RRDENG_OPCODE_MRG_LOAD, "mrg tier load");

2324

2325

2326 worker_register_job_name(RRDENG_OPCODE_MAX, "get opcode");

@@ -2407,6 +2336,7 @@ void dbengine_event_loop(void* arg) {

2336 worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_FLUSH_DIRTY, "ctx flush dirty cb");

2337 worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_QUIESCE, "ctx quiesce cb");

2338 worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_PARALLEL_WEIGHT, "parallel weight cb");

2339 + worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_MRG_LOAD, "mrg tier load cb");

2340

2341 // special jobs

2342 worker_register_job_name(RRDENG_RETENTION_TIMER_CB, "retention timer");

@@ -2431,9 +2361,6 @@ void dbengine_event_loop(void* arg) {

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

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

2363 mlt[i].sem = &sem;

2434 - mlt[i].max_threads = cpus;

2435 - mlt[i].busy = false;

2436 - mlt[i].finished = false;

2364 }

2365

2366 #if defined(OS_WINDOWS)

@@ -2452,6 +2379,10 @@ void dbengine_event_loop(void* arg) {

2379 worker_is_busy(opcode);

2380

2381 switch (opcode) {

2382 + case RRDENG_OPCODE_MRG_LOAD:

2383 + work_dispatch(NULL, cmd.data, cmd.completion, cmd.opcode, tier_mrg_load, NULL);

2384 + break;

2385 +

2386 case RRDENG_OPCODE_PARALLEL_WEIGHT:;

2387

2388 work_dispatch(NULL, cmd.data, cmd.completion, cmd.opcode, weights_worker, after_weights_worker);

src/database/engine/rrdengine.h

@@ -284,6 +284,7 @@ enum rrdeng_opcode {

284 RRDENG_OPCODE_CTX_POPULATE_MRG,

285 RRDENG_OPCODE_SHUTDOWN_EVLOOP,

286 RRDENG_OPCODE_PARALLEL_WEIGHT,

287 + RRDENG_OPCODE_MRG_LOAD,

288 RRDENG_OPCODE_CLEANUP,

289

290 RRDENG_OPCODE_MAX