Do not run queries synchronously in the event loop (#19448)
Do not execute aclk commands (http api calls in the event loop)
Stelios Fragkakis committed
Jan 21, 2025 at 01:31 UTC
4033840c20307b1e53ca77ebd71757db6956508c
1 file changed
+89
-17
src/database/sqlite/sqlite_aclk.c
+89
-17
@@ -587,6 +587,26 @@ static void start_alert_push(uv_work_t *req __maybe_unused)
587
588
#define MAX_BATCH_SIZE (64)
589
590
+// Take a query, and try to schedule it in a worker
591
+// Update config->aclk_queries_running if success
592
+// config->aclk_queries_running is only accessed from the vent loop
593
+// On failure: free the payload
594
+
595
+int schedule_query_in_worker(uv_loop_t *loop, struct aclk_sync_config_s *config, aclk_query_t query) {
596
+ struct aclk_query_payload *payload = mallocz(sizeof(*payload));
597
+ payload->request.data = payload;
598
+ payload->config = config;
599
+ payload->data = query;
600
+ config->aclk_queries_running++;
601
+ int rc = uv_queue_work(loop, &payload->request, aclk_run_query_job, after_aclk_run_query_job);
602
+ if (rc) {
603
+ config->aclk_queries_running--;
604
+ freez(payload);
605
+ }
606
+ return rc;
607
+}
608
+
609
+
610
static void aclk_synchronization(void *arg)
611
{
612
struct aclk_sync_config_s *config = arg;
@@ -624,6 +644,9 @@ static void aclk_synchronization(void *arg)
644
struct alert_push_data *data;
645
aclk_query_t query;
646
struct judy_list_t *aclk_query_batch = NULL;
647
+ struct judy_list_t *aclk_query_execute = callocz(1, sizeof(*aclk_query_execute));;
648
+ size_t pending_queries = 0;
649
+
650
Pvoid_t *Pvalue;
651
struct aclk_query_payload *payload;
652
@@ -650,11 +673,17 @@ static void aclk_synchronization(void *arg)
673
if(likely(opcode != ACLK_DATABASE_NOOP && opcode != ACLK_QUERY_EXECUTE))
674
worker_is_busy(opcode);
675
676
+ // Check if we have pending commands to execute
677
+ if (opcode == ACLK_DATABASE_NOOP && pending_queries && config->aclk_queries_running < query_thread_count) {
678
+ opcode = ACLK_QUERY_EXECUTE;
679
+ cmd.param[0] = NULL;
680
+ }
681
+
682
switch (opcode) {
683
case ACLK_DATABASE_NOOP:
684
/* the command queue was empty, do nothing */
685
break;
657
-// NODE STATE
686
+ // NODE STATE
687
case ACLK_DATABASE_NODE_STATE:;
688
RRDHOST *host = cmd.param[0];
689
struct aclk_sync_cfg_t *ahc = host->aclk_config;
@@ -712,28 +741,71 @@ static void aclk_synchronization(void *arg)
741
}
742
break;
743
case ACLK_MQTT_WSS_CLIENT:
715
- config->client = (mqtt_wss_client) cmd.param[0];
744
+ config->client = (mqtt_wss_client)cmd.param[0];
745
break;
746
747
case ACLK_QUERY_EXECUTE:
748
query = (aclk_query_t)cmd.param[0];
720
- payload = NULL;
721
- config->aclk_queries_running++;
722
- bool execute_now = (config->aclk_queries_running > query_thread_count);
723
- if (!execute_now) {
724
- payload = mallocz(sizeof(*payload));
725
- payload->request.data = payload;
726
- payload->config = config;
727
- payload->data = query;
728
- execute_now = uv_queue_work(loop, &payload->request, aclk_run_query_job, after_aclk_run_query_job);
749
+
750
+ bool too_busy = (config->aclk_queries_running >= query_thread_count);
751
+
752
+ // If we are busy and it's just a ping to run, leave
753
+ if (too_busy && !query)
754
+ break;
755
+
756
+ // if we are busy (we have a query) store it and leave
757
+ if (too_busy) {
758
+ Pvalue = JudyLIns(&aclk_query_execute->JudyL, ++aclk_query_execute->count, PJE0);
759
+ if (Pvalue != PJERR) {
760
+ *Pvalue = query;
761
+ pending_queries++;
762
+ } else
763
+ nd_log_daemon(NDLP_ERR, "Failed to add ACLK command to the pending commands Judy");
764
+ break;
765
}
766
731
- if (execute_now) {
732
- worker_is_busy(ACLK_QUERY_EXECUTE_SYNC);
733
- aclk_run_query(config, query, false);
734
- freez(payload);
735
- config->aclk_queries_running--;
736
- cmd_batch_size = MAX_BATCH_SIZE;
767
+ // Here: we are not busy
768
+ // If we have query it was a normal incoming command
769
+ // if we dont, it was a ping from the callback
770
+
771
+ // Lets try to queue as many of the pending commands
772
+ while(!too_busy && pending_queries && config->aclk_queries_running < query_thread_count) {
773
+
774
+ Word_t Index = 0;
775
+ Pvalue = JudyLFirst(aclk_query_execute->JudyL, &Index, PJE0);
776
+
777
+ // We have nothing, leave
778
+ if (Pvalue == NULL)
779
+ break;
780
+ aclk_query_t query_in_queue = *Pvalue;
781
+
782
+ // Schedule it and increase running
783
+ too_busy = schedule_query_in_worker(loop, config, query_in_queue);
784
+
785
+ // It was scheduled in worker, remove it from pending
786
+ if (!too_busy) {
787
+ pending_queries--;
788
+ (void)JudyLDel(&aclk_query_execute->JudyL, Index, PJE0);
789
+ }
790
+ }
791
+
792
+ // Was it just a ping to run? leave
793
+ if (!query)
794
+ break;
795
+
796
+ // We have a query, if not busy lets run it
797
+ if (!too_busy)
798
+ too_busy = schedule_query_in_worker(loop, config, query);
799
+
800
+ // We were either busy, or failed to start worker, schedule for later
801
+ if (too_busy) {
802
+ Pvalue = JudyLIns(&aclk_query_execute->JudyL, ++aclk_query_execute->count, PJE0);
803
+ if (Pvalue != PJERR) {
804
+ *Pvalue = query;
805
+ pending_queries++;
806
+ }
807
+ else
808
+ nd_log_daemon(NDLP_ERR, "Failed to add ACLK command to the pending commands Judy");
809
}
810
break;
811