From 2cc9a685488ed7a9508a0652d39279bd53dc94e9 Mon Sep 17 00:00:00 2001 From: "xiang.zhou" Date: Thu, 3 Jul 2025 18:06:45 +0800 Subject: [PATCH] mem op when driver report callback --- include/neuron/msg.h | 77 +++++++++++++++++++++++++------------ src/adapter/driver/cache.c | 67 ++++++++++++++++++++------------ src/adapter/driver/cache.h | 9 ++--- src/adapter/driver/driver.c | 23 +++++++---- 4 files changed, 115 insertions(+), 61 deletions(-) diff --git a/include/neuron/msg.h b/include/neuron/msg.h index 153f57ab2..5c588c6e9 100644 --- a/include/neuron/msg.h +++ b/include/neuron/msg.h @@ -733,10 +733,11 @@ typedef struct neu_resp_tag_value { typedef neu_resp_tag_value_t neu_tag_value_t; typedef struct neu_resp_tag_value_meta { - char tag[NEU_TAG_NAME_LEN]; - neu_dvalue_t value; - neu_tag_meta_t metas[NEU_TAG_META_SIZE]; - neu_datatag_t datatag; + char tag[NEU_TAG_NAME_LEN]; + neu_dvalue_t value; + neu_tag_meta_t *metas; + int n_meta; + neu_datatag_t datatag; } neu_resp_tag_value_meta_t; static inline UT_icd *neu_resp_tag_value_meta_icd() @@ -797,10 +798,11 @@ typedef struct { } neu_resp_read_group_t; typedef struct neu_resp_tag_value_meta_paginate { - char tag[NEU_TAG_NAME_LEN]; - neu_dvalue_t value; - neu_tag_meta_t metas[NEU_TAG_META_SIZE]; - neu_datatag_t datatag; + char tag[NEU_TAG_NAME_LEN]; + neu_dvalue_t value; + neu_tag_meta_t *metas; + int n_meta; + neu_datatag_t datatag; } neu_resp_tag_value_meta_paginate_t; static inline UT_icd *neu_resp_tag_value_meta_paginate_icd() @@ -831,6 +833,13 @@ static inline void neu_resp_read_free(neu_resp_read_group_t *resp) tag_value->value.type < NEU_TYPE_ARRAY_STRING) { free(tag_value->value.value.bools.bools); } + + if (tag_value->metas != NULL) { + for (int i = 0; i < tag_value->n_meta; i++) { + neu_free_dvalue(&tag_value->metas[i].value); + } + free(tag_value->metas); + } } free(resp->driver); free(resp->group); @@ -848,6 +857,13 @@ neu_resp_read_paginate_free(neu_resp_read_group_paginate_t *resp) tag_value->value.type < NEU_TYPE_ARRAY_STRING) { free(tag_value->value.value.bools.bools); } + + if (tag_value->metas != NULL) { + for (int i = 0; i < tag_value->n_meta; i++) { + neu_free_dvalue(&tag_value->metas[i].value); + } + free(tag_value->metas); + } } free(resp->driver); free(resp->group); @@ -925,6 +941,13 @@ static inline void neu_trans_data_free(neu_reqresp_trans_data_t *data) tag_value->value.type < NEU_TYPE_ARRAY_STRING) { free(tag_value->value.value.bools.bools); } + + if (tag_value->metas != NULL) { + for (int i = 0; i < tag_value->n_meta; i++) { + neu_free_dvalue(&tag_value->metas[i].value); + } + free(tag_value->metas); + } } utarray_free(data->tags); free(data->group); @@ -943,18 +966,21 @@ static inline void neu_tag_value_to_json(neu_resp_tag_value_meta_t *tag_value, tag_json->name = tag_value->tag; tag_json->error = 0; - for (int k = 0; k < NEU_TAG_META_SIZE; k++) { - if (strlen(tag_value->metas[k].name) > 0) { - tag_json->n_meta++; - } else { - break; - } - } + // for (int k = 0; k < NEU_TAG_META_SIZE; k++) { + // if (strlen(tag_value->metas[k].name) > 0) { + // tag_json->n_meta++; + // } else { + // break; + // } + // } + + tag_json->n_meta = tag_value->n_meta; + if (tag_json->n_meta > 0) { tag_json->metas = (neu_json_tag_meta_t *) calloc( tag_json->n_meta, sizeof(neu_json_tag_meta_t)); } - neu_json_metas_to_json(tag_value->metas, NEU_TAG_META_SIZE, tag_json); + neu_json_metas_to_json(tag_value->metas, tag_value->n_meta, tag_json); tag_json->datatag.bias = tag_value->datatag.bias; @@ -1152,18 +1178,21 @@ neu_tag_value_to_json_paginate(neu_resp_tag_value_meta_paginate_t *tag_value, memcpy(tag_json->datatag.meta, tag_value->datatag.meta, NEU_TAG_META_LENGTH); - for (int k = 0; k < NEU_TAG_META_SIZE; k++) { - if (strlen(tag_value->metas[k].name) > 0) { - tag_json->n_meta++; - } else { - break; - } - } + // for (int k = 0; k < NEU_TAG_META_SIZE; k++) { + // if (strlen(tag_value->metas[k].name) > 0) { + // tag_json->n_meta++; + // } else { + // break; + // } + // } + + tag_json->n_meta = tag_value->n_meta; + if (tag_json->n_meta > 0) { tag_json->metas = (neu_json_tag_meta_t *) calloc( tag_json->n_meta, sizeof(neu_json_tag_meta_t)); } - neu_json_metas_to_json_paginate(tag_value->metas, NEU_TAG_META_SIZE, + neu_json_metas_to_json_paginate(tag_value->metas, tag_value->n_meta, tag_json); switch (tag_value->value.type) { diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index 03e16c2c3..aec580286 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -45,7 +45,8 @@ struct elem { neu_dvalue_t value; neu_dvalue_t value_old; - neu_tag_meta_t metas[NEU_TAG_META_SIZE]; + neu_tag_meta_t *metas; + int n_meta; tkey_t key; UT_hash_handle hh; @@ -115,6 +116,13 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache) free(elem->value.value.bools.bools); } + if (elem->metas != NULL) { + for (int i = 0; i < elem->n_meta; i++) { + neu_free_dvalue(&elem->metas[i].value); + } + free(elem->metas); + } + free(elem); } @@ -764,9 +772,16 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, } elem->value.type = value.type; - 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)); + if (metas != NULL) { + if (elem->metas != NULL) { + free(elem->metas); + } + elem->metas = calloc(n_meta, sizeof(neu_tag_meta_t)); + for (int i = 0; i < n_meta; i++) { + memcpy(&elem->metas[i], &metas[i], sizeof(neu_tag_meta_t)); + } + + elem->n_meta = n_meta; } } @@ -784,7 +799,7 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group, int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group, const char *tag, neu_driver_cache_value_t *value, - neu_tag_meta_t *metas, int n_meta) + neu_tag_meta_t **metas, int *n_meta) { struct elem *elem = NULL; int ret = -1; @@ -798,8 +813,14 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group, value->value.type = elem->value.type; value->value.precision = elem->value.precision; - assert(n_meta <= NEU_TAG_META_SIZE); - memcpy(metas, elem->metas, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE); + // assert(n_meta <= NEU_TAG_META_SIZE); + if (elem->metas) { + *metas = calloc(elem->n_meta, sizeof(neu_tag_meta_t)); + memcpy(*metas, elem->metas, sizeof(neu_tag_meta_t) * elem->n_meta); + } else { + *metas = NULL; + } + *n_meta = elem->n_meta; switch (elem->value.type) { case NEU_TYPE_INT8: @@ -941,13 +962,6 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group, break; } - for (int i = 0; i < NEU_TAG_META_SIZE; i++) { - if (strlen(elem->metas[i].name) > 0) { - memcpy(&value->metas[i], &elem->metas[i], - sizeof(neu_tag_meta_t)); - } - } - ret = 0; } @@ -959,7 +973,7 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group, int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache, const char *group, const char *tag, neu_driver_cache_value_t *value, - neu_tag_meta_t *metas, int n_meta) + neu_tag_meta_t **metas, int *n_meta) { struct elem *elem = NULL; int ret = -1; @@ -973,8 +987,14 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache, value->value.type = elem->value.type; value->value.precision = elem->value.precision; - assert(n_meta <= NEU_TAG_META_SIZE); - memcpy(metas, elem->metas, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE); + // assert(n_meta <= NEU_TAG_META_SIZE); + if (elem->metas) { + *metas = calloc(elem->n_meta, sizeof(neu_tag_meta_t)); + memcpy(*metas, elem->metas, sizeof(neu_tag_meta_t) * elem->n_meta); + } else { + *metas = NULL; + } + *n_meta = elem->n_meta; switch (elem->value.type) { case NEU_TYPE_INT8: @@ -1116,13 +1136,6 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache, break; } - for (int i = 0; i < NEU_TAG_META_SIZE; i++) { - if (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; } @@ -1165,6 +1178,12 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group, elem->value.type < NEU_TYPE_ARRAY_STRING) { free(elem->value.value.bools.bools); } + if (elem->metas != NULL) { + for (int i = 0; i < elem->n_meta; i++) { + neu_free_dvalue(&elem->metas[i].value); + } + free(elem->metas); + } free(elem); } diff --git a/src/adapter/driver/cache.h b/src/adapter/driver/cache.h index cfff71829..3468a9a21 100644 --- a/src/adapter/driver/cache.h +++ b/src/adapter/driver/cache.h @@ -50,17 +50,16 @@ void neu_driver_cache_update_trace(neu_driver_cache_t *cache, const char *group, void *neu_driver_cache_get_trace(neu_driver_cache_t *cache, const char *group); typedef struct { - neu_dvalue_t value; - int64_t timestamp; - neu_tag_meta_t metas[NEU_TAG_META_SIZE]; + neu_dvalue_t value; + int64_t timestamp; } neu_driver_cache_value_t; int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group, const char *tag, neu_driver_cache_value_t *value, - neu_tag_meta_t *metas, int n_meta); + neu_tag_meta_t **metas, int *n_meta); int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache, const char *group, const char *tag, neu_driver_cache_value_t *value, - neu_tag_meta_t *metas, int n_meta); + neu_tag_meta_t **metas, int *n_meta); #endif diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 694c24958..482c3ce67 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -2222,6 +2222,13 @@ static int report_callback(void *usr_data) } else { neu_free_dvalue(&tag_value->value); } + + if (tag_value->metas != NULL) { + for (int i = 0; i < tag_value->n_meta; i++) { + neu_free_dvalue(&tag_value->metas[i].value); + } + free(tag_value->metas); + } } utarray_free(data->tags); free(data->group); @@ -2426,15 +2433,15 @@ static void read_report_group(int64_t timestamp, int64_t timeout, if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE)) { if (neu_driver_cache_meta_get_changed(cache, group, tag->name, - &value, tag_value.metas, - NEU_TAG_META_SIZE) != 0) { + &value, &tag_value.metas, + &tag_value.n_meta) != 0) { nlog_debug("tag: %s not changed", tag->name); continue; } } else { if (neu_driver_cache_meta_get(cache, group, tag->name, &value, - tag_value.metas, - NEU_TAG_META_SIZE) != 0) { + &tag_value.metas, + &tag_value.n_meta) != 0) { strcpy(tag_value.tag, tag->name); tag_value.value.type = NEU_TYPE_ERROR; tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; @@ -2621,8 +2628,8 @@ static void read_group(int64_t timestamp, int64_t timeout, tag_value.datatag.bias = tag->bias; if (neu_driver_cache_meta_get(cache, group, tag->name, &value, - tag_value.metas, - NEU_TAG_META_SIZE) != 0) { + &tag_value.metas, + &tag_value.n_meta) != 0) { tag_value.value.type = NEU_TYPE_ERROR; tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; @@ -2809,8 +2816,8 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout, } if (neu_driver_cache_meta_get(cache, group, tag->name, &value, - tag_value.metas, - NEU_TAG_META_SIZE) != 0) { + &tag_value.metas, + &tag_value.n_meta) != 0) { tag_value.value.type = NEU_TYPE_ERROR; tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY;