From 3e448e7139e6f00806606454f71fd67436d41a8d Mon Sep 17 00:00:00 2001 From: hxy7yx <1595670487@qq.com> Date: Wed, 21 May 2025 19:54:23 -0700 Subject: [PATCH] datalayers:remove doget in excute --- plugins/datalayers/datalayers_handle.c | 22 +++++++------- plugins/datalayers/datalayers_plugin.h | 4 +-- plugins/datalayers/datalayers_plugin_intf.c | 30 +++++++------------- src/persist/datalayers/flight_sql_client.cpp | 4 +-- 4 files changed, 26 insertions(+), 34 deletions(-) diff --git a/plugins/datalayers/datalayers_handle.c b/plugins/datalayers/datalayers_handle.c index 65228252..a37f7b6b 100644 --- a/plugins/datalayers/datalayers_handle.c +++ b/plugins/datalayers/datalayers_handle.c @@ -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; } diff --git a/plugins/datalayers/datalayers_plugin.h b/plugins/datalayers/datalayers_plugin.h index 693ab5b6..a376b48c 100644 --- a/plugins/datalayers/datalayers_plugin.h +++ b/plugins/datalayers/datalayers_plugin.h @@ -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); diff --git a/plugins/datalayers/datalayers_plugin_intf.c b/plugins/datalayers/datalayers_plugin_intf.c index 7e63524b..7e41ac5b 100644 --- a/plugins/datalayers/datalayers_plugin_intf.c +++ b/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_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); diff --git a/src/persist/datalayers/flight_sql_client.cpp b/src/persist/datalayers/flight_sql_client.cpp index ece28568..9aea25c5 100644 --- a/src/persist/datalayers/flight_sql_client.cpp +++ b/src/persist/datalayers/flight_sql_client.cpp @@ -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> @@ -104,7 +104,7 @@ arrow::Status Client::Execute(const std::string &sql) if (!table_result.ok()) { return table_result.status(); } - } + }*/ return arrow::Status::OK(); }