From 1b60bed1e6417f26a1bcd255915199dc4f11af49 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Sat, 11 May 2024 13:00:17 +0800 Subject: [PATCH] feat(core): support write batch --- include/neuron/errcodes.h | 1 + src/adapter/adapter.c | 19 +++++++ src/adapter/driver/driver.c | 83 ++++++++++++++++++++++++++++ src/adapter/driver/driver_internal.h | 2 + src/base/msg.c | 6 +- src/core/manager.c | 48 ++++++++++++++++ 6 files changed, 157 insertions(+), 2 deletions(-) diff --git a/include/neuron/errcodes.h b/include/neuron/errcodes.h index 715ecdec..86aa53eb 100644 --- a/include/neuron/errcodes.h +++ b/include/neuron/errcodes.h @@ -135,6 +135,7 @@ typedef enum { NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS = 3017, NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC = 3018, NEU_ERR_PLUGIN_TYPE_NOT_SUPPORT = 3019, + NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_BATCH = 3021, NEU_ERR_MQTT_FAILURE = 4000, NEU_ERR_MQTT_NO_CERTFILESET = 4001, diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 28eadace..b2a4b145 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -707,6 +707,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQRESP_NODES_STATE: case NEU_REQ_PRGFILE_PROCESS: case NEU_RESP_PRGFILE_PROCESS: + case NEU_RESP_WRITE_BATCH: adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); neu_msg_free(msg); @@ -803,6 +804,24 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } break; } + case NEU_REQ_WRITE_BATCH: { + neu_reqresp_write_batch_t *cmd = + (neu_reqresp_write_batch_t *) &header[1]; + neu_write_batch_t * batch = cmd->batch; + neu_write_batch_driver_t *driver = &batch->drivers[cmd->index_]; + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + driver->error = NEU_ERR_GROUP_NOT_ALLOW; + header->type = NEU_RESP_WRITE_BATCH; + neu_msg_exchange(header); + reply(adapter, header, cmd); + } else { + neu_adapter_driver_write_batch((neu_adapter_driver_t *) adapter, + header); + } + + break; + } case NEU_REQ_NODE_SETTING: { neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) &header[1]; neu_resp_error_t error = { 0 }; diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 1b3d8102..4aa1835d 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -162,6 +162,12 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error) neu_req_write_gtags_fini((neu_req_write_gtags_t *) &req[1]); } + if (NEU_REQ_WRITE_BATCH == req->type) { + req->type = NEU_RESP_WRITE_BATCH; + adapter->cb_funs.response(adapter, req, &req[1]); + return; + } + req->type = NEU_RESP_ERROR; nlog_notice("write tag response <%p>", req->ctx); @@ -979,6 +985,83 @@ void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver, store_write_tag(first_g, &wtag); } +void neu_adapter_driver_write_batch(neu_adapter_driver_t *driver, + neu_reqresp_head_t * req) +{ + neu_reqresp_write_batch_t *cmd = (neu_reqresp_write_batch_t *) &req[1]; + neu_write_batch_driver_t * batch_driver = &cmd->batch->drivers[cmd->index_]; + + if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { + batch_driver->error = NEU_ERR_PLUGIN_NOT_RUNNING; + driver->adapter.cb_funs.driver.write_response( + &driver->adapter, req, NEU_ERR_PLUGIN_NOT_RUNNING); + return; + } + + if (driver->adapter.module->intf_funs->driver.write_batch == NULL) { + batch_driver->error = NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_BATCH; + driver->adapter.cb_funs.driver.write_response( + &driver->adapter, req, NEU_ERR_PLUGIN_NOT_SUPPORT_WRITE_TAGS); + return; + } + + bool need_write = false; + for (int i = 0; i < batch_driver->n_group; i++) { + neu_write_batch_group_t *batch_group = &batch_driver->groups[i]; + group_t * g = find_group(driver, batch_group->group); + if (NULL == g) { + // mark all group tags error + batch_group->error = NEU_ERR_GROUP_NOT_EXIST; + continue; + } + + for (int j = 0; j < batch_group->n_tag; j++) { + neu_datatag_t *tag = + neu_group_find_tag(g->group, batch_group->tags[j].name); + + if (NULL == tag) { + neu_write_batch_group_set_error(batch_group, j, + NEU_ERR_TAG_NOT_EXIST); + continue; + } + + if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_WRITE) || + neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC)) { + neu_write_batch_group_set_error( + batch_group, j, NEU_ERR_PLUGIN_TAG_NOT_ALLOW_WRITE); + continue; + } + + if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) { + if (batch_group->tags[j].value.type == NEU_TYPE_INT64) { + batch_group->tags[j].value.value.d64 = + (double) batch_group->tags[j].value.value.i64; + } + } + + if (tag->decimal != 0) { + cal_decimal(tag->type, batch_group->tags[j].value.type, + &batch_group->tags[j].value.value, tag->decimal); + } + + fix_value(tag, batch_group->tags[j].value.type, + &batch_group->tags[j].value); + + batch_group->tags[j].tag = tag; + need_write = true; + } + } + + if (!need_write) { + // nothing to request plugin write + driver->adapter.cb_funs.driver.write_response(&driver->adapter, req, 0); + return; + } + + driver->adapter.module->intf_funs->driver.write_batch( + driver->adapter.plugin, req, batch_driver); +} + void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, neu_reqresp_head_t * req) { diff --git a/src/adapter/driver/driver_internal.h b/src/adapter/driver/driver_internal.h index 0d52c048..625d0123 100644 --- a/src/adapter/driver/driver_internal.h +++ b/src/adapter/driver/driver_internal.h @@ -42,6 +42,8 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver, neu_reqresp_head_t * req); void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver, neu_reqresp_head_t * req); +void neu_adapter_driver_write_batch(neu_adapter_driver_t *driver, + neu_reqresp_head_t * req); int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name, uint32_t interval); diff --git a/src/base/msg.c b/src/base/msg.c index 36d20e39..04683517 100644 --- a/src/base/msg.c +++ b/src/base/msg.c @@ -26,5 +26,7 @@ void neu_msg_gen(neu_reqresp_head_t *header, void *data) { size_t data_size = neu_reqresp_size(header->type); assert(header->len >= sizeof(neu_reqresp_head_t) + data_size); - memcpy((uint8_t *) &header[1], data, data_size); -} \ No newline at end of file + if (data && &header[1] != data) { + memcpy((uint8_t *) &header[1], data, data_size); + } +} diff --git a/src/core/manager.c b/src/core/manager.c index d3e9b881..4acdf699 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -1058,6 +1058,54 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } + case NEU_REQ_WRITE_BATCH: { + neu_reqresp_write_batch_t *cmd = + (neu_reqresp_write_batch_t *) &header[1]; + neu_write_batch_t *batch = cmd->batch; + + for (int i = 0; i < batch->n_driver; ++i) { + const char *driver = batch->drivers[i].driver; + + if (NULL == neu_node_manager_find(manager->node_manager, driver)) { + // mark errors and done + batch->n_done_ += 1; + batch->drivers[i].error = NEU_ERR_NODE_NOT_EXIST; + continue; + } + + cmd->index_ = i; + if (i + 1 == batch->n_driver) { + forward_msg(manager, header, driver); // save a copy + } else { + forward_msg_copy(manager, header, driver); + } + } + + if (batch->n_done_ == batch->n_driver) { + // all done, reply response + header->type = NEU_RESP_WRITE_BATCH; + neu_msg_exchange(header); + forward_msg(manager, header, header->receiver); + } + + break; + } + case NEU_RESP_WRITE_BATCH: { + neu_reqresp_write_batch_t *cmd = + (neu_reqresp_write_batch_t *) &header[1]; + neu_write_batch_t *batch = cmd->batch; + + batch->n_done_ += 1; + if (batch->n_done_ == batch->n_driver) { + // all done, reply response + header->type = NEU_RESP_WRITE_BATCH; + forward_msg(manager, header, header->receiver); + } else { + neu_msg_free(msg); + } + + break; + } case NEU_REQ_PRGFILE_PROCESS: case NEU_REQ_PRGFILE_UPLOAD: case NEU_REQ_GET_NODE_STATE: {