From da7623fed1e7ddb88c4c86f7128e3904190d44af Mon Sep 17 00:00:00 2001 From: hxy7yx <1595670487@qq.com> Date: Tue, 20 May 2025 01:41:53 -0700 Subject: [PATCH] fix(datalayers):exit gracefully --- plugins/datalayers/datalayers_handle.c | 83 ++++++++++++++++----- plugins/datalayers/datalayers_plugin.h | 2 + plugins/datalayers/datalayers_plugin_intf.c | 34 +++++++-- 3 files changed, 97 insertions(+), 22 deletions(-) diff --git a/plugins/datalayers/datalayers_handle.c b/plugins/datalayers/datalayers_handle.c index 7319e3a9..a11ec582 100644 --- a/plugins/datalayers/datalayers_handle.c +++ b/plugins/datalayers/datalayers_handle.c @@ -48,15 +48,22 @@ static inline db_write_task_t *task_new() static inline void task_free(db_write_task_t *task) { - if (task) { - utarray_free(task->int_tags); - utarray_free(task->float_tags); - utarray_free(task->bool_tags); - utarray_free(task->string_tags); - free(task); + if (task == NULL) { + return; } -} + if (task->freed) { + return; + } + + task->freed = true; + + utarray_free(task->int_tags); + utarray_free(task->float_tags); + utarray_free(task->bool_tags); + utarray_free(task->string_tags); + free(task); +} void tasks_free(task_queue_t *queue) { db_write_task_t *task = queue->head; @@ -99,7 +106,7 @@ void task_queue_push(neu_plugin_t *plugin, db_write_task_t *task) db_write_task_t *task_queue_pop(neu_plugin_t *plugin, task_queue_t *queue) { - if (queue->size == 0) { + if (queue == NULL || queue->head == NULL || queue->size == 0) { return NULL; } @@ -118,38 +125,59 @@ db_write_task_t *task_queue_pop(neu_plugin_t *plugin, task_queue_t *queue) static void db_write_task_cb(db_write_task_t *task, neu_plugin_t *plugin) { - int ret = 0; + int ret; + if (utarray_len(task->int_tags) > 0) { ret = client_insert(plugin->client, INT_TYPE, utarray_eltptr(task->int_tags, 0), utarray_len(task->int_tags)); + if (ret != 0) { + plog_error(plugin, "Failed to insert INT_TYPE data, disconnected"); + goto handle_error; + } } if (utarray_len(task->float_tags) > 0) { ret = client_insert(plugin->client, FLOAT_TYPE, utarray_eltptr(task->float_tags, 0), utarray_len(task->float_tags)); + if (ret != 0) { + plog_error(plugin, + "Failed to insert FLOAT_TYPE data, disconnected"); + goto handle_error; + } } if (utarray_len(task->bool_tags) > 0) { ret = client_insert(plugin->client, BOOL_TYPE, utarray_eltptr(task->bool_tags, 0), utarray_len(task->bool_tags)); + if (ret != 0) { + plog_error(plugin, "Failed to insert BOOL_TYPE data, disconnected"); + goto handle_error; + } } if (utarray_len(task->string_tags) > 0) { ret = client_insert(plugin->client, STRING_TYPE, utarray_eltptr(task->string_tags, 0), utarray_len(task->string_tags)); - } - - if (ret != 0) { - plog_error(plugin, "insert data to datalayers failed, disconnected"); - if (plugin->client) { - client_destroy(plugin->client); - plugin->client = NULL; + if (ret != 0) { + plog_error(plugin, + "Failed to insert STRING_TYPE data, disconnected"); + goto handle_error; } } + + return; + +handle_error: + pthread_mutex_lock(&plugin->plugin_mutex); + if (plugin->client) { + client_destroy(plugin->client); + plugin->client = NULL; + } + pthread_mutex_unlock(&plugin->plugin_mutex); } void db_write_task_consumer(neu_plugin_t *plugin) @@ -157,15 +185,22 @@ void db_write_task_consumer(neu_plugin_t *plugin) db_write_task_t *task = NULL; while (1) { + pthread_mutex_lock(&plugin->plugin_mutex); + if (plugin->consumer_thread_stop_flag) { + pthread_mutex_unlock(&plugin->plugin_mutex); + break; + } + pthread_mutex_unlock(&plugin->plugin_mutex); + task = task_queue_pop(plugin, &plugin->task_queue); if (task) { db_write_task_cb(task, plugin); - task_free(task); } else { usleep(1000); } } + plog_notice(plugin, "Consumer thread exiting gracefully.\n"); } static void tag_array_copy(void *dst, const void *src) @@ -369,12 +404,16 @@ int handle_trans_data(neu_plugin_t * plugin, utarray_new(bool_tags, &ut_datatag_icd); utarray_new(string_tags, &ut_datatag_icd); + bool has_valid_tags = false; + utarray_foreach(trans_data->tags, neu_resp_tag_value_meta_t *, tag_meta) { if (tag_meta->value.type == NEU_TYPE_ERROR) { continue; } + has_valid_tags = true; + switch (tag_meta->value.type) { case NEU_TYPE_BIT: case NEU_TYPE_INT8: @@ -440,6 +479,16 @@ int handle_trans_data(neu_plugin_t * plugin, } } + if (!has_valid_tags) { + utarray_free(int_tags); + utarray_free(float_tags); + utarray_free(bool_tags); + utarray_free(string_tags); + + pthread_mutex_unlock(&plugin->plugin_mutex); + return rv; + } + db_write_task_t *task = task_new(); task->int_tags = int_tags; task->float_tags = float_tags; diff --git a/plugins/datalayers/datalayers_plugin.h b/plugins/datalayers/datalayers_plugin.h index 6cf53908..693ab5b6 100644 --- a/plugins/datalayers/datalayers_plugin.h +++ b/plugins/datalayers/datalayers_plugin.h @@ -49,6 +49,7 @@ typedef struct db_write_task_s { UT_array * bool_tags; UT_array * string_tags; struct db_write_task_s *next; + bool freed; } db_write_task_t; typedef struct { @@ -68,6 +69,7 @@ struct neu_plugin { pthread_t consumer_thread; pthread_mutex_t plugin_mutex; + bool consumer_thread_stop_flag; int (*parse_config)(neu_plugin_t *plugin, const char *setting, datalayers_config_t *config); diff --git a/plugins/datalayers/datalayers_plugin_intf.c b/plugins/datalayers/datalayers_plugin_intf.c index e82dea85..af5e27dc 100644 --- a/plugins/datalayers/datalayers_plugin_intf.c +++ b/plugins/datalayers/datalayers_plugin_intf.c @@ -31,6 +31,19 @@ extern const neu_plugin_module_t neu_plugin_module; +void create_consumer_thread(neu_plugin_t *plugin) +{ + pthread_mutex_lock(&plugin->plugin_mutex); + plugin->consumer_thread_stop_flag = false; + pthread_mutex_unlock(&plugin->plugin_mutex); + + int status = pthread_create(&plugin->consumer_thread, NULL, + (void *) db_write_task_consumer, plugin); + if (status != 0) { + plog_notice(plugin, "Failed to create consumer thread: %d\n", status); + } +} + neu_plugin_t *datalayers_plugin_open(void) { neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t)); @@ -39,10 +52,17 @@ neu_plugin_t *datalayers_plugin_open(void) return plugin; } -void stop_consumer_thread(pthread_t *consumer_thread) +void stop_consumer_thread(pthread_t *consumer_thread, neu_plugin_t *plugin) { - pthread_cancel(*consumer_thread); - pthread_join(*consumer_thread, NULL); + pthread_mutex_lock(&plugin->plugin_mutex); + plugin->consumer_thread_stop_flag = true; + pthread_mutex_unlock(&plugin->plugin_mutex); + + int join_status = pthread_join(*consumer_thread, NULL); + if (join_status != 0) { + plog_notice(plugin, "Failed to join consumer thread: %d\n", + join_status); + } } int datalayers_plugin_close(neu_plugin_t *plugin) @@ -53,7 +73,7 @@ int datalayers_plugin_close(neu_plugin_t *plugin) plog_notice(plugin, "success to free plugin:%s", name); - stop_consumer_thread(&plugin->consumer_thread); + stop_consumer_thread(&plugin->consumer_thread, plugin); tasks_free(&plugin->task_queue); @@ -90,7 +110,7 @@ int datalayers_plugin_uninit(neu_plugin_t *plugin) { pthread_mutex_lock(&plugin->plugin_mutex); - stop_consumer_thread(&plugin->consumer_thread); + stop_consumer_thread(&plugin->consumer_thread, plugin); tasks_free(&plugin->task_queue); datalayers_config_fini(&plugin->config); @@ -136,7 +156,11 @@ int datalayers_plugin_config(neu_plugin_t *plugin, const char *setting) const char * plugin_name = neu_plugin_module.module_name; datalayers_config_t config = { 0 }; if (plugin->client != NULL) { + stop_consumer_thread(&plugin->consumer_thread, plugin); + tasks_free(&plugin->task_queue); client_destroy(plugin->client); + plugin->client = NULL; + create_consumer_thread(plugin); } rv = plugin->parse_config(plugin, setting, &config);