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

Merge pull request #660 from fengzeroz/main

Browse Source 
fix: replace large stack var with heap var
This commit is contained in:
fengzeroz
2022-07-26 10:58:02 +08:00
committed by GitHub
parent 756736ae97 d54fb5772e
commit 49574082b2
7 changed files with 117 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+8 -3
View File
@@ -96,11 +96,16 @@ jobs:
needs: [cross-build, build-debug-sdk]
steps:
- name: Extract branch name
shell: bash
run: echo "##[set-output name=branch;]$(echo ${GITHUB_REF#refs/heads/})"
id: extract_branch
- name: delete old daily-pre-release
uses: dev-drprasad/delete-tag-and-release@v0.2.0
with:
delete_release: true
tag_name: daily
tag_name: ${{ steps.extract_branch.outputs.branch }}-daily
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
@@ -110,8 +115,8 @@ jobs:
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
with:
tag_name: daily
release_name: Daily Release
tag_name: ${{ steps.extract_branch.outputs.branch }}-daily
release_name: ${{ steps.extract_branch.outputs.branch }} Daily Release
draft: false
prerelease: true
include/neuron/connection/neu_connection.h
-2
View File
@@ -112,11 +112,9 @@ void neu_conn_disconnect(neu_conn_t *conn);
int neu_conn_tcp_server_accept(neu_conn_t *conn);
int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd);
void neu_conn_flush(neu_conn_t *conn);
ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len);
ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len);
void neu_conn_tcp_server_flush(neu_conn_t *conn, int fd);
ssize_t neu_conn_tcp_server_send(neu_conn_t *conn, int fd, uint8_t *buf,
ssize_t len);
ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
include/neuron/utils/log.h
+42 -39
View File
@@ -64,46 +64,49 @@ extern zlog_category_t *neuron;
zlog((plugin)->common.log, __FILE__, sizeof(__FILE__) - 1, __func__, \
sizeof(__func__) - 1, __LINE__, ZLOG_LEVEL_DEBUG, __VA_ARGS__)
#define plog_send_protocol(plugin, bytes, n_byte) \
static __thread char log_protocol_buf[65535] = { 0 }; \
int log_offset = 0; \
uint16_t tmp_n_byte = n_byte; \
memset(log_protocol_buf, 0, sizeof(log_protocol_buf)); \
log_offset = snprintf(log_protocol_buf, sizeof(log_protocol_buf), \
">>(%d)", tmp_n_byte); \
for (int i = 0; i < n_byte; i++) { \
log_offset += snprintf(log_protocol_buf + log_offset, \
sizeof(log_protocol_buf) - log_offset, \
" 0x%02hhX", (bytes)[i]); \
} \
plog_debug(plugin, "%s", log_protocol_buf);
#define plog_send_protocol(plugin, bytes, n_byte) \
uint16_t log_protocol_buf_size = n_byte * 5 + 20; \
char * log_protocol_buf = calloc(log_protocol_buf_size, 1); \
int log_offset = 0; \
uint16_t tmp_n_byte = n_byte; \
log_offset = snprintf(log_protocol_buf, log_protocol_buf_size, ">>(%d)", \
tmp_n_byte); \
for (int i = 0; i < n_byte; i++) { \
log_offset += snprintf(log_protocol_buf + log_offset, \
log_protocol_buf_size - log_offset, \
" 0x%02hhX", (bytes)[i]); \
} \
plog_debug(plugin, "%s", log_protocol_buf); \
free(log_protocol_buf);
#define plog_recv_protocol(plugin, bytes, n_byte) \
static __thread char log_protocol_buf[65535] = { 0 }; \
int log_offset = 0; \
uint16_t tmp_n_byte = n_byte; \
memset(log_protocol_buf, 0, sizeof(log_protocol_buf)); \
log_offset = snprintf(log_protocol_buf, sizeof(log_protocol_buf), \
"<<(%d)", tmp_n_byte); \
for (int i = 0; i < n_byte; i++) { \
log_offset += snprintf(log_protocol_buf + log_offset, \
sizeof(log_protocol_buf) - log_offset, \
" 0x%02hhX", (bytes)[i]); \
} \
plog_debug(plugin, "%s", log_protocol_buf);
#define plog_recv_protocol(plugin, bytes, n_byte) \
uint16_t log_protocol_buf_size = n_byte * 5 + 20; \
char * log_protocol_buf = calloc(log_protocol_buf_size, 1); \
int log_offset = 0; \
uint16_t tmp_n_byte = n_byte; \
log_offset = snprintf(log_protocol_buf, log_protocol_buf_size, "<<(%d)", \
tmp_n_byte); \
for (int i = 0; i < n_byte; i++) { \
log_offset += snprintf(log_protocol_buf + log_offset, \
log_protocol_buf_size - log_offset, \
" 0x%02hhX", (bytes)[i]); \
} \
plog_debug(plugin, "%s", log_protocol_buf); \
free(log_protocol_buf);
#define zlog_recv_protocol(log, bytes, n_byte) \
static __thread char log_protocol_buf[65535] = { 0 }; \
int log_offset = 0; \
uint16_t tmp_n_byte = n_byte; \
memset(log_protocol_buf, 0, sizeof(log_protocol_buf)); \
log_offset = snprintf(log_protocol_buf, sizeof(log_protocol_buf), \
"<<(%d)", tmp_n_byte); \
for (int i = 0; i < n_byte; i++) { \
log_offset += snprintf(log_protocol_buf + log_offset, \
sizeof(log_protocol_buf) - log_offset, \
" 0x%02hhX", (bytes)[i]); \
} \
zlog_debug(log, "%s", log_protocol_buf);
#define zlog_recv_protocol(log, bytes, n_byte) \
uint16_t log_protocol_buf_size = n_byte * 5 + 20; \
char * log_protocol_buf = calloc(log_protocol_buf_size, 1); \
int log_offset = 0; \
uint16_t tmp_n_byte = n_byte; \
log_offset = snprintf(log_protocol_buf, log_protocol_buf_size, "<<(%d)", \
tmp_n_byte); \
for (int i = 0; i < n_byte; i++) { \
log_offset += snprintf(log_protocol_buf + log_offset, \
log_protocol_buf_size - log_offset, \
" 0x%02hhX", (bytes)[i]); \
} \
zlog_debug(log, "%s", log_protocol_buf); \
free(log_protocol_buf);
#endif
plugins/modbus/modbus_req.c
+7 -6
View File
@@ -387,17 +387,18 @@ static void plugin_group_free(neu_plugin_group_t *pgp)
static int process_protocol_buf(neu_plugin_t *plugin, uint16_t response_size)
{
static __thread uint8_t recv_buf[65535] = { 0 };
neu_protocol_unpack_buf_t pbuf = { 0 };
ssize_t ret = 0;
uint8_t * recv_buf = calloc(256, 1);
neu_protocol_unpack_buf_t pbuf = { 0 };
ssize_t ret = 0;
ret = neu_conn_recv(plugin->conn, recv_buf, response_size);
if (ret > 0) {
neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
plog_recv_protocol(plugin, recv_buf, ret);
return modbus_stack_recv(plugin->stack, &pbuf);
} else {
return -1;
ret = modbus_stack_recv(plugin->stack, &pbuf);
}
free(recv_buf);
return ret;
}
plugins/modbus/modbus_stack.c
+9 -6
View File
@@ -111,7 +111,7 @@ int modbus_stack_read(modbus_stack_t *stack, uint8_t slave_id,
enum modbus_area area, uint16_t start_address,
uint16_t n_reg, uint16_t *response_size)
{
static __thread uint8_t buf[32] = { 0 };
static __thread uint8_t buf[16] = { 0 };
static __thread neu_protocol_pack_buf_t pbuf = { 0 };
neu_protocol_pack_buf_init(&pbuf, buf, sizeof(buf));
@@ -152,10 +152,11 @@ int modbus_stack_write(modbus_stack_t *stack, void *req, uint8_t slave_id,
uint16_t n_reg, uint8_t *bytes, uint8_t n_byte,
uint16_t *response_size)
{
static __thread uint8_t buf[256] = { 0 };
static __thread neu_protocol_pack_buf_t pbuf = { 0 };
uint8_t * buf = calloc(256, 1);
neu_protocol_pack_buf_t pbuf = { 0 };
int ret = 0;
neu_protocol_pack_buf_init(&pbuf, buf, sizeof(buf));
neu_protocol_pack_buf_init(&pbuf, buf, 256);
switch (area) {
case MODBUS_AREA_COIL:
@@ -182,6 +183,8 @@ int modbus_stack_write(modbus_stack_t *stack, void *req, uint8_t slave_id,
*response_size += sizeof(struct modbus_header);
write_req = req;
return stack->send_fn(stack->ctx, neu_protocol_pack_buf_used_size(&pbuf),
neu_protocol_pack_buf_get(&pbuf));
ret = stack->send_fn(stack->ctx, neu_protocol_pack_buf_used_size(&pbuf),
neu_protocol_pack_buf_get(&pbuf));
free(buf);
return ret;
}
src/connection/connection.c
+50 -82
View File
@@ -57,6 +57,10 @@ struct neu_conn {
int n_client;
bool is_listen;
} tcp_server;
uint8_t *buf;
uint16_t buf_size;
uint16_t offset;
};
static void conn_tcp_server_add_client(neu_conn_t *conn, int fd,
@@ -85,6 +89,10 @@ neu_conn_t *neu_conn_new(neu_conn_param_t *param, void *data,
conn->disconnected = disconnected;
conn->connected = connected;
conn->buf_size = 2048;
conn->buf = calloc(conn->buf_size, 1);
conn->offset = 0;
conn_tcp_server_listen(conn);
pthread_mutex_init(&conn->mtx, NULL);
@@ -121,6 +129,7 @@ void neu_conn_destory(neu_conn_t *conn)
pthread_mutex_destroy(&conn->mtx);
free(conn->buf);
free(conn);
}
@@ -308,41 +317,6 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
return ret;
}
void neu_conn_tcp_server_flush(neu_conn_t *conn, int fd)
{
(void) conn;
ssize_t ret = 1;
char buf[64] = { 0 };
while (ret > 0) {
ret = recv(fd, &buf, sizeof(buf), 0);
}
}
void neu_conn_flush(neu_conn_t *conn)
{
ssize_t ret = 1;
char buf[64] = { 0 };
switch (conn->param.type) {
case NEU_CONN_TCP_SERVER:
zlog_fatal(conn->param.log,
"neu_conn_flush cann't flush tcp server msg");
assert(1 == 0);
break;
case NEU_CONN_TCP_CLIENT:
case NEU_CONN_UDP:
while (ret > 0) {
pthread_mutex_lock(&conn->mtx);
ret = recv(conn->fd, buf, sizeof(buf), 0);
pthread_mutex_unlock(&conn->mtx);
}
break;
case NEU_CONN_TTY_CLIENT:
break;
}
}
void neu_conn_disconnect(neu_conn_t *conn)
{
conn_disconnect(conn);
@@ -725,6 +699,7 @@ static void conn_disconnect(neu_conn_t *conn)
conn->disconnected(conn->data, conn->fd);
conn->callback_trigger = false;
}
conn->offset = 0;
}
static void conn_tcp_server_add_client(neu_conn_t *conn, int fd,
@@ -760,39 +735,37 @@ static void conn_tcp_server_del_client(neu_conn_t *conn, int fd)
void neu_conn_stream_consume(neu_conn_t *conn, void *context,
neu_conn_stream_consume_fn fn)
{
static __thread uint8_t recv_buf[2048] = { 0 };
static __thread uint16_t offset = 0;
if (fn == NULL) {
offset = 0;
memset(recv_buf, 0, sizeof(recv_buf));
conn->offset = 0;
memset(conn->buf, 0, conn->buf_size);
return;
}
ssize_t ret =
neu_conn_recv(conn, recv_buf + offset, sizeof(recv_buf) - offset);
ssize_t ret = neu_conn_recv(conn, conn->buf + conn->offset,
conn->buf_size - conn->offset);
if (ret > 0) {
zlog_recv_protocol(conn->param.log, recv_buf + offset, ret);
offset += ret;
zlog_recv_protocol(conn->param.log, conn->buf + conn->offset, ret);
conn->offset += ret;
neu_protocol_unpack_buf_t protocol_buf = { 0 };
neu_protocol_unpack_buf_init(&protocol_buf, recv_buf, offset);
neu_protocol_unpack_buf_init(&protocol_buf, conn->buf, conn->offset);
while (neu_protocol_unpack_buf_unused_size(&protocol_buf) > 0) {
int used = fn(context, &protocol_buf);
zlog_debug(conn->param.log, "buf used: %d offset: %d", used,
offset);
conn->offset);
if (used == 0) {
break;
} else if (used == -1) {
memset(recv_buf, 0, sizeof(recv_buf));
offset = 0;
memset(conn->buf, 0, conn->buf_size);
conn->offset = 0;
break;
}
}
if (offset != 0) {
offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
memmove(recv_buf,
recv_buf + neu_protocol_unpack_buf_used_size(&protocol_buf),
offset);
if (conn->offset != 0) {
conn->offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
memmove(conn->buf,
conn->buf +
neu_protocol_unpack_buf_used_size(&protocol_buf),
conn->offset);
}
}
}
@@ -800,30 +773,29 @@ 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)
{
static __thread uint8_t recv_buf[2048] = { 0 };
static __thread uint16_t offset = 0;
ssize_t ret = neu_conn_tcp_server_recv(conn, fd, recv_buf + offset,
sizeof(recv_buf) - offset);
ssize_t ret = neu_conn_tcp_server_recv(conn, fd, conn->buf + conn->offset,
conn->buf_size - conn->offset);
if (ret > 0) {
offset += ret;
conn->offset += ret;
neu_protocol_unpack_buf_t protocol_buf = { 0 };
neu_protocol_unpack_buf_init(&protocol_buf, recv_buf, offset);
neu_protocol_unpack_buf_init(&protocol_buf, conn->buf, conn->offset);
while (neu_protocol_unpack_buf_unused_size(&protocol_buf) > 0) {
int used = fn(context, &protocol_buf);
if (used == 0) {
break;
} else if (used == -1) {
memset(recv_buf, 0, sizeof(recv_buf));
offset = 0;
memset(conn->buf, 0, conn->buf_size);
conn->offset = 0;
break;
}
}
if (offset != 0) {
offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
memmove(recv_buf,
recv_buf + neu_protocol_unpack_buf_used_size(&protocol_buf),
offset);
if (conn->offset != 0) {
conn->offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
memmove(conn->buf,
conn->buf +
neu_protocol_unpack_buf_used_size(&protocol_buf),
conn->offset);
}
}
}
@@ -831,19 +803,17 @@ void neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
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);
ssize_t ret = neu_conn_recv(conn, conn->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);
zlog_recv_protocol(conn->param.log, conn->buf + conn->offset, ret);
conn->offset += ret;
neu_protocol_unpack_buf_init(&pbuf, conn->buf, conn->offset);
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);
assert(result.need <= conn->buf_size - conn->offset);
ret = neu_conn_recv(conn, conn->buf + conn->offset, result.need);
} else {
return result.used;
}
@@ -855,19 +825,17 @@ 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);
ssize_t ret = neu_conn_tcp_server_recv(conn, fd, conn->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);
zlog_recv_protocol(conn->param.log, conn->buf + conn->offset, ret);
conn->offset += ret;
neu_protocol_unpack_buf_init(&pbuf, conn->buf, conn->offset);
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,
assert(result.need <= conn->buf_size - conn->offset);
ret = neu_conn_tcp_server_recv(conn, fd, conn->buf + conn->offset,
result.need);
} else {
return result.used;
version
+1 -1
View File
@@ -1 +1 @@
2.1.1
2.2.0