@cryptotaxi247 / netdata-1 / commits / ecae70fe8

Calculate weights for multiple nodes in parallel (#21184)

Stelios Fragkakis committed Oct 22, 2025 at 17:22 UTC ecae70fe8270095ffd99cd58019786fc5b9418b3

6 files changed +365 -11

src/daemon/libuv_workers.c

@@ -19,6 +19,8 @@ static void register_libuv_worker_jobs_internal(void) {

19 worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_PAGE_LOOKUP, "page lookup");

20 worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_PAGE_POPULATION, "page populate");

21 worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_PAGE_ALLOCATION, "page allocate");

22 + // Weights

23 + worker_register_job_name(UV_EVENT_WEIGHTS_CALCULATION, "weights calculation");

25 // flushing related

26 worker_register_job_name(UV_EVENT_DBENGINE_FLUSH_MAIN_CACHE, "flush main");

src/daemon/libuv_workers.h

@@ -17,6 +17,8 @@ enum event_loop_job {

17 UV_EVENT_DBENGINE_EXTENT_PAGE_LOOKUP,

18 UV_EVENT_DBENGINE_EXTENT_PAGE_POPULATION,

19 UV_EVENT_DBENGINE_EXTENT_PAGE_ALLOCATION,

20 + // Metrics calculation

21 + UV_EVENT_WEIGHTS_CALCULATION,

23 // flushing related

24 UV_EVENT_DBENGINE_FLUSH_MAIN_CACHE,

src/database/engine/rrdengine.c

+27

@@ -1015,6 +1015,25 @@ datafile_extent_build(struct rrdengine_instance *ctx, struct page_descr_with_dat

1015 return xt_io_descr;

1016 }

1017

1018 +

1019 +static void after_weights_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* uv_work_req __maybe_unused, int status __maybe_unused)

1020 +{

1021 + ;

1022 +}

1023 +

1024 +static void *weights_worker(

1025 + struct rrdengine_instance *ctx __maybe_unused,

1026 + void *data,

1027 + struct completion *completion,

1028 + uv_work_t *req __maybe_unused)

1029 +{

1030 + worker_is_busy(UV_EVENT_WEIGHTS_CALCULATION);

1031 + query_weights_worker_thread(data);

1032 + completion_mark_complete(completion);

1033 + worker_is_idle();

1034 + return NULL;

1035 +}

1036 +

1037 static void after_extent_write(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* uv_work_req __maybe_unused, int status __maybe_unused)

1038 {

1039 check_and_schedule_db_rotation(ctx);

@@ -2372,6 +2391,8 @@ void dbengine_event_loop(void* arg) {

2391 worker_register_job_name(RRDENG_OPCODE_CTX_FLUSH_HOT_DIRTY, "ctx flush all");

2392 worker_register_job_name(RRDENG_OPCODE_CTX_QUIESCE, "ctx quiesce");

2393 worker_register_job_name(RRDENG_OPCODE_SHUTDOWN_EVLOOP, "dbengine shutdown");

2394 + worker_register_job_name(RRDENG_OPCODE_PARALLEL_WEIGHT, "parallel weight");

2395 +

2396

2397 worker_register_job_name(RRDENG_OPCODE_MAX, "get opcode");

2398

@@ -2385,6 +2406,7 @@ void dbengine_event_loop(void* arg) {

2406 worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_SHUTDOWN, "ctx shutdown cb");

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

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

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

2410

2411 // special jobs

2412 worker_register_job_name(RRDENG_RETENTION_TIMER_CB, "retention timer");

@@ -2430,6 +2452,11 @@ void dbengine_event_loop(void* arg) {

2452 worker_is_busy(opcode);

2453

2454 switch (opcode) {

2455 + case RRDENG_OPCODE_PARALLEL_WEIGHT:;

2456 +

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

2458 + break;

2459 +

2460 case RRDENG_OPCODE_EXTENT_READ:

2461 worker_dispatch_extent_read(cmd, false);

2462 break;

src/database/engine/rrdengine.h

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

282 RRDENG_OPCODE_CTX_QUIESCE,

283 RRDENG_OPCODE_CTX_POPULATE_MRG,

284 RRDENG_OPCODE_SHUTDOWN_EVLOOP,

285 + RRDENG_OPCODE_PARALLEL_WEIGHT,

286 RRDENG_OPCODE_CLEANUP,

287

288 RRDENG_OPCODE_MAX

src/web/api/queries/weights.c

+332 -11

@@ -66,6 +66,11 @@ struct register_result {

66 };

68 static DICTIONARY *register_result_init() {

69 + DICTIONARY *results = dictionary_create_advanced(DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct register_result));

70 + return results;

71 +}

72 +

73 +static DICTIONARY *register_result_init_single_threaded() {

74 DICTIONARY *results = dictionary_create_advanced(DICT_OPTION_SINGLE_THREADED | DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct register_result));

75 return results;

76 }

@@ -74,6 +79,35 @@ static void register_result_destroy(DICTIONARY *results) {

79 dictionary_destroy(results);

80 }

82 +// Merge results from local dictionary into main dictionary

83 +static void merge_results_dictionaries(DICTIONARY *main_results, DICTIONARY *local_results) {

84 + if (!local_results || !main_results)

85 + return;

86 +

87 + struct register_result *local_result;

88 + dfe_start_read(local_results, local_result) {

89 + // Try to get existing result in main dictionary

90 + struct register_result *main_result = dictionary_get(main_results, local_result_dfe.name);

91 + if (main_result) {

92 + // Merge the results - keep the higher weight

93 + if (local_result->value > main_result->value) {

94 + // Create a copy with the new values and replace the entire entry

95 + struct register_result merged_result = *local_result;

96 + dictionary_set(main_results, local_result_dfe.name, &merged_result, sizeof(struct register_result));

97 + }

98 + // If local value is not higher, keep the existing main result (do nothing)

99 + } else {

100 + // Insert new result - copy the entire structure

101 + dictionary_set(main_results, local_result_dfe.name, local_result, sizeof(struct register_result));

102 + }

103 + }

104 + dfe_done(local_result);

105 +}

106 +

107 +// Forward declarations

108 +static ssize_t weights_do_node_callback(void *data, RRDHOST *host, bool queryable);

109 +static ssize_t weights_do_context_callback(void *data, RRDCONTEXT_ACQUIRED *rca, bool queryable_context);

110 +

111 static void register_result(DICTIONARY *results, RRDHOST *host, RRDCONTEXT_ACQUIRED *rca, RRDINSTANCE_ACQUIRED *ria,

112 RRDMETRIC_ACQUIRED *rma, NETDATA_DOUBLE value, RESULT_FLAGS flags,

113 STORAGE_POINT *highlighted, STORAGE_POINT *baseline, WEIGHTS_STATS *stats,

@@ -108,7 +142,7 @@ static void register_result(DICTIONARY *results, RRDHOST *host, RRDCONTEXT_ACQUI

142 if(baseline)

143 t.baseline = *baseline;

144

111 - // we can use the pointer address or RMA as a unique key for each metric

145 + // Use the original pointer address approach - revert the stable key change

146 char buf[20 + 1];

147 ssize_t len = snprintfz(buf, sizeof(buf) - 1, "%p", rma);

148 dictionary_set_advanced(results, buf, len, &t, sizeof(struct register_result), NULL);

@@ -301,6 +335,13 @@ static size_t registered_results_to_json_contexts(DICTIONARY *results, BUFFER *w

335 return total_dimensions;

336 }

337

338 +// Workload statistics for progress tracking and thread optimization

339 +struct workload_stats {

340 + size_t nodes;

341 + size_t contexts;

342 + size_t metrics;

343 +};

344 +

345 struct query_weights_data {

346 QUERY_WEIGHTS_REQUEST *qwr;

347

@@ -315,7 +356,7 @@ struct query_weights_data {

356 SIMPLE_PATTERN *dimensions_sp;

357 SIMPLE_PATTERN *labels_sp;

358 SIMPLE_PATTERN *alerts_sp;

318 -

359 +

360 struct pattern_array *scope_labels_pa;

361 struct pattern_array *labels_pa;

362

@@ -330,12 +371,135 @@ struct query_weights_data {

371

372 DICTIONARY *results;

373 WEIGHTS_STATS stats;

374 + RRDHOST **hosts_array;

375 + size_t total_hosts;

376 + size_t hosts_array_capacity;

377

378 uint32_t shifts;

379

380 struct query_versions versions;

381 + struct workload_stats total_workload; // Overall workload statistics for progress tracking

382 +};

383 +

384 +// Thread-local data for parallel processing

385 +struct query_weights_thread_data {

386 + struct query_weights_data *main_qwd;

387 + DICTIONARY *local_results;

388 + WEIGHTS_STATS local_stats;

389 + size_t local_examined_dimensions;

390 + struct query_versions local_versions;

391 + RRDHOST **hosts;

392 + struct completion completion;

393 + size_t host_count;

394 + size_t thread_id;

395 };

396

397 +// Worker thread function for parallel host processing

398 +void query_weights_worker_thread(void *arg)

399 +{

400 + struct query_weights_thread_data *thread_data = (struct query_weights_thread_data *)arg;

401 + struct query_weights_data *main_qwd = thread_data->main_qwd;

402 +

403 + // Initialize local statistics

404 + memset(&thread_data->local_stats, 0, sizeof(WEIGHTS_STATS));

405 + thread_data->local_examined_dimensions = 0;

406 + memset(&thread_data->local_versions, 0, sizeof(struct query_versions));

407 +

408 + // Process assigned hosts

409 + for (size_t i = 0; i < thread_data->host_count; i++) {

410 + RRDHOST *host = thread_data->hosts[i];

411 + if (!host) continue;

412 +

413 + // Check for timeout/interruption

414 + if (__atomic_load_n(&main_qwd->timed_out, __ATOMIC_RELAXED) ||

415 + __atomic_load_n(&main_qwd->interrupted, __ATOMIC_RELAXED)) {

416 + break;

417 + }

418 +

419 + // Check timeout

420 + if (now_monotonic_usec() > (main_qwd->timings.received_ut + main_qwd->timeout_us)) {

421 + __atomic_store_n(&main_qwd->timed_out, true, __ATOMIC_RELAXED);

422 + break;

423 + }

424 +

425 + // Check interruption callback

426 + if (main_qwd->qwr->interrupt_callback &&

427 + main_qwd->qwr->interrupt_callback(main_qwd->qwr->interrupt_callback_data)) {

428 + __atomic_store_n(&main_qwd->interrupted, true, __ATOMIC_RELAXED);

429 + break;

430 + }

431 +

432 + // Create a local query_weights_data for this thread

433 + struct query_weights_data local_qwd = *main_qwd;

434 + local_qwd.results = thread_data->local_results;

435 + local_qwd.stats = thread_data->local_stats;

436 + local_qwd.examined_dimensions = thread_data->local_examined_dimensions;

437 + local_qwd.versions = thread_data->local_versions;

438 +

439 + char uuid[UUID_STR_LEN];

440 + if(!UUIDiszero(host->node_id))

441 + uuid_unparse_lower(host->node_id.uuid, uuid);

442 + else

443 + uuid[0] = '\0';

444 +

445 + SIMPLE_PATTERN_RESULT match = SP_MATCHED_POSITIVE;

446 + if(main_qwd->scope_nodes_sp) {

447 + match = simple_pattern_matches_string_extract(main_qwd->scope_nodes_sp, host->hostname, NULL, 0);

448 + if(match == SP_NOT_MATCHED) {

449 + match = simple_pattern_matches_extract(main_qwd->scope_nodes_sp, host->machine_guid, NULL, 0);

450 + if(match == SP_NOT_MATCHED && *uuid)

451 + match = simple_pattern_matches_extract(main_qwd->scope_nodes_sp, uuid, NULL, 0);

452 + }

453 + }

454 +

455 + if(match != SP_MATCHED_POSITIVE)

456 + continue;

457 +

458 + if(main_qwd->nodes_sp) {

459 + match = simple_pattern_matches_string_extract(main_qwd->nodes_sp, host->hostname, NULL, 0);

460 + if(match == SP_NOT_MATCHED) {

461 + match = simple_pattern_matches_extract(main_qwd->nodes_sp, host->machine_guid, NULL, 0);

462 + if(match == SP_NOT_MATCHED && *uuid)

463 + match = simple_pattern_matches_extract(main_qwd->nodes_sp, uuid, NULL, 0);

464 + }

465 + }

466 +

467 + bool queryable_host = (match == SP_MATCHED_POSITIVE);

468 +

469 + // Update local version hashes

470 + thread_data->local_versions.contexts_hard_hash += dictionary_version(host->rrdctx.contexts);

471 + thread_data->local_versions.contexts_soft_hash += rrdcontext_queue_version(&host->rrdctx.hub_queue);

472 + thread_data->local_versions.alerts_hard_hash += dictionary_version(host->rrdcalc_root_index);

473 + thread_data->local_versions.alerts_soft_hash += __atomic_load_n(&host->health_transitions, __ATOMIC_RELAXED);

474 +

475 + // Process the host using the callback

476 + ssize_t ret = weights_do_node_callback(&local_qwd, host, queryable_host);

477 + if (ret < 0)

478 + break;

479 +

480 + // Update thread-local counters

481 + thread_data->local_examined_dimensions = local_qwd.examined_dimensions;

482 + thread_data->local_stats = local_qwd.stats;

483 + }

484 +}

485 +

486 +// Thread-safe statistics merging - use simple addition since we're in single-threaded merge

487 +static void merge_weights_stats(WEIGHTS_STATS *dest, const WEIGHTS_STATS *src) {

488 + dest->db_queries += src->db_queries;

489 + dest->db_points += src->db_points;

490 + dest->result_points += src->result_points;

491 + dest->binary_searches += src->binary_searches;

492 +

493 + // Update max ratio if needed

494 + if (src->max_base_high_ratio > dest->max_base_high_ratio) {

495 + dest->max_base_high_ratio = src->max_base_high_ratio;

496 + }

497 +

498 + for(size_t tier = 0; tier < RRD_STORAGE_TIERS; tier++) {

499 + dest->db_points_per_tier[tier] += src->db_points_per_tier[tier];

500 + }

501 +}

502 +

503 #define AGGREGATED_WEIGHT_EMPTY (struct aggregated_weight) { \

504 .min = NAN, \

505 .max = NAN, \

@@ -1875,6 +2039,19 @@ static size_t registered_results_to_json_mcp(

2039 return state.count;

2040 }

2041

2042 +static ssize_t weights_count_for_rrdmetric(

2043 + void *data,

2044 + RRDHOST *host __maybe_unused,

2045 + RRDCONTEXT_ACQUIRED *rca __maybe_unused,

2046 + RRDINSTANCE_ACQUIRED *ria __maybe_unused,

2047 + RRDMETRIC_ACQUIRED *rma __maybe_unused)

2048 +{

2049 + struct query_weights_data *qwd = data;

2050 +

2051 + __atomic_fetch_add(&qwd->total_workload.metrics, 1, __ATOMIC_RELAXED);

2052 + return 1;

2053 +}

2054 +

2055 // ----------------------------------------------------------------------------

2056 // The main function

2057

@@ -1883,11 +2060,11 @@ static ssize_t weights_for_rrdmetric(void *data, RRDHOST *host, RRDCONTEXT_ACQUI

2060 QUERY_WEIGHTS_REQUEST *qwr = qwd->qwr;

2061

2062 if(qwd->qwr->interrupt_callback && qwd->qwr->interrupt_callback(qwd->qwr->interrupt_callback_data)) {

1886 - qwd->interrupted = true;

2063 + __atomic_store_n(&qwd->interrupted, true, __ATOMIC_RELAXED);

2064 return -1;

2065 }

2066

1890 - qwd->examined_dimensions++;

2067 + __atomic_fetch_add(&qwd->examined_dimensions, 1, __ATOMIC_RELAXED);

2068

2069 switch(qwr->method) {

2070 case WEIGHTS_METHOD_VALUE:

@@ -1946,6 +2123,67 @@ static ssize_t weights_for_rrdmetric(void *data, RRDHOST *host, RRDCONTEXT_ACQUI

2123 return 1;

2124 }

2125

2126 +static ssize_t weights_count_context_callback(void *data, RRDCONTEXT_ACQUIRED *rca, bool queryable_context) {

2127 + if(!queryable_context)

2128 + return false;

2129 +

2130 + struct query_weights_data *qwd = data;

2131 +

2132 + bool has_retention = false;

2133 + switch(qwd->qwr->method) {

2134 + case WEIGHTS_METHOD_VALUE:

2135 + case WEIGHTS_METHOD_ANOMALY_RATE:

2136 + has_retention = rrdcontext_retention_match(rca, qwd->qwr->after, qwd->qwr->before);

2137 + break;

2138 +

2139 + case WEIGHTS_METHOD_MC_KS2:

2140 + case WEIGHTS_METHOD_MC_VOLUME:

2141 + has_retention = rrdcontext_retention_match(rca, qwd->qwr->after, qwd->qwr->before);

2142 + if(has_retention)

2143 + has_retention = rrdcontext_retention_match(rca, qwd->qwr->baseline_after, qwd->qwr->baseline_before);

2144 + break;

2145 + }

2146 +

2147 + if(!has_retention)

2148 + return 0;

2149 +

2150 + __atomic_fetch_add(&qwd->total_workload.contexts, 1, __ATOMIC_RELAXED);

2151 + ssize_t ret = weights_foreach_rrdmetric_in_context(rca,

2152 + qwd->scope_instances_sp,

2153 + qwd->scope_labels_pa,

2154 + qwd->scope_dimensions_sp,

2155 + qwd->instances_sp,

2156 + NULL,

2157 + qwd->labels_pa,

2158 + qwd->alerts_sp,

2159 + qwd->dimensions_sp,

2160 + true, true, qwd->qwr->version,

2161 + weights_count_for_rrdmetric, qwd);

2162 + if (ret >= 1)

2163 + return 1;

2164 + else

2165 + return 0;

2166 +}

2167 +

2168 +static ssize_t weights_count_node_callback(void *data, RRDHOST *host, bool queryable) {

2169 + if(!queryable)

2170 + return 0;

2171 +

2172 + struct query_weights_data *qwd = data;

2173 + if (qwd->total_hosts >= qwd->hosts_array_capacity) {

2174 + qwd->hosts_array_capacity *= 2;

2175 + qwd->hosts_array = reallocz(qwd->hosts_array, sizeof(RRDHOST *) * qwd->hosts_array_capacity);

2176 + }

2177 + qwd->hosts_array[qwd->total_hosts++] = host;

2178 +

2179 + __atomic_fetch_add(&qwd->total_workload.nodes, 1, __ATOMIC_RELAXED);

2180 + ssize_t ret = query_scope_foreach_context(host, qwd->qwr->scope_contexts,

2181 + qwd->scope_contexts_sp, qwd->contexts_sp,

2182 + weights_count_context_callback, queryable, qwd);

2183 +

2184 + return ret;

2185 +}

2186 +

2187 static ssize_t weights_do_context_callback(void *data, RRDCONTEXT_ACQUIRED *rca, bool queryable_context) {

2188 if(!queryable_context)

2189 return false;

@@ -1984,7 +2222,92 @@ static ssize_t weights_do_context_callback(void *data, RRDCONTEXT_ACQUIRED *rca,

2222 return ret;

2223 }

2224

1987 -ssize_t weights_do_node_callback(void *data, RRDHOST *host, bool queryable) {

2225 +// Parallel version of query_scope_foreach_host

2226 +static ssize_t query_scope_foreach_host_parallel(SIMPLE_PATTERN *scope_hosts_sp, SIMPLE_PATTERN *hosts_sp,

2227 + struct query_weights_data *qwd)

2228 +{

2229 + size_t host_count = dictionary_entries(rrdhost_root_index);

2230 + qwd->hosts_array = mallocz(sizeof(RRDHOST *) * host_count);

2231 + qwd->hosts_array_capacity = host_count;

2232 + qwd->total_hosts = 0;

2233 +

2234 + (void) query_scope_foreach_host(scope_hosts_sp, hosts_sp, weights_count_node_callback, qwd, &qwd->versions, NULL);

2235 +

2236 + size_t active_hosts = qwd->total_hosts;

2237 +

2238 + size_t num_threads = netdata_conf_cpus();

2239 + if (num_threads < 1) num_threads = 1;

2240 +

2241 + // If we have fewer hosts than threads, reduce thread count

2242 + if (active_hosts < num_threads) {

2243 + num_threads = active_hosts;

2244 + }

2245 +

2246 + if (num_threads <= 1 || active_hosts <= 1) {

2247 + // Fall back to single-threaded processing

2248 + freez(qwd->hosts_array);

2249 + return query_scope_foreach_host(scope_hosts_sp, hosts_sp,

2250 + weights_do_node_callback, qwd,

2251 + &qwd->versions, NULL);

2252 + }

2253 +

2254 + // Calculate hosts per thread

2255 + size_t hosts_per_thread = active_hosts / num_threads;

2256 + size_t remaining_hosts = active_hosts % num_threads;

2257 +

2258 + // Prepare thread data

2259 + struct query_weights_thread_data *thread_data = mallocz(sizeof(struct query_weights_thread_data) * num_threads);

2260 + ND_THREAD **threads = mallocz(sizeof(ND_THREAD *) * num_threads);

2261 +

2262 + size_t current_host_idx = 0;

2263 + for (size_t i = 0; i < num_threads; i++) {

2264 + thread_data[i].main_qwd = qwd;

2265 + thread_data[i].local_results = register_result_init_single_threaded();

2266 + thread_data[i].thread_id = i;

2267 + thread_data[i].hosts = &qwd->hosts_array[current_host_idx];

2268 +

2269 + // Distribute hosts evenly, giving extra hosts to first threads

2270 + thread_data[i].host_count = hosts_per_thread + (i < remaining_hosts ? 1 : 0);

2271 + current_host_idx += thread_data[i].host_count;

2272 +

2273 + completion_init(&thread_data[i].completion);

2274 + rrdeng_enq_cmd(NULL, RRDENG_OPCODE_PARALLEL_WEIGHT, &thread_data[i], &thread_data[i].completion, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

2275 + }

2276 +

2277 + // Wait for all threads to complete

2278 + ssize_t total_added = 0;

2279 + for (size_t i = 0; i < num_threads; i++) {

2280 + completion_wait_for(&thread_data[i].completion);

2281 + completion_destroy(&thread_data[i].completion);

2282 +

2283 + // Merge results from this thread

2284 + merge_results_dictionaries(qwd->results, thread_data[i].local_results);

2285 + merge_weights_stats(&qwd->stats, &thread_data[i].local_stats);

2286 +

2287 + // Accumulate examined dimensions

2288 + __atomic_fetch_add(&qwd->examined_dimensions, thread_data[i].local_examined_dimensions, __ATOMIC_RELAXED);

2289 +

2290 + // Merge version hashes

2291 + qwd->versions.contexts_hard_hash += thread_data[i].local_versions.contexts_hard_hash;

2292 + qwd->versions.contexts_soft_hash += thread_data[i].local_versions.contexts_soft_hash;

2293 + qwd->versions.alerts_hard_hash += thread_data[i].local_versions.alerts_hard_hash;

2294 + qwd->versions.alerts_soft_hash += thread_data[i].local_versions.alerts_soft_hash;

2295 +

2296 + // Clean up thread data

2297 + register_result_destroy(thread_data[i].local_results);

2298 + }

2299 +

2300 + total_added = (ssize_t) dictionary_entries(qwd->results);

2301 +

2302 + // Cleanup

2303 + freez(thread_data);

2304 + freez(threads);

2305 + freez(qwd->hosts_array);

2306 +

2307 + return total_added;

2308 +}

2309 +

2310 +static ssize_t weights_do_node_callback(void *data, RRDHOST *host, bool queryable) {

2311 if(!queryable)

2312 return 0;

2313

@@ -2035,6 +2358,7 @@ int web_api_v12_weights(BUFFER *wb, QUERY_WEIGHTS_REQUEST *qwr) {

2358 .results = register_result_init(),

2359 .stats = {},

2360 .shifts = 0,

2361 + .total_workload = {0}, // Initialize workload statistics

2362 .timings = {

2363 .received_ut = now_monotonic_usec(),

2364 }

@@ -2137,10 +2461,7 @@ int web_api_v12_weights(BUFFER *wb, QUERY_WEIGHTS_REQUEST *qwr) {

2461 rrdset_weights_multi_dimensional_value(&qwd);

2462 }

2463 else {

2140 - query_scope_foreach_host(qwd.scope_nodes_sp, qwd.nodes_sp,

2141 - weights_do_node_callback, &qwd,

2142 - &qwd.versions,

2143 - NULL);

2464 + query_scope_foreach_host_parallel(qwd.scope_nodes_sp, qwd.nodes_sp, &qwd);

2465 }

2466 }

2467

@@ -2149,13 +2470,13 @@ int web_api_v12_weights(BUFFER *wb, QUERY_WEIGHTS_REQUEST *qwr) {

2470 qwr->options |= RRDR_OPTION_NONZERO;

2471 }

2472

2152 - if(qwd.timed_out) {

2473 + if(__atomic_load_n(&qwd.timed_out, __ATOMIC_RELAXED)) {

2474 error = "timed out";

2475 resp = HTTP_RESP_GATEWAY_TIMEOUT;

2476 goto cleanup;

2477 }

2478

2158 - if(qwd.interrupted) {

2479 + if(__atomic_load_n(&qwd.interrupted, __ATOMIC_RELAXED)) {

2480 error = "interrupted";

2481 resp = HTTP_RESP_CLIENT_CLOSED_REQUEST;

2482 goto cleanup;

src/web/api/queries/weights.h

@@ -67,6 +67,7 @@ typedef struct query_weights_request {

67 int web_api_v12_weights(BUFFER *wb, QUERY_WEIGHTS_REQUEST *qwr);

69 WEIGHTS_METHOD weights_string_to_method(const char *method);

70 +void query_weights_worker_thread(void *arg);

71 const char *weights_method_to_string(WEIGHTS_METHOD method);

72 int mc_unittest(void);