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

Merge pull request #809 from eeff/eeff/dev

Browse Source 
perf: optimize persistence speed when adding tags
This commit is contained in:
fengzeroz
2022-09-22 13:37:48 +08:00
committed by GitHub
parent f70e8ac151 e7ecfe794e
commit 44e35b252c
5 changed files with 144 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+7 -3
View File
@@ -597,16 +597,20 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter,
cmd->group, &cmd->tags[i]);
if (ret == 0) {
adapter_storage_add_tag(adapter->persister, cmd->driver,
cmd->group, &cmd->tags[i]);
resp.index += 1;
} else {
resp.error = ret;
break;
}
}
}
if (resp.index) {
// we have added some tags, try to persist
adapter_storage_add_tags(adapter->persister, cmd->driver,
cmd->group, cmd->tags, resp.index);
}
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i].address);
free(cmd->tags[i].name);
src/adapter/storage.c
+19
View File
@@ -77,6 +77,25 @@ void adapter_storage_add_tag(neu_persister_t *persister, const char *node,
}
}
void adapter_storage_add_tags(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tags,
size_t n)
{
if (0 == n) {
return;
}
if (1 == n) {
return adapter_storage_add_tag(persister, node, group, &tags[0]);
}
int rv = neu_persister_store_tags(persister, node, group, tags, n);
if (0 != rv) {
nlog_error("fail store %zu tags:[%s ... %s] adapter:%s grp:%s", n,
tags->name, tags[n - 1].name, node, group);
}
}
void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tag)
{
src/adapter/storage.h
+3
View File
@@ -36,6 +36,9 @@ void adapter_storage_del_group(neu_persister_t *persister, const char *node,
const char *group);
void adapter_storage_add_tag(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tag);
void adapter_storage_add_tags(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tags,
size_t n);
void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tag);
void adapter_storage_del_tag(neu_persister_t *persister, const char *node,
src/persist/persist.c
+111
View File
@@ -629,6 +629,14 @@ neu_persister_t *neu_persister_create(const char *dir_name)
sqlite3_close(persister->db);
goto error;
}
if (SQLITE_OK !=
sqlite3_exec(persister->db, "PRAGMA journal_mode=WAL", NULL, NULL,
NULL)) {
nlog_error("db journal_mode WAL fail: %s",
sqlite3_errmsg(persister->db));
sqlite3_close(persister->db);
goto error;
}
if (0 != apply_schemas(persister->db, g_config_dir)) {
goto error;
}
@@ -908,6 +916,109 @@ int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
tag->description);
}
int neu_persister_store_tags(neu_persister_t *persister,
const char *driver_name, const char *group_name,
const neu_datatag_t *tags, size_t n)
{
sqlite3_stmt *stmt = NULL;
const char * query = "INSERT INTO tags (driver_name, group_name, name, "
"address, attribute, precision, type, decimal, "
"description) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)";
if (SQLITE_OK != sqlite3_exec(persister->db, "BEGIN", NULL, NULL, NULL)) {
nlog_error("begin transaction fail: %s", sqlite3_errmsg(persister->db));
return NEU_ERR_EINTERNAL;
}
if (SQLITE_OK !=
sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query,
sqlite3_errmsg(persister->db));
goto error;
}
for (size_t i = 0; i < n; ++i) {
const neu_datatag_t *tag = &tags[i];
sqlite3_reset(stmt);
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
nlog_error("bind `%s` with driver_name=`%s` fail: %s", query,
driver_name, sqlite3_errmsg(persister->db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
nlog_error("bind `%s` with group_name=`%s` fail: %s", query,
group_name, sqlite3_errmsg(persister->db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 3, tag->name, -1, NULL)) {
nlog_error("bind `%s` with name=`%s` fail: %s", query, tag->name,
sqlite3_errmsg(persister->db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 4, tag->address, -1, NULL)) {
nlog_error("bind `%s` with address=`%s` fail: %s", query,
tag->address, sqlite3_errmsg(persister->db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 5, tag->attribute)) {
nlog_error("bind `%s` with attribute=`%i` fail: %s", query,
tag->attribute, sqlite3_errmsg(persister->db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 6, tag->precision)) {
nlog_error("bind `%s` with precision=`%i` fail: %s", query,
tag->precision, sqlite3_errmsg(persister->db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 7, tag->type)) {
nlog_error("bind `%s` with type=`%i` fail: %s", query, tag->type,
sqlite3_errmsg(persister->db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_double(stmt, 8, tag->decimal)) {
nlog_error("bind `%s` with decimal=`%f` fail: %s", query,
tag->decimal, sqlite3_errmsg(persister->db));
goto error;
}
if (SQLITE_OK !=
sqlite3_bind_text(stmt, 9, tag->description, -1, NULL)) {
nlog_error("bind `%s` with description=`%s` fail: %s", query,
tag->description, sqlite3_errmsg(persister->db));
goto error;
}
if (SQLITE_DONE != sqlite3_step(stmt)) {
nlog_error("sqlite3_step fail: %s", sqlite3_errmsg(persister->db));
goto error;
}
}
if (SQLITE_OK != sqlite3_exec(persister->db, "COMMIT", NULL, NULL, NULL)) {
nlog_error("commit transaction fail: %s",
sqlite3_errmsg(persister->db));
goto error;
}
sqlite3_finalize(stmt);
return 0;
error:
nlog_info("rollback transaction");
sqlite3_exec(persister->db, "ROLLBACK", NULL, NULL, NULL);
sqlite3_finalize(stmt);
return NEU_ERR_EINTERNAL;
}
int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
const char *group_name, UT_array **tags)
{
src/persist/persist.h
+4
View File
@@ -172,6 +172,10 @@ int neu_persister_load_plugins(neu_persister_t *persister,
int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
const char *group_name, const neu_datatag_t *tag);
int neu_persister_store_tags(neu_persister_t *persister,
const char *driver_name, const char *group_name,
const neu_datatag_t *tags, size_t n);
/**
* Load node tag infos.
* @param persister persiter object.