From c20a1468ed00bf18fec3894fd9c7af753ccf5a0f Mon Sep 17 00:00:00 2001 From: ZhouBox Date: Thu, 12 Nov 2026 18:57:51 +0800 Subject: [PATCH 1/2] core: fix empty array write memory leak --- include/neuron/msg.h | 35 +++++++ plugins/restful/rw_handle.c | 200 +++++++++++++++++++----------------- src/adapter/adapter.c | 4 + src/adapter/driver/driver.c | 116 +++++++++++++-------- src/core/manager.c | 6 ++ src/parser/neu_json_rw.c | 3 + 6 files changed, 228 insertions(+), 136 deletions(-) diff --git a/include/neuron/msg.h b/include/neuron/msg.h index 95e3248de..83f3ce100 100644 --- a/include/neuron/msg.h +++ b/include/neuron/msg.h @@ -26,6 +26,7 @@ extern "C" { #include #include +#include #include "json/neu_json_rw.h" @@ -1104,6 +1105,12 @@ static inline void neu_req_write_tag_fini(neu_req_write_tag_t *req) free(req->tag); } +static inline void neu_req_write_tag_value_fini(neu_req_write_tag_t *req) +{ + neu_free_dvalue(&req->value); + memset(&req->value, 0, sizeof(req->value)); +} + typedef struct neu_resp_tag_value { char tag[NEU_TAG_NAME_LEN]; neu_dvalue_t value; @@ -1150,6 +1157,17 @@ static inline void neu_req_write_tags_fini(neu_req_write_tags_t *req) free(req->tags); } +static inline void neu_req_write_tags_value_fini(neu_req_write_tags_t *req) +{ + if (req->tags == NULL) { + return; + } + for (int i = 0; i < req->n_tag; ++i) { + neu_free_dvalue(&req->tags[i].value); + memset(&req->tags[i].value, 0, sizeof(req->tags[i].value)); + } +} + typedef struct { char *group; @@ -1179,6 +1197,23 @@ static inline void neu_req_write_gtags_fini(neu_req_write_gtags_t *req) free(req->groups); } +static inline void neu_req_write_gtags_value_fini(neu_req_write_gtags_t *req) +{ + if (req->groups == NULL) { + return; + } + for (int i = 0; i < req->n_group; ++i) { + if (req->groups[i].tags == NULL) { + continue; + } + for (int k = 0; k < req->groups[i].n_tag; ++k) { + neu_free_dvalue(&req->groups[i].tags[k].value); + memset(&req->groups[i].tags[k].value, 0, + sizeof(req->groups[i].tags[k].value)); + } + } +} + typedef struct { char *driver; char *group; diff --git a/plugins/restful/rw_handle.c b/plugins/restful/rw_handle.c index f61a1a903..6a0089b54 100644 --- a/plugins/restful/rw_handle.c +++ b/plugins/restful/rw_handle.c @@ -331,43 +331,50 @@ void handle_write(nng_aio *aio) case NEU_JSON_ARRAY_INT64: cmd.value.type = NEU_TYPE_ARRAY_INT64; cmd.value.value.i64s.length = req->value.val_array_int64.length; - cmd.value.value.i64s.i64s = - calloc(req->value.val_array_int64.length, sizeof(int64_t)); - for (size_t i = 0; i < req->value.val_array_int64.length; i++) { - cmd.value.value.i64s.i64s[i] = - req->value.val_array_int64.i64s[i]; + if (cmd.value.value.i64s.length > 0) { + cmd.value.value.i64s.i64s = + calloc(cmd.value.value.i64s.length, sizeof(int64_t)); + for (size_t i = 0; i < cmd.value.value.i64s.length; i++) { + cmd.value.value.i64s.i64s[i] = + req->value.val_array_int64.i64s[i]; + } } break; case NEU_JSON_ARRAY_DOUBLE: cmd.value.type = NEU_TYPE_ARRAY_DOUBLE; cmd.value.value.f64s.length = req->value.val_array_double.length; - cmd.value.value.f64s.f64s = - calloc(req->value.val_array_double.length, sizeof(double)); - for (size_t i = 0; i < req->value.val_array_double.length; - i++) { - cmd.value.value.f64s.f64s[i] = - req->value.val_array_double.f64s[i]; + if (cmd.value.value.f64s.length > 0) { + cmd.value.value.f64s.f64s = + calloc(cmd.value.value.f64s.length, sizeof(double)); + for (size_t i = 0; i < cmd.value.value.f64s.length; i++) { + cmd.value.value.f64s.f64s[i] = + req->value.val_array_double.f64s[i]; + } } break; case NEU_JSON_ARRAY_BOOL: cmd.value.type = NEU_TYPE_ARRAY_BOOL; cmd.value.value.bools.length = req->value.val_array_bool.length; - cmd.value.value.bools.bools = - calloc(req->value.val_array_bool.length, sizeof(bool)); - for (size_t i = 0; i < req->value.val_array_bool.length; i++) { - cmd.value.value.bools.bools[i] = - req->value.val_array_bool.bools[i]; + if (cmd.value.value.bools.length > 0) { + cmd.value.value.bools.bools = + calloc(cmd.value.value.bools.length, sizeof(bool)); + for (size_t i = 0; i < cmd.value.value.bools.length; i++) { + cmd.value.value.bools.bools[i] = + req->value.val_array_bool.bools[i]; + } } break; case NEU_JSON_ARRAY_STR: cmd.value.type = NEU_TYPE_ARRAY_STRING; cmd.value.value.strs.length = req->value.val_array_str.length; - cmd.value.value.strs.strs = - calloc(req->value.val_array_str.length, sizeof(char *)); - for (size_t i = 0; i < req->value.val_array_str.length; i++) { - const char *s = req->value.val_array_str.p_strs[i]; - cmd.value.value.strs.strs[i] = strdup(s ? s : ""); + if (cmd.value.value.strs.length > 0) { + cmd.value.value.strs.strs = + calloc(cmd.value.value.strs.length, sizeof(char *)); + for (size_t i = 0; i < cmd.value.value.strs.length; i++) { + const char *s = req->value.val_array_str.p_strs[i]; + cmd.value.value.strs.strs[i] = strdup(s ? s : ""); + } } break; case NEU_JSON_OBJECT: @@ -389,6 +396,7 @@ void handle_write(nng_aio *aio) } if (ret != 0) { + neu_req_write_tag_value_fini(&cmd); neu_req_write_tag_fini(&cmd); NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); @@ -549,50 +557,58 @@ void handle_write_tags(nng_aio *aio) cmd.tags[i].value.type = NEU_TYPE_ARRAY_INT64; cmd.tags[i].value.value.i64s.length = req->tags[i].value.val_array_int64.length; - cmd.tags[i].value.value.i64s.i64s = - calloc(req->tags[i].value.val_array_int64.length, - sizeof(int64_t)); - memcpy(cmd.tags[i].value.value.i64s.i64s, - req->tags[i].value.val_array_int64.i64s, - sizeof(int64_t) * - req->tags[i].value.val_array_int64.length); + if (cmd.tags[i].value.value.i64s.length > 0) { + cmd.tags[i].value.value.i64s.i64s = + calloc(cmd.tags[i].value.value.i64s.length, + sizeof(int64_t)); + memcpy(cmd.tags[i].value.value.i64s.i64s, + req->tags[i].value.val_array_int64.i64s, + sizeof(int64_t) * + cmd.tags[i].value.value.i64s.length); + } break; case NEU_JSON_ARRAY_DOUBLE: cmd.tags[i].value.type = NEU_TYPE_ARRAY_DOUBLE; cmd.tags[i].value.value.f64s.length = req->tags[i].value.val_array_double.length; - cmd.tags[i].value.value.f64s.f64s = - calloc(req->tags[i].value.val_array_double.length, - sizeof(double)); - memcpy(cmd.tags[i].value.value.f64s.f64s, - req->tags[i].value.val_array_double.f64s, - sizeof(double) * - req->tags[i].value.val_array_double.length); + if (cmd.tags[i].value.value.f64s.length > 0) { + cmd.tags[i].value.value.f64s.f64s = + calloc(cmd.tags[i].value.value.f64s.length, + sizeof(double)); + memcpy(cmd.tags[i].value.value.f64s.f64s, + req->tags[i].value.val_array_double.f64s, + sizeof(double) * + cmd.tags[i].value.value.f64s.length); + } break; case NEU_JSON_ARRAY_BOOL: cmd.tags[i].value.type = NEU_TYPE_ARRAY_BOOL; cmd.tags[i].value.value.bools.length = req->tags[i].value.val_array_bool.length; - cmd.tags[i].value.value.bools.bools = calloc( - req->tags[i].value.val_array_bool.length, sizeof(bool)); - memcpy(cmd.tags[i].value.value.bools.bools, - req->tags[i].value.val_array_bool.bools, - sizeof(bool) * - req->tags[i].value.val_array_bool.length); + if (cmd.tags[i].value.value.bools.length > 0) { + cmd.tags[i].value.value.bools.bools = calloc( + cmd.tags[i].value.value.bools.length, sizeof(bool)); + memcpy(cmd.tags[i].value.value.bools.bools, + req->tags[i].value.val_array_bool.bools, + sizeof(bool) * + cmd.tags[i].value.value.bools.length); + } break; case NEU_JSON_ARRAY_STR: cmd.tags[i].value.type = NEU_TYPE_ARRAY_STRING; cmd.tags[i].value.value.strs.length = req->tags[i].value.val_array_str.length; - cmd.tags[i].value.value.strs.strs = - calloc(req->tags[i].value.val_array_str.length, - sizeof(char *)); - for (size_t k = 0; - k < req->tags[i].value.val_array_str.length; k++) { - const char *s = - req->tags[i].value.val_array_str.p_strs[k]; - cmd.tags[i].value.value.strs.strs[k] = - strdup(s ? s : ""); + if (cmd.tags[i].value.value.strs.length > 0) { + cmd.tags[i].value.value.strs.strs = + calloc(cmd.tags[i].value.value.strs.length, + sizeof(char *)); + for (size_t k = 0; + k < cmd.tags[i].value.value.strs.length; k++) { + const char *s = + req->tags[i].value.val_array_str.p_strs[k]; + cmd.tags[i].value.value.strs.strs[k] = + strdup(s ? s : ""); + } } break; case NEU_JSON_OBJECT: @@ -616,6 +632,7 @@ void handle_write_tags(nng_aio *aio) } if (ret != 0) { + neu_req_write_tags_value_fini(&cmd); neu_req_write_tags_fini(&cmd); NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); @@ -683,64 +700,62 @@ static void trans(neu_json_write_gtags_req_t *req, neu_req_write_gtags_t *cmd) cmd->groups[i].tags[k].value.type = NEU_TYPE_ARRAY_INT64; cmd->groups[i].tags[k].value.value.i64s.length = req->groups[i].tags[k].value.val_array_int64.length; - - cmd->groups[i].tags[k].value.value.i64s.i64s = - calloc(req->groups[i].tags[k].value.val_array_int64.length, - sizeof(int64_t)); - - memcpy(cmd->groups[i].tags[k].value.value.i64s.i64s, - req->groups[i].tags[k].value.val_array_int64.i64s, - sizeof(int64_t) * - req->groups[i].tags[k].value.val_array_int64.length); - + if (cmd->groups[i].tags[k].value.value.i64s.length > 0) { + cmd->groups[i].tags[k].value.value.i64s.i64s = + calloc(cmd->groups[i].tags[k].value.value.i64s.length, + sizeof(int64_t)); + memcpy(cmd->groups[i].tags[k].value.value.i64s.i64s, + req->groups[i].tags[k].value.val_array_int64.i64s, + sizeof(int64_t) * + cmd->groups[i].tags[k].value.value.i64s.length); + } break; case NEU_JSON_ARRAY_DOUBLE: cmd->groups[i].tags[k].value.type = NEU_TYPE_ARRAY_DOUBLE; cmd->groups[i].tags[k].value.value.f64s.length = req->groups[i].tags[k].value.val_array_double.length; - - cmd->groups[i].tags[k].value.value.f64s.f64s = calloc( - - req->groups[i].tags[k].value.val_array_double.length, - sizeof(double)); - memcpy( - cmd->groups[i].tags[k].value.value.f64s.f64s, - req->groups[i].tags[k].value.val_array_double.f64s, - sizeof(double) * - req->groups[i].tags[k].value.val_array_double.length); - + if (cmd->groups[i].tags[k].value.value.f64s.length > 0) { + cmd->groups[i].tags[k].value.value.f64s.f64s = + calloc(cmd->groups[i].tags[k].value.value.f64s.length, + sizeof(double)); + memcpy(cmd->groups[i].tags[k].value.value.f64s.f64s, + req->groups[i].tags[k].value.val_array_double.f64s, + sizeof(double) * + cmd->groups[i].tags[k].value.value.f64s.length); + } break; case NEU_JSON_ARRAY_BOOL: cmd->groups[i].tags[k].value.type = NEU_TYPE_ARRAY_BOOL; cmd->groups[i].tags[k].value.value.bools.length = req->groups[i].tags[k].value.val_array_bool.length; - - cmd->groups[i].tags[k].value.value.bools.bools = - calloc(sizeof(bool), - req->groups[i].tags[k].value.val_array_bool.length); - memcpy(cmd->groups[i].tags[k].value.value.bools.bools, - req->groups[i].tags[k].value.val_array_bool.bools, - sizeof(bool) * - req->groups[i].tags[k].value.val_array_bool.length); - + if (cmd->groups[i].tags[k].value.value.bools.length > 0) { + cmd->groups[i].tags[k].value.value.bools.bools = + calloc(cmd->groups[i].tags[k].value.value.bools.length, + sizeof(bool)); + memcpy(cmd->groups[i].tags[k].value.value.bools.bools, + req->groups[i].tags[k].value.val_array_bool.bools, + sizeof(bool) * + cmd->groups[i].tags[k].value.value.bools.length); + } break; case NEU_JSON_ARRAY_STR: cmd->groups[i].tags[k].value.type = NEU_TYPE_ARRAY_STRING; cmd->groups[i].tags[k].value.value.strs.length = req->groups[i].tags[k].value.val_array_str.length; - - cmd->groups[i].tags[k].value.value.strs.strs = - calloc(req->groups[i].tags[k].value.val_array_str.length, - sizeof(char *)); - for (size_t j = 0; - j < req->groups[i].tags[k].value.val_array_str.length; - j++) { - const char *s = - req->groups[i].tags[k].value.val_array_str.p_strs[j]; - cmd->groups[i].tags[k].value.value.strs.strs[j] = - strdup(s ? s : ""); + if (cmd->groups[i].tags[k].value.value.strs.length > 0) { + cmd->groups[i].tags[k].value.value.strs.strs = + calloc(cmd->groups[i].tags[k].value.value.strs.length, + sizeof(char *)); + for (size_t j = 0; + j < cmd->groups[i].tags[k].value.value.strs.length; + j++) { + const char *s = req->groups[i] + .tags[k] + .value.val_array_str.p_strs[j]; + cmd->groups[i].tags[k].value.value.strs.strs[j] = + strdup(s ? s : ""); + } } - break; case NEU_JSON_OBJECT: cmd->groups[i].tags[k].value.type = NEU_TYPE_CUSTOM; @@ -839,6 +854,7 @@ void handle_write_gtags(nng_aio *aio) } if (ret != 0) { + neu_req_write_gtags_value_fini(&cmd); neu_req_write_gtags_fini(&cmd); NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 6606fa80f..85afb332a 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -1233,6 +1233,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } } } else { + neu_req_write_tag_value_fini((neu_req_write_tag_t *) &header[1]); neu_req_write_tag_fini((neu_req_write_tag_t *) &header[1]); error.error = NEU_ERR_GROUP_NOT_ALLOW; header->type = NEU_RESP_ERROR; @@ -1279,6 +1280,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) bool re_flag = false; if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + neu_req_write_tags_value_fini((neu_req_write_tags_t *) &header[1]); neu_req_write_tags_fini((neu_req_write_tags_t *) &header[1]); error.error = NEU_ERR_GROUP_NOT_ALLOW; header->type = NEU_RESP_ERROR; @@ -1344,6 +1346,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) bool re_flag = false; if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + neu_req_write_gtags_value_fini( + (neu_req_write_gtags_t *) &header[1]); neu_req_write_gtags_fini((neu_req_write_gtags_t *) &header[1]); error.error = NEU_ERR_GROUP_NOT_ALLOW; header->type = NEU_RESP_ERROR; diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index b31dcf994..831d55911 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -1184,7 +1184,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, case NEU_TYPE_ARRAY_INT8: if (value->type == NEU_TYPE_ARRAY_INT64) { int64_t *i64s = value->value.i64s.i64s; - value->value.i8s.i8s = calloc(value->value.i64s.length, 1); + value->value.i8s.i8s = value->value.i64s.length + ? calloc(value->value.i64s.length, 1) + : NULL; for (size_t i = 0; i < value->value.i64s.length; i++) { value->value.i8s.i8s[i] = (int8_t) i64s[i]; } @@ -1194,7 +1196,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, case NEU_TYPE_ARRAY_UINT8: if (value->type == NEU_TYPE_ARRAY_INT64) { int64_t *i64s = value->value.i64s.i64s; - value->value.u8s.u8s = calloc(value->value.i64s.length, 1); + value->value.u8s.u8s = value->value.i64s.length + ? calloc(value->value.i64s.length, 1) + : NULL; for (size_t i = 0; i < value->value.i64s.length; i++) { value->value.u8s.u8s[i] = (uint8_t) i64s[i]; } @@ -1204,7 +1208,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, case NEU_TYPE_ARRAY_INT16: if (value->type == NEU_TYPE_ARRAY_INT64) { int64_t *i64s = value->value.i64s.i64s; - value->value.i16s.i16s = calloc(value->value.i64s.length, 2); + value->value.i16s.i16s = value->value.i64s.length + ? calloc(value->value.i64s.length, 2) + : NULL; for (size_t i = 0; i < value->value.i64s.length; i++) { value->value.i16s.i16s[i] = (int16_t) i64s[i]; } @@ -1214,7 +1220,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, case NEU_TYPE_ARRAY_UINT16: if (value->type == NEU_TYPE_ARRAY_INT64) { int64_t *i64s = value->value.i64s.i64s; - value->value.u16s.u16s = calloc(value->value.i64s.length, 2); + value->value.u16s.u16s = value->value.i64s.length + ? calloc(value->value.i64s.length, 2) + : NULL; for (size_t i = 0; i < value->value.i64s.length; i++) { value->value.u16s.u16s[i] = (uint16_t) i64s[i]; } @@ -1224,7 +1232,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, case NEU_TYPE_ARRAY_INT32: if (value->type == NEU_TYPE_ARRAY_INT64) { int64_t *i64s = value->value.i64s.i64s; - value->value.i32s.i32s = calloc(value->value.i64s.length, 4); + value->value.i32s.i32s = value->value.i64s.length + ? calloc(value->value.i64s.length, 4) + : NULL; for (size_t i = 0; i < value->value.i64s.length; i++) { value->value.i32s.i32s[i] = (int32_t) i64s[i]; } @@ -1234,7 +1244,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, case NEU_TYPE_ARRAY_UINT32: if (value->type == NEU_TYPE_ARRAY_INT64) { int64_t *i64s = value->value.i64s.i64s; - value->value.u32s.u32s = calloc(value->value.i64s.length, 4); + value->value.u32s.u32s = value->value.i64s.length + ? calloc(value->value.i64s.length, 4) + : NULL; for (size_t i = 0; i < value->value.i64s.length; i++) { value->value.u32s.u32s[i] = (uint32_t) i64s[i]; } @@ -1244,7 +1256,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, case NEU_TYPE_ARRAY_UINT64: if (value->type == NEU_TYPE_ARRAY_INT64) { int64_t *i64s = value->value.i64s.i64s; - value->value.u64s.u64s = calloc(value->value.i64s.length, 8); + value->value.u64s.u64s = value->value.i64s.length + ? calloc(value->value.i64s.length, 8) + : NULL; for (size_t i = 0; i < value->value.i64s.length; i++) { value->value.u64s.u64s[i] = (uint64_t) i64s[i]; } @@ -1254,7 +1268,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, case NEU_TYPE_ARRAY_FLOAT: if (value->type == NEU_TYPE_ARRAY_INT64) { int64_t *i64s = value->value.i64s.i64s; - value->value.f32s.f32s = calloc(value->value.i64s.length, 4); + value->value.f32s.f32s = value->value.i64s.length + ? calloc(value->value.i64s.length, 4) + : NULL; for (size_t i = 0; i < value->value.i64s.length; i++) { value->value.f32s.f32s[i] = (float) i64s[i]; } @@ -1262,7 +1278,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, } if (value->type == NEU_TYPE_ARRAY_DOUBLE) { double *f64s = value->value.f64s.f64s; - value->value.f32s.f32s = calloc(value->value.f64s.length, 4); + value->value.f32s.f32s = value->value.f64s.length + ? calloc(value->value.f64s.length, 4) + : NULL; for (size_t i = 0; i < value->value.f64s.length; i++) { value->value.f32s.f32s[i] = (float) f64s[i]; } @@ -1272,7 +1290,9 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, case NEU_TYPE_ARRAY_DOUBLE: if (value->type == NEU_TYPE_ARRAY_INT64) { int64_t *i64s = value->value.i64s.i64s; - value->value.f64s.f64s = calloc(value->value.i64s.length, 8); + value->value.f64s.f64s = value->value.i64s.length + ? calloc(value->value.i64s.length, 8) + : NULL; for (size_t i = 0; i < value->value.i64s.length; i++) { value->value.f64s.f64s[i] = (double) i64s[i]; } @@ -1587,20 +1607,45 @@ static bool write_types_valid(neu_adapter_driver_t *driver, UT_array *types) return true; } +// Report an error for a write request that has not reached the driver plugin. +// The plugin owns dynamically allocated write values once it is called, so the +// core must release them itself on these pre-plugin failure paths. +static void write_tag_error(neu_adapter_driver_t *driver, + neu_reqresp_head_t *req, neu_error error) +{ + neu_req_write_tag_t *cmd = (neu_req_write_tag_t *) &req[1]; + neu_req_write_tag_value_fini(cmd); + driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, error); +} + +static void write_tags_error(neu_adapter_driver_t *driver, + neu_reqresp_head_t *req, neu_error error) +{ + neu_req_write_tags_t *cmd = (neu_req_write_tags_t *) &req[1]; + neu_req_write_tags_value_fini(cmd); + driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, error); +} + +static void write_gtags_error(neu_adapter_driver_t *driver, + neu_reqresp_head_t *req, neu_error error) +{ + neu_req_write_gtags_t *cmd = (neu_req_write_gtags_t *) &req[1]; + neu_req_write_gtags_value_fini(cmd); + driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, error); +} + int neu_adapter_driver_write_tags(neu_adapter_driver_t *driver, neu_reqresp_head_t * req) { neu_req_write_tags_t *cmd = (neu_req_write_tags_t *) &req[1]; if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { - driver->adapter.cb_funs.driver.write_response( - &driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING); + write_tags_error(driver, req, NEU_ERR_PLUGIN_NOT_RUNNING); return NEU_ERR_PLUGIN_NOT_RUNNING; } if (driver->adapter.module->intf_funs->driver.write_tags == NULL) { - driver->adapter.cb_funs.driver.write_response( - &driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS); + write_tags_error(driver, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS); return NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS; } @@ -1609,8 +1654,7 @@ int neu_adapter_driver_write_tags(neu_adapter_driver_t *driver, if (g == NULL) { nlog_warn("driver write tags group not exist, driver: %s, group: %s", driver->adapter.name, cmd->group); - driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, - NEU_ERR_GROUP_NOT_EXIST); + write_tags_error(driver, req, NEU_ERR_GROUP_NOT_EXIST); return NEU_ERR_GROUP_NOT_EXIST; } @@ -1653,13 +1697,10 @@ int neu_adapter_driver_write_tags(neu_adapter_driver_t *driver, if (value_err != NEU_ERR_SUCCESS) { nlog_warn("driver write tags value error(%d), driver: %s, group: %s", value_err, driver->adapter.name, cmd->group); - driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, - value_err); + write_tags_error(driver, req, value_err); utarray_foreach(tags, neu_plugin_tag_value_t *, tv) { neu_tag_free(tv->tag); - neu_dvalue_t dvalue = { .type = tv->type, .value = tv->value }; - neu_free_dvalue(&dvalue); } utarray_free(tags); if (types != NULL) { @@ -1672,13 +1713,10 @@ int neu_adapter_driver_write_tags(neu_adapter_driver_t *driver, nlog_warn("driver write tags some tag not exist, driver: %s, group: " "%s", driver->adapter.name, cmd->group); - driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, - NEU_ERR_TAG_NOT_EXIST); + write_tags_error(driver, req, NEU_ERR_TAG_NOT_EXIST); utarray_foreach(tags, neu_plugin_tag_value_t *, tv) { neu_tag_free(tv->tag); - neu_dvalue_t dvalue = { .type = tv->type, .value = tv->value }; - neu_free_dvalue(&dvalue); } utarray_free(tags); if (types != NULL) { @@ -1722,14 +1760,12 @@ int neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver, group_t * first_g = NULL; if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { - driver->adapter.cb_funs.driver.write_response( - &driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING); + write_gtags_error(driver, req, NEU_ERR_PLUGIN_NOT_RUNNING); return NEU_ERR_PLUGIN_NOT_RUNNING; } if (driver->adapter.module->intf_funs->driver.write_tags == NULL) { - driver->adapter.cb_funs.driver.write_response( - &driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS); + write_gtags_error(driver, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS); return NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS; } @@ -1737,8 +1773,7 @@ int neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver, group_t *g = find_group(driver, cmd->groups[i].group); if (g == NULL) { - driver->adapter.cb_funs.driver.write_response( - &driver->adapter, req, NEU_ERR_GROUP_NOT_EXIST); + write_gtags_error(driver, req, NEU_ERR_GROUP_NOT_EXIST); return NEU_ERR_GROUP_NOT_EXIST; } @@ -1787,8 +1822,7 @@ int neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver, } if (value_err != NEU_ERR_SUCCESS) { - driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, - value_err); + write_gtags_error(driver, req, value_err); utarray_foreach(tags, neu_plugin_tag_value_t *, tv) { neu_tag_free(tv->tag); @@ -1806,8 +1840,7 @@ int neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver, } if (utarray_len(tags) != (unsigned int) n_tag) { - driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, - NEU_ERR_TAG_NOT_EXIST); + write_gtags_error(driver, req, NEU_ERR_TAG_NOT_EXIST); utarray_foreach(tags, neu_plugin_tag_value_t *, tv) { neu_tag_free(tv->tag); @@ -1857,8 +1890,7 @@ int neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, neu_reqresp_head_t * req) { if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { - driver->adapter.cb_funs.driver.write_response( - &driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING); + write_tag_error(driver, req, NEU_ERR_PLUGIN_NOT_RUNNING); return NEU_ERR_PLUGIN_NOT_RUNNING; } @@ -1868,8 +1900,7 @@ int neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, if (g == NULL) { nlog_warn("driver write tag group not exist, driver: %s, group: %s", driver->adapter.name, cmd->group); - driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, - NEU_ERR_GROUP_NOT_EXIST); + write_tag_error(driver, req, NEU_ERR_GROUP_NOT_EXIST); return NEU_ERR_GROUP_NOT_EXIST; } neu_datatag_t *tag = neu_group_find_tag(g->group, cmd->tag); @@ -1877,16 +1908,14 @@ int neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, if (tag == NULL) { nlog_warn("driver write tag not exist, driver: %s, group: %s, tag: %s", driver->adapter.name, cmd->group, cmd->tag); - driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, - NEU_ERR_TAG_NOT_EXIST); + write_tag_error(driver, req, NEU_ERR_TAG_NOT_EXIST); return NEU_ERR_TAG_NOT_EXIST; } else { if ((tag->attribute & NEU_ATTRIBUTE_WRITE) != NEU_ATTRIBUTE_WRITE) { nlog_warn("driver write tag not allow write, driver: %s, group: " "%s, tag: %s", driver->adapter.name, cmd->group, cmd->tag); - driver->adapter.cb_funs.driver.write_response( - &driver->adapter, req, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE); + write_tag_error(driver, req, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE); neu_tag_free(tag); return NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE; } @@ -1899,8 +1928,7 @@ int neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, nlog_warn("driver write tag value error(%d), driver: %s, group: " "%s, tag: %s", value_check, driver->adapter.name, cmd->group, cmd->tag); - driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, - value_check); + write_tag_error(driver, req, value_check); if (types != NULL) { utarray_free(types); } diff --git a/src/core/manager.c b/src/core/manager.c index c1b28e939..ba1022751 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -1488,10 +1488,16 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_req_read_group_paginate_fini( (neu_req_read_group_paginate_t *) &header[1]); } else if (NEU_REQ_WRITE_TAG == header->type) { + neu_req_write_tag_value_fini( + (neu_req_write_tag_t *) &header[1]); neu_req_write_tag_fini((neu_req_write_tag_t *) &header[1]); } else if (NEU_REQ_WRITE_TAGS == header->type) { + neu_req_write_tags_value_fini( + (neu_req_write_tags_t *) &header[1]); neu_req_write_tags_fini((neu_req_write_tags_t *) &header[1]); } else if (NEU_REQ_WRITE_GTAGS == header->type) { + neu_req_write_gtags_value_fini( + (neu_req_write_gtags_t *) &header[1]); neu_req_write_gtags_fini((neu_req_write_gtags_t *) &header[1]); } neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; diff --git a/src/parser/neu_json_rw.c b/src/parser/neu_json_rw.c index 765c5d308..1d2f44a96 100644 --- a/src/parser/neu_json_rw.c +++ b/src/parser/neu_json_rw.c @@ -1390,6 +1390,9 @@ static int decode_write_gtags_req_json(void * json_obj, req->groups[i].tags[k].tag = v_elems[0].v.val_str; req->groups[i].tags[k].t = v_elems[1].t; req->groups[i].tags[k].value = v_elems[1].v; + if (req->groups[i].tags[k].t == NEU_JSON_OBJECT) { + json_incref(req->groups[i].tags[k].value.val_object); + } } } From 3b72cc7d7434f105da0795a7ff4b7134f4f27765 Mon Sep 17 00:00:00 2001 From: ZhouBox Date: Fri, 13 Nov 2026 11:03:35 +0800 Subject: [PATCH 2/2] fix(driver): accept UINT64 write values for numeric tags Sparkplug B encodes UInt64 metrics as an unsigned 64-bit value, which cannot be represented by an INT64 write value. Accept NEU_TYPE_UINT64 writes for numeric tags, evaluate the range with the unsigned value, and extend cal_decimal and float conversion accordingly. INT64 write values keep their signed semantics. --- src/adapter/driver/driver.c | 30 ++++++++++++++++++++++-------- tests/ut/driver_auto_test.cc | 2 ++ 2 files changed, 24 insertions(+), 8 deletions(-) diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 831d55911..b178db543 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -1308,9 +1308,14 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type, static void cal_decimal(neu_type_e tag_type, neu_type_e value_type, neu_value_u *value, double decimal, double bias) { - long double result = value_type == NEU_TYPE_INT64 - ? (long double) value->i64 - : (long double) value->d64; + long double result; + if (value_type == NEU_TYPE_INT64) { + result = (long double) value->i64; + } else if (value_type == NEU_TYPE_UINT64) { + result = (long double) value->u64; + } else { + result = (long double) value->d64; + } result -= (long double) bias; if (decimal != 0) { result /= (long double) decimal; @@ -1473,7 +1478,7 @@ static int check_write_value(neu_type_e tag_type, int64_t value, double value_d, return valid ? NEU_ERR_SUCCESS : NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH; } } - if (write_type != NEU_TYPE_INT64 && + if (write_type != NEU_TYPE_INT64 && write_type != NEU_TYPE_UINT64 && (write_type != NEU_TYPE_DOUBLE || (integer && decimal == 0 && bias == 0))) { return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH; @@ -1482,8 +1487,14 @@ static int check_write_value(neu_type_e tag_type, int64_t value, double value_d, (!isfinite(decimal) || !isfinite(bias))) { return NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE; } - long double result = write_type == NEU_TYPE_INT64 ? (long double) value - : (long double) value_d; + long double result; + if (write_type == NEU_TYPE_INT64) { + result = (long double) value; + } else if (write_type == NEU_TYPE_UINT64) { + result = (long double) (uint64_t) value; + } else { + result = (long double) value_d; + } result -= (long double) bias; if (decimal != 0) { result /= (long double) decimal; @@ -1567,8 +1578,11 @@ static int prepare_write_tag(neu_adapter_driver_t *driver, group_t *group, cal_decimal(tag->type, value->type, &value->value, tag->decimal, tag->bias); } else if ((tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) && - value->type == NEU_TYPE_INT64) { - value->value.d64 = (double) value->value.i64; + (value->type == NEU_TYPE_INT64 || + value->type == NEU_TYPE_UINT64)) { + value->value.d64 = value->type == NEU_TYPE_INT64 + ? (double) value->value.i64 + : (double) value->value.u64; } if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) { value->type = NEU_TYPE_DOUBLE; diff --git a/tests/ut/driver_auto_test.cc b/tests/ut/driver_auto_test.cc index 4b0a69142..73909140b 100644 --- a/tests/ut/driver_auto_test.cc +++ b/tests/ut/driver_auto_test.cc @@ -184,6 +184,8 @@ TEST(DriverAuto, NonfiniteAndWideBounds) is_value_in_range(NEU_TYPE_INT64, INT64_MAX, 0, NEU_TYPE_INT64, 0.5)); EXPECT_EQ(NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE, is_value_in_range(NEU_TYPE_UINT64, -1, 0, NEU_TYPE_INT64, 0)); + EXPECT_EQ(0, is_value_in_range(NEU_TYPE_UINT64, -1, 0, NEU_TYPE_UINT64, 0)); + EXPECT_EQ(0, is_value_in_range(NEU_TYPE_FLOAT, -1, 0, NEU_TYPE_UINT64, 0)); EXPECT_EQ(0, is_value_in_range(NEU_TYPE_FLOAT, 0, 1e30, NEU_TYPE_DOUBLE, 0)); EXPECT_EQ(NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE,