fix(datalayers):queue

plugins/datalayers/datalayers_handle.c

@@ -185,21 +185,27 @@ void db_write_task_consumer(neu_plugin_t *plugin)
     db_write_task_t *task = NULL;
 
     while (1) {
-        pthread_rwlock_rdlock(&plugin->plugin_mutex);
-        if (plugin->consumer_thread_stop_flag) {
-            pthread_rwlock_unlock(&plugin->plugin_mutex);
+        pthread_mutex_lock(&plugin->queue_mutex);
+
+        while (plugin->task_queue.size == 0 &&
+               !plugin->consumer_thread_stop_flag) {
+            pthread_cond_wait(&plugin->queue_not_empty, &plugin->queue_mutex);
+        }
+
+        if (plugin->consumer_thread_stop_flag && plugin->task_queue.size == 0) {
+            pthread_mutex_unlock(&plugin->queue_mutex);
             break;
         }
-        pthread_rwlock_unlock(&plugin->plugin_mutex);
 
         task = task_queue_pop(plugin, &plugin->task_queue);
+        pthread_mutex_unlock(&plugin->queue_mutex);
+
         if (task) {
             db_write_task_cb(task, plugin);
             task_free(task);
-        } else {
-            usleep(1000);
         }
     }
+
     plog_notice(plugin, "Consumer thread exiting gracefully.\n");
 }
 
@@ -558,10 +564,13 @@ int handle_trans_data(neu_plugin_t *            plugin,
     task->bool_tags       = bool_tags;
     task->string_tags     = string_tags;
 
-    task_queue_push(plugin, task);
-
     pthread_rwlock_unlock(&plugin->plugin_mutex);
 
+    pthread_mutex_lock(&plugin->queue_mutex);
+    task_queue_push(plugin, task);
+    pthread_cond_signal(&plugin->queue_not_empty);
+    pthread_mutex_unlock(&plugin->queue_mutex);
+
     return rv;
 }
 

plugins/datalayers/datalayers_plugin.h

@@ -68,6 +68,9 @@ struct neu_plugin {
     task_queue_t task_queue;
     pthread_t    consumer_thread;
 
+    pthread_mutex_t queue_mutex;
+    pthread_cond_t  queue_not_empty;
+
     pthread_rwlock_t plugin_mutex;
     bool             consumer_thread_stop_flag;
 

plugins/datalayers/datalayers_plugin_intf.c

@@ -33,9 +33,9 @@ extern const neu_plugin_module_t neu_plugin_module;
 
 void create_consumer_thread(neu_plugin_t *plugin)
 {
-    pthread_rwlock_wrlock(&plugin->plugin_mutex);
+    pthread_mutex_lock(&plugin->queue_mutex);
     plugin->consumer_thread_stop_flag = false;
-    pthread_rwlock_unlock(&plugin->plugin_mutex);
+    pthread_mutex_unlock(&plugin->queue_mutex);
 
     int status = pthread_create(&plugin->consumer_thread, NULL,
                                 (void *) db_write_task_consumer, plugin);
@@ -54,9 +54,10 @@ neu_plugin_t *datalayers_plugin_open(void)
 
 void stop_consumer_thread(pthread_t *consumer_thread, neu_plugin_t *plugin)
 {
-    pthread_rwlock_wrlock(&plugin->plugin_mutex);
+    pthread_mutex_lock(&plugin->queue_mutex);
     plugin->consumer_thread_stop_flag = true;
-    pthread_rwlock_unlock(&plugin->plugin_mutex);
+    pthread_cond_broadcast(&plugin->queue_not_empty);
+    pthread_mutex_unlock(&plugin->queue_mutex);
 
     int join_status = pthread_join(*consumer_thread, NULL);
     if (join_status != 0) {
@@ -81,11 +82,17 @@ int datalayers_plugin_init(neu_plugin_t *plugin, bool load)
 
     pthread_rwlock_init(&plugin->plugin_mutex, NULL);
 
+    pthread_mutex_init(&plugin->queue_mutex, NULL);
+    pthread_cond_init(&plugin->queue_not_empty, NULL);
+
+    pthread_mutex_lock(&plugin->queue_mutex);
+    plugin->consumer_thread_stop_flag = false;
+    pthread_mutex_unlock(&plugin->queue_mutex);
+
     task_queue_init(&plugin->task_queue);
 
     pthread_create(&plugin->consumer_thread, NULL,
                    (void *) db_write_task_consumer, plugin);
-    pthread_detach(plugin->consumer_thread);
 
     NEU_PLUGIN_REGISTER_CACHED_QUEUE_SIZE_METRIC(plugin);
     NEU_PLUGIN_REGISTER_MAX_CACHED_QUEUE_SIZE_METRIC(plugin);
@@ -101,7 +108,11 @@ int datalayers_plugin_uninit(neu_plugin_t *plugin)
     pthread_rwlock_wrlock(&plugin->plugin_mutex);
 
     stop_consumer_thread(&plugin->consumer_thread, plugin);
+    pthread_mutex_lock(&plugin->queue_mutex);
     tasks_free(&plugin->task_queue);
+    pthread_mutex_unlock(&plugin->queue_mutex);
+    pthread_mutex_destroy(&plugin->queue_mutex);
+    pthread_cond_destroy(&plugin->queue_not_empty);
     datalayers_config_fini(&plugin->config);
 
     route_tbl_free(plugin->route_tbl);
@@ -149,7 +160,9 @@ int datalayers_plugin_config(neu_plugin_t *plugin, const char *setting)
     datalayers_config_t config      = { 0 };
     if (plugin->client != NULL) {
         stop_consumer_thread(&plugin->consumer_thread, plugin);
+        pthread_mutex_lock(&plugin->queue_mutex);
         tasks_free(&plugin->task_queue);
+        pthread_mutex_unlock(&plugin->queue_mutex);
         client_destroy(plugin->client);
         plugin->client = NULL;
         create_consumer_thread(plugin);