From da3746d1350650fefbe0794151b2f341ebfc3310 Mon Sep 17 00:00:00 2001 From: gc87 Date: Thu, 28 Oct 2021 10:14:07 +0800 Subject: [PATCH] Fix the timing problem of request context - Thread lock handling for running status flags --- plugins/mqtt/command/command_rw.c | 18 ++++++++---------- plugins/mqtt/command/command_rw.h | 4 ++-- plugins/mqtt/mqttc_client.c | 18 +++++++++++++++++- plugins/mqtt/plugin_mqtt.c | 8 ++++++-- 4 files changed, 33 insertions(+), 15 deletions(-) diff --git a/plugins/mqtt/command/command_rw.c b/plugins/mqtt/command/command_rw.c index 85c4079d..a129915e 100644 --- a/plugins/mqtt/command/command_rw.c +++ b/plugins/mqtt/command/command_rw.c @@ -24,7 +24,7 @@ #include "command_rw.h" int command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, - neu_parse_read_req_t *req) + neu_parse_read_req_t *req, uint32_t req_id) { log_info("READ uuid:%s, node id:%u", mqtt->uuid, req->node_id); int rc = common_has_node(plugin, req->node_id); @@ -40,9 +40,8 @@ int command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, return -2; } - uint32_t req_id = neu_plugin_get_event_id(plugin); neu_plugin_send_read_cmd(plugin, req_id, req->node_id, config); - return req_id; + return 0; } static int wrap_read_response_json_object(neu_fixed_array_t * array, @@ -230,7 +229,7 @@ char *command_read_cycle_response(neu_plugin_t * plugin, static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id, const char * group_config_name, - neu_data_val_t *write_val) + neu_data_val_t *write_val, uint32_t req_id) { int rc = common_has_node(plugin, dest_node_id); if (0 != rc) { @@ -244,13 +243,12 @@ static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id, return -2; } - uint32_t req_id = neu_plugin_get_event_id(plugin); neu_plugin_send_write_cmd(plugin, req_id, dest_node_id, config, write_val); - return req_id; + return 0; } int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, - neu_parse_write_req_t *write_req) + neu_parse_write_req_t *write_req, uint32_t req_id) { log_info("WRITE uuid:%s, group config name:%s", mqtt->uuid, write_req->group_config_name); @@ -312,10 +310,10 @@ int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, } neu_dvalue_init_move_array(write_val, NEU_DTYPE_INT_VAL, array); - int req_id = write_command(plugin, write_req->node_id, - write_req->group_config_name, write_val); + write_command(plugin, write_req->node_id, write_req->group_config_name, + write_val, req_id); neu_dvalue_free(write_val); - return req_id; + return 0; } static int wrap_write_response_json_object(neu_fixed_array_t * array, diff --git a/plugins/mqtt/command/command_rw.h b/plugins/mqtt/command/command_rw.h index 3c5e2bd2..e718dd22 100644 --- a/plugins/mqtt/command/command_rw.h +++ b/plugins/mqtt/command/command_rw.h @@ -29,14 +29,14 @@ extern "C" { #include "common.h" int command_read_once_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, - neu_parse_read_req_t *req); + neu_parse_read_req_t *req, uint32_t req_id); char *command_read_once_response(neu_plugin_t * plugin, neu_parse_mqtt_t *parse_header, neu_data_val_t * resp_val); char *command_read_cycle_response(neu_plugin_t * plugin, neu_data_val_t *resp_val); int command_write_request(neu_plugin_t *plugin, neu_parse_mqtt_t *mqtt, - neu_parse_write_req_t *write_req); + neu_parse_write_req_t *write_req, uint32_t req_id); char *command_write_response(neu_plugin_t * plugin, neu_parse_mqtt_t *parse_header, neu_data_val_t * resp_val); diff --git a/plugins/mqtt/mqttc_client.c b/plugins/mqtt/mqttc_client.c index e3bf1591..0dc5a3ca 100644 --- a/plugins/mqtt/mqttc_client.c +++ b/plugins/mqtt/mqttc_client.c @@ -54,6 +54,7 @@ struct mqttc_client { uint8_t send_buf[SEND_BUF_SIZE]; uint8_t recv_buf[RECV_BUF_SIZE]; pthread_t client_daemon; + pthread_mutex_t mutex; void * user_data; }; @@ -207,7 +208,17 @@ static void *mqtt_refresher(void *context) { mqttc_client_t * client = (mqttc_client_t *) context; struct mqtt_client *mqtt = &client->mqtt; - while (client->running) { + + bool run_flag = true; + while (1) { + pthread_mutex_lock(&client->mutex); + run_flag = client->running; + pthread_mutex_unlock(&client->mutex); + + if (!run_flag) { + break; + } + mqtt_sync((struct mqtt_client *) mqtt); usleep(INTERVAL); } @@ -224,6 +235,7 @@ mqttc_client_t *mqttc_client_create(option_t *option, void *context) } memset(client, 0, sizeof(mqttc_client_t)); + pthread_mutex_init(&client->mutex, NULL); client->option = option; client->running = true; client->user_data = context; @@ -354,6 +366,7 @@ client_error_e mqttc_client_disconnect(mqttc_client_t *client) client_error_e mqttc_client_destroy(mqttc_client_t *client) { + pthread_mutex_destroy(&client->mutex); free(client); return ClientSuccess; } @@ -492,7 +505,10 @@ client_error_e mqttc_client_close(mqttc_client_t *client) return ClientIsNULL; } + pthread_mutex_lock(&client->mutex); client->running = false; + pthread_mutex_unlock(&client->mutex); + pthread_join(client->client_daemon, NULL); client_error_e rc = mqttc_client_disconnect(client); rc = mqttc_client_destroy(client); diff --git a/plugins/mqtt/plugin_mqtt.c b/plugins/mqtt/plugin_mqtt.c index fff1e5b5..852a0268 100644 --- a/plugins/mqtt/plugin_mqtt.c +++ b/plugins/mqtt/plugin_mqtt.c @@ -198,10 +198,12 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, neu_parse_read_req_t *req = NULL; rc = neu_parse_decode_read(json_str, &req); if (0 == rc) { - int req_id = command_read_once_request(plugin, mqtt, req); + uint32_t req_id = neu_plugin_get_event_id(plugin); if (0 < req_id) { context_list_add(&plugin->context_list, req_id, mqtt); } + + command_read_once_request(plugin, mqtt, req, req_id); neu_parse_decode_read_free(req); } break; @@ -210,10 +212,12 @@ static void plugin_response_handle(const char *topic_name, size_t topic_len, neu_parse_write_req_t *req = NULL; rc = neu_parse_decode_write(json_str, &req); if (0 == rc) { - int req_id = command_write_request(plugin, mqtt, req); + uint32_t req_id = neu_plugin_get_event_id(plugin); if (0 < req_id) { context_list_add(&plugin->context_list, req_id, mqtt); } + + command_write_request(plugin, mqtt, req, req_id); neu_parse_decode_write_free(req); } break;