Add check for spurious wakeups (#9751)

7 files changed, 17 insertions, 4 deletions

diff --git a/exporting/aws_kinesis/aws_kinesis.c b/exporting/aws_kinesis/aws_kinesis.c
index 791cad436c..00600fa9bb 100644
--- a/exporting/aws_kinesis/aws_kinesis.c
+++ b/exporting/aws_kinesis/aws_kinesis.c
@@ -103,7 +103,10 @@ void aws_kinesis_connector_worker(void *instance_p)
         struct stats *stats = &instance->stats;
 
         uv_mutex_lock(&instance->mutex);
-        uv_cond_wait(&instance->cond_var, &instance->mutex);
+        while (!instance->data_is_ready)
+            uv_cond_wait(&instance->cond_var, &instance->mutex);
+        instance->data_is_ready = 0;
+
         if (unlikely(instance->engine->exit)) {
             uv_mutex_unlock(&instance->mutex);
             break;
diff --git a/exporting/exporting_engine.c b/exporting/exporting_engine.c
index c617a97b26..6a1320cd11 100644
--- a/exporting/exporting_engine.c
+++ b/exporting/exporting_engine.c
@@ -67,6 +67,7 @@ static void exporting_main_cleanup(void *ptr)
             found++;
             info("stopping worker for instance %s", instance->config.name);
             uv_mutex_unlock(&instance->mutex);
+            instance->data_is_ready = 1;
             uv_cond_signal(&instance->cond_var);
         } else
             info("found stopped worker for instance %s", instance->config.name);
diff --git a/exporting/exporting_engine.h b/exporting/exporting_engine.h
index 12fdb766c1..b436f5d809 100644
--- a/exporting/exporting_engine.h
+++ b/exporting/exporting_engine.h
@@ -165,6 +165,7 @@ struct instance {
     uv_thread_t thread;
     uv_mutex_t mutex;
     uv_cond_t cond_var;
+    int data_is_ready;
 
     int (*start_batch_formatting)(struct instance *instance);
     int (*start_host_formatting)(struct instance *instance, RRDHOST *host);
diff --git a/exporting/mongodb/mongodb.c b/exporting/mongodb/mongodb.c
index 38e1d01378..c2cb72f9e3 100644
--- a/exporting/mongodb/mongodb.c
+++ b/exporting/mongodb/mongodb.c
@@ -284,7 +284,9 @@ void mongodb_connector_worker(void *instance_p)
         struct stats *stats = &instance->stats;
 
         uv_mutex_lock(&instance->mutex);
-        uv_cond_wait(&instance->cond_var, &instance->mutex);
+        while (!instance->data_is_ready)
+            uv_cond_wait(&instance->cond_var, &instance->mutex);
+        instance->data_is_ready = 0;
 
         if (unlikely(instance->engine->exit)) {
             uv_mutex_unlock(&instance->mutex);
diff --git a/exporting/process_data.c b/exporting/process_data.c
index a418e1b0da..00087a22b3 100644
--- a/exporting/process_data.c
+++ b/exporting/process_data.c
@@ -294,6 +294,7 @@ void end_batch_formatting(struct engine *engine)
                 continue;
             }
             uv_mutex_unlock(&instance->mutex);
+            instance->data_is_ready = 1;
             uv_cond_signal(&instance->cond_var);
 
             instance->scheduled = 0;
diff --git a/exporting/pubsub/pubsub.c b/exporting/pubsub/pubsub.c
index 95a87422a6..ead44ed2ac 100644
--- a/exporting/pubsub/pubsub.c
+++ b/exporting/pubsub/pubsub.c
@@ -103,7 +103,10 @@ void pubsub_connector_worker(void *instance_p)
         char error_message[ERROR_LINE_MAX + 1] = "";
 
         uv_mutex_lock(&instance->mutex);
-        uv_cond_wait(&instance->cond_var, &instance->mutex);
+        while (!instance->data_is_ready)
+            uv_cond_wait(&instance->cond_var, &instance->mutex);
+        instance->data_is_ready = 0;
+
 
         if (unlikely(instance->engine->exit)) {
             uv_mutex_unlock(&instance->mutex);
diff --git a/exporting/send_data.c b/exporting/send_data.c
index e7fda72331..cc107ea2dd 100644
--- a/exporting/send_data.c
+++ b/exporting/send_data.c
@@ -340,7 +340,9 @@ void simple_connector_worker(void *instance_p)
         // if we are connected, send our buffer to the data collecting server
 
         uv_mutex_lock(&instance->mutex);
-        uv_cond_wait(&instance->cond_var, &instance->mutex);
+        while (!instance->data_is_ready)
+            uv_cond_wait(&instance->cond_var, &instance->mutex);
+        instance->data_is_ready = 0;
 
         if (unlikely(instance->engine->exit)) {
             uv_mutex_unlock(&instance->mutex);