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

Send a Exit message to manager for ensure quit the manager

Browse Source 
* Create a main adapter for send message to neuron manager
This commit is contained in:
Parker Liu
2022-01-05 18:32:59 +08:00
parent 2928a7a46a
commit aeb020e885
3 changed files with 154 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/neu_manager.c
+71 -10
View File
@@ -59,12 +59,14 @@ typedef struct manager_bind_info {
typedef enum manager_state {
MANAGER_STATE_NULL,
MANAGER_STATE_IDLE,
MANAGER_STATE_READY,
MANAGER_STATE_RUNNING,
} manager_state_e;
struct neu_manager {
const char * listen_url;
nng_mtx * mtx;
nng_cv * cv;
manager_state_e state;
bool stop;
nng_thread * thrd;
@@ -450,7 +452,6 @@ static void manager_bind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg)
static void manager_unbind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg)
{
neu_adapter_t * adapter;
neu_adapter_t * cur_adapter;
neu_manager_t * manager;
adapter_reg_entity_t *reg_entity;
@@ -461,13 +462,12 @@ static void manager_unbind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg)
* adapter, so we can not use adapter_id of this adapter to find registred
* adapter entity. We must use pipe to find current unbind adapter.
*/
adapter = (neu_adapter_t *) arg;
manager = neu_adapter_get_manager(adapter);
manager = (neu_manager_t *) arg;
nng_mtx_lock(manager->adapters_mtx);
reg_entity = find_reg_adapter_by_pipe(&manager->reg_adapters, p);
cur_adapter = reg_entity->adapter;
reg_entity = find_reg_adapter_by_pipe(&manager->reg_adapters, p);
nng_mtx_unlock(manager->adapters_mtx);
if (reg_entity != NULL) {
cur_adapter = reg_entity->adapter;
log_info("The manager unbind the adapter(%s)",
neu_adapter_get_name(cur_adapter));
nng_mtx_lock(manager->bind_info.mtx);
@@ -608,6 +608,28 @@ static int manager_unreg_adapter(neu_manager_t *manager, adapter_id_t id,
return NEU_ERR_SUCCESS;
}
int neu_manager_init_main_adapter(neu_manager_t * manager,
bind_notify_fun bind_adapter,
bind_notify_fun unbind_adapter)
{
int rv = 0;
manager_bind_info_t *bind_info;
bind_info = &manager->bind_info;
nng_mtx_lock(bind_info->mtx);
while (bind_info->bind_count != bind_info->expect_bind_count) {
nng_cv_wait(bind_info->cv);
}
nng_mtx_unlock(bind_info->mtx);
rv = nng_pipe_notify(manager->bind_info.mng_sock, NNG_PIPE_EV_ADD_POST,
bind_adapter, manager);
rv = nng_pipe_notify(manager->bind_info.mng_sock, NNG_PIPE_EV_REM_POST,
unbind_adapter, manager);
return rv;
}
int neu_manager_init_adapter(neu_manager_t *manager, neu_adapter_t *adapter)
{
int rv = 0;
@@ -627,7 +649,7 @@ int neu_manager_init_adapter(neu_manager_t *manager, neu_adapter_t *adapter)
rv = nng_pipe_notify(manager->bind_info.mng_sock, NNG_PIPE_EV_ADD_POST,
manager_bind_adapter, adapter);
rv = nng_pipe_notify(manager->bind_info.mng_sock, NNG_PIPE_EV_REM_POST,
manager_unbind_adapter, adapter);
manager_unbind_adapter, manager);
neu_adapter_init(adapter);
return rv;
}
@@ -1064,9 +1086,18 @@ static void manager_loop(void *arg)
}
nng_mtx_lock(manager->mtx);
manager->state = MANAGER_STATE_RUNNING;
manager->state = MANAGER_STATE_READY;
nng_cv_wake(manager->cv);
nng_mtx_unlock(manager->mtx);
// Wait the main process connect to manager
nng_mtx_lock(manager->mtx);
while (manager->state != MANAGER_STATE_RUNNING) {
nng_cv_wait(manager->cv);
}
nng_mtx_unlock(manager->mtx);
log_info("Register and start all default adapters");
register_default_plugins(manager);
reg_and_start_default_adapters(manager);
add_default_grp_configs(manager);
@@ -1446,9 +1477,11 @@ neu_manager_t *neu_manager_create()
neu_panic("Can't allocate mutex for manager");
}
rv = nng_cv_alloc(&manager->adapters_cv, manager->adapters_mtx);
if (rv != 0) {
neu_panic("Failed to initialize condition(cv) for vector of adapters");
rv = nng_cv_alloc(&manager->cv, manager->mtx);
rv1 = nng_cv_alloc(&manager->adapters_cv, manager->adapters_mtx);
if (rv != 0 || rv1 != 0) {
neu_panic("Failed to initialize condition(cv) for manager"
"and vector of adapters");
}
rv = vector_init(&manager->reg_adapters, DEFAULT_ADAPTER_REG_COUNT,
@@ -1482,12 +1515,40 @@ void neu_manager_destroy(neu_manager_t *manager)
vector_uninit(&manager->reg_adapters);
nng_cv_free(manager->adapters_cv);
nng_cv_free(manager->cv);
nng_mtx_free(manager->adapters_mtx);
nng_mtx_free(manager->mtx);
free(manager);
return;
}
void neu_manager_wait_ready(neu_manager_t *manager)
{
if (manager == NULL) {
return;
}
nng_mtx_lock(manager->mtx);
while (manager->state != MANAGER_STATE_READY) {
nng_cv_wait(manager->cv);
}
nng_mtx_unlock(manager->mtx);
return;
}
void neu_manager_trigger_running(neu_manager_t *manager)
{
if (manager == NULL) {
return;
}
nng_mtx_lock(manager->mtx);
manager->state = MANAGER_STATE_RUNNING;
nng_cv_wake(manager->cv);
nng_mtx_unlock(manager->mtx);
return;
}
const char *neu_manager_get_url(neu_manager_t *manager)
{
if (manager == NULL) {
src/core/neu_manager.h
+7
View File
@@ -27,11 +27,15 @@
typedef struct neu_manager neu_manager_t;
typedef void (*bind_notify_fun)(nng_pipe p, nng_pipe_ev ev, void *arg);
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);
void neu_manager_wait_ready(neu_manager_t *manager);
void neu_manager_trigger_running(neu_manager_t *manager);
int neu_manager_add_node(neu_manager_t *manager, neu_cmd_add_node_t *cmd,
neu_node_id_t *p_node_id);
@@ -80,4 +84,7 @@ int neu_manager_adapter_get_sub_grp_configs(neu_manager_t *manager,
neu_node_id_t node_id,
vector_t ** result_sgc);
int neu_manager_init_main_adapter(neu_manager_t * manager,
bind_notify_fun bind_adapter,
bind_notify_fun unbind_adapter);
#endif
src/main.c
+76 -3
View File
@@ -25,17 +25,19 @@
#include <unistd.h>
#include <nng/nng.h>
#include <nng/protocol/bus0/bus.h>
#include <nng/protocol/pair1/pair.h>
#include <nng/supplemental/util/platform.h>
#include "config.h"
#include "core/message.h"
#include "core/neu_manager.h"
#include "idhash.h"
#include "neu_log.h"
#include "neu_panic.h"
#include "restful/rest.h"
static neu_manager_t *manager;
static neu_manager_t *manager = NULL;
static nng_socket main_sock = { 0 };
static nng_mtx *log_mtx;
FILE * g_logfile;
@@ -84,8 +86,56 @@ static int read_neuron_config()
static void sig_handler(int sig)
{
// Receive abort signal, exit the neuron
log_warn("recv sig: %d", sig);
neu_manager_stop(manager);
int rv;
size_t msg_size;
nng_msg *msg;
msg_size = msg_inplace_data_get_size(sizeof(uint32_t));
rv = nng_msg_alloc(&msg, msg_size);
if (rv != 0) {
neu_panic("Can't allocate msg for stop neuron manager");
}
void * buf_ptr;
message_t *pay_msg;
pay_msg = (message_t *) nng_msg_body(msg);
msg_inplace_data_init(pay_msg, MSG_CMD_EXIT_LOOP, sizeof(uint32_t));
buf_ptr = msg_get_buf_ptr(pay_msg);
*(uint32_t *) buf_ptr = 0; // exit_code is 0
if (main_sock.id != 0) {
nng_sendmsg(main_sock, msg, 0);
} else {
nng_msg_free(msg);
}
if (manager != NULL) {
neu_manager_stop(manager);
}
}
static void bind_main_adapter(nng_pipe p, nng_pipe_ev ev, void *arg)
{
neu_manager_t *manager;
(void) ev;
manager = (neu_manager_t *) arg;
neu_manager_trigger_running(manager);
log_info("Bind the main adapter to neuron manager with pipe(%d)", p);
return;
}
static void unbind_main_adapter(nng_pipe p, nng_pipe_ev ev, void *arg)
{
(void) ev;
(void) arg;
log_info("Unbind the main adapter from neuron manager with pipe(%d)", p);
return;
}
int main(int argc, char *argv[])
@@ -143,8 +193,31 @@ int main(int argc, char *argv[])
rv = -1;
goto main_end;
}
neu_manager_wait_ready(manager);
nng_duration recv_timeout = 100;
rv = nng_pair1_open(&main_sock);
if (rv != 0) {
neu_panic("The main process can't open pipe");
}
nng_setopt(main_sock, NNG_OPT_RECVTIMEO, &recv_timeout,
sizeof(recv_timeout));
rv = neu_manager_init_main_adapter(manager, bind_main_adapter,
unbind_main_adapter);
const char *manager_url;
nng_dialer dialer = { 0 };
manager_url = neu_manager_get_url(manager);
rv = nng_dial(main_sock, manager_url, &dialer, 0);
if (rv != 0) {
neu_panic("The main process can't dial to %s", manager_url);
}
log_info("neuron main process wait for exit");
neu_manager_destroy(manager);
log_info("neuron main process close dialer");
nng_dialer_close(dialer);
main_end:
uninit();