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

Merge pull request #2457 from hxy7yx/main-1

Browse Source 
datalayers:remove doget in excute
This commit is contained in:
hxy
2025-05-22 11:02:06 +08:00
committed by GitHub
parent 5ea8f5531d 3e448e7139
commit 562c74307e
4 changed files with 26 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/datalayers/datalayers_handle.c
+11 -11
View File
@@ -172,12 +172,12 @@ static void db_write_task_cb(db_write_task_t *task, neu_plugin_t *plugin)
return;
handle_error:
pthread_mutex_lock(&plugin->plugin_mutex);
pthread_rwlock_wrlock(&plugin->plugin_mutex);
if (plugin->client) {
client_destroy(plugin->client);
plugin->client = NULL;
}
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
}
void db_write_task_consumer(neu_plugin_t *plugin)
@@ -185,12 +185,12 @@ void db_write_task_consumer(neu_plugin_t *plugin)
db_write_task_t *task = NULL;
while (1) {
pthread_mutex_lock(&plugin->plugin_mutex);
pthread_rwlock_rdlock(&plugin->plugin_mutex);
if (plugin->consumer_thread_stop_flag) {
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
break;
}
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
task = task_queue_pop(plugin, &plugin->task_queue);
if (task) {
@@ -376,15 +376,15 @@ int handle_trans_data(neu_plugin_t * plugin,
{
int rv = 0;
pthread_mutex_lock(&plugin->plugin_mutex);
pthread_rwlock_rdlock(&plugin->plugin_mutex);
if (NULL == plugin->client) {
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
return NEU_ERR_DATALAYERS_IS_NULL;
}
if (plugin->common.link_state != NEU_NODE_LINK_STATE_CONNECTED) {
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
return NEU_ERR_PLUGIN_NOT_RUNNING;
}
@@ -393,7 +393,7 @@ int handle_trans_data(neu_plugin_t * plugin,
if (NULL == route) {
plog_error(plugin, "no route for driver:%s group:%s",
trans_data->driver, trans_data->group);
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
return NEU_ERR_GROUP_NOT_SUBSCRIBE;
}
@@ -548,7 +548,7 @@ int handle_trans_data(neu_plugin_t * plugin,
utarray_free(bool_tags);
utarray_free(string_tags);
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
return rv;
}
@@ -560,7 +560,7 @@ int handle_trans_data(neu_plugin_t * plugin,
task_queue_push(plugin, task);
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
return rv;
}
plugins/datalayers/datalayers_plugin.h
+2 -2
View File
@@ -68,8 +68,8 @@ struct neu_plugin {
task_queue_t task_queue;
pthread_t consumer_thread;
pthread_mutex_t plugin_mutex;
bool consumer_thread_stop_flag;
pthread_rwlock_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
+11 -19
View File
@@ -33,9 +33,9 @@ extern const neu_plugin_module_t neu_plugin_module;
void create_consumer_thread(neu_plugin_t *plugin)
{
pthread_mutex_lock(&plugin->plugin_mutex);
pthread_rwlock_wrlock(&plugin->plugin_mutex);
plugin->consumer_thread_stop_flag = false;
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
int status = pthread_create(&plugin->consumer_thread, NULL,
(void *) db_write_task_consumer, plugin);
@@ -54,9 +54,9 @@ neu_plugin_t *datalayers_plugin_open(void)
void stop_consumer_thread(pthread_t *consumer_thread, neu_plugin_t *plugin)
{
pthread_mutex_lock(&plugin->plugin_mutex);
pthread_rwlock_wrlock(&plugin->plugin_mutex);
plugin->consumer_thread_stop_flag = true;
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_rwlock_unlock(&plugin->plugin_mutex);
int join_status = pthread_join(*consumer_thread, NULL);
if (join_status != 0) {
@@ -68,20 +68,10 @@ void stop_consumer_thread(pthread_t *consumer_thread, neu_plugin_t *plugin)
int datalayers_plugin_close(neu_plugin_t *plugin)
{
const char *name = neu_plugin_module.module_name;
pthread_mutex_lock(&plugin->plugin_mutex);
plog_notice(plugin, "success to free plugin:%s", name);
stop_consumer_thread(&plugin->consumer_thread, plugin);
tasks_free(&plugin->task_queue);
client_destroy(plugin->client);
free(plugin);
pthread_mutex_unlock(&plugin->plugin_mutex);
return NEU_ERR_SUCCESS;
}
@@ -89,7 +79,7 @@ int datalayers_plugin_init(neu_plugin_t *plugin, bool load)
{
(void) load;
pthread_mutex_init(&plugin->plugin_mutex, NULL);
pthread_rwlock_init(&plugin->plugin_mutex, NULL);
task_queue_init(&plugin->task_queue);
@@ -108,18 +98,20 @@ int datalayers_plugin_init(neu_plugin_t *plugin, bool load)
int datalayers_plugin_uninit(neu_plugin_t *plugin)
{
pthread_mutex_lock(&plugin->plugin_mutex);
pthread_rwlock_wrlock(&plugin->plugin_mutex);
stop_consumer_thread(&plugin->consumer_thread, plugin);
tasks_free(&plugin->task_queue);
datalayers_config_fini(&plugin->config);
route_tbl_free(plugin->route_tbl);
plugin->route_tbl = NULL;
pthread_mutex_unlock(&plugin->plugin_mutex);
pthread_mutex_destroy(&plugin->plugin_mutex);
client_destroy(plugin->client);
plugin->client = NULL;
pthread_rwlock_unlock(&plugin->plugin_mutex);
pthread_rwlock_destroy(&plugin->plugin_mutex);
plog_notice(plugin, "uninitialize plugin `%s` success",
neu_plugin_module.module_name);
src/persist/datalayers/flight_sql_client.cpp
+2 -2
View File
@@ -88,7 +88,7 @@ arrow::Status Client::Execute(const std::string &sql)
if (!flight_info_result.ok()) {
return flight_info_result.status();
}
auto flight_info = std::move(flight_info_result.ValueOrDie());
/*auto flight_info = std::move(flight_info_result.ValueOrDie());
for (const flight::FlightEndpoint &endpoint : flight_info->endpoints()) {
arrow::Result<std::unique_ptr<flight::FlightStreamReader>>
@@ -104,7 +104,7 @@ arrow::Status Client::Execute(const std::string &sql)
if (!table_result.ok()) {
return table_result.status();
}
}
}*/
return arrow::Status::OK();
}