GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

optimize modbus

Browse Source
This commit is contained in:
hxy7yx
2023-07-18 14:04:53 +08:00
parent 6f012790e5
commit 928289eaf1
5 changed files with 278 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/modbus/modbus.h
+17 -8
View File
@@ -40,14 +40,23 @@ typedef enum modbus_action {
} modbus_action_e;
typedef enum modbus_function {
MODBUS_READ_COIL = 0x1,
MODBUS_READ_INPUT = 0x02,
MODBUS_READ_HOLD_REG = 0x03,
MODBUS_READ_INPUT_REG = 0x04,
MODBUS_WRITE_S_COIL = 0x05,
MODBUS_WRITE_S_HOLD_REG = 0x06,
MODBUS_WRITE_M_HOLD_REG = 0x10,
MODBUS_WRITE_M_COIL = 0x0F
MODBUS_READ_COIL = 0x1,
MODBUS_READ_INPUT = 0x02,
MODBUS_READ_HOLD_REG = 0x03,
MODBUS_READ_INPUT_REG = 0x04,
MODBUS_WRITE_S_COIL = 0x05,
MODBUS_WRITE_S_HOLD_REG = 0x06,
MODBUS_WRITE_M_HOLD_REG = 0x10,
MODBUS_WRITE_M_COIL = 0x0F,
MODBUS_READ_COIL_ERR = 0x81,
MODBUS_READ_INPUT_ERR = 0x82,
MODBUS_READ_HOLD_REG_ERR = 0x83,
MODBUS_READ_INPUT_REG_ERR = 0x84,
MODBUS_WRITE_S_COIL_ERR = 0x85,
MODBUS_WRITE_S_HOLD_REG_ERR = 0x86,
MODBUS_WRITE_M_HOLD_REG_ERR = 0x90,
MODBUS_WRITE_M_COIL_ERR = 0x8F,
MODBUS_DEVICE_ERR = -2
} modbus_function_e;
typedef enum modbus_area {
plugins/modbus/modbus_req.c
+223 -33
View File
@@ -30,7 +30,8 @@ struct modbus_group_data {
};
static void plugin_group_free(neu_plugin_group_t *pgp);
static int process_protocol_buf(neu_plugin_t *plugin, uint16_t response_size);
static int process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
uint16_t response_size);
void modbus_conn_connected(void *data, int fd)
{
@@ -162,29 +163,137 @@ int modbus_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group,
plugin->cmd_idx = i;
uint16_t response_size = 0;
uint64_t read_tms = neu_time_ms();
int ret = modbus_stack_read(
int ret_buf = 0;
int ret_r = modbus_stack_read(
plugin->stack, gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].area, gd->cmd_sort->cmd[i].start_address,
gd->cmd_sort->cmd[i].n_register, &response_size);
if (ret > 0) {
ret = process_protocol_buf(plugin, response_size);
rtt = neu_time_ms() - read_tms;
}
if (ret <= 0) {
if (ret_r > 0) {
ret_buf = process_protocol_buf(
plugin, gd->cmd_sort->cmd[i].slave_id, response_size);
if (ret_buf > 0) {
rtt = neu_time_ms() - read_tms;
} else if (ret_buf == 0) {
for (uint16_t j = 0; j < plugin->max_retries; j++) {
ret_r = modbus_stack_read_retry(plugin, gd, i, j,
&response_size);
if (ret_r > 0) {
ret_buf = process_protocol_buf(
plugin, gd->cmd_sort->cmd[i].slave_id,
response_size);
if (ret_buf > 0) {
rtt = neu_time_ms() - read_tms;
break;
} else if (ret_buf < 0) {
if (ret_buf == -1) {
modbus_value_handle(
plugin, gd->cmd_sort->cmd[i].slave_id, 0,
NULL,
NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE);
plog_error(
plugin,
"modbus message error, skip, %hhu!%hu",
gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].start_address);
} else if (ret_buf == -2) {
modbus_value_handle(
plugin, gd->cmd_sort->cmd[i].slave_id, 0,
NULL, NEU_ERR_PLUGIN_READ_FAILURE);
plog_error(
plugin,
"modbus device response error, skip, "
"%hhu!%hu",
gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].start_address);
}
rtt = neu_time_ms() - read_tms;
break;
}
} else {
modbus_value_handle(plugin,
gd->cmd_sort->cmd[i].slave_id, 0,
NULL, NEU_ERR_PLUGIN_DISCONNECTED);
rtt = NEU_METRIC_LAST_RTT_MS_MAX;
neu_conn_disconnect(plugin->conn);
break;
}
}
if (ret_r > 0 && ret_buf == 0) {
modbus_value_handle(plugin, gd->cmd_sort->cmd[i].slave_id,
0, NULL,
NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE);
plog_warn(plugin,
"no modbus response received, skip, %hhu!%hu",
gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].start_address);
}
} else if (ret_buf < 0) {
if (ret_buf == -1) {
modbus_value_handle(plugin, gd->cmd_sort->cmd[i].slave_id,
0, NULL,
NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE);
plog_error(plugin, "modbus message error, skip, %hhu!%hu",
gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].start_address);
} else if (ret_buf == -2) {
modbus_value_handle(plugin, gd->cmd_sort->cmd[i].slave_id,
0, NULL, NEU_ERR_PLUGIN_READ_FAILURE);
plog_error(plugin,
"modbus device response error, skip, %hhu!%hu",
gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].start_address);
}
rtt = neu_time_ms() - read_tms;
}
} else {
for (uint16_t j = 0; j < plugin->max_retries; j++) {
ret = modbus_stack_read_retry(plugin, gd, i, j, &response_size);
if (ret > 0) {
ret = process_protocol_buf(plugin, response_size);
rtt = neu_time_ms() - read_tms;
break;
ret_r =
modbus_stack_read_retry(plugin, gd, i, j, &response_size);
if (ret_r > 0) {
ret_buf = process_protocol_buf(
plugin, gd->cmd_sort->cmd[i].slave_id, response_size);
if (ret_buf > 0) {
rtt = neu_time_ms() - read_tms;
break;
} else if (ret_buf < 0) {
if (ret_buf == -1) {
modbus_value_handle(
plugin, gd->cmd_sort->cmd[i].slave_id, 0, NULL,
NEU_ERR_PLUGIN_PROTOCOL_DECODE_FAILURE);
plog_error(plugin,
"modbus message error, skip, %hhu!%hu",
gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].start_address);
} else if (ret_buf == -2) {
modbus_value_handle(
plugin, gd->cmd_sort->cmd[i].slave_id, 0, NULL,
NEU_ERR_PLUGIN_READ_FAILURE);
plog_error(plugin,
"modbus device response error, skip, "
"%hhu!%hu",
gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].start_address);
}
rtt = neu_time_ms() - read_tms;
break;
} else if (ret_buf == 0) {
modbus_value_handle(
plugin, gd->cmd_sort->cmd[i].slave_id, 0, NULL,
NEU_ERR_PLUGIN_DEVICE_NOT_RESPONSE);
plog_warn(plugin,
"no modbus response received, skip, %hhu!%hu",
gd->cmd_sort->cmd[i].slave_id,
gd->cmd_sort->cmd[i].start_address);
continue;
}
}
}
if (ret <= 0) {
if (ret_r <= 0) {
modbus_value_handle(plugin, gd->cmd_sort->cmd[i].slave_id, 0,
NULL, NEU_ERR_PLUGIN_DISCONNECTED);
rtt = NEU_METRIC_LAST_RTT_MS_MAX;
neu_conn_disconnect(plugin->conn);
break;
}
}
if (plugin->interval > 0) {
@@ -216,7 +325,7 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
uint16_t start_address = gd->cmd_sort->cmd[plugin->cmd_idx].start_address;
uint16_t n_register = gd->cmd_sort->cmd[plugin->cmd_idx].n_register;
if (error != NEU_ERR_SUCCESS) {
if (error == NEU_ERR_PLUGIN_DISCONNECTED) {
neu_dvalue_t dvalue = { 0 };
dvalue.type = NEU_TYPE_ERROR;
@@ -224,6 +333,17 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
plugin->common.adapter_callbacks->driver.update(
plugin->common.adapter, gd->group, NULL, dvalue);
return 0;
} else if (error != NEU_ERR_SUCCESS) {
utarray_foreach(gd->cmd_sort->cmd[plugin->cmd_idx].tags,
modbus_point_t **, p_tag)
{
neu_dvalue_t dvalue = { 0 };
dvalue.type = NEU_TYPE_ERROR;
dvalue.value.i32 = error;
plugin->common.adapter_callbacks->driver.update(
plugin->common.adapter, gd->group, (*p_tag)->name, dvalue);
}
return 0;
}
utarray_foreach(gd->cmd_sort->cmd[plugin->cmd_idx].tags, modbus_point_t **,
@@ -380,7 +500,7 @@ int modbus_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
point.start_address, point.n_register, value.bytes,
n_byte, &response_size);
if (ret > 0) {
process_protocol_buf(plugin, response_size);
process_protocol_buf(plugin, point.slave_id, response_size);
}
return 0;
@@ -409,29 +529,99 @@ static void plugin_group_free(neu_plugin_group_t *pgp)
free(gd);
}
static int process_protocol_buf(neu_plugin_t *plugin, uint16_t response_size)
static int process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
uint16_t response_size)
{
uint8_t * recv_buf = calloc(response_size, 1);
neu_protocol_unpack_buf_t pbuf = { 0 };
ssize_t ret = 0;
if (plugin->is_server && plugin->protocol == MODBUS_PROTOCOL_TCP) {
ret = neu_conn_tcp_server_recv(plugin->conn, plugin->client_fd,
recv_buf, response_size);
} else {
ret = neu_conn_recv(plugin->conn, recv_buf, response_size);
}
if (ret == response_size) {
if (response_size < 512) {
plog_recv_protocol(plugin, recv_buf, ret);
if (plugin->protocol == MODBUS_PROTOCOL_TCP) {
if (plugin->is_server) {
ret = neu_conn_tcp_server_recv(plugin->conn, plugin->client_fd,
recv_buf,
sizeof(struct modbus_header));
} else {
ret = neu_conn_recv(plugin->conn, recv_buf,
sizeof(struct modbus_header));
}
neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
ret = modbus_stack_recv(plugin->stack, &pbuf);
} else {
ret = -1;
}
if (ret == 0 || ret == -1) {
free(recv_buf);
return 0;
}
if (ret == sizeof(struct modbus_header)) {
struct modbus_header *header = (struct modbus_header *) recv_buf;
int ret1 = 0;
if (htons(header->len) >
response_size - sizeof(struct modbus_header)) {
free(recv_buf);
return -1;
}
plog_recv_protocol(plugin, recv_buf, ret);
if (plugin->is_server) {
ret1 = neu_conn_tcp_server_recv(
plugin->conn, plugin->client_fd,
recv_buf + sizeof(struct modbus_header),
htons(header->len));
} else {
ret1 = neu_conn_recv(plugin->conn,
recv_buf + sizeof(struct modbus_header),
htons(header->len));
}
if (ret1 != htons(header->len)) {
free(recv_buf);
return -1;
}
ret += ret1;
} else {
free(recv_buf);
return -1;
}
if (ret > 0) {
if (ret < 512) {
plog_recv_protocol(plugin, recv_buf, ret);
}
neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
int ret_s = modbus_stack_recv(plugin->stack, slave_id, &pbuf);
if (ret_s == MODBUS_DEVICE_ERR) {
ret = ret_s;
} else {
if (ret != response_size) {
ret = -1;
} else {
ret = ret_s;
}
}
}
} else if (plugin->protocol == MODBUS_PROTOCOL_RTU) {
ret = neu_conn_recv(plugin->conn, recv_buf, response_size);
if (ret == 0 || ret == -1) {
free(recv_buf);
return 0;
}
if (ret > 0) {
if (response_size < 512) {
plog_recv_protocol(plugin, recv_buf, ret);
}
neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
int ret_s = modbus_stack_recv(plugin->stack, 0, &pbuf);
if (ret_s == MODBUS_DEVICE_ERR) {
ret = ret_s;
} else {
if (ret != response_size) {
ret = -1;
} else {
ret = ret_s;
}
}
}
}
free(recv_buf);
return ret;
}
plugins/modbus/modbus_rtu.c
+3 -3
View File
@@ -66,11 +66,11 @@ const neu_plugin_module_t neu_plugin_module = {
.module_name = "Modbus RTU",
.module_descr =
"This plugin is used to connect devices using modbus rtu protocol. "
"Support serial,TCP and UDP communication methods. Selecting "
"Support serial and TCP communication methods. Selecting "
"client/server is also supported.",
.module_descr_zh =
"该插件用于连接使用 modbus rtu 协议的设备。支持串口, TCP 和 UDP "
"三种通信方式。也支持选择客户端/服务端。",
"该插件用于连接使用 modbus rtu 协议的设备。支持串口, TCP "
"两种通信方式。也支持选择客户端/服务端。",
.intf_funs = &plugin_intf_funs,
.kind = NEU_PLUGIN_KIND_SYSTEM,
.type = NEU_NA_TYPE_DRIVER,
plugins/modbus/modbus_stack.c
+33 -6
View File
@@ -62,7 +62,8 @@ void modbus_stack_destroy(modbus_stack_t *stack)
free(stack);
}
int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf)
int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
neu_protocol_unpack_buf_t *buf)
{
struct modbus_header header = { 0 };
struct modbus_code code = { 0 };
@@ -74,13 +75,19 @@ int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf)
plog_warn((neu_plugin_t *) stack->ctx, "try modbus rtu driver");
}
if (ret <= 0) {
return ret;
return -1;
}
}
ret = modbus_code_unwrap(buf, &code);
if (ret <= 0) {
return ret;
return -1;
}
if (stack->protocol == MODBUS_PROTOCOL_TCP) {
if (code.slave_id != slave_id) {
return -1;
}
}
switch (code.function) {
@@ -92,7 +99,7 @@ int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf)
uint8_t * bytes = NULL;
ret = modbus_data_unwrap(buf, &data);
if (ret <= 0) {
return ret;
return -1;
}
if (stack->protocol == MODBUS_PROTOCOL_TCP && data.n_byte == 0xff) {
@@ -116,18 +123,38 @@ int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf)
case MODBUS_WRITE_M_COIL: {
struct modbus_address address = { 0 };
ret = modbus_address_unwrap(buf, &address);
if (ret <= 0) {
return -1;
}
break;
}
case MODBUS_READ_COIL_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_READ_INPUT_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_READ_HOLD_REG_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_READ_INPUT_REG_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_S_COIL_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_S_HOLD_REG_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_M_HOLD_REG_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_M_COIL_ERR:
return MODBUS_DEVICE_ERR;
case MODBUS_WRITE_S_HOLD_REG:
default:
break;
default:
return -1;
}
if (stack->protocol == MODBUS_PROTOCOL_RTU) {
struct modbus_crc crc = { 0 };
ret = modbus_crc_unwrap(buf, &crc);
if (ret <= 0) {
return ret;
return -1;
}
}
plugins/modbus/modbus_stack.h
+2 -1
View File
@@ -43,7 +43,8 @@ modbus_stack_t *modbus_stack_create(void *ctx, modbus_protocol_e protocol,
modbus_stack_write_resp write_resp);
void modbus_stack_destroy(modbus_stack_t *stack);
int modbus_stack_recv(modbus_stack_t *stack, neu_protocol_unpack_buf_t *buf);
int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
neu_protocol_unpack_buf_t *buf);
int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
enum modbus_area area, uint16_t start_address,