diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index c2545fd4..9f114a0d 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -32,6 +32,7 @@ #include "adapter.h" #include "adapter_internal.h" +#include "base/msg_internal.h" #include "driver/driver_internal.h" #include "errcodes.h" #include "persist/persist.h" @@ -313,23 +314,24 @@ int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name) void neu_adapter_init(neu_adapter_t *adapter, neu_node_running_state_e state) { - memset(adapter->buf, 0, sizeof(adapter->buf)); - neu_reqresp_head_t *header = (neu_reqresp_head_t *) adapter->buf; - neu_req_node_init_t init = { 0 }; - - header->type = NEU_REQ_NODE_INIT; + neu_req_node_init_t init = { 0 }; + init.state = state; + strcpy(init.node, adapter->name); + neu_msg_t *msg = neu_msg_new(NEU_REQ_NODE_INIT, NULL, &init); + if (NULL == msg) { + nlog_error("failed alloc msg for %s", adapter->name); + return; + } + neu_reqresp_head_t *header = neu_msg_get_header(msg); strcpy(header->sender, adapter->name); strcpy(header->receiver, "manager"); - strcpy(init.node, adapter->name); - init.state = state; - neu_msg_gen(header, &init); - - int ret = send(adapter->control_fd, header, header->len, 0); - if (ret != (int) header->len) { + int ret = neu_send_msg(adapter->control_fd, msg); + if (0 != ret) { nlog_error("%s failed to send init msg to manager, ret: %d, errno: %d", adapter->name, ret, errno); + neu_msg_free(msg); } } @@ -418,11 +420,13 @@ static void adapter_reset_metrics(neu_adapter_t *adapter) static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, void *data) { - int ret = 0; - uint8_t buf[NEU_MSG_MAX_SIZE] = { 0 }; + int ret = 0; - neu_reqresp_head_t *pheader = (neu_reqresp_head_t *) buf; - *pheader = header; + neu_msg_t *msg = neu_msg_new(header.type, header.ctx, data); + if (NULL == msg) { + return NEU_ERR_EINTERNAL; + } + neu_reqresp_head_t *pheader = neu_msg_get_header(msg); strcpy(pheader->sender, adapter->name); switch (pheader->type) { @@ -507,15 +511,14 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, break; } - neu_msg_gen(pheader, data); - - ret = send(adapter->control_fd, pheader, pheader->len, 0); - if (ret != (int) pheader->len) { + ret = neu_send_msg(adapter->control_fd, msg); + if (0 != ret) { nlog_error( "adapter: %s send %d command %s failed, ret: %d-%d, errno: %s(%d)", adapter->name, adapter->control_fd, neu_reqresp_type_string(pheader->type), ret, pheader->len, strerror(errno), errno); + neu_msg_free(msg); return -1; } else { return 0; @@ -529,11 +532,13 @@ static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, neu_msg_exchange(header); neu_msg_gen(header, data); - int ret = send(adapter->control_fd, header, header->len, 0); - if (ret <= 0) { + neu_msg_t *msg = (neu_msg_t *) header; + int ret = neu_send_msg(adapter->control_fd, msg); + if (0 != ret) { nlog_error("adapter: %s send response %s failed, ret: %d, errno: %d", adapter->name, neu_reqresp_type_string(header->type), ret, errno); + neu_msg_free(msg); } return ret; @@ -544,15 +549,20 @@ static int adapter_responseto(neu_adapter_t * adapter, struct sockaddr_in dst) { assert(header->type == NEU_REQRESP_TRANS_DATA); - strcpy(header->sender, adapter->name); - neu_msg_gen(header, data); - int ret = sendto(adapter->control_fd, header, header->len, 0, - (struct sockaddr *) &dst, sizeof(dst)); - if (ret <= 0) { + neu_msg_t *msg = neu_msg_new(header->type, header->ctx, data); + if (NULL == msg) { + return NEU_ERR_EINTERNAL; + } + neu_reqresp_head_t *pheader = neu_msg_get_header(msg); + strcpy(pheader->sender, adapter->name); + + int ret = neu_send_msg_to(adapter->control_fd, &dst, msg); + if (0 != ret) { nlog_error("adapter: %s send responseto %s failed, ret: %d, errno: %d", adapter->name, neu_reqresp_type_string(header->type), ret, errno); + neu_msg_free(msg); } return ret; @@ -568,17 +578,16 @@ static int adapter_trans_data(enum neu_event_io_type type, int fd, return 0; } - memset(adapter->recv_buf, 0, sizeof(adapter->recv_buf)); - neu_reqresp_head_t *header = (neu_reqresp_head_t *) adapter->recv_buf; - - int rv = recv(adapter->trans_data_fd, adapter->recv_buf, - sizeof(adapter->recv_buf), 0); - if (rv <= 0) { + neu_msg_t *msg = NULL; + int rv = neu_recv_msg(adapter->trans_data_fd, &msg); + if (0 != rv) { nlog_warn("adapter: %s recv trans data failed, ret: %d, errno: %s(%d)", adapter->name, rv, strerror(errno), errno); return 0; } + neu_reqresp_head_t *header = neu_msg_get_header(msg); + nlog_debug("adapter(%s) recv msg from: %s %p, type: %s", adapter->name, header->sender, header->ctx, neu_reqresp_type_string(header->type)); @@ -587,6 +596,7 @@ static int adapter_trans_data(enum neu_event_io_type type, int fd, header->type != NEU_RESP_ERROR) { nlog_warn("adapter: %s recv msg type error, type: %s", adapter->name, neu_reqresp_type_string(header->type)); + neu_msg_free(msg); return 0; } @@ -595,13 +605,13 @@ static int adapter_trans_data(enum neu_event_io_type type, int fd, if (header->type == NEU_REQRESP_TRANS_DATA) { neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]); } + neu_msg_free(msg); return 0; } static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) { - neu_adapter_t * adapter = (neu_adapter_t *) usr_data; - static __thread uint8_t recv_buf[2048] = { 0 }; + neu_adapter_t *adapter = (neu_adapter_t *) usr_data; if (type != NEU_EVENT_IO_READ) { nlog_warn("adapter: %s recv close, exit loop, fd: %d", adapter->name, @@ -609,16 +619,16 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) return 0; } - memset(recv_buf, 0, sizeof(recv_buf)); - neu_reqresp_head_t *header = (neu_reqresp_head_t *) recv_buf; - - int rv = recv(adapter->control_fd, recv_buf, sizeof(recv_buf), 0); - if (rv <= 0) { + neu_msg_t *msg = NULL; + int rv = neu_recv_msg(adapter->control_fd, &msg); + if (0 != rv) { nlog_warn("adapter: %s recv failed, ret: %d, errno: %s(%d)", adapter->name, rv, strerror(errno), errno); return 0; } + neu_reqresp_head_t *header = neu_msg_get_header(msg); + nlog_info("adapter(%s) recv msg from: %s %p, type: %s", adapter->name, header->sender, header->ctx, neu_reqresp_type_string(header->type)); @@ -632,6 +642,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); } + neu_msg_free(msg); break; } case NEU_REQ_UNSUBSCRIBE_GROUP: { @@ -643,6 +654,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); } + neu_msg_free(msg); break; } case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: @@ -683,11 +695,13 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_UPDATE_TAG_EVENT: adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); + neu_msg_free(msg); break; case NEU_RESP_READ_GROUP: adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); neu_resp_read_free((neu_resp_read_group_t *) &header[1]); + neu_msg_free(msg); break; case NEU_REQ_READ_GROUP: { neu_resp_error_t error = { 0 }; @@ -709,9 +723,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_resp_error_t error = { 0 }; if (adapter->module->type == NEU_NA_TYPE_DRIVER) { - neu_reqresp_head_t *msg_dump = neu_msg_dup(header); neu_adapter_driver_write_tag((neu_adapter_driver_t *) adapter, - msg_dump); + header); } else { neu_req_write_tag_fini((neu_req_write_tag_t *) &header[1]); error.error = NEU_ERR_GROUP_NOT_ALLOW; @@ -732,9 +745,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_msg_exchange(header); reply(adapter, header, &error); } else { - neu_reqresp_head_t *msg_dump = neu_msg_dup(header); neu_adapter_driver_write_tags((neu_adapter_driver_t *) adapter, - msg_dump); + header); } break; } @@ -748,9 +760,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_msg_exchange(header); reply(adapter, header, &error); } else { - neu_reqresp_head_t *msg_dump = neu_msg_dup(header); neu_adapter_driver_write_gtags((neu_adapter_driver_t *) adapter, - msg_dump); + header); } break; } @@ -786,7 +797,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) strcpy(resp.node, adapter->name); reply(adapter, header, &resp); } - break; } case NEU_REQ_GET_NODE_STATE: { @@ -1177,19 +1187,22 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) strcpy(name, adapter->name); strcpy(receiver, header->receiver); - int ret = send(adapter->control_fd, header, header->len, 0); - if (ret != (int) header->len) { + int ret = neu_send_msg(adapter->control_fd, msg); + if (0 != ret) { nlog_error("%s %d send uninit msg to %s error: %s(%d)", name, adapter->control_fd, receiver, strerror(errno), errno); + neu_msg_free(msg); } else { nlog_notice("%s send uninit msg to %s failed", name, receiver); } break; } case NEU_REQ_ADD_NDRIVER_MAP: { + neu_msg_free(msg); break; } case NEU_REQ_DEL_NDRIVER_MAP: { + neu_msg_free(msg); break; } case NEU_REQ_UPDATE_NDRIVER_TAG_PARAM: { @@ -1590,9 +1603,8 @@ inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data) { neu_msg_gen(header, data); - - int ret = send(adapter->control_fd, header, header->len, 0); - if (ret <= 0) { + int ret = neu_send_msg(adapter->control_fd, (neu_msg_t *) header); + if (0 != ret) { nlog_warn("%s reply %s to %s, error: %s(%d)", header->sender, neu_reqresp_type_string(header->type), header->receiver, strerror(errno), errno); @@ -1602,19 +1614,20 @@ inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, inline static void notify_monitor(neu_adapter_t * adapter, neu_reqresp_type_e event, void *data) { - memset(adapter->buf, 0, sizeof(adapter->buf)); - neu_reqresp_head_t *header = (neu_reqresp_head_t *) adapter->buf; + neu_msg_t *msg = neu_msg_new(event, NULL, data); + if (NULL == msg) { + return; + } + neu_reqresp_head_t *header = neu_msg_get_header(msg); strcpy(header->receiver, "manager"); strncpy(header->sender, adapter->name, NEU_NODE_NAME_LEN); - header->type = event; - neu_msg_gen(header, data); - - int ret = send(adapter->control_fd, header, header->len, 0); - if (ret <= 0) { + int ret = neu_send_msg(adapter->control_fd, msg); + if (0 != ret) { nlog_warn("notify %s of %s, error: %s(%d)", header->receiver, neu_reqresp_type_string(header->type), strerror(errno), errno); + neu_msg_free(msg); } } diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index 7fc9ab5c..f2efb33c 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -39,9 +39,6 @@ struct neu_adapter { neu_plugin_module_t *module; neu_plugin_t * plugin; - uint8_t buf[NEU_MSG_MAX_SIZE]; - uint8_t recv_buf[NEU_MSG_MAX_SIZE]; - neu_event_io_t *control_io; neu_event_io_t *trans_data_io; diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 76b3588b..2db0fef1 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -120,7 +120,6 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error) nlog_notice("write tag response <%p>", req->ctx); adapter->cb_funs.response(adapter, req, &nerror); - free(req); } static void update_with_meta(neu_adapter_t *adapter, const char *group, @@ -177,9 +176,7 @@ static void update_im(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value, neu_tag_meta_t *metas, int n_meta) { - neu_adapter_driver_t * driver = (neu_adapter_driver_t *) adapter; - static __thread uint8_t buf[1024] = { 0 }; - neu_reqresp_head_t * header = (neu_reqresp_head_t *) buf; + neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter; if (tag == NULL || value.type == NEU_TYPE_ERROR) { nlog_warn("update_im tag is null or value is error %d", @@ -215,8 +212,9 @@ static void update_im(neu_adapter_t *adapter, const char *group, driver->adapter.name, group, tag, neu_type_string(value.type), global_timestamp); - memset(buf, 0, sizeof(buf)); - header->type = NEU_REQRESP_TRANS_DATA; + neu_reqresp_head_t header = { + .type = NEU_REQRESP_TRANS_DATA, + }; neu_reqresp_trans_data_t *data = calloc(1, sizeof(neu_reqresp_trans_data_t)); @@ -241,7 +239,7 @@ static void update_im(neu_adapter_t *adapter, const char *group, utarray_foreach(find->apps, sub_app_t *, app) { if (driver->adapter.cb_funs.responseto( - &driver->adapter, header, data, app->addr) <= 0) { + &driver->adapter, &header, data, app->addr) != 0) { neu_trans_data_free(data); } } @@ -1361,16 +1359,15 @@ UT_array *neu_adapter_driver_get_ptag(neu_adapter_driver_t *driver, static int report_callback(void *usr_data) { - group_t * group = (group_t *) usr_data; - static __thread uint8_t buf[1024] = { 0 }; - neu_reqresp_head_t * header = (neu_reqresp_head_t *) buf; - neu_node_running_state_e state = group->driver->adapter.state; + group_t * group = (group_t *) usr_data; + neu_node_running_state_e state = group->driver->adapter.state; if (state != NEU_NODE_RUNNING_STATE_RUNNING) { return 0; } - memset(buf, 0, sizeof(buf)); - header->type = NEU_REQRESP_TRANS_DATA; + neu_reqresp_head_t header = { + .type = NEU_REQRESP_TRANS_DATA, + }; UT_array *tags = neu_adapter_driver_get_read_tag(group->driver, group->name); @@ -1398,7 +1395,7 @@ static int report_callback(void *usr_data) utarray_foreach(group->apps, sub_app_t *, app) { if (group->driver->adapter.cb_funs.responseto( - &group->driver->adapter, header, data, app->addr) <= + &group->driver->adapter, &header, data, app->addr) != 0) { neu_trans_data_free(data); } diff --git a/src/base/msg.c b/src/base/msg.c index 0dcf948c..866188e6 100644 --- a/src/base/msg.c +++ b/src/base/msg.c @@ -25,9 +25,8 @@ void neu_msg_gen(neu_reqresp_head_t *header, void *data) { size_t data_size = neu_reqresp_size(header->type); - assert(NEU_MSG_MAX_SIZE >= sizeof(neu_reqresp_head_t) + data_size); + assert(header->len >= sizeof(neu_reqresp_head_t) + data_size); memcpy((uint8_t *) &header[1], data, data_size); - header->len = sizeof(neu_reqresp_head_t) + data_size; } neu_reqresp_head_t *neu_msg_dup(neu_reqresp_head_t *header) diff --git a/src/base/msg_internal.h b/src/base/msg_internal.h index 583d47b6..a12efdc2 100644 --- a/src/base/msg_internal.h +++ b/src/base/msg_internal.h @@ -24,6 +24,9 @@ extern "C" { #endif +#include +#include + #include "msg.h" #define NEU_REQRESP_TYPE_MAP(XX) \ @@ -158,9 +161,46 @@ typedef struct neu_msg_s neu_msg_t; static inline neu_msg_t *neu_msg_new(neu_reqresp_type_e t, void *ctx, void *data) { - size_t data_size = neu_reqresp_size(t); - size_t total = sizeof(neu_msg_t) + data_size; - neu_msg_t *msg = calloc(1, total); + size_t data_size = neu_reqresp_size(t); + + // NOTE: ensure enough space to reuse message + size_t body_size = 0; + switch (t) { + case NEU_REQ_GET_PLUGIN: + body_size = neu_reqresp_size(NEU_RESP_GET_PLUGIN); + break; + case NEU_REQ_UPDATE_GROUP: + case NEU_REQ_UPDATE_DRIVER_GROUP: + body_size = neu_reqresp_size(NEU_RESP_UPDATE_DRIVER_GROUP); + break; + case NEU_REQ_UPDATE_NODE: + case NEU_REQ_NODE_RENAME: + body_size = neu_reqresp_size(NEU_RESP_NODE_RENAME); + break; + case NEU_REQ_DEL_NODE: + body_size = neu_reqresp_size(NEU_RESP_NODE_UNINIT); + break; + case NEU_REQ_GET_NODE_SETTING: + body_size = neu_reqresp_size(NEU_RESP_GET_NODE_SETTING); + break; + case NEU_REQ_GET_NODES_STATE: + body_size = neu_reqresp_size(NEU_RESP_GET_NODES_STATE); + break; + case NEU_REQ_GET_TEMPLATE: + body_size = neu_reqresp_size(NEU_RESP_GET_TEMPLATE); + break; + case NEU_REQ_GET_TEMPLATES: + body_size = neu_reqresp_size(NEU_RESP_GET_TEMPLATES); + break; + case NEU_REQ_READ_GROUP: + body_size = neu_reqresp_size(NEU_RESP_READ_GROUP); + break; + default: + body_size = data_size; + } + + size_t total = sizeof(neu_msg_t) + body_size; + neu_msg_t *msg = calloc(1, total); if (msg) { msg->head.type = t; msg->head.len = total; @@ -208,6 +248,47 @@ static inline void *neu_msg_get_body(neu_msg_t *msg) return &msg->body; } +inline static int neu_send_msg(int fd, neu_msg_t *msg) +{ + int ret = send(fd, &msg, sizeof(neu_msg_t *), 0); + return sizeof(neu_msg_t *) == ret ? 0 : ret; +} + +inline static int neu_recv_msg(int fd, neu_msg_t **msg_p) +{ + neu_msg_t *msg = NULL; + int ret = recv(fd, &msg, sizeof(neu_msg_t *), 0); + if (sizeof(neu_msg_t *) != ret) { + // recv may return 0 bytes + return 0 == ret ? -1 : ret; + } + *msg_p = msg; + return 0; +} + +inline static int neu_send_msg_to(int fd, struct sockaddr_in *addr, + neu_msg_t *msg) +{ + int ret = sendto(fd, &msg, sizeof(neu_msg_t *), 0, (struct sockaddr *) addr, + sizeof(*addr)); + return sizeof(neu_msg_t *) == ret ? 0 : ret; +} + +inline static int neu_recv_msg_from(int fd, struct sockaddr_in *addr, + neu_msg_t **msg_p) +{ + neu_msg_t *msg = NULL; + socklen_t addr_len = sizeof(struct sockaddr_in); + int ret = recvfrom(fd, &msg, sizeof(neu_msg_t *), 0, + (struct sockaddr *) addr, &addr_len); + if (sizeof(neu_msg_t *) != ret) { + // recvfrom may return 0 bytes + return 0 == ret ? -1 : ret; + } + *msg_p = msg; + return 0; +} + #ifdef __cplusplus } #endif diff --git a/src/core/manager.c b/src/core/manager.c index 25231496..51680e7e 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -35,6 +35,7 @@ #include "adapter/adapter_internal.h" #include "adapter/driver/driver_internal.h" #include "argparse.h" +#include "base/msg_internal.h" #include "errcodes.h" #include "node_manager.h" @@ -55,6 +56,9 @@ inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header, void *data); inline static void forward_msg(neu_manager_t * manager, neu_reqresp_head_t *header, const char *node); +inline static void forward_msg_copy(neu_manager_t * manager, + neu_reqresp_head_t *header, + const char * node); inline static void notify_monitors(neu_manager_t * manager, neu_reqresp_type_e event, void *data); @@ -159,25 +163,21 @@ neu_manager_t *neu_manager_create() void neu_manager_destroy(neu_manager_t *manager) { - memset(manager->buf, 0, sizeof(manager->buf)); - neu_reqresp_head_t *header = (neu_reqresp_head_t *) manager->buf; neu_req_node_init_t uninit = { 0 }; UT_array *addrs = neu_node_manager_get_addrs_all(manager->node_manager); neu_event_del_timer(manager->events, manager->timer_timestamp); - header->type = NEU_REQ_NODE_UNINIT; - strcpy(header->sender, "manager"); - utarray_foreach(addrs, struct sockaddr_in *, addr) { - neu_msg_gen(header, &uninit); - if (sendto(manager->server_fd, manager->buf, header->len, 0, - (struct sockaddr *) addr, - sizeof(struct sockaddr_in)) != (int) header->len) { + neu_msg_t *msg = neu_msg_new(NEU_REQ_NODE_UNINIT, NULL, &uninit); + neu_reqresp_head_t *header = neu_msg_get_header(msg); + strcpy(header->sender, "manager"); + if (0 != neu_send_msg_to(manager->server_fd, addr, msg)) { nlog_error("manager -> %s uninit msg send fail", inet_ntoa((*addr).sin_addr)); + neu_msg_free(msg); } } utarray_free(addrs); @@ -206,8 +206,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) { int rv = 0; neu_manager_t * manager = (neu_manager_t *) usr_data; - socklen_t addr_len = sizeof(struct sockaddr_in); struct sockaddr_in src_addr = { 0 }; + neu_msg_t * msg = NULL; neu_reqresp_head_t *header = NULL; if (type == NEU_EVENT_IO_CLOSED || type == NEU_EVENT_IO_HUP) { @@ -215,16 +215,13 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) return 0; } - memset(manager->recv_buf, 0, sizeof(manager->recv_buf)); - rv = recvfrom(manager->server_fd, manager->recv_buf, - sizeof(manager->recv_buf), 0, (struct sockaddr *) &src_addr, - &addr_len); + rv = neu_recv_msg_from(manager->server_fd, &src_addr, &msg); if (rv == -1) { nlog_warn("manager recv msg error: %s(%d)", strerror(errno), errno); return 0; } - header = (neu_reqresp_head_t *) manager->recv_buf; + header = neu_msg_get_header(msg); nlog_info("manager recv msg from: %s to %s, type: %s", header->sender, header->receiver, neu_reqresp_type_string(header->type)); @@ -237,6 +234,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) src_addr)) { nlog_warn("bind node %s to src addr(%d) fail", init->node, ntohs(src_addr.sin_port)); + neu_msg_free(msg); break; } @@ -252,6 +250,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) nlog_notice("bind node %s to src addr(%d)", init->node, ntohs(src_addr.sin_port)); + neu_msg_free(msg); break; } case NEU_REQ_ADD_PLUGIN: { @@ -992,19 +991,22 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_subscribe_manager_unsub_all(manager->subscribe_manager, cmd->node); } + notify_monitors(manager, NEU_REQ_DEL_NODE_EVENT, cmd); + neu_reqresp_node_deleted_t resp = { 0 }; + strcpy(resp.node, header->receiver); header->type = NEU_REQ_NODE_UNINIT; forward_msg(manager, header, header->receiver); - notify_monitors(manager, NEU_REQ_DEL_NODE_EVENT, cmd); if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_DRIVER) { UT_array *apps = neu_subscribe_manager_find_by_driver( - manager->subscribe_manager, cmd->node); - - neu_reqresp_node_deleted_t resp = { 0 }; - header->type = NEU_REQRESP_NODE_DELETED; - strcpy(resp.node, header->receiver); + manager->subscribe_manager, resp.node); utarray_foreach(apps, neu_app_subscribe_t *, app) { + msg = neu_msg_new(NEU_REQRESP_NODE_DELETED, NULL, &resp); + if (NULL == msg) { + break; + } + header = neu_msg_get_header(msg); strcpy(header->receiver, app->app_name); strcpy(header->sender, "manager"); reply(manager, header, &resp); @@ -1023,6 +1025,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_resp_error_t error = { 0 }; header->type = NEU_RESP_ERROR; reply(manager, header, &error); + } else { + neu_msg_free(msg); } break; } @@ -1047,8 +1051,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (error.error == NEU_ERR_SUCCESS) { cmd->port = app_port; - forward_msg(manager, header, cmd->app); - forward_msg(manager, header, cmd->driver); + forward_msg_copy(manager, header, cmd->app); + forward_msg_copy(manager, header, cmd->driver); manager_storage_subscribe(manager, cmd->app, cmd->driver, cmd->group, cmd->params); } else { @@ -1100,7 +1104,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) manager, cmd->app, cmd->driver, cmd->group, cmd->params); if (error.error == NEU_ERR_SUCCESS) { - forward_msg(manager, header, cmd->app); + forward_msg_copy(manager, header, cmd->app); manager_storage_update_subscribe(manager, cmd->app, cmd->driver, cmd->group, cmd->params); } else { @@ -1120,8 +1124,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_manager_unsubscribe(manager, cmd->app, cmd->driver, cmd->group); if (error.error == NEU_ERR_SUCCESS) { - forward_msg(manager, header, cmd->app); - forward_msg(manager, header, cmd->driver); + forward_msg_copy(manager, header, cmd->app); + forward_msg_copy(manager, header, cmd->driver); manager_storage_unsubscribe(manager, cmd->app, cmd->driver, cmd->group); } @@ -1266,7 +1270,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) utarray_foreach(apps, neu_app_subscribe_t *, app) { header->type = NEU_REQ_UPDATE_NODE; - forward_msg(manager, header, app->app_name); + forward_msg_copy(manager, header, app->app_name); } utarray_free(apps); @@ -1293,7 +1297,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) utarray_foreach(apps, neu_app_subscribe_t *, app) { header->type = NEU_REQ_UPDATE_GROUP; - forward_msg(manager, header, app->app_name); + forward_msg_copy(manager, header, app->app_name); } utarray_free(apps); @@ -1317,6 +1321,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_DEL_TAG_EVENT: case NEU_REQ_UPDATE_TAG_EVENT: { notify_monitors(manager, header->type, &header[1]); + neu_msg_free(msg); break; } @@ -1364,14 +1369,14 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (neu_node_manager_find(manager->node_manager, header->receiver) == NULL) { - neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; - header->type = NEU_RESP_ERROR; - neu_msg_exchange(header); - reply(manager, header, &e); for (int i = 0; i < cmd->n_tag; i++) { free(cmd->tags[i]); } free(cmd->tags); + neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(manager, header, &e); } else { forward_msg(manager, header, header->receiver); } @@ -1384,14 +1389,14 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (neu_node_manager_find(manager->node_manager, header->receiver) == NULL) { - neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; - header->type = NEU_RESP_ERROR; - neu_msg_exchange(header); - reply(manager, header, &e); for (int i = 0; i < cmd->n_tag; i++) { neu_tag_fini(&cmd->tags[i]); } free(cmd->tags); + neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(manager, header, &e); } else { forward_msg(manager, header, header->receiver); } @@ -1403,10 +1408,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (neu_node_manager_find(manager->node_manager, header->receiver) == NULL) { - neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; - header->type = NEU_RESP_ERROR; - neu_msg_exchange(header); - reply(manager, header, &e); for (int i = 0; i < cmd->n_group; i++) { for (int j = 0; j < cmd->groups[i].n_tag; j++) { neu_tag_fini(&cmd->groups[i].tags[j]); @@ -1414,6 +1415,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) free(cmd->groups[i].tags); } free(cmd->groups); + neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(manager, header, &e); } else { forward_msg(manager, header, header->receiver); } @@ -1425,11 +1430,11 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (neu_node_manager_find(manager->node_manager, header->receiver) == NULL) { + free(cmd->setting); neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; header->type = NEU_RESP_ERROR; neu_msg_exchange(header); reply(manager, header, &e); - free(cmd->setting); } else { forward_msg(manager, header, header->receiver); } @@ -1473,7 +1478,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) error.error = neu_manager_add_ndriver_map(manager, cmd->ndriver, cmd->driver, cmd->group); if (error.error == NEU_ERR_SUCCESS) { - forward_msg(manager, header, cmd->ndriver); + forward_msg_copy(manager, header, cmd->ndriver); manager_storage_add_ndriver_map(manager, cmd->ndriver, cmd->driver, cmd->group); } @@ -1490,7 +1495,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) error.error = neu_manager_del_ndriver_map(manager, cmd->ndriver, cmd->driver, cmd->group); if (error.error == NEU_ERR_SUCCESS) { - forward_msg(manager, header, cmd->ndriver); + forward_msg_copy(manager, header, cmd->ndriver); manager_storage_del_ndriver_map(manager, cmd->ndriver, cmd->driver, cmd->group); } @@ -1591,18 +1596,29 @@ inline static void forward_msg(neu_manager_t * manager, struct sockaddr_in addr = neu_node_manager_get_addr(manager->node_manager, node); - int ret = sendto(manager->server_fd, header, header->len, 0, - (struct sockaddr *) &addr, sizeof(addr)); - if (ret != (int) header->len) { - nlog_warn("forward msg %s to node (%s)%d fail", - neu_reqresp_type_string(header->type), node, - ntohs(addr.sin_port)); + neu_reqresp_type_e t = header->type; + char receiver[NEU_NODE_NAME_LEN] = { 0 }; + strncpy(receiver, header->receiver, sizeof(receiver)); + + neu_msg_t *msg = (neu_msg_t *) header; + int ret = neu_send_msg_to(manager->server_fd, &addr, msg); + if (0 == ret) { + nlog_info("forward msg %s to %s", neu_reqresp_type_string(t), receiver); } else { - nlog_info("forward msg %s to %s", neu_reqresp_type_string(header->type), - node); + nlog_warn("forward msg %s to node (%s)%d fail", + neu_reqresp_type_string(t), receiver, ntohs(addr.sin_port)); + neu_msg_free(msg); } } +inline static void forward_msg_copy(neu_manager_t * manager, + neu_reqresp_head_t *header, + const char * node) +{ + neu_msg_t *msg = neu_msg_copy((neu_msg_t *) header); + forward_msg(manager, neu_msg_get_header(msg), node); +} + struct notify_monitor_ctx { neu_manager_t * manager; neu_reqresp_type_e event; @@ -1622,10 +1638,21 @@ static int notify_monitor(const char *name, struct sockaddr_in addr, void *arg) return 0; } + // message header + neu_msg_t *msg = neu_msg_new(ctx->event, NULL, data); + if (NULL == msg) { + return -1; + } + + neu_reqresp_head_t *header = neu_msg_get_header(msg); + strcpy(header->sender, "manager"); + strcpy(header->receiver, name); + // copy message data switch (ctx->event) { case NEU_REQ_NODE_SETTING_EVENT: { if (0 != neu_req_node_setting_copy(&setting, ctx->data)) { + neu_msg_free(msg); return -1; } data = &setting; @@ -1634,6 +1661,7 @@ static int notify_monitor(const char *name, struct sockaddr_in addr, void *arg) case NEU_REQ_ADD_TAG_EVENT: case NEU_REQ_UPDATE_TAG_EVENT: { if (0 != neu_req_add_tag_copy(&mod_tag, ctx->data)) { + neu_msg_free(msg); return -1; } data = &mod_tag; @@ -1641,6 +1669,7 @@ static int notify_monitor(const char *name, struct sockaddr_in addr, void *arg) } case NEU_REQ_DEL_TAG_EVENT: { if (0 != neu_req_del_tag_copy(&del_tag, ctx->data)) { + neu_msg_free(msg); return -1; } data = &del_tag; @@ -1650,23 +1679,12 @@ static int notify_monitor(const char *name, struct sockaddr_in addr, void *arg) data = ctx->data; } - // message header - memset(ctx->manager->buf, 0, sizeof(ctx->manager->buf)); - neu_reqresp_head_t *header = (neu_reqresp_head_t *) ctx->manager->buf; - - strcpy(header->sender, "manager"); - strcpy(header->receiver, name); - header->type = ctx->event; - - // send message - neu_msg_gen(header, data); - - int ret = sendto(ctx->manager->server_fd, header, header->len, 0, - (struct sockaddr *) &addr, sizeof(addr)); - if (ret <= 0) { + int ret = neu_send_msg_to(ctx->manager->server_fd, &addr, msg); + if (0 != ret) { nlog_warn("notify %s of %s, error: %s(%d)", header->receiver, neu_reqresp_type_string(header->type), strerror(errno), errno); + neu_msg_free(msg); } return 0; @@ -1747,16 +1765,20 @@ inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header, struct sockaddr_in addr = neu_node_manager_get_addr(manager->node_manager, header->receiver); - int ret = sendto(manager->server_fd, header, header->len, 0, - (struct sockaddr *) &addr, sizeof(addr)); + neu_reqresp_type_e t = header->type; + void * ctx = header->ctx; + char receiver[NEU_NODE_NAME_LEN] = { 0 }; + strncpy(receiver, header->receiver, sizeof(receiver)); - if (ret != (int) header->len) { - nlog_warn("reply %s to %s, error: %d", - neu_reqresp_type_string(header->type), header->receiver, ret); + neu_msg_t *msg = (neu_msg_t *) header; + int ret = neu_send_msg_to(manager->server_fd, &addr, msg); + if (0 == ret) { + nlog_notice("reply %s to %s(%d) %p", neu_reqresp_type_string(t), + receiver, ntohs(addr.sin_port), ctx); } else { - nlog_notice("reply %s to %s(%d) %p", - neu_reqresp_type_string(header->type), header->receiver, - ntohs(addr.sin_port), header->ctx); + nlog_warn("reply %s to %s, error: %d", neu_reqresp_type_string(t), + receiver, ret); + neu_msg_free(msg); } } diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 06ef3a53..2b282e65 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -26,6 +26,7 @@ #include "adapter/adapter_internal.h" #include "adapter/driver/driver_internal.h" +#include "base/msg_internal.h" #include "manager_internal.h" #include "template_manager.h" @@ -730,12 +731,6 @@ int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, uint16_t app_port, const char *params) { - memset(manager->buf, 0, sizeof(manager->buf)); - neu_reqresp_head_t *header = (neu_reqresp_head_t *) manager->buf; - header->type = NEU_REQ_SUBSCRIBE_GROUP; - strcpy(header->sender, "manager"); - strcpy(header->receiver, app); - neu_req_subscribe_t cmd = { 0 }; strcpy(cmd.app, app); strcpy(cmd.driver, driver); @@ -746,35 +741,47 @@ int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, return NEU_ERR_EINTERNAL; } + neu_msg_t *msg = neu_msg_new(NEU_REQ_SUBSCRIBE_GROUP, NULL, &cmd); + if (NULL == msg) { + free(cmd.params); + return NEU_ERR_EINTERNAL; + } + neu_reqresp_head_t *header = neu_msg_get_header(msg); + strcpy(header->sender, "manager"); + strcpy(header->receiver, app); + struct sockaddr_in addr = neu_node_manager_get_addr(manager->node_manager, app); - neu_msg_gen(header, &cmd); - - int ret = sendto(manager->server_fd, header, header->len, 0, - (struct sockaddr *) &addr, sizeof(addr)); - if (ret != (int) header->len) { + int ret = neu_send_msg_to(manager->server_fd, &addr, msg); + if (0 != ret) { nlog_warn("send %s to %s app failed", neu_reqresp_type_string(header->type), app); free(cmd.params); + neu_msg_free(msg); } else { nlog_notice("send %s to %s app", neu_reqresp_type_string(header->type), app); } + cmd.params = NULL; + msg = neu_msg_new(NEU_REQ_SUBSCRIBE_GROUP, NULL, &cmd); + if (NULL == msg) { + return NEU_ERR_EINTERNAL; + } + header = neu_msg_get_header(msg); + strcpy(header->sender, "manager"); strcpy(header->receiver, driver); addr = neu_node_manager_get_addr(manager->node_manager, driver); - neu_msg_gen(header, &cmd); - - ret = sendto(manager->server_fd, header, header->len, 0, - (struct sockaddr *) &addr, sizeof(addr)); - if (ret != (int) header->len) { + ret = neu_send_msg_to(manager->server_fd, &addr, msg); + if (0 != ret) { nlog_warn("send %s to %s driver failed", - neu_reqresp_type_string(header->type), app); + neu_reqresp_type_string(NEU_REQ_SUBSCRIBE_GROUP), driver); + neu_msg_free(msg); } else { nlog_notice("send %s to %s driver", - neu_reqresp_type_string(header->type), app); + neu_reqresp_type_string(NEU_REQ_SUBSCRIBE_GROUP), driver); } return 0; diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index ea39d136..ba8d66a0 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -29,9 +29,7 @@ #include "template_manager.h" typedef struct neu_manager { - int server_fd; - uint8_t buf[NEU_MSG_MAX_SIZE]; - uint8_t recv_buf[NEU_MSG_MAX_SIZE]; + int server_fd; neu_events_t * events; neu_event_io_t *loop;