diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 329de8a2..1fa29aa6 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -26,6 +26,8 @@ extern "C" { #include +#include + #include "define.h" #include "metrics.h" #include "msg.h" @@ -47,7 +49,7 @@ typedef struct adapter_callbacks { int (*response)(neu_adapter_t *adapter, neu_reqresp_head_t *head, void *data); int (*responseto)(neu_adapter_t *adapter, neu_reqresp_head_t *head, - void *data, struct sockaddr_in dst); + void *data, struct sockaddr_un dst); neu_adapter_register_metric_cb_t register_metric; neu_adapter_update_metric_cb_t update_metric; diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 18489ae4..69b08253 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -26,6 +26,7 @@ #include #include #include +#include #include #include "utils/log.h" @@ -49,7 +50,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data); static int adapter_responseto(neu_adapter_t * adapter, neu_reqresp_head_t *header, void *data, - struct sockaddr_in dst); + struct sockaddr_un dst); static int adapter_register_metric(neu_adapter_t *adapter, const char *name, const char *help, neu_metric_type_e type, uint64_t init); @@ -151,14 +152,12 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load) break; } - adapter->control_fd = - socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP); + adapter->control_fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK, 0); if (adapter->control_fd <= 0) { free(adapter); return NULL; } - adapter->trans_data_fd = - socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP); + adapter->trans_data_fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK, 0); if (adapter->trans_data_fd <= 0) { close(adapter->control_fd); free(adapter); @@ -179,13 +178,24 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load) adapter->trans_data_port = 0; adapter->log_level = ZLOG_LEVEL_NOTICE; - struct sockaddr_in remote = { - .sin_family = AF_INET, - .sin_port = htons(7788), - .sin_addr.s_addr = inet_addr("127.0.0.1"), + // use port number to distinguish each Linux abstract domain socket + uint16_t port = neu_manager_get_port(); + struct sockaddr_un local = { + .sun_family = AF_UNIX, }; + snprintf(local.sun_path, sizeof(local.sun_path), "%cneuron-%" PRIu16, '\0', + port); + rv = bind(adapter->control_fd, (struct sockaddr *) &local, + sizeof(struct sockaddr_un)); + assert(rv == 0); + + struct sockaddr_un remote = { + .sun_family = AF_UNIX, + .sun_path = "#neuron-manager", + }; + remote.sun_path[0] = '\0'; rv = connect(adapter->control_fd, (struct sockaddr *) &remote, - sizeof(struct sockaddr_in)); + sizeof(struct sockaddr_un)); assert(rv == 0); switch (info->module->type) { @@ -200,14 +210,12 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load) pthread_create(&adapter->consumer_tid, NULL, adapter_consumer, (void *) adapter); while (true) { - uint16_t port = neu_manager_get_port(); - struct sockaddr_in local = { - .sin_family = AF_INET, - .sin_port = htons(port), - .sin_addr.s_addr = inet_addr("127.0.0.1"), - }; + // use port number to distinguish each Linux abstract domain socket + port = neu_manager_get_port(); + snprintf(local.sun_path, sizeof(local.sun_path), + "%cneuron-%" PRIu16, '\0', port); if (bind(adapter->trans_data_fd, (struct sockaddr *) &local, - sizeof(struct sockaddr_in)) == 0) { + sizeof(struct sockaddr_un)) == 0) { adapter->trans_data_port = port; break; } @@ -541,7 +549,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, static int adapter_responseto(neu_adapter_t * adapter, neu_reqresp_head_t *header, void *data, - struct sockaddr_in dst) + struct sockaddr_un dst) { assert(header->type == NEU_REQRESP_TRANS_DATA); diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index a19f5f13..97672f5e 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -45,7 +45,7 @@ typedef struct to_be_write_tag { typedef struct { char app[NEU_NODE_NAME_LEN]; - struct sockaddr_in addr; + struct sockaddr_un addr; } sub_app_t; typedef struct group { @@ -81,7 +81,7 @@ struct neu_adapter_driver { }; static void report_to_app(neu_adapter_driver_t *driver, group_t *group, - struct sockaddr_in dst); + struct sockaddr_un dst); static int report_callback(void *usr_data); static int read_callback(void *usr_data); static int write_callback(void *usr_data); @@ -1457,7 +1457,7 @@ UT_array *neu_adapter_driver_get_ptag(neu_adapter_driver_t *driver, } static void report_to_app(neu_adapter_driver_t *driver, group_t *group, - struct sockaddr_in dst) + struct sockaddr_un dst) { neu_reqresp_head_t header = { .type = NEU_REQRESP_TRANS_DATA, @@ -2053,9 +2053,9 @@ void neu_adapter_driver_subscribe(neu_adapter_driver_t *driver, } strcpy(sub_app.app, req->app); - sub_app.addr.sin_family = AF_INET; - sub_app.addr.sin_addr.s_addr = inet_addr("127.0.0.1"); - sub_app.addr.sin_port = htons(req->port); + sub_app.addr.sun_family = AF_UNIX; + snprintf(sub_app.addr.sun_path, sizeof(sub_app.addr.sun_path), + "%cneuron-%" PRIu16, '\0', req->port); utarray_push_back(find->apps, &sub_app); pthread_mutex_unlock(&find->apps_mtx); diff --git a/src/base/msg_internal.h b/src/base/msg_internal.h index 0f772f00..de6b9c25 100644 --- a/src/base/msg_internal.h +++ b/src/base/msg_internal.h @@ -26,6 +26,7 @@ extern "C" { #include #include +#include #include "msg.h" @@ -224,7 +225,7 @@ inline static int neu_recv_msg(int fd, neu_msg_t **msg_p) return 0; } -inline static int neu_send_msg_to(int fd, struct sockaddr_in *addr, +inline static int neu_send_msg_to(int fd, struct sockaddr_un *addr, neu_msg_t *msg) { int ret = sendto(fd, &msg, sizeof(neu_msg_t *), 0, (struct sockaddr *) addr, @@ -232,11 +233,11 @@ inline static int neu_send_msg_to(int fd, struct sockaddr_in *addr, return sizeof(neu_msg_t *) == ret ? 0 : ret; } -inline static int neu_recv_msg_from(int fd, struct sockaddr_in *addr, +inline static int neu_recv_msg_from(int fd, struct sockaddr_un *addr, neu_msg_t **msg_p) { neu_msg_t *msg = NULL; - socklen_t addr_len = sizeof(struct sockaddr_in); + socklen_t addr_len = sizeof(struct sockaddr_un); int ret = recvfrom(fd, &msg, sizeof(neu_msg_t *), 0, (struct sockaddr *) addr, &addr_len); if (sizeof(neu_msg_t *) != ret) { diff --git a/src/core/manager.c b/src/core/manager.c index d139736e..573eb06b 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -21,6 +21,7 @@ #include #include #include +#include #include #include @@ -98,17 +99,17 @@ neu_manager_t *neu_manager_create() manager->subscribe_manager = neu_subscribe_manager_create(); manager->log_level = ZLOG_LEVEL_NOTICE; - manager->server_fd = - socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP); + manager->server_fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK, 0); assert(manager->server_fd > 0); - struct sockaddr_in local = { - .sin_family = AF_INET, - .sin_port = htons(7788), - .sin_addr.s_addr = inet_addr("127.0.0.1"), + // abstract domain socket is a Linux extension, thus not portable. + // use abstract domain socket here to avoid polluting the file system + struct sockaddr_un local = { + .sun_family = AF_UNIX, + .sun_path = "#neuron-manager", }; - rv = bind(manager->server_fd, (struct sockaddr *) &local, - sizeof(struct sockaddr_in)); + local.sun_path[0] = '\0'; + rv = bind(manager->server_fd, (struct sockaddr *) &local, sizeof(local)); assert(rv == 0); param.fd = manager->server_fd; @@ -156,14 +157,14 @@ void neu_manager_destroy(neu_manager_t *manager) neu_event_del_timer(manager->events, manager->timer_timestamp); - utarray_foreach(addrs, struct sockaddr_in *, addr) + utarray_foreach(addrs, struct sockaddr_un *, addr) { 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)); + &addr->sun_path[1]); neu_msg_free(msg); } } @@ -192,7 +193,7 @@ 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; - struct sockaddr_in src_addr = { 0 }; + struct sockaddr_un src_addr = { 0 }; neu_msg_t * msg = NULL; neu_reqresp_head_t *header = NULL; @@ -218,8 +219,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (0 != neu_node_manager_update(manager->node_manager, init->node, src_addr)) { - nlog_warn("bind node %s to src addr(%d) fail", init->node, - ntohs(src_addr.sin_port)); + nlog_warn("bind node %s to src addr(%s) fail", init->node, + &src_addr.sun_path[1]); neu_msg_free(msg); break; } @@ -234,8 +235,8 @@ 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)); + nlog_notice("bind node %s to src addr(%s)", init->node, + &src_addr.sun_path[1]); neu_msg_free(msg); break; } @@ -1337,7 +1338,7 @@ inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header, void *data) { neu_msg_gen(header, data); - struct sockaddr_in addr = + struct sockaddr_un addr = neu_node_manager_get_addr(manager->node_manager, header->receiver); neu_reqresp_type_e t = header->type; @@ -1348,8 +1349,8 @@ inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header, 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); + nlog_notice("reply %s to %s(%s) %p", neu_reqresp_type_string(t), + receiver, &addr.sun_path[1], ctx); } else { nlog_warn("reply %s to %s, error: %d", neu_reqresp_type_string(t), receiver, ret); diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index ce17f257..5efe8af5 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -209,7 +209,7 @@ static inline int manager_subscribe(neu_manager_t *manager, const char *app, const char *params) { int ret = NEU_ERR_SUCCESS; - struct sockaddr_in addr = { 0 }; + struct sockaddr_un addr = { 0 }; neu_adapter_t * adapter = neu_node_manager_find(manager->node_manager, driver); @@ -291,7 +291,7 @@ int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, strcpy(header->sender, "manager"); strcpy(header->receiver, app); - struct sockaddr_in addr = + struct sockaddr_un addr = neu_node_manager_get_addr(manager->node_manager, app); int ret = neu_send_msg_to(manager->server_fd, &addr, msg); diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 12f71fc2..035abc62 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -90,7 +90,7 @@ int neu_manager_add_drivers(neu_manager_t * manager, inline static void forward_msg(neu_manager_t * manager, neu_reqresp_head_t *header, const char *node) { - struct sockaddr_in addr = + struct sockaddr_un addr = neu_node_manager_get_addr(manager->node_manager, node); neu_reqresp_type_e t = header->type; @@ -102,8 +102,8 @@ inline static void forward_msg(neu_manager_t * manager, if (0 == ret) { nlog_info("forward msg %s to %s", neu_reqresp_type_string(t), receiver); } else { - nlog_warn("forward msg %s to node (%s)%d fail", - neu_reqresp_type_string(t), receiver, ntohs(addr.sin_port)); + nlog_warn("forward msg %s to node (%s)%s fail", + neu_reqresp_type_string(t), receiver, &addr.sun_path[1]); neu_msg_free(msg); } } diff --git a/src/core/node_manager.c b/src/core/node_manager.c index 10db9a67..a9cb0901 100644 --- a/src/core/node_manager.c +++ b/src/core/node_manager.c @@ -32,7 +32,7 @@ typedef struct node_entity { bool is_static; bool display; bool single; - struct sockaddr_in addr; + struct sockaddr_un addr; UT_hash_handle hh; } node_entity_t; @@ -127,7 +127,7 @@ int neu_node_manager_update_name(neu_node_manager_t *mgr, const char *node_name, } int neu_node_manager_update(neu_node_manager_t *mgr, const char *name, - struct sockaddr_in addr) + struct sockaddr_un addr) { node_entity_t *node = NULL; @@ -163,7 +163,7 @@ bool neu_node_manager_exist_uninit(neu_node_manager_t *mgr) HASH_ITER(hh, mgr->nodes, el, tmp) { - if (el->addr.sin_port == 0) { + if (el->addr.sun_path[1] == 0) { return true; } } @@ -303,7 +303,7 @@ bool neu_node_manager_is_driver(neu_node_manager_t *mgr, const char *name) UT_array *neu_node_manager_get_addrs(neu_node_manager_t *mgr, int type) { - UT_icd icd = { sizeof(struct sockaddr_in), NULL, NULL, NULL }; + UT_icd icd = { sizeof(struct sockaddr_un), NULL, NULL, NULL }; UT_array * addrs = NULL; node_entity_t *el = NULL, *tmp = NULL; @@ -313,8 +313,7 @@ UT_array *neu_node_manager_get_addrs(neu_node_manager_t *mgr, int type) { if (!el->is_static) { if (el->adapter->module->type & type) { - struct sockaddr_in addr = el->addr; - utarray_push_back(addrs, &addr); + utarray_push_back(addrs, &el->addr); } } } @@ -324,25 +323,21 @@ UT_array *neu_node_manager_get_addrs(neu_node_manager_t *mgr, int type) UT_array *neu_node_manager_get_addrs_all(neu_node_manager_t *mgr) { - UT_icd icd = { sizeof(struct sockaddr_in), NULL, NULL, NULL }; + UT_icd icd = { sizeof(struct sockaddr_un), NULL, NULL, NULL }; UT_array * addrs = NULL; node_entity_t *el = NULL, *tmp = NULL; utarray_new(addrs, &icd); - HASH_ITER(hh, mgr->nodes, el, tmp) - { - struct sockaddr_in addr = el->addr; - utarray_push_back(addrs, &addr); - } + HASH_ITER(hh, mgr->nodes, el, tmp) { utarray_push_back(addrs, &el->addr); } return addrs; } -struct sockaddr_in neu_node_manager_get_addr(neu_node_manager_t *mgr, +struct sockaddr_un neu_node_manager_get_addr(neu_node_manager_t *mgr, const char * name) { - struct sockaddr_in addr = { 0 }; + struct sockaddr_un addr = { 0 }; node_entity_t * node = NULL; HASH_FIND_STR(mgr->nodes, name, node); diff --git a/src/core/node_manager.h b/src/core/node_manager.h index 4e8ad82b..2ebf4cd5 100644 --- a/src/core/node_manager.h +++ b/src/core/node_manager.h @@ -21,6 +21,7 @@ #define _NEU_NODE_MANAGER_H_ #include +#include #include "define.h" #include "utils/utextend.h" @@ -40,7 +41,7 @@ int neu_node_manager_add_single(neu_node_manager_t *mgr, neu_adapter_t *adapter, int neu_node_manager_update_name(neu_node_manager_t *mgr, const char *node_name, const char *new_node_name); int neu_node_manager_update(neu_node_manager_t *mgr, const char *name, - struct sockaddr_in addr); + struct sockaddr_un addr); bool neu_node_manager_exist_uninit(neu_node_manager_t *mgr); void neu_node_manager_del(neu_node_manager_t *mgr, const char *name); uint16_t neu_node_manager_size(neu_node_manager_t *mgr); @@ -60,7 +61,7 @@ bool neu_node_manager_is_driver(neu_node_manager_t *mgr, const char *name); // addr array UT_array *neu_node_manager_get_addrs(neu_node_manager_t *mgr, int type); UT_array *neu_node_manager_get_addrs_all(neu_node_manager_t *mgr); -struct sockaddr_in neu_node_manager_get_addr(neu_node_manager_t *mgr, +struct sockaddr_un neu_node_manager_get_addr(neu_node_manager_t *mgr, const char * name); // neu_nodes_state_t array diff --git a/src/core/subscribe.c b/src/core/subscribe.c index e2b71b7c..f40255f1 100644 --- a/src/core/subscribe.c +++ b/src/core/subscribe.c @@ -173,7 +173,7 @@ void neu_subscribe_manager_unsub_all(neu_subscribe_mgr_t *mgr, const char *app) int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver, const char *app, const char *group, - const char *params, struct sockaddr_in addr) + const char *params, struct sockaddr_un addr) { sub_elem_t * find = NULL; sub_elem_key_t key = { 0 }; diff --git a/src/core/subscribe.h b/src/core/subscribe.h index 9b8d5a1d..fe6e4e48 100644 --- a/src/core/subscribe.h +++ b/src/core/subscribe.h @@ -21,6 +21,7 @@ #define _NEU_SUBSCRIBE_H_ #include +#include #include "define.h" #include "utils/utextend.h" @@ -31,7 +32,7 @@ typedef struct neu_app_subscribe { char app_name[NEU_NODE_NAME_LEN]; char *params; - struct sockaddr_in addr; + struct sockaddr_un addr; } neu_app_subscribe_t; static inline void neu_app_subscribe_fini(neu_app_subscribe_t *app_sub) @@ -54,7 +55,7 @@ size_t neu_subscribe_manager_group_count(const neu_subscribe_mgr_t *mgr, void neu_subscribe_manager_unsub_all(neu_subscribe_mgr_t *mgr, const char *app); int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver, const char *app, const char *group, - const char *params, struct sockaddr_in addr); + const char *params, struct sockaddr_un addr); int neu_subscribe_manager_unsub(neu_subscribe_mgr_t *mgr, const char *driver, const char *app, const char *group); void neu_subscribe_manager_remove(neu_subscribe_mgr_t *mgr, const char *driver,