From 7ea2bbf35effa0636d99d7df78aa6f3f624172c9 Mon Sep 17 00:00:00 2001 From: fengzero Date: Fri, 21 Jul 2023 08:28:51 +0000 Subject: [PATCH] feat(driver): new interface for updating tags, tags can carry metadata --- include/neuron/adapter.h | 3 ++ include/neuron/tag.h | 11 +++- src/adapter/driver/cache.c | 25 ++++++++- src/adapter/driver/cache.h | 8 +-- src/adapter/driver/driver.c | 102 ++++++++++++++++++++---------------- 5 files changed, 99 insertions(+), 50 deletions(-) diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index fa625462..2ed0a497 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -791,6 +791,9 @@ typedef struct adapter_callbacks { struct { void (*update)(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value); + void (*update_with_meta)(neu_adapter_t *adapter, const char *group, + const char *tag, neu_dvalue_t value, + neu_tag_meta_t *metas, int n_meta); void (*write_response)(neu_adapter_t *adapter, void *req, int error); void (*update_im)(neu_adapter_t *adapter, const char *group, diff --git a/include/neuron/tag.h b/include/neuron/tag.h index ad08015e..e18e414b 100644 --- a/include/neuron/tag.h +++ b/include/neuron/tag.h @@ -23,11 +23,13 @@ config_ **/ #include #include +#include "define.h" #include "type.h" #include "utils/utextend.h" #include "json/json.h" -#define NEU_TAG_META_SIZE 20 +#define NEU_TAG_META_LENGTH 20 +#define NEU_TAG_META_SIZE 3 typedef enum { NEU_ATTRIBUTE_READ = 1, @@ -84,9 +86,14 @@ typedef struct { double decimal; char * description; neu_datatag_addr_option_u option; - uint8_t meta[NEU_TAG_META_SIZE]; + uint8_t meta[NEU_TAG_META_LENGTH]; } neu_datatag_t; +typedef struct neu_tag_meta { + char name[NEU_TAG_NAME_LEN]; + neu_dvalue_t value; +} neu_tag_meta_t; + UT_icd *neu_tag_get_icd(); neu_datatag_t *neu_tag_dup(const neu_datatag_t *tag); diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index 0e514d5b..f692ebef 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -42,6 +42,8 @@ struct elem { neu_dvalue_t value; + neu_tag_meta_t metas[NEU_TAG_META_SIZE]; + tkey_t key; UT_hash_handle hh; }; @@ -125,7 +127,8 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group, void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group, const char *tag, int64_t timestamp, - neu_dvalue_t value) + neu_dvalue_t value, neu_tag_meta_t *metas, + int n_meta) { struct elem *elem = NULL; tkey_t key = to_key(group, tag); @@ -187,6 +190,10 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group, elem->value.type = value.type; elem->value.value = value.value; + memset(elem->metas, 0, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE); + for (int i = 0; i < n_meta; i++) { + memcpy(&elem->metas[i], &metas[i], sizeof(neu_tag_meta_t)); + } } nng_mtx_unlock(cache->mtx); @@ -241,6 +248,14 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group, break; } + for (int i = 0; i < NEU_TAG_META_SIZE; i++) { + if (elem->metas[i].name != NULL && + strlen(elem->metas[i].name) > 0) { + memcpy(&value->metas[i], &elem->metas[i], + sizeof(neu_tag_meta_t)); + } + } + ret = 0; } @@ -299,6 +314,14 @@ int neu_driver_cache_get_changed(neu_driver_cache_t *cache, const char *group, break; } + for (int i = 0; i < NEU_TAG_META_SIZE; i++) { + if (elem->metas[i].name != NULL && + strlen(elem->metas[i].name) > 0) { + memcpy(&value->metas[i], &elem->metas[i], + sizeof(neu_tag_meta_t)); + } + } + if (elem->value.type != NEU_TYPE_ERROR) { elem->changed = false; } diff --git a/src/adapter/driver/cache.h b/src/adapter/driver/cache.h index 0df8b7a6..d1d96e13 100644 --- a/src/adapter/driver/cache.h +++ b/src/adapter/driver/cache.h @@ -33,14 +33,16 @@ 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, - neu_dvalue_t value); + neu_dvalue_t value, neu_tag_meta_t *metas, + int n_meta); void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group, const char *tag); typedef struct { - neu_dvalue_t value; - int64_t timestamp; + neu_dvalue_t value; + int64_t timestamp; + neu_tag_meta_t metas[NEU_TAG_META_SIZE]; } 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 3c84fa23..3bf3c95c 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -85,6 +85,11 @@ static int read_report_group(int64_t timestamp, int64_t timeout, UT_array *tags, neu_resp_tag_value_t *datas); static void update(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value); +static void update_im(neu_adapter_t *adapter, const char *group, + const char *tag, neu_dvalue_t value); +static void update_with_meta(neu_adapter_t *adapter, const char *group, + const char *tag, neu_dvalue_t value, + neu_tag_meta_t *metas, int n_meta); static void write_response(neu_adapter_t *adapter, void *r, neu_error error); static group_t *find_group(neu_adapter_driver_t *driver, const char *name); static void store_write_tag(group_t *group, to_be_write_tag_t *tag); @@ -100,6 +105,49 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error) free(req); } +static void update_with_meta(neu_adapter_t *adapter, const char *group, + const char *tag, neu_dvalue_t value, + neu_tag_meta_t *metas, int n_meta) +{ + neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter; + + 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); + uint64_t err_count = 0; + + utarray_foreach(tags, neu_datatag_t *, t) + { + if (neu_tag_attribute_test(t, NEU_ATTRIBUTE_STATIC)) { + continue; + } + neu_driver_cache_update(driver->cache, group, t->name, + global_timestamp, value, NULL, 0); + ++err_count; + } + driver->adapter.cb_funs.update_metric( + &driver->adapter, NEU_METRIC_TAG_READS_TOTAL, err_count, NULL); + driver->adapter.cb_funs.update_metric( + &driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, err_count, + NULL); + utarray_free(tags); + } + } else { + neu_driver_cache_update(driver->cache, group, tag, global_timestamp, + value, metas, n_meta); + driver->adapter.cb_funs.update_metric( + &driver->adapter, NEU_METRIC_TAG_READS_TOTAL, 1, NULL); + driver->adapter.cb_funs.update_metric( + &driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, + NEU_TYPE_ERROR == value.type, NULL); + } + nlog_info( + "update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64, + driver->adapter.name, group, tag, neu_type_string(value.type), + global_timestamp); +} + static void update_im(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value) { @@ -112,7 +160,8 @@ static void update_im(neu_adapter_t *adapter, const char *group, } UT_array *tags = neu_adapter_driver_get_ptag(driver, group, tag); - neu_driver_cache_update(driver->cache, group, tag, global_timestamp, value); + neu_driver_cache_update(driver->cache, group, tag, global_timestamp, value, + NULL, 0); driver->adapter.cb_funs.update_metric(&driver->adapter, NEU_METRIC_TAG_READS_TOTAL, 1, NULL); neu_datatag_t *first = utarray_front(tags); @@ -155,54 +204,19 @@ static void update_im(neu_adapter_t *adapter, const char *group, 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; - - 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); - uint64_t err_count = 0; - - utarray_foreach(tags, neu_datatag_t *, t) - { - if (neu_tag_attribute_test(t, NEU_ATTRIBUTE_STATIC)) { - continue; - } - neu_driver_cache_update(driver->cache, group, t->name, - global_timestamp, value); - ++err_count; - } - driver->adapter.cb_funs.update_metric( - &driver->adapter, NEU_METRIC_TAG_READS_TOTAL, err_count, NULL); - driver->adapter.cb_funs.update_metric( - &driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, err_count, - NULL); - utarray_free(tags); - } - } else { - neu_driver_cache_update(driver->cache, group, tag, global_timestamp, - value); - driver->adapter.cb_funs.update_metric( - &driver->adapter, NEU_METRIC_TAG_READS_TOTAL, 1, NULL); - driver->adapter.cb_funs.update_metric( - &driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, - NEU_TYPE_ERROR == value.type, NULL); - } - nlog_info( - "update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64, - driver->adapter.name, group, tag, neu_type_string(value.type), - global_timestamp); + update_with_meta(adapter, group, tag, value, NULL, 0); } neu_adapter_driver_t *neu_adapter_driver_create() { neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t)); - driver->cache = neu_driver_cache_new(); - driver->driver_events = neu_event_new(); - driver->adapter.cb_funs.driver.update = update; - driver->adapter.cb_funs.driver.write_response = write_response; - driver->adapter.cb_funs.driver.update_im = update_im; + driver->cache = neu_driver_cache_new(); + driver->driver_events = neu_event_new(); + driver->adapter.cb_funs.driver.update = update; + driver->adapter.cb_funs.driver.write_response = write_response; + driver->adapter.cb_funs.driver.update_im = update_im; + driver->adapter.cb_funs.driver.update_with_meta = update_with_meta; return driver; } @@ -610,7 +624,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC)) { neu_driver_cache_update(g->driver->cache, g->name, tag->name, - global_timestamp, cmd->value); + global_timestamp, cmd->value, NULL, 0); neu_tag_set_static_value(tag, &cmd->value.value); neu_group_update_tag(g->group, tag); adapter_storage_update_tag_value(cmd->driver, cmd->group, tag);