GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(datalayers):exit gracefully

Browse Source
This commit is contained in:
hxy7yx
2025-05-20 01:41:53 -07:00
parent 853cceb63c
commit da7623fed1
3 changed files with 97 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/datalayers/datalayers_handle.c
+66 -17
View File
@@ -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;
plugins/datalayers/datalayers_plugin.h
+2
View File
@@ -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);
plugins/datalayers/datalayers_plugin_intf.c
+29 -5
View File
@@ -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);