diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index 82580acb..5aaa1c57 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -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) { diff --git a/src/core/neu_manager.h b/src/core/neu_manager.h index 836c1cc9..184bf714 100644 --- a/src/core/neu_manager.h +++ b/src/core/neu_manager.h @@ -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 diff --git a/src/main.c b/src/main.c index d5be0ba7..ed5eb220 100644 --- a/src/main.c +++ b/src/main.c @@ -25,17 +25,19 @@ #include #include -#include +#include #include #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();