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

feat: use abstract domain socket instead of UDP

Browse Source
This commit is contained in:
eeff
2024-02-21 16:33:49 +08:00
parent 91b847dc3e
commit b0d12b07a5
11 changed files with 79 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+3 -1
View File
@@ -26,6 +26,8 @@ extern "C" {
#include <stdint.h>
#include <sys/un.h>
#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;
src/adapter/adapter.c
+26 -18
View File
@@ -26,6 +26,7 @@
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <sys/un.h>
#include <unistd.h>
#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);
src/adapter/driver/driver.c
+6 -6
View File
@@ -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);
src/base/msg_internal.h
+4 -3
View File
@@ -26,6 +26,7 @@ extern "C" {
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/un.h>
#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) {
src/core/manager.c
+19 -18
View File
@@ -21,6 +21,7 @@
#include <string.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/un.h>
#include <unistd.h>
#include <arpa/inet.h>
@@ -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);
src/core/manager_internal.c
+2 -2
View File
@@ -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);
src/core/manager_internal.h
+3 -3
View File
@@ -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);
}
}
src/core/node_manager.c
+9 -14
View File
@@ -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);
src/core/node_manager.h
+3 -2
View File
@@ -21,6 +21,7 @@
#define _NEU_NODE_MANAGER_H_
#include <arpa/inet.h>
#include <sys/un.h>
#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
src/core/subscribe.c
+1 -1
View File
@@ -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 };
src/core/subscribe.h
+3 -2
View File
@@ -21,6 +21,7 @@
#define _NEU_SUBSCRIBE_H_
#include <netinet/in.h>
#include <sys/un.h>
#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,