From e48335ddd2cd192f5826532a7ce03763a8faeab2 Mon Sep 17 00:00:00 2001 From: fengzero Date: Tue, 16 Aug 2022 08:10:50 +0000 Subject: [PATCH] feat(driver): compare whether tag value changed on precision --- plugins/modbus/modbus_req.c | 7 +- src/adapter/driver/cache.c | 224 +++++++++++++++++++++++------------- src/adapter/driver/cache.h | 14 +-- src/adapter/driver/driver.c | 91 +++++---------- 4 files changed, 179 insertions(+), 157 deletions(-) diff --git a/plugins/modbus/modbus_req.c b/plugins/modbus/modbus_req.c index 30b0e97e..73576129 100644 --- a/plugins/modbus/modbus_req.c +++ b/plugins/modbus/modbus_req.c @@ -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: diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index 703fe392..f0f1c95d 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -18,6 +18,7 @@ **/ #include +#include #include #include @@ -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); -} diff --git a/src/adapter/driver/cache.h b/src/adapter/driver/cache.h index 1110aca6..0df8b7a6 100644 --- a/src/adapter/driver/cache.h +++ b/src/adapter/driver/cache.h @@ -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, diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 1be089d1..534620f2 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -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) {