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

feat(driver): compare whether tag value changed on precision

Browse Source
This commit is contained in:
fengzero
2022-08-16 08:10:50 +00:00
parent 7a54325804
commit e48335ddd2
4 changed files with 179 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/modbus/modbus_req.c
+1 -6
View File
@@ -150,13 +150,13 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
}
}
dvalue.type = (*p_tag)->type;
switch ((*p_tag)->type) {
case NEU_TYPE_UINT16:
case NEU_TYPE_INT16:
if ((*p_tag)->option.value16.endian == NEU_DATATAG_ENDIAN_L16) {
dvalue.value.u16 = htons(dvalue.value.u16);
}
dvalue.type = NEU_TYPE_INT16;
break;
case NEU_TYPE_FLOAT: {
uint16_t v1 = ((uint16_t *) dvalue.value.bytes)[0];
@@ -181,7 +181,6 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
break;
}
dvalue.type = NEU_TYPE_FLOAT;
dvalue.value.u32 = v;
break;
}
@@ -209,9 +208,7 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
break;
}
dvalue.type = NEU_TYPE_UINT32;
dvalue.value.u32 = v;
break;
}
case NEU_TYPE_BIT: {
@@ -229,7 +226,6 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
break;
}
dvalue.type = NEU_TYPE_BIT;
break;
}
case NEU_TYPE_STRING: {
@@ -252,7 +248,6 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
dvalue.value.str[1] = 0;
}
dvalue.type = NEU_TYPE_STRING;
break;
}
default:
src/adapter/driver/cache.c
+144 -80
View File
@@ -18,6 +18,7 @@
**/
#include <assert.h>
#include <math.h>
#include <string.h>
#include <nng/nng.h>
@@ -36,12 +37,10 @@ typedef struct {
} tkey_t;
struct elem {
int32_t error;
int64_t timestamp;
bool changed;
bool changed;
uint16_t n_byte;
uint8_t *bytes;
neu_dvalue_t value;
tkey_t key;
UT_hash_handle hh;
@@ -53,8 +52,8 @@ struct neu_driver_cache {
struct elem *table;
};
static void update_tag_error(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp, int error);
// static void update_tag_error(neu_driver_cache_t *cache, const char *group,
// const char *tag, int64_t timestamp, int error);
inline static tkey_t to_key(const char *group, const char *tag)
{
@@ -84,9 +83,6 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
HASH_ITER(hh, cache->table, elem, tmp)
{
HASH_DEL(cache->table, elem);
if (elem->n_byte > 0) {
free(elem->bytes);
}
free(elem);
}
nng_mtx_unlock(cache->mtx);
@@ -96,15 +92,14 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
free(cache);
}
void neu_driver_cache_error(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp, int32_t error)
{
update_tag_error(cache, group, tag, timestamp, error);
}
// void neu_driver_cache_error(neu_driver_cache_t *cache, const char *group,
// const char *tag, int64_t timestamp, int32_t error)
//{
// update_tag_error(cache, group, tag, timestamp, error);
//}
void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp,
uint16_t n_byte, uint8_t *bytes)
void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
const char *tag, neu_dvalue_t value)
{
struct elem *elem = NULL;
tkey_t key = to_key(group, tag);
@@ -118,45 +113,78 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
strcpy(elem->key.group, group);
strcpy(elem->key.tag, tag);
assert(n_byte != 0);
assert(n_byte <= NEU_VALUE_SIZE);
elem->n_byte = n_byte;
elem->bytes = calloc(elem->n_byte, 1);
elem->changed = true;
HASH_ADD(hh, cache->table, key, sizeof(tkey_t), elem);
}
if (elem->error != 0) {
elem->changed = true;
}
elem->timestamp = 0;
elem->changed = false;
elem->value = value;
elem->error = 0;
elem->timestamp = timestamp;
nng_mtx_unlock(cache->mtx);
}
if (elem->n_byte == 0) {
assert(n_byte != 0);
assert(n_byte <= NEU_VALUE_SIZE);
elem->n_byte = n_byte;
elem->bytes = calloc(elem->n_byte, n_byte);
elem->changed = true;
} else {
if (n_byte != elem->n_byte) {
elem->n_byte = n_byte;
elem->bytes = realloc(elem->bytes, n_byte);
memset(elem->bytes, 0, elem->n_byte);
void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp,
neu_dvalue_t value)
{
struct elem *elem = NULL;
tkey_t key = to_key(group, tag);
nng_mtx_lock(cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL) {
elem->timestamp = timestamp;
if (elem->value.type != value.type) {
elem->changed = true;
} else {
if (memcmp(elem->bytes, bytes, elem->n_byte) != 0) {
switch (value.type) {
case NEU_TYPE_INT8:
case NEU_TYPE_UINT8:
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32:
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
case NEU_TYPE_BIT:
case NEU_TYPE_BOOL:
case NEU_TYPE_STRING:
case NEU_TYPE_BYTES:
if (memcmp(&elem->value.value, &value.value,
sizeof(value.value)) != 0) {
elem->changed = true;
}
break;
case NEU_TYPE_FLOAT:
if (elem->value.precision == 0) {
elem->changed = elem->value.value.f32 != value.value.f32;
} else {
if (fabs(elem->value.value.f32 - value.value.f32) >
pow(0.1, elem->value.precision)) {
elem->changed = true;
}
}
break;
case NEU_TYPE_DOUBLE:
if (elem->value.precision == 0) {
elem->changed = elem->value.value.d64 != value.value.d64;
} else {
if (fabs(elem->value.value.d64 - value.value.d64) >
pow(0.1, elem->value.precision)) {
elem->changed = true;
}
}
break;
case NEU_TYPE_ERROR:
elem->changed = true;
break;
}
}
}
memcpy(elem->bytes, bytes, elem->n_byte);
elem->value.type = value.type;
elem->value.value = value.value;
}
nng_mtx_unlock(cache->mtx);
}
@@ -172,10 +200,40 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL) {
value->error = elem->error;
value->n_byte = elem->n_byte;
value->timestamp = elem->timestamp;
memcpy(value->value.bytes, elem->bytes, elem->n_byte);
value->timestamp = elem->timestamp;
value->value.type = elem->value.type;
value->value.precision = elem->value.precision;
switch (elem->value.type) {
case NEU_TYPE_INT8:
case NEU_TYPE_UINT8:
case NEU_TYPE_BIT:
value->value.value.u8 = elem->value.value.u8;
break;
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
value->value.value.u16 = elem->value.value.u16;
break;
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32:
case NEU_TYPE_FLOAT:
case NEU_TYPE_ERROR:
value->value.value.u32 = elem->value.value.u32;
break;
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
case NEU_TYPE_DOUBLE:
value->value.value.u64 = elem->value.value.u64;
break;
case NEU_TYPE_BOOL:
value->value.value.boolean = elem->value.value.boolean;
break;
case NEU_TYPE_STRING:
case NEU_TYPE_BYTES:
memcpy(value->value.value.str, elem->value.value.str,
sizeof(elem->value.value.str));
break;
}
ret = 0;
}
@@ -197,13 +255,45 @@ int neu_driver_cache_get_changed(neu_driver_cache_t *cache, const char *group,
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL && elem->changed) {
value->error = elem->error;
value->n_byte = elem->n_byte;
value->timestamp = elem->timestamp;
memcpy(value->value.bytes, elem->bytes, elem->n_byte);
value->timestamp = elem->timestamp;
value->value.type = elem->value.type;
value->value.precision = elem->value.precision;
elem->changed = false;
ret = 0;
switch (elem->value.type) {
case NEU_TYPE_INT8:
case NEU_TYPE_UINT8:
case NEU_TYPE_BIT:
value->value.value.u8 = elem->value.value.u8;
break;
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
value->value.value.u16 = elem->value.value.u16;
break;
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32:
case NEU_TYPE_FLOAT:
case NEU_TYPE_ERROR:
value->value.value.u32 = elem->value.value.u32;
break;
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
case NEU_TYPE_DOUBLE:
value->value.value.u64 = elem->value.value.u64;
break;
case NEU_TYPE_BOOL:
value->value.value.boolean = elem->value.value.boolean;
break;
case NEU_TYPE_STRING:
case NEU_TYPE_BYTES:
memcpy(value->value.value.str, elem->value.value.str,
sizeof(elem->value.value.str));
break;
}
if (elem->value.type != NEU_TYPE_ERROR) {
elem->changed = false;
}
ret = 0;
}
nng_mtx_unlock(cache->mtx);
@@ -222,34 +312,8 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
if (elem != NULL) {
HASH_DEL(cache->table, elem);
if (elem->n_byte != 0) {
free(elem->bytes);
}
free(elem);
}
nng_mtx_unlock(cache->mtx);
}
static void update_tag_error(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp, int error)
{
struct elem *elem = NULL;
tkey_t key = to_key(group, tag);
nng_mtx_lock(cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem == NULL) {
elem = calloc(1, sizeof(struct elem));
strcpy(elem->key.group, group);
strcpy(elem->key.tag, tag);
HASH_ADD(hh, cache->table, key, sizeof(tkey_t), elem);
}
elem->error = error;
elem->timestamp = timestamp;
elem->changed = true;
nng_mtx_unlock(cache->mtx);
}
src/adapter/driver/cache.h
+6 -8
View File
@@ -29,20 +29,18 @@ typedef struct neu_driver_cache neu_driver_cache_t;
neu_driver_cache_t *neu_driver_cache_new();
void neu_driver_cache_destroy(neu_driver_cache_t *cache);
void neu_driver_cache_error(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp, int32_t error);
void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
const char *tag, neu_dvalue_t value);
void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp,
uint16_t n_byte, uint8_t *bytes);
neu_dvalue_t value);
void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
const char *tag);
typedef struct {
uint16_t n_byte;
neu_value_u value;
int32_t error;
int64_t timestamp;
neu_dvalue_t value;
int64_t timestamp;
} neu_driver_cache_value_t;
int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
src/adapter/driver/driver.c
+28 -63
View File
@@ -87,63 +87,22 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
neu_dvalue_t value)
{
neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter;
uint16_t n_byte = 0;
switch (value.type) {
case NEU_TYPE_INT8:
case NEU_TYPE_UINT8:
case NEU_TYPE_BIT:
case NEU_TYPE_BOOL:
n_byte = 1;
break;
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
n_byte = 2;
break;
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32:
case NEU_TYPE_FLOAT:
n_byte = 4;
break;
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
case NEU_TYPE_DOUBLE:
n_byte = 8;
break;
case NEU_TYPE_STRING:
n_byte = sizeof(value.value.str);
break;
case NEU_TYPE_BYTES:
break;
case NEU_TYPE_ERROR:
break;
default:
assert(value.type == NEU_TYPE_INT8);
break;
}
if (value.type == NEU_TYPE_ERROR && tag == NULL) {
group_t *g = find_group(driver, group);
if (g != NULL) {
UT_array *tags = neu_group_get_read_tag(g->group);
if (value.type == NEU_TYPE_ERROR) {
if (tag != NULL) {
neu_driver_cache_error(driver->cache, group, tag,
driver->adapter.timestamp, value.value.i32);
} else {
group_t *g = find_group(driver, group);
if (g != NULL) {
UT_array *tags = neu_group_get_read_tag(g->group);
utarray_foreach(tags, neu_datatag_t *, t)
{
neu_driver_cache_error(driver->cache, group, t->name,
driver->adapter.timestamp,
value.value.i32);
}
utarray_free(tags);
utarray_foreach(tags, neu_datatag_t *, t)
{
neu_driver_cache_update(driver->cache, group, t->name,
driver->adapter.timestamp, value);
}
utarray_free(tags);
}
} else {
neu_driver_cache_update(driver->cache, group, tag,
driver->adapter.timestamp, n_byte,
value.value.bytes);
driver->adapter.timestamp, value);
}
nlog_debug(
"update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64,
@@ -631,6 +590,18 @@ static void group_change(void *arg, int64_t timestamp, UT_array *tags,
neu_driver_cache_del(group->driver->cache, group->name, tag->name);
}
utarray_foreach(tags, neu_datatag_t *, tag)
{
neu_dvalue_t value = { 0 };
value.precision = tag->precision;
value.type = NEU_TYPE_ERROR;
value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY;
neu_driver_cache_add(group->driver->cache, group->name, tag->name,
value);
}
neu_plugin_group_t grp = {
.group_name = strdup(group->name),
.tags = NULL,
@@ -696,9 +667,8 @@ static int read_report_group(int64_t timestamp, int64_t timeout,
}
strcpy(datas[index].tag, tag->name);
if (value.error != 0) {
datas[index].value.type = NEU_TYPE_ERROR;
datas[index].value.value.i32 = value.error;
if (value.value.type == NEU_TYPE_ERROR) {
datas[index].value = value.value;
index += 1;
continue;
}
@@ -707,9 +677,7 @@ static int read_report_group(int64_t timestamp, int64_t timeout,
datas[index].value.type = NEU_TYPE_ERROR;
datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED;
} else {
datas[index].value.type = tag->type;
datas[index].value.value = value.value;
datas[index].value.precision = tag->precision;
datas[index].value = value.value;
if (tag->decimal != 0) {
datas[index].value.type = NEU_TYPE_DOUBLE;
switch (tag->type) {
@@ -770,9 +738,8 @@ static void read_group(int64_t timestamp, int64_t timeout,
continue;
}
if (value.error != 0) {
datas[index].value.type = NEU_TYPE_ERROR;
datas[index].value.value.i32 = value.error;
if (value.value.type == NEU_TYPE_ERROR) {
datas[index].value = value.value;
continue;
}
@@ -780,9 +747,7 @@ static void read_group(int64_t timestamp, int64_t timeout,
datas[index].value.type = NEU_TYPE_ERROR;
datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED;
} else {
datas[index].value.type = tag->type;
datas[index].value.value = value.value;
datas[index].value.precision = tag->precision;
datas[index].value = value.value;
if (tag->decimal != 0) {
datas[index].value.type = NEU_TYPE_DOUBLE;
switch (tag->type) {