@cryptotaxi247 / netdata-1 / commits / 58467c686

Make sure ACLK synchronization event loop runs frequently (#19446)

* Process MAX_BATCH_SIZE commands before running the event loop again * Increase command count Make sure to exit and run event loop on sync command execution

Stelios Fragkakis committed Jan 20, 2025 at 20:22 UTC 58467c686c213017a6aa001c6e959ce92b6c93fa

1 file changed +11

src/database/sqlite/sqlite_aclk.c

+11

@@ -585,6 +585,8 @@ static void start_alert_push(uv_work_t *req __maybe_unused)

585

586 #define MAX_ACLK_BATCH_JOBS_IN_QUEUE (20)

587

588 +#define MAX_BATCH_SIZE (64)

589 +

590 static void aclk_synchronization(void *arg)

591 {

592 struct aclk_sync_config_s *config = arg;

@@ -599,6 +601,8 @@ static void aclk_synchronization(void *arg)

601 worker_register_job_name(ACLK_QUERY_EXECUTE_SYNC, "aclk query execute sync");

602 worker_register_job_name(ACLK_QUERY_BATCH_EXECUTE, "aclk batch execute");

603 worker_register_job_name(ACLK_QUERY_BATCH_ADD, "aclk batch add");

604 + worker_register_job_name(ACLK_MQTT_WSS_CLIENT, "config mqtt client");

605 + worker_register_job_name(ACLK_DATABASE_NODE_UNREGISTER, "unregister node");

606

607 uv_loop_t *loop = &config->loop;

608 fatal_assert(0 == uv_loop_init(loop));

@@ -623,18 +627,24 @@ static void aclk_synchronization(void *arg)

627 Pvoid_t *Pvalue;

628 struct aclk_query_payload *payload;

629

630 + unsigned cmd_batch_size;

631 while (likely(service_running(SERVICE_ACLK))) {

632 enum aclk_database_opcode opcode;

633 worker_is_idle();

634 uv_run(loop, UV_RUN_DEFAULT);

635

636 /* wait for commands */

637 + cmd_batch_size = 0;

638 do {

639 + if (unlikely(cmd_batch_size >= MAX_BATCH_SIZE))

640 + break;

641 +

642 struct aclk_database_cmd cmd = aclk_database_deq_cmd();

643

644 if (unlikely(!service_running(SERVICE_ACLK)))

645 break;

646

647 + ++cmd_batch_size;

648 opcode = cmd.opcode;

649

650 if(likely(opcode != ACLK_DATABASE_NOOP && opcode != ACLK_QUERY_EXECUTE))

@@ -723,6 +733,7 @@ static void aclk_synchronization(void *arg)

733 aclk_run_query(config, query, false);

734 freez(payload);

735 config->aclk_queries_running--;

736 + cmd_batch_size = MAX_BATCH_SIZE;

737 }

738 break;

739