From eacfc6720a76df3ffb1afa7ced3447d80a77644e Mon Sep 17 00:00:00 2001 From: fengzero Date: Thu, 14 Apr 2022 09:52:39 +0000 Subject: [PATCH] perf(adapter): use neu_event to monitor nng-fd event --- src/adapter/adapter.c | 1000 +++++++++++++++----------------- src/adapter/adapter_internal.h | 6 + src/adapter/driver.c | 6 +- src/adapter/driver_internal.h | 2 +- 4 files changed, 484 insertions(+), 530 deletions(-) diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index bfd79c89..ea83c490 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -751,428 +751,6 @@ static uint32_t adapter_get_req_id(neu_adapter_t *adapter) return req_id; } -static void adapter_loop(void *arg) -{ - int rv; - nng_dialer dialer = { 0 }; - neu_adapter_t *adapter = (neu_adapter_t *) arg; - - const char *manager_url; - manager_url = neu_manager_get_url(adapter->manager); - rv = nng_dial(adapter->sock, manager_url, &dialer, 0); - if (rv != 0) { - neu_panic("The adapter can't dial to %s", manager_url); - } - - const char *adapter_str = adapter->name; - nng_msg * out_msg; - size_t msg_size; - msg_size = msg_inplace_data_get_size(strlen(adapter_str) + 1); - rv = nng_msg_alloc(&out_msg, msg_size); - if (rv == 0) { - message_t *msg_ptr; - char * buf_ptr; - msg_ptr = (message_t *) nng_msg_body(out_msg); - msg_inplace_data_init(msg_ptr, MSG_EVENT_NODE_PING, msg_size); - buf_ptr = msg_get_buf_ptr(msg_ptr); - memcpy(buf_ptr, adapter_str, strlen(adapter_str)); - buf_ptr[strlen(adapter_str)] = 0; - nng_sendmsg(adapter->sock, out_msg, 0); - } - - while (1) { - nng_msg *msg; - - nng_mtx_lock(adapter->mtx); - if (adapter->stop) { - adapter->state = ADAPTER_STATE_IDLE; - nng_mtx_unlock(adapter->mtx); - log_info("Exit loop of the adapter(%s)", adapter->name); - break; - } - nng_mtx_unlock(adapter->mtx); - - rv = nng_recvmsg(adapter->sock, &msg, 0); - if (rv != 0) { - if (rv != NNG_ETIMEDOUT) - log_warn("Manage pipe no message received %d", rv); - continue; - } - - message_t *pay_msg = nng_msg_body(msg); - msg_type_e pay_msg_type = msg_get_type(pay_msg); - switch (pay_msg_type) { - case MSG_CMD_RESP_PONG: { - char *buf_ptr; - - buf_ptr = msg_get_buf_ptr(pay_msg); - log_info("Adapter(%s) received pong: %s", adapter->name, buf_ptr); - break; - } - - case MSG_DATA_NEURON_TRANS_DATA: { - neuron_trans_data_t *trans_data; - trans_data = (neuron_trans_data_t *) msg_get_buf_ptr(pay_msg); - - const neu_plugin_intf_funs_t *intf_funs; - bool is_need_free; - neu_trans_buf_t * trans_buf; - neu_request_t req; - neu_reqresp_data_t data_req; - data_req.grp_config = trans_data->grp_config; - trans_buf = &trans_data->trans_buf; - data_req.data_val = - neu_trans_buf_get_data_val(trans_buf, &is_need_free); - if (data_req.data_val == NULL) { - neu_trans_buf_uninit(trans_buf); - log_error("Failed to get data value from trans data"); - break; - } - - intf_funs = adapter->plugin_module->intf_funs; - req.req_id = adapter_get_req_id(adapter); - req.req_type = NEU_REQRESP_TRANS_DATA; - req.buf_len = sizeof(neu_reqresp_data_t); - req.buf = (void *) &data_req; - req.sender_id = trans_data->sender_id; - req.node_name = trans_data->node_name; - intf_funs->request(adapter->plugin, &req); - if (is_need_free) { - neu_dvalue_free(data_req.data_val); - } - neu_trans_buf_uninit(trans_buf); - break; - } - - case MSG_CMD_PERSISTENCE_LOAD: { - persister_singleton_load_data(adapter); - break; - } - - case MSG_EVENT_ADD_NODE: - // fall through - - case MSG_EVENT_DEL_NODE: { - const char *node_name = msg_get_buf_ptr(pay_msg); - persister_singleton_handle_nodes(adapter, pay_msg_type, node_name); - break; - } - - case MSG_EVENT_UPDATE_NODE: { - neu_event_update_node_t *event = msg_get_buf_ptr(pay_msg); - persister_singleton_handle_update_node(adapter, event); - free(event->src_name); - free(event->dst_name); - break; - } - - case MSG_EVENT_SET_NODE_SETTING: { - neu_cmd_set_node_setting_t *cmd = msg_get_buf_ptr(pay_msg); - neu_persister_t * persister = persister_singleton_get(); - char * adapter_name = NULL; - rv = neu_manager_get_node_name_by_id(adapter->manager, cmd->node_id, - &adapter_name); - if (0 != rv) { - free((char *) cmd->setting); - break; - } - rv = neu_persister_store_adapter_setting(persister, adapter_name, - cmd->setting); - if (0 != rv) { - log_error("%s fail to store adapter:%s setting:%s", - adapter->name, adapter_name, cmd->setting); - } - free((char *) cmd->setting); - free(adapter_name); - break; - } - - case MSG_EVENT_ADD_GRP_CONFIG: { - neu_cmd_add_grp_config_t *cmd = msg_get_buf_ptr(pay_msg); - persister_singleton_handle_grp_config( - adapter, pay_msg_type, cmd->node_id, cmd->grp_config); - neu_taggrp_cfg_free(cmd->grp_config); - break; - } - - case MSG_EVENT_UPDATE_GRP_CONFIG: { - neu_cmd_update_grp_config_t *cmd = msg_get_buf_ptr(pay_msg); - persister_singleton_handle_grp_config( - adapter, pay_msg_type, cmd->node_id, cmd->grp_config); - neu_taggrp_cfg_free(cmd->grp_config); - break; - } - - case MSG_EVENT_DEL_GRP_CONFIG: { - neu_cmd_del_grp_config_t *cmd = msg_get_buf_ptr(pay_msg); - persister_singleton_handle_grp_config( - adapter, pay_msg_type, cmd->node_id, cmd->config_name); - free(cmd->config_name); - break; - } - - case MSG_EVENT_SUBSCRIBE_NODE: - // fall through - - case MSG_EVENT_UNSUBSCRIBE_NODE: { - const neu_node_id_t *node_id_p = msg_get_buf_ptr(pay_msg); - persister_singleton_handle_subscriptions(adapter, *node_id_p); - break; - } - - case MSG_EVENT_ADD_TAGS: - // fall through - - case MSG_EVENT_UPDATE_TAGS: - // fall through - - case MSG_EVENT_DEL_TAGS: { - neu_event_tags_t *event = msg_get_buf_ptr(pay_msg); - persister_singleton_handle_datatags(adapter, event->node_id, - event->grp_config_name); - free(event->grp_config_name); - break; - } - - case MSG_EVENT_ADD_PLUGIN: - // fall through - - case MSG_EVENT_UPDATE_PLUGIN: - // fall through - - case MSG_EVENT_DEL_PLUGIN: { - persister_singleton_handle_plugins(adapter, pay_msg_type); - break; - } - - case MSG_EVENT_UPDATE_LICENSE: { - const neu_plugin_intf_funs_t *intf_funs = NULL; - neu_request_t req = {}; - - intf_funs = adapter->plugin_module->intf_funs; - req.req_id = adapter_get_req_id(adapter); - req.req_type = NEU_REQRESP_UPDATE_LICENSE; - req.sender_id = 0; - req.buf_len = 0; - req.buf = NULL; - intf_funs->request(adapter->plugin, &req); - break; - } - - case MSG_CMD_SUBSCRIBE_NODE: { - subscribe_node_cmd_t *cmd_ptr; - cmd_ptr = (subscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg); - - const neu_plugin_intf_funs_t *intf_funs; - neu_request_t req; - neu_reqresp_subscribe_node_t sub_node_req; - - sub_node_req.grp_config = cmd_ptr->grp_config; - sub_node_req.dst_node_id = cmd_ptr->dst_node_id; - sub_node_req.src_node_id = cmd_ptr->src_node_id; - - intf_funs = adapter->plugin_module->intf_funs; - req.req_id = adapter_get_req_id(adapter); - req.req_type = NEU_REQRESP_SUBSCRIBE_NODE; - req.sender_id = cmd_ptr->sender_id; - req.buf_len = sizeof(neu_reqresp_subscribe_node_t); - req.buf = (void *) &sub_node_req; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { - neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - &req); - } else { - intf_funs->request(adapter->plugin, &req); - } - break; - } - - case MSG_CMD_UNSUBSCRIBE_NODE: { - unsubscribe_node_cmd_t *cmd_ptr; - cmd_ptr = (unsubscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg); - - const neu_plugin_intf_funs_t * intf_funs; - neu_request_t req; - neu_reqresp_unsubscribe_node_t unsub_node_req; - - unsub_node_req.grp_config = cmd_ptr->grp_config; - unsub_node_req.dst_node_id = cmd_ptr->dst_node_id; - unsub_node_req.src_node_id = cmd_ptr->src_node_id; - - intf_funs = adapter->plugin_module->intf_funs; - req.req_id = adapter_get_req_id(adapter); - req.req_type = NEU_REQRESP_UNSUBSCRIBE_NODE; - req.sender_id = cmd_ptr->sender_id; - req.buf_len = sizeof(neu_reqresp_unsubscribe_node_t); - req.buf = (void *) &unsub_node_req; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { - neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - &req); - } else { - intf_funs->request(adapter->plugin, &req); - } - break; - } - - case MSG_CMD_READ_DATA: { - read_data_cmd_t *cmd_ptr; - cmd_ptr = (read_data_cmd_t *) msg_get_buf_ptr(pay_msg); - - const neu_plugin_intf_funs_t *intf_funs; - neu_request_t req; - neu_reqresp_read_t read_req; - - read_req.grp_config = cmd_ptr->grp_config; - read_req.dst_node_id = cmd_ptr->dst_node_id; - - intf_funs = adapter->plugin_module->intf_funs; - req.req_id = cmd_ptr->req_id; - req.req_type = NEU_REQRESP_READ_DATA; - req.sender_id = cmd_ptr->sender_id; - req.buf_len = sizeof(neu_reqresp_read_t); - req.buf = (void *) &read_req; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { - neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - &req); - } else { - intf_funs->request(adapter->plugin, &req); - } - break; - } - - case MSG_DATA_READ_RESP: { - read_data_resp_t *cmd_ptr; - cmd_ptr = (read_data_resp_t *) msg_get_buf_ptr(pay_msg); - - const neu_plugin_intf_funs_t *intf_funs; - bool is_need_free; - neu_trans_buf_t * trans_buf; - neu_request_t req; - neu_reqresp_read_resp_t read_resp; - - read_resp.grp_config = cmd_ptr->grp_config; - trans_buf = &cmd_ptr->trans_buf; - read_resp.data_val = - neu_trans_buf_get_data_val(trans_buf, &is_need_free); - if (read_resp.data_val == NULL) { - neu_trans_buf_uninit(trans_buf); - log_error("Failed to get data value from read response"); - break; - } - - intf_funs = adapter->plugin_module->intf_funs; - req.req_id = cmd_ptr->req_id; - req.req_type = NEU_REQRESP_READ_RESP; - req.sender_id = cmd_ptr->sender_id; - req.buf_len = sizeof(neu_reqresp_read_resp_t); - req.buf = (void *) &read_resp; - intf_funs->request(adapter->plugin, &req); - if (is_need_free) { - neu_dvalue_free(read_resp.data_val); - } - neu_trans_buf_uninit(trans_buf); - break; - } - - case MSG_CMD_WRITE_DATA: { - write_data_cmd_t *cmd_ptr; - cmd_ptr = (write_data_cmd_t *) msg_get_buf_ptr(pay_msg); - - const neu_plugin_intf_funs_t *intf_funs; - bool is_need_free; - neu_trans_buf_t * trans_buf; - neu_request_t req; - neu_reqresp_write_t write_req; - - write_req.grp_config = cmd_ptr->grp_config; - write_req.dst_node_id = cmd_ptr->dst_node_id; - trans_buf = &cmd_ptr->trans_buf; - write_req.data_val = - neu_trans_buf_get_data_val(trans_buf, &is_need_free); - if (write_req.data_val == NULL) { - neu_trans_buf_uninit(trans_buf); - log_error("Failed to get data value from write request"); - break; - } - - intf_funs = adapter->plugin_module->intf_funs; - req.req_id = cmd_ptr->req_id; - req.req_type = NEU_REQRESP_WRITE_DATA; - req.sender_id = cmd_ptr->sender_id; - req.buf_len = sizeof(neu_reqresp_write_t); - req.buf = (void *) &write_req; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { - neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - &req); - } else { - intf_funs->request(adapter->plugin, &req); - } - if (is_need_free) { - neu_dvalue_free(write_req.data_val); - } - neu_trans_buf_uninit(trans_buf); - break; - } - - case MSG_DATA_WRITE_RESP: { - write_data_resp_t *cmd_ptr; - cmd_ptr = (write_data_resp_t *) msg_get_buf_ptr(pay_msg); - - const neu_plugin_intf_funs_t *intf_funs; - bool is_need_free; - neu_trans_buf_t * trans_buf; - neu_request_t req; - neu_reqresp_write_resp_t write_resp; - - write_resp.grp_config = cmd_ptr->grp_config; - trans_buf = &cmd_ptr->trans_buf; - write_resp.data_val = - neu_trans_buf_get_data_val(trans_buf, &is_need_free); - if (write_resp.data_val == NULL) { - neu_trans_buf_uninit(trans_buf); - log_error("Failed to get data value from write request"); - break; - } - - intf_funs = adapter->plugin_module->intf_funs; - req.req_id = cmd_ptr->req_id; - req.req_type = NEU_REQRESP_WRITE_RESP; - req.sender_id = cmd_ptr->sender_id; - req.buf_len = sizeof(neu_reqresp_write_resp_t); - req.buf = (void *) &write_resp; - intf_funs->request(adapter->plugin, &req); - if (is_need_free) { - neu_dvalue_free(write_resp.data_val); - } - neu_trans_buf_uninit(trans_buf); - break; - } - - case MSG_CMD_EXIT_LOOP: { - uint32_t exit_code; - - exit_code = *(uint32_t *) msg_get_buf_ptr(pay_msg); - log_info("adapter(%s) exit loop by exit_code=%d", adapter->name, - exit_code); - nng_mtx_lock(adapter->mtx); - adapter->state = ADAPTER_STATE_IDLE; - adapter->stop = true; - nng_mtx_unlock(adapter->mtx); - break; - } - - default: - log_warn("Receive a not supported message(type: %d)", - msg_get_type(pay_msg)); - break; - } - - nng_msg_free(msg); - } - - nng_dialer_close(dialer); - return; -} - static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, neu_response_t **p_result) { @@ -1802,33 +1380,26 @@ static const adapter_callbacks_t callback_funs = { neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, neu_manager_t * manager) { - neu_adapter_t *adapter; - nng_duration recv_timeout = 100; + neu_adapter_t *adapter = NULL; + int rv = 0; - if (manager == NULL) { - log_error("Create adapter with NULL manager"); - return NULL; + assert(manager != NULL); + assert(info->plugin_lib_name != NULL); + assert(info->name != NULL); + + switch (info->type) { + case ADAPTER_TYPE_DRIVERX: + adapter = (neu_adapter_t *) neu_adapter_driver_create(); + break; + default: + adapter = calloc(1, sizeof(neu_adapter_t)); + break; } - adapter = calloc(1, sizeof(neu_adapter_t)); - if (adapter == NULL) { - log_error("Out of memeory for create adapter"); - return NULL; - } - - int rv; - adapter->stop = false; - if ((rv = nng_mtx_alloc(&adapter->mtx)) != 0) { - log_error("Can't allocate mutex for adapter"); - free(adapter); - return NULL; - } - - if ((rv = nng_mtx_alloc(&adapter->sub_grp_mtx)) != 0) { - log_error("Can't allocate mutex for adapter sub grp config"); - free(adapter); - return NULL; - } + adapter->plugin_lib = + load_plugin_library((char *) info->plugin_lib_name, info->plugin_kind, + &adapter->plugin_module); + assert(adapter->plugin_lib != NULL); adapter->id = info->id; adapter->type = info->type; @@ -1840,77 +1411,37 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, adapter->manager = manager; adapter->trans_kind = NEURON_TRANS_DATAVAL; adapter->node_id = neu_manager_adapter_id_to_node_id(manager, info->id); + adapter->stop = false; + adapter->cb_funs = callback_funs; - if (adapter->name == NULL || info->plugin_lib_name == NULL) { - if (adapter->name != NULL) { - free(adapter->name); - } - nng_mtx_free(adapter->mtx); - free(adapter); - log_error("Failed duplicate string for create adapter"); - return NULL; - } + rv = nng_mtx_alloc(&adapter->mtx); + assert(rv == 0); + rv = nng_mtx_alloc(&adapter->sub_grp_mtx); + assert(rv == 0); vector_init(&adapter->sub_grp_configs, 0, sizeof(neu_sub_grp_config_t)); - void * handle; - neu_plugin_module_t *plugin_module; - handle = load_plugin_library((char *) info->plugin_lib_name, - adapter->plugin_kind, &plugin_module); - if (handle == NULL) { - neu_panic("Can't to load library(%s) for plugin(%s)", - info->plugin_lib_name, adapter->name); - } + adapter->plugin = + adapter->plugin_module->intf_funs->open(adapter, &adapter->cb_funs); + assert(adapter->plugin != NULL); + assert(neu_plugin_common_check(adapter->plugin)); - adapter->cb_funs = callback_funs; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { - adapter = (neu_adapter_t *) neu_adapter_driver_create(adapter); - } - - neu_plugin_t *plugin; - adapter->plugin_lib = handle; - adapter->plugin_module = plugin_module; - plugin = plugin_module->intf_funs->open(adapter, &adapter->cb_funs); - if (plugin == NULL) { - log_error("Can't to open plugin(%s)", plugin_module->module_name); - free(adapter->name); - nng_mtx_free(adapter->mtx); - nng_mtx_free(adapter->sub_grp_mtx); - vector_uninit(&adapter->sub_grp_configs); - free(adapter); - return NULL; - } - - if (!neu_plugin_common_check(plugin)) { - neu_panic("Failed to check if plugin is valid, %s", - plugin_module->module_name); - } - - adapter->plugin = plugin; - - rv = nng_pair1_open(&adapter->sock); - if (rv != 0) { - neu_panic("The adapter(%s) can't open pipe", adapter->name); - } - - nng_setopt(adapter->sock, NNG_OPT_RECVTIMEO, &recv_timeout, - sizeof(recv_timeout)); + adapter->events = neu_event_new(); log_info("Success to create adapter: %s", adapter->name); + return adapter; } void neu_adapter_destroy(neu_adapter_t *adapter) { - if (adapter == NULL) { - neu_panic("Destroied adapter is NULL"); - } + assert(adapter != NULL); if (adapter->type == ADAPTER_TYPE_DRIVERX) { neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter); } - nng_close(adapter->sock); + // nng_close(adapter->sock); if (adapter->plugin_module != NULL) { adapter->plugin_module->intf_funs->close(adapter->plugin); } @@ -1929,32 +1460,449 @@ void neu_adapter_destroy(neu_adapter_t *adapter) nng_mtx_free(adapter->mtx); nng_mtx_free(adapter->sub_grp_mtx); + neu_event_close(adapter->events); free(adapter); return; } +int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) +{ + neu_adapter_t *adapter = (neu_adapter_t *) usr_data; + int rv = 0; + nng_msg * msg; + message_t * pay_msg; + msg_type_e pay_msg_type; + + switch (type) { + case NEU_EVENT_IO_READ: + rv = nng_recvmsg(adapter->sock, &msg, 0); + if (rv == NNG_ECLOSED) { + nng_mtx_lock(adapter->mtx); + adapter->state = ADAPTER_STATE_IDLE; + adapter->stop = true; + nng_mtx_unlock(adapter->mtx); + + log_warn("nng socket closed."); + return 0; + } + + pay_msg = nng_msg_body(msg); + pay_msg_type = msg_get_type(pay_msg); + + switch (pay_msg_type) { + case MSG_CMD_RESP_PONG: { + char *buf_ptr = msg_get_buf_ptr(pay_msg); + + adapter->plugin_module->intf_funs->init(adapter->plugin); + adapter->state = ADAPTER_STATE_INIT; + log_info("Adapter(%s) received pong: %s\n", adapter->name, buf_ptr); + break; + } + case MSG_CMD_EXIT_LOOP: { + uint32_t exit_code = *(uint32_t *) msg_get_buf_ptr(pay_msg); + + adapter->plugin_module->intf_funs->uninit(adapter->plugin); + nng_mtx_lock(adapter->mtx); + adapter->state = ADAPTER_STATE_IDLE; + adapter->stop = true; + nng_mtx_unlock(adapter->mtx); + + log_info("adapter(%s) exit loop by exit_code=%d", adapter->name, + exit_code); + break; + } + case MSG_DATA_NEURON_TRANS_DATA: { + neuron_trans_data_t *trans_data; + trans_data = (neuron_trans_data_t *) msg_get_buf_ptr(pay_msg); + + const neu_plugin_intf_funs_t *intf_funs; + bool is_need_free; + neu_trans_buf_t * trans_buf; + neu_request_t req; + neu_reqresp_data_t data_req; + data_req.grp_config = trans_data->grp_config; + trans_buf = &trans_data->trans_buf; + data_req.data_val = + neu_trans_buf_get_data_val(trans_buf, &is_need_free); + if (data_req.data_val == NULL) { + neu_trans_buf_uninit(trans_buf); + log_error("Failed to get data value from trans data"); + break; + } + + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = adapter_get_req_id(adapter); + req.req_type = NEU_REQRESP_TRANS_DATA; + req.buf_len = sizeof(neu_reqresp_data_t); + req.buf = (void *) &data_req; + req.sender_id = trans_data->sender_id; + req.node_name = trans_data->node_name; + intf_funs->request(adapter->plugin, &req); + if (is_need_free) { + neu_dvalue_free(data_req.data_val); + } + neu_trans_buf_uninit(trans_buf); + break; + } + + case MSG_CMD_PERSISTENCE_LOAD: { + persister_singleton_load_data(adapter); + break; + } + + case MSG_EVENT_ADD_NODE: + // fall through + + case MSG_EVENT_DEL_NODE: { + const char *node_name = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_nodes(adapter, pay_msg_type, node_name); + break; + } + + case MSG_EVENT_UPDATE_NODE: { + neu_event_update_node_t *event = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_update_node(adapter, event); + free(event->src_name); + free(event->dst_name); + break; + } + + case MSG_EVENT_SET_NODE_SETTING: { + neu_cmd_set_node_setting_t *cmd = msg_get_buf_ptr(pay_msg); + neu_persister_t * persister = persister_singleton_get(); + char * adapter_name = NULL; + rv = neu_manager_get_node_name_by_id(adapter->manager, cmd->node_id, + &adapter_name); + if (0 != rv) { + free((char *) cmd->setting); + break; + } + rv = neu_persister_store_adapter_setting(persister, adapter_name, + cmd->setting); + if (0 != rv) { + log_error("%s fail to store adapter:%s setting:%s", + adapter->name, adapter_name, cmd->setting); + } + free((char *) cmd->setting); + free(adapter_name); + break; + } + + case MSG_EVENT_ADD_GRP_CONFIG: { + neu_cmd_add_grp_config_t *cmd = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_grp_config( + adapter, pay_msg_type, cmd->node_id, cmd->grp_config); + neu_taggrp_cfg_free(cmd->grp_config); + break; + } + + case MSG_EVENT_UPDATE_GRP_CONFIG: { + neu_cmd_update_grp_config_t *cmd = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_grp_config( + adapter, pay_msg_type, cmd->node_id, cmd->grp_config); + neu_taggrp_cfg_free(cmd->grp_config); + break; + } + + case MSG_EVENT_DEL_GRP_CONFIG: { + neu_cmd_del_grp_config_t *cmd = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_grp_config( + adapter, pay_msg_type, cmd->node_id, cmd->config_name); + free(cmd->config_name); + break; + } + + case MSG_EVENT_SUBSCRIBE_NODE: + // fall through + + case MSG_EVENT_UNSUBSCRIBE_NODE: { + const neu_node_id_t *node_id_p = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_subscriptions(adapter, *node_id_p); + break; + } + + case MSG_EVENT_ADD_TAGS: + // fall through + + case MSG_EVENT_UPDATE_TAGS: + // fall through + + case MSG_EVENT_DEL_TAGS: { + neu_event_tags_t *event = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_datatags(adapter, event->node_id, + event->grp_config_name); + free(event->grp_config_name); + break; + } + + case MSG_EVENT_ADD_PLUGIN: + // fall through + + case MSG_EVENT_UPDATE_PLUGIN: + // fall through + + case MSG_EVENT_DEL_PLUGIN: { + persister_singleton_handle_plugins(adapter, pay_msg_type); + break; + } + + case MSG_EVENT_UPDATE_LICENSE: { + const neu_plugin_intf_funs_t *intf_funs = NULL; + neu_request_t req = {}; + + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = adapter_get_req_id(adapter); + req.req_type = NEU_REQRESP_UPDATE_LICENSE; + req.sender_id = 0; + req.buf_len = 0; + req.buf = NULL; + intf_funs->request(adapter->plugin, &req); + break; + } + + case MSG_CMD_SUBSCRIBE_NODE: { + subscribe_node_cmd_t *cmd_ptr; + cmd_ptr = (subscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg); + + const neu_plugin_intf_funs_t *intf_funs; + neu_request_t req; + neu_reqresp_subscribe_node_t sub_node_req; + + sub_node_req.grp_config = cmd_ptr->grp_config; + sub_node_req.dst_node_id = cmd_ptr->dst_node_id; + sub_node_req.src_node_id = cmd_ptr->src_node_id; + + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = adapter_get_req_id(adapter); + req.req_type = NEU_REQRESP_SUBSCRIBE_NODE; + req.sender_id = cmd_ptr->sender_id; + req.buf_len = sizeof(neu_reqresp_subscribe_node_t); + req.buf = (void *) &sub_node_req; + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + &req); + } else { + intf_funs->request(adapter->plugin, &req); + } + break; + } + + case MSG_CMD_UNSUBSCRIBE_NODE: { + unsubscribe_node_cmd_t *cmd_ptr; + cmd_ptr = (unsubscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg); + + const neu_plugin_intf_funs_t * intf_funs; + neu_request_t req; + neu_reqresp_unsubscribe_node_t unsub_node_req; + + unsub_node_req.grp_config = cmd_ptr->grp_config; + unsub_node_req.dst_node_id = cmd_ptr->dst_node_id; + unsub_node_req.src_node_id = cmd_ptr->src_node_id; + + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = adapter_get_req_id(adapter); + req.req_type = NEU_REQRESP_UNSUBSCRIBE_NODE; + req.sender_id = cmd_ptr->sender_id; + req.buf_len = sizeof(neu_reqresp_unsubscribe_node_t); + req.buf = (void *) &unsub_node_req; + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + &req); + } else { + intf_funs->request(adapter->plugin, &req); + } + break; + } + + case MSG_CMD_READ_DATA: { + read_data_cmd_t *cmd_ptr; + cmd_ptr = (read_data_cmd_t *) msg_get_buf_ptr(pay_msg); + + const neu_plugin_intf_funs_t *intf_funs; + neu_request_t req; + neu_reqresp_read_t read_req; + + read_req.grp_config = cmd_ptr->grp_config; + read_req.dst_node_id = cmd_ptr->dst_node_id; + + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = cmd_ptr->req_id; + req.req_type = NEU_REQRESP_READ_DATA; + req.sender_id = cmd_ptr->sender_id; + req.buf_len = sizeof(neu_reqresp_read_t); + req.buf = (void *) &read_req; + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + &req); + } else { + intf_funs->request(adapter->plugin, &req); + } + break; + } + + case MSG_DATA_READ_RESP: { + read_data_resp_t *cmd_ptr; + cmd_ptr = (read_data_resp_t *) msg_get_buf_ptr(pay_msg); + + const neu_plugin_intf_funs_t *intf_funs; + bool is_need_free; + neu_trans_buf_t * trans_buf; + neu_request_t req; + neu_reqresp_read_resp_t read_resp; + + read_resp.grp_config = cmd_ptr->grp_config; + trans_buf = &cmd_ptr->trans_buf; + read_resp.data_val = + neu_trans_buf_get_data_val(trans_buf, &is_need_free); + if (read_resp.data_val == NULL) { + neu_trans_buf_uninit(trans_buf); + log_error("Failed to get data value from read response"); + break; + } + + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = cmd_ptr->req_id; + req.req_type = NEU_REQRESP_READ_RESP; + req.sender_id = cmd_ptr->sender_id; + req.buf_len = sizeof(neu_reqresp_read_resp_t); + req.buf = (void *) &read_resp; + intf_funs->request(adapter->plugin, &req); + if (is_need_free) { + neu_dvalue_free(read_resp.data_val); + } + neu_trans_buf_uninit(trans_buf); + break; + } + + case MSG_CMD_WRITE_DATA: { + write_data_cmd_t *cmd_ptr; + cmd_ptr = (write_data_cmd_t *) msg_get_buf_ptr(pay_msg); + + const neu_plugin_intf_funs_t *intf_funs; + bool is_need_free; + neu_trans_buf_t * trans_buf; + neu_request_t req; + neu_reqresp_write_t write_req; + + write_req.grp_config = cmd_ptr->grp_config; + write_req.dst_node_id = cmd_ptr->dst_node_id; + trans_buf = &cmd_ptr->trans_buf; + write_req.data_val = + neu_trans_buf_get_data_val(trans_buf, &is_need_free); + if (write_req.data_val == NULL) { + neu_trans_buf_uninit(trans_buf); + log_error("Failed to get data value from write request"); + break; + } + + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = cmd_ptr->req_id; + req.req_type = NEU_REQRESP_WRITE_DATA; + req.sender_id = cmd_ptr->sender_id; + req.buf_len = sizeof(neu_reqresp_write_t); + req.buf = (void *) &write_req; + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + &req); + } else { + intf_funs->request(adapter->plugin, &req); + } + if (is_need_free) { + neu_dvalue_free(write_req.data_val); + } + neu_trans_buf_uninit(trans_buf); + break; + } + + case MSG_DATA_WRITE_RESP: { + write_data_resp_t *cmd_ptr; + cmd_ptr = (write_data_resp_t *) msg_get_buf_ptr(pay_msg); + + const neu_plugin_intf_funs_t *intf_funs; + bool is_need_free; + neu_trans_buf_t * trans_buf; + neu_request_t req; + neu_reqresp_write_resp_t write_resp; + + write_resp.grp_config = cmd_ptr->grp_config; + trans_buf = &cmd_ptr->trans_buf; + write_resp.data_val = + neu_trans_buf_get_data_val(trans_buf, &is_need_free); + if (write_resp.data_val == NULL) { + neu_trans_buf_uninit(trans_buf); + log_error("Failed to get data value from write request"); + break; + } + + intf_funs = adapter->plugin_module->intf_funs; + req.req_id = cmd_ptr->req_id; + req.req_type = NEU_REQRESP_WRITE_RESP; + req.sender_id = cmd_ptr->sender_id; + req.buf_len = sizeof(neu_reqresp_write_resp_t); + req.buf = (void *) &write_resp; + intf_funs->request(adapter->plugin, &req); + if (is_need_free) { + neu_dvalue_free(write_resp.data_val); + } + neu_trans_buf_uninit(trans_buf); + break; + } + default: + log_warn("Receive a not supported message(type: %d)", + msg_get_type(pay_msg)); + break; + } + + nng_msg_free(msg); + + break; + case NEU_EVENT_IO_CLOSED: + case NEU_EVENT_IO_HUP: + log_warn("adapter %s, fd: %d, recv closed or hup.", adapter->name, fd); + break; + } + + return 0; +} + int neu_adapter_init(neu_adapter_t *adapter) { - int rv = 0; + int rv = nng_pair1_open(&adapter->sock); - if (adapter == NULL) { - log_error("Start adapter with NULL adapter"); - return (-1); - } + neu_event_io_param_t param = { + .usr_data = (void *) adapter, + .cb = adapter_loop, + }; + + assert(rv == 0); + + nng_socket_get_int(adapter->sock, NNG_OPT_RECVFD, &adapter->recv_fd); + param.fd = adapter->recv_fd; + + adapter->nng_io = neu_event_add_io(adapter->events, param); if (adapter->type == ADAPTER_TYPE_DRIVERX) { neu_adapter_driver_init((neu_adapter_driver_t *) adapter); } - if (adapter->plugin_module != NULL) { - const neu_plugin_intf_funs_t *intf_funs; - intf_funs = adapter->plugin_module->intf_funs; - intf_funs->init(adapter->plugin); - } + rv = nng_dial(adapter->sock, neu_manager_get_url(adapter->manager), + &adapter->dialer, 0); + assert(rv == 0); - adapter->stop = false; - adapter->state = ADAPTER_STATE_INIT; - nng_thread_create(&adapter->thrd, adapter_loop, adapter); + nng_msg *out_msg; + size_t msg_size; + msg_size = msg_inplace_data_get_size(strlen(adapter->name) + 1); + rv = nng_msg_alloc(&out_msg, msg_size); + if (rv == 0) { + message_t *msg_ptr; + char * buf_ptr; + msg_ptr = (message_t *) nng_msg_body(out_msg); + msg_inplace_data_init(msg_ptr, MSG_EVENT_NODE_PING, msg_size); + buf_ptr = msg_get_buf_ptr(msg_ptr); + memcpy(buf_ptr, adapter->name, strlen(adapter->name)); + buf_ptr[strlen(adapter->name)] = 0; + nng_sendmsg(adapter->sock, out_msg, 0); + } return rv; } @@ -1962,27 +1910,27 @@ int neu_adapter_uninit(neu_adapter_t *adapter) { int rv = 0; - if (adapter == NULL) { - log_error("Stop adapter with NULL adapter"); - return -1; - } - if (adapter->type == ADAPTER_TYPE_DRIVERX) { - neu_adapter_driver_init((neu_adapter_driver_t *) adapter); + neu_adapter_driver_uninit((neu_adapter_driver_t *) adapter); } - log_info("Stop the adapter(%s)", adapter->name); nng_mtx_lock(adapter->mtx); adapter->stop = true; nng_mtx_unlock(adapter->mtx); - nng_thread_destroy(adapter->thrd); - if (adapter->plugin_module != NULL) { - const neu_plugin_intf_funs_t *intf_funs; - intf_funs = adapter->plugin_module->intf_funs; - intf_funs->uninit(adapter->plugin); - } + neu_event_del_io(adapter->events, adapter->nng_io); + // nng_thread_destroy(adapter->thrd); + // if (adapter->plugin_module != NULL) { + // const neu_plugin_intf_funs_t *intf_funs; + // intf_funs = adapter->plugin_module->intf_funs; + // intf_funs->uninit(adapter->plugin); + //} + + nng_dialer_close(adapter->dialer); + nng_close(adapter->sock); + + log_info("Stop the adapter(%s)", adapter->name); return rv; } diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index dc46a9a0..65c10d49 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -24,6 +24,7 @@ #include #include +#include "event/event.h" #include "plugin.h" #include "plugin_info.h" @@ -53,6 +54,8 @@ struct neu_adapter { neu_manager_t * manager; nng_pipe pipe; nng_socket sock; + nng_dialer dialer; + int recv_fd; nng_thread * thrd; uint32_t new_req_id; plugin_id_t plugin_id; @@ -64,6 +67,9 @@ struct neu_adapter { adapter_callbacks_t cb_funs; neu_config_t node_setting; vector_t sub_grp_configs; // neu_sub_grp_config_t + + neu_events_t * events; + neu_event_io_t *nng_io; }; neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, diff --git a/src/adapter/driver.c b/src/adapter/driver.c index 5a16e72f..133cb66c 100644 --- a/src/adapter/driver.c +++ b/src/adapter/driver.c @@ -108,12 +108,12 @@ static int update(neu_adapter_t *adapter, const char *name, neu_dvalue_t value) return 0; } -neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter) +neu_adapter_driver_t *neu_adapter_driver_create() { neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t)); - driver->adapter = *adapter; - free(adapter); + // driver->adapter = *adapter; + // free(adapter); pthread_mutex_init(&driver->grp_mtx, NULL); driver->cache = neu_driver_cache_new(); diff --git a/src/adapter/driver_internal.h b/src/adapter/driver_internal.h index 9a9087de..306ffa10 100644 --- a/src/adapter/driver_internal.h +++ b/src/adapter/driver_internal.h @@ -23,7 +23,7 @@ #include "adapter.h" #include "core/message.h" -neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter); +neu_adapter_driver_t *neu_adapter_driver_create(); void neu_adapter_driver_destroy(neu_adapter_driver_t *driver); int neu_adapter_driver_start(neu_adapter_driver_t *driver);