From 1fbebd38725c671b4c21bb64ba40874b02a0debd Mon Sep 17 00:00:00 2001 From: fengzero Date: Wed, 13 Jul 2022 07:46:25 +0000 Subject: [PATCH 1/2] feat(connection): wait fixed msg --- ft/resource/error.resource | 5 ++- include/neuron/connection/neu_connection.h | 7 +++ include/neuron/errcodes.h | 3 +- include/neuron/tag.h | 31 ++++++++----- include/neuron/utils/protocol_buf.h | 5 +++ plugins/restful/http.c | 1 + src/adapter/adapter.c | 2 +- src/adapter/driver/driver.c | 8 ++-- src/base/tag.c | 3 ++ src/connection/connection.c | 51 ++++++++++++++++++++++ 10 files changed, 98 insertions(+), 18 deletions(-) diff --git a/ft/resource/error.resource b/ft/resource/error.resource index a8e4ad4f..dcdb75de 100644 --- a/ft/resource/error.resource +++ b/ft/resource/error.resource @@ -60,5 +60,6 @@ ${NEU_ERR_PLUGIN_DISCONNECTED} 3002 ${NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ} 3003 ${NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE} 3004 ${NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH} 3007 -${NEU_ERR_PLUGIN_TAG_EXPIRED} 3008 -${NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE} 3009, \ No newline at end of file +${NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED} 3008 +${NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE} 3009 +${NEU_ERR_PLUGIN_NOT_RUNNING} 3010 \ No newline at end of file diff --git a/include/neuron/connection/neu_connection.h b/include/neuron/connection/neu_connection.h index ea0a37a5..996b1014 100644 --- a/include/neuron/connection/neu_connection.h +++ b/include/neuron/connection/neu_connection.h @@ -148,4 +148,11 @@ void neu_conn_stream_consume(neu_conn_t *conn, void *context, void neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context, neu_conn_stream_consume_fn fn); +typedef neu_buf_result_t (*neu_conn_process_msg)( + void *context, neu_protocol_unpack_buf_t *protocol_buf); +int neu_conn_wait_msg(neu_conn_t *conn, void *context, uint16_t n_byte, + neu_conn_process_msg fn); +int neu_conn_tcp_server_wait_msg(neu_conn_t *conn, int fd, void *context, + uint16_t n_byte, neu_conn_process_msg fn); + #endif diff --git a/include/neuron/errcodes.h b/include/neuron/errcodes.h index bc393fae..426a3f28 100644 --- a/include/neuron/errcodes.h +++ b/include/neuron/errcodes.h @@ -91,8 +91,9 @@ typedef enum { NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ = 3003, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE = 3004, NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH = 3007, - NEU_ERR_PLUGIN_TAG_EXPIRED = 3008, + NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED = 3008, NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE = 3009, + NEU_ERR_PLUGIN_NOT_RUNNING = 3010, NEU_ERR_MQTT_FAILURE = 4000, NEU_ERR_MQTT_NO_CERTFILESET = 4001, diff --git a/include/neuron/tag.h b/include/neuron/tag.h index a2f81a84..9e7e4352 100644 --- a/include/neuron/tag.h +++ b/include/neuron/tag.h @@ -27,22 +27,14 @@ config_ **/ #include "type.h" +#define NEU_TAG_META_SIZE 20 + typedef enum { NEU_ATTRIBUTE_READ = 1, NEU_ATTRIBUTE_WRITE = 2, NEU_ATTRIBUTE_SUBSCRIBE = 4, } neu_attribute_e; -typedef struct { - char * name; - char * address; - char * description; - neu_attribute_e attribute; - neu_type_e type; -} neu_datatag_t; - -UT_icd *neu_tag_get_icd(); - typedef enum { NEU_DATATAG_ENDIAN_L16 = 0, // #L NEU_DATATAG_ENDIAN_B16 = 1, // #B @@ -72,10 +64,29 @@ typedef union { } string; struct { + bool op; uint8_t bit; } bit; } neu_datatag_addr_option_u; +typedef struct { + char * name; + char * address; + char * description; + neu_attribute_e attribute; + neu_type_e type; + neu_datatag_addr_option_u option; + uint8_t meta[NEU_TAG_META_SIZE]; +} neu_datatag_t; + +inline static bool neu_tag_attribute_test(neu_datatag_t * tag, + neu_attribute_e attribute) +{ + return (tag->attribute & attribute) == attribute; +} + +UT_icd *neu_tag_get_icd(); + int neu_datatag_parse_addr_option(neu_datatag_t * datatag, neu_datatag_addr_option_u *option); diff --git a/include/neuron/utils/protocol_buf.h b/include/neuron/utils/protocol_buf.h index ac298ded..ad2f05e9 100644 --- a/include/neuron/utils/protocol_buf.h +++ b/include/neuron/utils/protocol_buf.h @@ -24,6 +24,11 @@ config_ **/ #include #include +typedef struct { + int used; + uint16_t need; +} neu_buf_result_t; + typedef struct { uint8_t *base; uint16_t size; diff --git a/plugins/restful/http.c b/plugins/restful/http.c index 6c5ce400..f956795c 100644 --- a/plugins/restful/http.c +++ b/plugins/restful/http.c @@ -285,6 +285,7 @@ int http_response(nng_aio *aio, neu_err_code_e code, char *content) case NEU_ERR_GROUP_ALREADY_SUBSCRIBED: case NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH: case NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE: + case NEU_ERR_PLUGIN_NOT_RUNNING: status = NNG_HTTP_STATUS_OK; break; case NEU_ERR_EINTERNAL: diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 3f593a0a..d3aafb41 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -85,7 +85,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start) rv = nng_pair1_open(&adapter->sock); assert(rv == 0); - nng_socket_set_ms(adapter->sock, NNG_OPT_SENDTIMEO, 100); + nng_socket_set_ms(adapter->sock, NNG_OPT_SENDTIMEO, 1000); nng_socket_get_int(adapter->sock, NNG_OPT_RECVFD, &adapter->recv_fd); switch (info->module->type) { diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 220c821f..523a2ce7 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -234,7 +234,7 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, int index = utarray_eltidx(tags, tag); strcpy(resp.tags[index].tag, tag->name); resp.tags[index].value.type = NEU_TYPE_ERROR; - resp.tags[index].value.value.i32 = NEU_ERR_NODE_NOT_RUNNING; + resp.tags[index].value.value.i32 = NEU_ERR_PLUGIN_NOT_RUNNING; } } else { @@ -257,8 +257,8 @@ void 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_NODE_NOT_RUNNING); + driver->adapter.cb_funs.driver.write_response( + &driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING); return; } @@ -678,7 +678,7 @@ static void read_group(int64_t timestamp, int64_t timeout, if ((timestamp - value.timestamp) > timeout) { datas[index].value.type = NEU_TYPE_ERROR; - datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_EXPIRED; + datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED; } else { datas[index].value.type = tag->type; datas[index].value.value = value.value; diff --git a/src/base/tag.c b/src/base/tag.c index b46055d8..3b945510 100644 --- a/src/base/tag.c +++ b/src/base/tag.c @@ -164,6 +164,9 @@ int neu_datatag_parse_addr_option(neu_datatag_t * datatag, if (op != NULL) { sscanf(op, ".%hhd", &option->bit.bit); + option->bit.op = true; + } else { + option->bit.op = false; } break; diff --git a/src/connection/connection.c b/src/connection/connection.c index 08eb5369..1406673a 100644 --- a/src/connection/connection.c +++ b/src/connection/connection.c @@ -824,4 +824,55 @@ void neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context, offset); } } +} + +int neu_conn_wait_msg(neu_conn_t *conn, void *context, uint16_t n_byte, + neu_conn_process_msg fn) +{ + uint8_t recv_buf[2048] = { 0 }; + uint16_t offset = 0; + ssize_t ret = neu_conn_recv(conn, recv_buf, n_byte); + neu_buf_result_t result = { 0 }; + neu_protocol_unpack_buf_t pbuf = { 0 }; + + while (ret > 0) { + zlog_recv_protocol(conn->param.log, recv_buf + offset, ret); + offset += ret; + neu_protocol_unpack_buf_init(&pbuf, recv_buf, offset); + result = fn(context, &pbuf); + if (result.need > 0) { + assert(result.need <= sizeof(recv_buf) - offset); + ret = neu_conn_recv(conn, recv_buf + offset, result.need); + } else { + return result.used; + } + } + + return ret; +} + +int neu_conn_tcp_server_wait_msg(neu_conn_t *conn, int fd, void *context, + uint16_t n_byte, neu_conn_process_msg fn) +{ + uint8_t recv_buf[2048] = { 0 }; + uint16_t offset = 0; + ssize_t ret = neu_conn_tcp_server_recv(conn, fd, recv_buf, n_byte); + neu_buf_result_t result = { 0 }; + neu_protocol_unpack_buf_t pbuf = { 0 }; + + while (ret > 0) { + zlog_recv_protocol(conn->param.log, recv_buf + offset, ret); + offset += ret; + neu_protocol_unpack_buf_init(&pbuf, recv_buf, offset); + result = fn(context, &pbuf); + if (result.need > 0) { + assert(result.need <= sizeof(recv_buf) - offset); + ret = neu_conn_tcp_server_recv(conn, fd, recv_buf + offset, + result.need); + } else { + return result.used; + } + } + + return ret; } \ No newline at end of file From a93796aa0ce7b37cdde22244e002958fff2392cb Mon Sep 17 00:00:00 2001 From: fengzero Date: Wed, 13 Jul 2022 08:04:12 +0000 Subject: [PATCH 2/2] fix(driver): memory leak --- src/adapter/driver/driver.c | 4 ++++ src/base/group.c | 4 ++++ src/base/tag.c | 6 ++++-- src/connection/connection.c | 16 +++++++--------- 4 files changed, 19 insertions(+), 11 deletions(-) diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 523a2ce7..52078acf 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -284,6 +284,10 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, if ((tag->attribute & NEU_ATTRIBUTE_WRITE) != NEU_ATTRIBUTE_WRITE) { driver->adapter.cb_funs.driver.write_response( &driver->adapter, req, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE); + free(tag->address); + free(tag->name); + free(tag->description); + free(tag); return; } switch (tag->type) { diff --git a/src/base/group.c b/src/base/group.c index d1af5d8c..05c47591 100644 --- a/src/base/group.c +++ b/src/base/group.c @@ -119,6 +119,8 @@ int neu_group_add_tag(neu_group_t *group, neu_datatag_t *tag) el->tag->description = strdup(tag->description); el->tag->type = tag->type; el->tag->attribute = tag->attribute; + el->tag->option = tag->option; + memcpy(el->tag->meta, tag->meta, sizeof(tag->meta)); HASH_ADD_STR(group->tags, name, el); update_timestamp(group); @@ -149,6 +151,8 @@ int neu_group_update_tag(neu_group_t *group, neu_datatag_t *tag) el->tag->description = strdup(tag->description); el->tag->type = tag->type; el->tag->attribute = tag->attribute; + el->tag->option = tag->option; + memcpy(el->tag->meta, tag->meta, sizeof(tag->meta)); HASH_ADD_STR(group->tags, name, el); update_timestamp(group); diff --git a/src/base/tag.c b/src/base/tag.c index 3b945510..0f6570b1 100644 --- a/src/base/tag.c +++ b/src/base/tag.c @@ -29,8 +29,10 @@ static void tag_array_copy(void *_dst, const void *_src) neu_datatag_t *dst = (neu_datatag_t *) _dst; neu_datatag_t *src = (neu_datatag_t *) _src; - dst->type = src->type; - dst->attribute = src->attribute; + dst->type = src->type; + dst->attribute = src->attribute; + dst->option = src->option; + memcpy(dst->meta, src->meta, sizeof(src->meta)); dst->address = strdup(src->address); dst->name = strdup(src->name); dst->description = strdup(src->description); diff --git a/src/connection/connection.c b/src/connection/connection.c index 1406673a..59847a61 100644 --- a/src/connection/connection.c +++ b/src/connection/connection.c @@ -831,15 +831,14 @@ int neu_conn_wait_msg(neu_conn_t *conn, void *context, uint16_t n_byte, { uint8_t recv_buf[2048] = { 0 }; uint16_t offset = 0; - ssize_t ret = neu_conn_recv(conn, recv_buf, n_byte); - neu_buf_result_t result = { 0 }; - neu_protocol_unpack_buf_t pbuf = { 0 }; + ssize_t ret = neu_conn_recv(conn, recv_buf, n_byte); + neu_protocol_unpack_buf_t pbuf = { 0 }; while (ret > 0) { zlog_recv_protocol(conn->param.log, recv_buf + offset, ret); offset += ret; neu_protocol_unpack_buf_init(&pbuf, recv_buf, offset); - result = fn(context, &pbuf); + neu_buf_result_t result = fn(context, &pbuf); if (result.need > 0) { assert(result.need <= sizeof(recv_buf) - offset); ret = neu_conn_recv(conn, recv_buf + offset, result.need); @@ -854,17 +853,16 @@ int neu_conn_wait_msg(neu_conn_t *conn, void *context, uint16_t n_byte, int neu_conn_tcp_server_wait_msg(neu_conn_t *conn, int fd, void *context, uint16_t n_byte, neu_conn_process_msg fn) { - uint8_t recv_buf[2048] = { 0 }; - uint16_t offset = 0; - ssize_t ret = neu_conn_tcp_server_recv(conn, fd, recv_buf, n_byte); - neu_buf_result_t result = { 0 }; + uint8_t recv_buf[2048] = { 0 }; + uint16_t offset = 0; + ssize_t ret = neu_conn_tcp_server_recv(conn, fd, recv_buf, n_byte); neu_protocol_unpack_buf_t pbuf = { 0 }; while (ret > 0) { zlog_recv_protocol(conn->param.log, recv_buf + offset, ret); offset += ret; neu_protocol_unpack_buf_init(&pbuf, recv_buf, offset); - result = fn(context, &pbuf); + neu_buf_result_t result = fn(context, &pbuf); if (result.need > 0) { assert(result.need <= sizeof(recv_buf) - offset); ret = neu_conn_tcp_server_recv(conn, fd, recv_buf + offset,