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

fix(adapter): crash when adapter auto start

Browse Source
This commit is contained in:
fengzero
2022-08-23 02:58:59 +00:00
parent 00a55af7da
commit d19f9562f2
10 changed files with 31 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/resource/api.resource
-1
View File
@@ -10,7 +10,6 @@ ${PLUGIN-EKUIPER} ekuiper
${NODE_DRIVER} 1
${NODE_APP} 2
${NODE_STATE_IDLE} 0
${NODE_STATE_INIT} 1
${NODE_STATE_READY} 2
${NODE_STATE_RUNNING} 3
include/neuron/adapter.h
+1
View File
@@ -175,6 +175,7 @@ typedef struct neu_resp_error {
typedef struct {
char node[NEU_NODE_NAME_LEN];
bool auto_start;
} neu_req_node_init_t, neu_req_node_uninit_t, neu_resp_node_uninit_t;
typedef struct neu_req_add_plugin {
include/neuron/define.h
-1
View File
@@ -51,7 +51,6 @@ typedef enum {
} neu_node_link_state_e;
typedef enum {
NEU_NODE_RUNNING_STATE_IDLE = 0,
NEU_NODE_RUNNING_STATE_INIT = 1,
NEU_NODE_RUNNING_STATE_READY = 2,
NEU_NODE_RUNNING_STATE_RUNNING = 3,
persistence/0000_2.2.0_initial.sql
+1 -1
View File
@@ -25,7 +25,7 @@ CREATE TABLE IF NOT EXISTS
nodes (
name TEXT PRIMARY KEY check(length(name) <= 32),
type integer(1) NOT NULL check(type IN (1, 2)),
state integer(1) NOT NULL check(state BETWEEN 0 AND 4),
state integer(1) NOT NULL check(state BETWEEN 1 AND 4),
plugin_name TEXT NOT NULL check(length(plugin_name) <= 32)
);
src/adapter/adapter.c
+7 -9
View File
@@ -54,7 +54,7 @@ static const adapter_callbacks_t callback_funs = {
.response = adapter_response,
};
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start)
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
{
int rv = 0;
neu_adapter_t * adapter = NULL;
@@ -119,10 +119,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start)
free(adapter->setting);
adapter->setting = NULL;
}
if (adapter->state == NEU_NODE_RUNNING_STATE_READY && start) {
adapter->module->intf_funs->start(adapter->plugin);
adapter->state = NEU_NODE_RUNNING_STATE_RUNNING;
}
}
if (info->module->type == NEU_NA_TYPE_DRIVER) {
@@ -147,14 +143,18 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start)
return adapter;
}
void neu_adapter_init(neu_adapter_t *adapter)
void neu_adapter_init(neu_adapter_t *adapter, bool auto_start)
{
neu_reqresp_head_t header = { .type = NEU_REQ_NODE_INIT };
neu_req_node_init_t init = { 0 };
nng_msg * init_msg = NULL;
strcpy(header.sender, adapter->name);
strcpy(header.receiver, "manager");
strcpy(init.node, adapter->name);
init_msg = neu_msg_gen(&header, &init);
init.auto_start = auto_start;
init_msg = neu_msg_gen(&header, &init);
nng_sendmsg(adapter->sock, init_msg, 0);
}
@@ -690,7 +690,6 @@ int neu_adapter_start(neu_adapter_t *adapter)
neu_err_code_e error = NEU_ERR_SUCCESS;
switch (adapter->state) {
case NEU_NODE_RUNNING_STATE_IDLE:
case NEU_NODE_RUNNING_STATE_INIT:
error = NEU_ERR_NODE_NOT_READY;
break;
@@ -730,7 +729,6 @@ int neu_adapter_stop(neu_adapter_t *adapter)
neu_err_code_e error = NEU_ERR_SUCCESS;
switch (adapter->state) {
case NEU_NODE_RUNNING_STATE_IDLE:
case NEU_NODE_RUNNING_STATE_INIT:
case NEU_NODE_RUNNING_STATE_READY:
error = NEU_ERR_NODE_NOT_RUNNING;
src/adapter/adapter_internal.h
+2 -2
View File
@@ -61,8 +61,8 @@ typedef struct adapter_msg_handler {
adapter_handler handler;
} adapter_msg_handler_t;
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start);
void neu_adapter_init(neu_adapter_t *adapter);
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info);
void neu_adapter_init(neu_adapter_t *adapter, bool auto_start);
int neu_adapter_start(neu_adapter_t *adapter);
int neu_adapter_start_single(neu_adapter_t *adapter);
src/core/manager.c
+11 -4
View File
@@ -223,6 +223,13 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
nng_pipe pipe = nng_msg_get_pipe(msg);
neu_node_manager_update(manager->node_manager, init->node, pipe);
if (init->auto_start) {
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, init->node);
neu_adapter_start(adapter);
}
nlog_notice("bind node %s to pipe(%d)", init->node, pipe.id);
break;
}
@@ -587,9 +594,9 @@ static void start_static_adapter(neu_manager_t *manager, const char *name)
adapter_info.handle = instance.handle;
adapter_info.module = instance.module;
adapter = neu_adapter_create(&adapter_info, manager);
adapter = neu_adapter_create(&adapter_info);
neu_node_manager_add_static(manager->node_manager, adapter);
neu_adapter_init(adapter);
neu_adapter_init(adapter, false);
neu_adapter_start(adapter);
}
@@ -610,10 +617,10 @@ static void start_single_adapter(neu_manager_t *manager, const char *name,
adapter_info.handle = instance.handle;
adapter_info.module = instance.module;
adapter = neu_adapter_create(&adapter_info, true);
adapter = neu_adapter_create(&adapter_info);
neu_node_manager_add_single(manager->node_manager, adapter, display);
neu_adapter_init(adapter);
neu_adapter_init(adapter, false);
neu_adapter_start_single(adapter);
}
src/core/manager_internal.c
+2 -2
View File
@@ -70,9 +70,9 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
adapter_info.handle = instance.handle;
adapter_info.module = instance.module;
adapter = neu_adapter_create(&adapter_info, start);
adapter = neu_adapter_create(&adapter_info);
neu_node_manager_add(manager->node_manager, adapter);
neu_adapter_init(adapter);
neu_adapter_init(adapter, start);
if (instance.module->type == NEU_NA_TYPE_APP) {
for (int i = 0; i < NEU_APP_SUBSCRIBE_MSG_SIZE; i++) {
src/core/storage.c
+3 -2
View File
@@ -120,8 +120,9 @@ int manager_load_node(neu_manager_t *manager)
manager, node_info->name, node_info->plugin_name,
node_info->state == NEU_NODE_RUNNING_STATE_RUNNING);
const char *ok_or_err = (0 == rv) ? "success" : "fail";
nlog_info("load adapter %s type:%d, name:%s plugin:%s", ok_or_err,
node_info->type, node_info->name, node_info->plugin_name);
nlog_info("load adapter %s type:%d, name:%s plugin:%s state:%d",
ok_or_err, node_info->type, node_info->name,
node_info->plugin_name, node_info->state);
}
utarray_free(node_infos);
src/persist/persist.c
+4 -12
View File
@@ -320,6 +320,8 @@ static inline int execute_sql(sqlite3 *db, const char *sql, ...)
if (SQLITE_OK != sqlite3_exec(db, query, NULL, NULL, &err_msg)) {
nlog_error("query `%s` fail: %s", query, err_msg);
rv = NEU_ERR_EINTERNAL;
} else {
nlog_info("query %s success", query);
}
sqlite3_free(err_msg);
@@ -561,18 +563,7 @@ neu_persister_t *neu_persister_create(const char *dir_name)
goto error;
}
int n = path_cat(path, dir_len, sizeof(path), "adapters");
if (sizeof(path) == n) {
nlog_error("path too long: %s", path);
goto error;
}
rv = create_dir(path);
if (rv != 0) {
nlog_error("failed to create directory: %s", path);
goto error;
}
n = path_cat(path, dir_len, sizeof(path), "plugins.json");
int n = path_cat(path, dir_len, sizeof(path), "plugins.json");
if (sizeof(path) == n) {
nlog_error("path too long: %s", path);
goto error;
@@ -607,6 +598,7 @@ neu_persister_t *neu_persister_create(const char *dir_name)
nlog_error("db `%s` fail: %s", path, sqlite3_errstr(rv));
goto error;
}
sqlite3_busy_timeout(persister->db, 100 * 1000);
// we rely on sqlite foreign key support
if (SQLITE_OK !=
sqlite3_exec(persister->db, "PRAGMA foreign_keys=ON", NULL, NULL,