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

refactor: send msg by pointer

Browse Source
This commit is contained in:
eeff
2023-11-23 14:36:23 +08:00
parent fc9cfcf40c
commit f1587359e5
8 changed files with 288 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+71 -58
View File
@@ -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);
}
}
src/adapter/adapter_internal.h
-3
View File
@@ -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;
src/adapter/driver/driver.c
+11 -14
View File
@@ -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);
}
src/base/msg.c
+1 -2
View File
@@ -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)
src/base/msg_internal.h
+84 -3
View File
@@ -24,6 +24,9 @@
extern "C" {
#endif
#include <sys/socket.h>
#include <sys/types.h>
#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
src/core/manager.c
+95 -73
View File
@@ -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);
}
}
src/core/manager_internal.c
+25 -18
View File
@@ -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;
src/core/manager_internal.h
+1 -3
View File
@@ -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;