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

Add signal processing

Browse Source 
Make the program exit normally and make libasan memory detection take
effect
This commit is contained in:
fengzero
2021-10-14 07:10:31 +00:00
committed by Parker Liu
parent 7572835b5b
commit 34647e8ae0
11 changed files with 94 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+9 -8
View File
@@ -191,7 +191,8 @@ static void adapter_loop(void *arg)
rv = nng_recvmsg(adapter->sock, &msg, 0);
if (rv != 0) {
log_warn("Manage pipe no message received");
if (rv != NNG_ETIMEDOUT)
log_warn("Manage pipe no message received %d", rv);
continue;
}
@@ -331,13 +332,11 @@ static void adapter_loop(void *arg)
const neu_plugin_intf_funs_t *intf_funs;
neu_request_t req;
if (adapter->plugin_module) {
ssize_t size;
neu_reqresp_read_resp_t read_resp;
read_resp.grp_config = cmd_ptr->grp_config;
void * buf = core_databuf_get_ptr(cmd_ptr->databuf);
size_t buf_len = core_databuf_get_len(cmd_ptr->databuf);
size =
neu_dvalue_deserialize(buf, buf_len, &read_resp.data_val);
neu_dvalue_deserialize(buf, buf_len, &read_resp.data_val);
intf_funs = adapter->plugin_module->intf_funs;
req.req_id = adapter_get_req_id(adapter);
@@ -388,13 +387,11 @@ static void adapter_loop(void *arg)
const neu_plugin_intf_funs_t *intf_funs;
neu_request_t req;
if (adapter->plugin_module) {
ssize_t size;
neu_reqresp_write_resp_t write_resp;
write_resp.grp_config = cmd_ptr->grp_config;
void * buf = core_databuf_get_ptr(cmd_ptr->databuf);
size_t buf_len = core_databuf_get_len(cmd_ptr->databuf);
size =
neu_dvalue_deserialize(buf, buf_len, &write_resp.data_val);
neu_dvalue_deserialize(buf, buf_len, &write_resp.data_val);
intf_funs = adapter->plugin_module->intf_funs;
req.req_id = adapter_get_req_id(adapter);
@@ -444,7 +441,7 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
return (-1);
}
log_info("Get command from plugin %d", cmd->req_type);
log_info("Get command from plugin %d, %s", cmd->req_type, adapter->name);
switch (cmd->req_type) {
case NEU_REQRESP_SUBSCRIBE_NODE: {
ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_SUBSCRIBE_NODE,
@@ -876,6 +873,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info,
neu_manager_t * manager)
{
neu_adapter_t *adapter;
nng_duration recv_timeout = 100;
if (manager == NULL) {
log_error("Create adapter with NULL manager");
@@ -948,6 +946,9 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info,
neu_panic("The adapter(%s) can't open pipe", adapter->name);
}
nng_setopt(adapter->sock, NNG_OPT_RECVTIMEO, &recv_timeout,
sizeof(recv_timeout));
log_info("Success to create adapter: %s", adapter->name);
return adapter;
}
src/core/neu_manager.c
+13 -7
View File
@@ -552,13 +552,12 @@ static int manager_unreg_adapter(neu_manager_t *manager, adapter_id_t id)
nng_mtx_lock(manager->adapters_mtx);
index = find_reg_adapter_index_by_id(reg_adapters, id);
if (index != SIZE_MAX) {
adapter_reg_entity_t *reg_entity;
// adapter_reg_entity_t *reg_entity;
reg_entity = (adapter_reg_entity_t *) vector_get(reg_adapters, index);
adapter = reg_entity->adapter;
datatag_mng = reg_entity->datatag_manager;
cv = reg_entity->cv;
vector_erase(reg_adapters, index);
// reg_entity = (adapter_reg_entity_t *) vector_get(reg_adapters,
// index); adapter = reg_entity->adapter; datatag_mng =
// reg_entity->datatag_manager; cv = reg_entity->cv;
// vector_erase(reg_adapters, index);
}
nng_mtx_unlock(manager->adapters_mtx);
@@ -1325,9 +1324,9 @@ static void manager_loop(void *arg)
}
log_info("End message loop of neu_manager");
remove_default_grp_configs(manager);
stop_and_unreg_bind_adapters(manager);
unregister_all_reg_plugins(manager);
remove_default_grp_configs(manager);
nng_close(manager_bind->mng_sock);
return;
}
@@ -1368,6 +1367,13 @@ neu_manager_t *neu_manager_create()
return manager;
}
void neu_manager_stop(neu_manager_t *manager)
{
nng_mtx_lock(manager->mtx);
manager->stop = true;
nng_mtx_unlock(manager->mtx);
}
void neu_manager_destroy(neu_manager_t *manager)
{
nng_thread_destroy(manager->thrd);
src/core/neu_manager.h
+1
View File
@@ -28,6 +28,7 @@
typedef struct neu_manager neu_manager_t;
neu_manager_t *neu_manager_create();
void neu_manager_stop(neu_manager_t *manager);
void neu_manager_destroy(neu_manager_t *manager);
const char *neu_manager_get_url(neu_manager_t *manager);
src/main.c
+13 -2
View File
@@ -18,6 +18,7 @@
**/
#include <getopt.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@@ -36,6 +37,7 @@
static nng_mtx *log_mtx;
FILE * g_logfile;
neu_manager_t * manager;
static void log_lock(bool lock, void *udata)
{
@@ -52,7 +54,7 @@ static void init()
nng_mtx_alloc(&log_mtx);
log_set_lock(log_lock, log_mtx);
log_set_level(LOG_DEBUG);
FILE *g_logfile = fopen("neuron.log", "a");
g_logfile = fopen("neuron.log", "a");
if (g_logfile == NULL) {
fprintf(stderr,
"Failed to open logfile when"
@@ -79,6 +81,12 @@ static int read_neuron_config()
return neu_config_init("./neuron.yaml");
}
static void sig_handler(int sig)
{
log_warn("recv sig: %d", sig);
neu_manager_stop(manager);
}
int main(int argc, char *argv[])
{
@@ -113,7 +121,10 @@ int main(int argc, char *argv[])
goto main_end;
}
neu_manager_t *manager;
signal(SIGINT, sig_handler);
signal(SIGTERM, sig_handler);
signal(SIGABRT, sig_handler);
log_info("running neuron main process, daemon: %d", is_daemon);
manager = neu_manager_create();
if (manager == NULL) {