Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feature: support udp large packet segmentation to send multiple times… #2351

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
Open

feature: support udp large packet segmentation to send multiple times… #2351

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
535f365
feature: support udp large packet segmentation to send multiple times…
Aug 21, 2024
cdb5ae6
Rollback doc commit
Aug 22, 2024
f5e3de5
Update ngx_http_lua_socket_udp.c
coffeei Aug 22, 2024
0bd6cd6
Update code format
coffeei Aug 22, 2024
078cf87
Update code format
coffeei Aug 22, 2024
a71f4a4
Update code format
coffeei Aug 22, 2024
f250915
Update 189-udp-buff.t
coffeei Aug 26, 2024
34a4895
Update 189-udp-buff.t
coffeei Nov 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
142 changes: 141 additions & 1 deletion src/ngx_http_lua_socket_udp.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
static int ngx_http_lua_socket_udp(lua_State *L);
static int ngx_http_lua_socket_udp_setpeername(lua_State *L);
static int ngx_http_lua_socket_udp_send(lua_State *L);
static int ngx_http_lua_socket_udp_buff_send(lua_State *L);
static int ngx_http_lua_socket_udp_receive(lua_State *L);
static int ngx_http_lua_socket_udp_settimeout(lua_State *L);
static void ngx_http_lua_socket_udp_finalize(ngx_http_request_t *r,
Expand Down Expand Up @@ -86,14 +87,17 @@ ngx_http_lua_inject_socket_udp_api(ngx_log_t *log, lua_State *L)
/* udp socket object metatable */
lua_pushlightuserdata(L, ngx_http_lua_lightudata_mask(
socket_udp_metatable_key));
lua_createtable(L, 0 /* narr */, 6 /* nrec */);
lua_createtable(L, 0 /* narr */, 7 /* nrec */);

lua_pushcfunction(L, ngx_http_lua_socket_udp_setpeername);
lua_setfield(L, -2, "setpeername"); /* ngx socket mt */

lua_pushcfunction(L, ngx_http_lua_socket_udp_send);
lua_setfield(L, -2, "send");

lua_pushcfunction(L, ngx_http_lua_socket_udp_buff_send);
lua_setfield(L, -2, "sendbuf");

lua_pushcfunction(L, ngx_http_lua_socket_udp_receive);
lua_setfield(L, -2, "receive");

Expand Down Expand Up @@ -961,6 +965,142 @@ ngx_http_lua_socket_udp_send(lua_State *L)
}


#define stack_diff 1


static int
ngx_http_lua_socket_udp_buff_send(lua_State *L)
{
ssize_t n;
ngx_http_request_t *r;
ngx_http_lua_socket_udp_upstream_t *u;
ngx_http_lua_loc_conf_t *llcf;
// add new variable for buffer send
ngx_str_t query;
size_t size;
size_t head_len;
size_t buf_len;
u_char *msg_tmp;
const char *header_str;
const char *buff_str;


if (lua_gettop(L) < 4 ) {
return luaL_error(L, "expecting at least 4 arguments, "
"but got %d (including the object)", lua_gettop(L));
}

r = ngx_http_lua_get_req(L);
if (r == NULL) {
return luaL_error(L, "request object not found");
}

luaL_checktype(L, 1, LUA_TTABLE);

lua_rawgeti(L, 1, SOCKET_CTX_INDEX);
u = lua_touserdata(L, -1);
lua_pop(L, 1);

if (u == NULL || u->udp_connection.connection == NULL) {
llcf = ngx_http_get_module_loc_conf(r, ngx_http_lua_module);

if (llcf->log_socket_errors) {
ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
"attempt to send data on a closed socket: u:%p, c:%p",
u, u ? u->udp_connection.connection : NULL);
}

lua_pushnil(L);
lua_pushliteral(L, "closed");
return 2;
}

if (u->request != r) {
return luaL_error(L, "bad request");
}

if (u->ft_type) {
u->ft_type = 0;
}

if (u->waiting) {
lua_pushnil(L);
lua_pushliteral(L, "socket busy");
return 2;
}

// check and calculate string len
int stack_size = lua_gettop(L);
size_t start = luaL_checklong(L, -2);
size_t end = luaL_checklong(L, -1);
buff_str = luaL_checklstring(L, -3, &buf_len);

if (start < 1 || end > buf_len || start > end || start > buf_len) {
lua_pushnil(L);
lua_pushliteral(L, "start or end index invalid");
return 2;
}

size = (end - start + 1);
buf_len = size;

if (stack_size > 3 + stack_diff) {
for (int i = (-1 * stack_size) + stack_diff; i <= -4; i++) {
luaL_checklstring(L, i, &head_len);
size += head_len;
}
}

// copy msg
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you should use the string.buffer instead of implement this feature.

Copy link
Author

@coffeei coffeei Aug 21, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

how to use string.buffer instead of implement this feature on lua land ,can you give some example? the copy occurs on the c land,but i need to split the big string on the lua land ,and also i need to make the header of the udp package on lua land. if i use the origin udpsock:send() method , sub string ( big_msg ) is inevitable ,.usually i can send big string
use udp like below，there is no need to make a new string pass to c land.

for idx = 1, totalPackage do
        local header_binary = struct.pack(">i2c1c32i4c1", packageSize + 42, '1', msg_id, totalPackage, zip_type or '0')
        local serial_binary
        if totalPackage == 1 then
            serial_binary = struct.pack(">i4", 0)
        else
            msgEndIndex = msgStartIndex + packageSize - 1
            serial_binary = struct.pack(">i4", idx - 1)
        end
        ok, err = udpsock:sendbuf(header_binary, serial_binary, big_msg, msgStartIndex, msgEndIndex)
        ---sub string use origin method
        ---ok, err = udpsock:send({ header_binary, serial_binary, string.sub(msg, msgStartIndex, msgEndIndex) })
        msgStartIndex = msgEndIndex + 1
    end

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://luajit.org/ext_buffer.html

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks a lot，but we can't assume that from the beginning the user can use string.buffer to build the string content on the lua land ,sometimes is really hard for it , and more scenario is that the user can only accept a large string which was already built before.

I think we still need to copy a string on lua land, event if we use the string buffer.that is because openresty only provides a udp send method which can accepts only a string or table object.

Limited by udp message size , it’s necessary to split an existing big string, so you need to avoid a new large number of substring , The new udp send method that i provided is to adapt to this scenario.

Thank you very much!

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

gotcha

query.data = lua_newuserdata(L, size);
query.len = size;
msg_tmp = query.data;
lua_pop(L, 1);

if (stack_size > 3 + stack_diff) {
for (int i = (-1 * stack_size) + stack_diff; i <= -4; i++) {
header_str = lua_tolstring(L, i, &head_len);
msg_tmp = ngx_cpymem(msg_tmp, (u_char *) header_str, head_len);
}
}

buff_str = buff_str + start - 1;
ngx_memcpy(msg_tmp, (u_char *) buff_str, buf_len);

u->ft_type = 0;

/* mimic ngx_http_upstream_init_request here */

#if 1
u->waiting = 0;
#endif

dd("sending query %.*s", (int) query.len, query.data);

n = ngx_send(u->udp_connection.connection, query.data, query.len);

dd("ngx_send returns %d (query len %d)", (int) n, (int) query.len);

if (n == NGX_ERROR || n == NGX_AGAIN) {
u->socket_errno = ngx_socket_errno;

return ngx_http_lua_socket_error_retval_handler(r, u, L);
}

if (n != (ssize_t) query.len) {
dd("not the while query was sent");

u->ft_type |= NGX_HTTP_LUA_SOCKET_FT_PARTIALWRITE;
return ngx_http_lua_socket_error_retval_handler(r, u, L);
}

dd("n == len");

lua_pushinteger(L, 1);
return 1;
}


static int
ngx_http_lua_socket_udp_receive(lua_State *L)
{
Expand Down
49 changes: 49 additions & 0 deletions t/189-udp-buff.t
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
# vim:set ft= ts=4 sw=4 et fdm=marker:

use Test::Nginx::Socket::Lua 'no_plan';

#worker_connections(1014);
master_on();
workers(2);
#log_level('warn');

repeat_each(2);

#no_diff();
no_long_string();
run_tests();

__DATA__

=== TEST 1: use udp send buffer
--- config
location /lua {
content_by_lua_block {
local udpsock = ngx.socket.udp()
udpsock:settimeout(500)
local host,port = "127.0.0.1",8080
local ok, err = udpsock:setpeername(host, port)
if not ok then
ngx.say("============failed to connect to udp server: ", host, ",err:", err)
return
end
local header_binary = "11"
local serial_binary = "22"
local log_msg_content = "3333333312345678901234567890123456789012345678901234567890"
local msgStartIndex = 10
local msgEndIndex = 20
ok, err = udpsock:sendbuf(header_binary, serial_binary, log_msg_content, msgStartIndex, msgEndIndex)
if not ok then
ngx.say("============failed to send: ", host, ",err:", err)
return
end
ngx.say("OK")

}
}
--- request
GET /lua
--- response_body_like
OK
--- no_error_log
[error]
Loading