diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index 1cb8dcc2..38f96a5c 100644 --- a/lualib-src/lua-socket.c +++ b/lualib-src/lua-socket.c @@ -525,41 +525,59 @@ concat_table(lua_State *L, int index, void *buffer, size_t tlen) { lua_pop(L,1); } -static void * -get_buffer(lua_State *L, int index, int *sz) { +static void +get_buffer(lua_State *L, int index, struct socket_sendbuffer *buf) { void *buffer; switch(lua_type(L, index)) { - const char * str; size_t len; case LUA_TUSERDATA: - case LUA_TLIGHTUSERDATA: - buffer = lua_touserdata(L,index); - *sz = luaL_checkinteger(L,index+1); + // lua full useobject must be a raw pointer, it can't be a socket object or a memory object. + buf->type = SOCKET_BUFFER_RAWPOINTER; + buf->buffer = lua_touserdata(L, index); + if (lua_isinteger(L, index+1)) { + buf->sz = lua_tointeger(L, index+1); + } else { + buf->sz = lua_rawlen(L, index+1); + } break; + case LUA_TLIGHTUSERDATA: { + int sz = -1; + if (lua_isinteger(L, index+1)) { + sz = lua_tointeger(L,index+1); + } + if (sz < 0) { + buf->type = SOCKET_BUFFER_OBJECT; + } else { + buf->type = SOCKET_BUFFER_MEMORY; + } + buf->buffer = lua_touserdata(L,index); + buf->sz = (size_t)sz; + break; + } case LUA_TTABLE: // concat the table as a string len = count_size(L, index); buffer = skynet_malloc(len); concat_table(L, index, buffer, len); - *sz = (int)len; + buf->type = SOCKET_BUFFER_MEMORY; + buf->buffer = buffer; + buf->sz = len; break; default: - str = luaL_checklstring(L, index, &len); - buffer = skynet_malloc(len); - memcpy(buffer, str, len); - *sz = (int)len; + buf->type = SOCKET_BUFFER_RAWPOINTER; + buf->buffer = luaL_checklstring(L, index, &buf->sz); break; } - return buffer; } static int lsend(lua_State *L) { struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); int id = luaL_checkinteger(L, 1); - int sz = 0; - void *buffer = get_buffer(L, 2, &sz); - int err = skynet_socket_send(ctx, id, buffer, sz); + struct socket_sendbuffer buf; + buf.id = id; + get_buffer(L, 2, &buf); + int err = skynet_socket_sendbuffer(ctx, &buf); lua_pushboolean(L, !err); return 1; } @@ -568,9 +586,10 @@ static int lsendlow(lua_State *L) { struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); int id = luaL_checkinteger(L, 1); - int sz = 0; - void *buffer = get_buffer(L, 2, &sz); - int err = skynet_socket_send_lowpriority(ctx, id, buffer, sz); + struct socket_sendbuffer buf; + buf.id = id; + get_buffer(L, 2, &buf); + int err = skynet_socket_sendbuffer_lowpriority(ctx, &buf); lua_pushboolean(L, !err); return 1; } @@ -645,9 +664,10 @@ ludp_send(lua_State *L) { struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); int id = luaL_checkinteger(L, 1); const char * address = luaL_checkstring(L, 2); - int sz = 0; - void *buffer = get_buffer(L, 3, &sz); - int err = skynet_socket_udp_send(ctx, id, address, buffer, sz); + struct socket_sendbuffer buf; + buf.id = id; + get_buffer(L, 3, &buf); + int err = skynet_socket_udp_sendbuffer(ctx, address, &buf); lua_pushboolean(L, !err); diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index acd82235..4e337344 100644 --- a/service-src/service_harbor.c +++ b/service-src/service_harbor.c @@ -332,13 +332,19 @@ send_remote(struct skynet_context * ctx, int fd, const char * buffer, size_t sz, skynet_error(ctx, "remote message from :%08x to :%08x is too large.", cookie->source, cookie->destination); return; } - uint8_t * sendbuf = skynet_malloc(sz_header+4); + uint8_t sendbuf[sz_header+4]; to_bigendian(sendbuf, (uint32_t)sz_header); memcpy(sendbuf+4, buffer, sz); header_to_message(cookie, sendbuf+4+sz); + struct socket_sendbuffer tmp; + tmp.id = fd; + tmp.type = SOCKET_BUFFER_RAWPOINTER; + tmp.buffer = sendbuf; + tmp.sz = sz_header+4; + // ignore send error, because if the connection is broken, the mainloop will recv a message. - skynet_socket_send(ctx, fd, sendbuf, sz_header+4); + skynet_socket_sendbuffer(ctx, &tmp); } static void @@ -580,9 +586,13 @@ remote_send_name(struct harbor *h, uint32_t source, const char name[GLOBALNAME_L static void handshake(struct harbor *h, int id) { struct slave *s = &h->s[id]; - uint8_t * handshake = skynet_malloc(1); - handshake[0] = (uint8_t)h->id; - skynet_socket_send(h->ctx, s->fd, handshake, 1); + uint8_t handshake[1] = { (uint8_t)h->id }; + struct socket_sendbuffer tmp; + tmp.id = s->fd; + tmp.type = SOCKET_BUFFER_RAWPOINTER; + tmp.buffer = handshake; + tmp.sz = 1; + skynet_socket_sendbuffer(h->ctx, &tmp); } static void diff --git a/skynet-src/skynet_socket.c b/skynet-src/skynet_socket.c index 1b56024c..25044b6a 100644 --- a/skynet-src/skynet_socket.c +++ b/skynet-src/skynet_socket.c @@ -117,13 +117,13 @@ skynet_socket_poll() { } int -skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz) { - return socket_server_send(SOCKET_SERVER, id, buffer, sz); +skynet_socket_sendbuffer(struct skynet_context *ctx, struct socket_sendbuffer *buffer) { + return socket_server_send(SOCKET_SERVER, buffer); } int -skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz) { - return socket_server_send_lowpriority(SOCKET_SERVER, id, buffer, sz); +skynet_socket_sendbuffer_lowpriority(struct skynet_context *ctx, struct socket_sendbuffer *buffer) { + return socket_server_send_lowpriority(SOCKET_SERVER, buffer); } int @@ -179,8 +179,8 @@ skynet_socket_udp_connect(struct skynet_context *ctx, int id, const char * addr, } int -skynet_socket_udp_send(struct skynet_context *ctx, int id, const char * address, const void *buffer, int sz) { - return socket_server_udp_send(SOCKET_SERVER, id, (const struct socket_udp_address *)address, buffer, sz); +skynet_socket_udp_sendbuffer(struct skynet_context *ctx, const char * address, struct socket_sendbuffer *buffer) { + return socket_server_udp_send(SOCKET_SERVER, (const struct socket_udp_address *)address, buffer); } const char * diff --git a/skynet-src/skynet_socket.h b/skynet-src/skynet_socket.h index 2fd38eb3..ff7e23fb 100644 --- a/skynet-src/skynet_socket.h +++ b/skynet-src/skynet_socket.h @@ -2,6 +2,7 @@ #define skynet_socket_h #include "socket_info.h" +#include "socket_buffer.h" struct skynet_context; @@ -26,8 +27,8 @@ void skynet_socket_free(); int skynet_socket_poll(); void skynet_socket_updatetime(); -int skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz); -int skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz); +int skynet_socket_sendbuffer(struct skynet_context *ctx, struct socket_sendbuffer *buffer); +int skynet_socket_sendbuffer_lowpriority(struct skynet_context *ctx, struct socket_sendbuffer *buffer); int skynet_socket_listen(struct skynet_context *ctx, const char *host, int port, int backlog); int skynet_socket_connect(struct skynet_context *ctx, const char *host, int port); int skynet_socket_bind(struct skynet_context *ctx, int fd); @@ -38,9 +39,40 @@ void skynet_socket_nodelay(struct skynet_context *ctx, int id); int skynet_socket_udp(struct skynet_context *ctx, const char * addr, int port); int skynet_socket_udp_connect(struct skynet_context *ctx, int id, const char * addr, int port); -int skynet_socket_udp_send(struct skynet_context *ctx, int id, const char * address, const void *buffer, int sz); +int skynet_socket_udp_sendbuffer(struct skynet_context *ctx, const char * address, struct socket_sendbuffer *buffer); const char * skynet_socket_udp_address(struct skynet_socket_message *, int *addrsz); struct socket_info * skynet_socket_info(); +// legacy APIs + +static inline void sendbuffer_init_(struct socket_sendbuffer *buf, int id, const void *buffer, int sz) { + buf->id = id; + buf->buffer = buffer; + if (sz < 0) { + buf->type = SOCKET_BUFFER_OBJECT; + } else { + buf->type = SOCKET_BUFFER_MEMORY; + } + buf->sz = (size_t)sz; +} + +static inline int skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz) { + struct socket_sendbuffer tmp; + sendbuffer_init_(&tmp, id, buffer, sz); + return skynet_socket_sendbuffer(ctx, &tmp); +} + +static inline int skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz) { + struct socket_sendbuffer tmp; + sendbuffer_init_(&tmp, id, buffer, sz); + return skynet_socket_sendbuffer_lowpriority(ctx, &tmp); +} + +static inline int skynet_socket_udp_send(struct skynet_context *ctx, int id, const char * address, const void *buffer, int sz) { + struct socket_sendbuffer tmp; + sendbuffer_init_(&tmp, id, buffer, sz); + return skynet_socket_udp_sendbuffer(ctx, address, &tmp); +} + #endif diff --git a/skynet-src/socket_buffer.h b/skynet-src/socket_buffer.h new file mode 100644 index 00000000..dbe81356 --- /dev/null +++ b/skynet-src/socket_buffer.h @@ -0,0 +1,17 @@ +#ifndef socket_buffer_h +#define socket_buffer_h + +#include + +#define SOCKET_BUFFER_MEMORY 0 +#define SOCKET_BUFFER_OBJECT 1 +#define SOCKET_BUFFER_RAWPOINTER 2 + +struct socket_sendbuffer { + int id; + int type; + const void *buffer; + size_t sz; +}; + +#endif diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c index ddee2943..1ebaa0bd 100644 --- a/skynet-src/socket_server.c +++ b/skynet-src/socket_server.c @@ -58,11 +58,13 @@ #define WARNING_SIZE (1024*1024) +#define USEROBJECT ((size_t)(~0)) + struct write_buffer { struct write_buffer * next; - void *buffer; + const void *buffer; char *ptr; - int sz; + size_t sz; bool userobject; uint8_t udp_address[UDP_ADDRESS_SIZE]; }; @@ -131,8 +133,8 @@ struct request_open { struct request_send { int id; - int sz; - char * buffer; + size_t sz; + const void * buffer; }; struct request_send_udp { @@ -225,8 +227,8 @@ union sockaddr_all { }; struct send_object { - void * buffer; - int sz; + const void * buffer; + size_t sz; void (*free_func)(void *); }; @@ -272,8 +274,8 @@ socket_unlock(struct socket_lock *sl) { } static inline bool -send_object_init(struct socket_server *ss, struct send_object *so, void *object, int sz) { - if (sz < 0) { +send_object_init(struct socket_server *ss, struct send_object *so, const void *object, size_t sz) { + if (sz == USEROBJECT) { so->buffer = ss->soi.buffer(object); so->sz = ss->soi.size(object); so->free_func = ss->soi.free; @@ -286,12 +288,40 @@ send_object_init(struct socket_server *ss, struct send_object *so, void *object, } } +static void +dummy_free(void *ptr) { + (void)ptr; +} + +static inline void +send_object_init_from_sendbuffer(struct socket_server *ss, struct send_object *so, struct socket_sendbuffer *buf) { + switch (buf->type) { + case SOCKET_BUFFER_MEMORY: + send_object_init(ss, so, (void *)buf->buffer, buf->sz); + break; + case SOCKET_BUFFER_OBJECT: + send_object_init(ss, so, (void *)buf->buffer, USEROBJECT); + break; + case SOCKET_BUFFER_RAWPOINTER: + so->buffer = (void *)buf->buffer; + so->sz = buf->sz; + so->free_func = dummy_free; + break; + default: + // never get here + so->buffer = NULL; + so->sz = 0; + so->free_func = NULL; + break; + } +} + static inline void write_buffer_free(struct socket_server *ss, struct write_buffer *wb) { if (wb->userobject) { - ss->soi.free(wb->buffer); + ss->soi.free((void *)wb->buffer); } else { - FREE(wb->buffer); + FREE((void *)wb->buffer); } FREE(wb); } @@ -400,10 +430,39 @@ free_wb_list(struct socket_server *ss, struct wb_list *list) { } static void -free_buffer(struct socket_server *ss, const void * buffer, int sz) { - struct send_object so; - send_object_init(ss, &so, (void *)buffer, sz); - so.free_func((void *)buffer); +free_buffer(struct socket_server *ss, struct socket_sendbuffer *buf) { + void *buffer = (void *)buf->buffer; + switch (buf->type) { + case SOCKET_BUFFER_MEMORY: + FREE((void *)buffer); + break; + case SOCKET_BUFFER_OBJECT: + ss->soi.free(buffer); + break; + case SOCKET_BUFFER_RAWPOINTER: + break; + } +} + +static const void * +clone_buffer(struct socket_sendbuffer *buf, size_t *sz) { + switch (buf->type) { + case SOCKET_BUFFER_MEMORY: + *sz = buf->sz; + return buf->buffer; + case SOCKET_BUFFER_OBJECT: + *sz = USEROBJECT; + return buf->buffer; + case SOCKET_BUFFER_RAWPOINTER: + // It's a raw pointer, we need make a copy + *sz = buf->sz; + void * tmp = MALLOC(*sz); + memcpy(tmp, buf->buffer, *sz); + return tmp; + } + // never get here + *sz = 0; + return NULL; } static void @@ -429,7 +488,12 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_lock *l, s } s->type = SOCKET_TYPE_INVALID; if (s->dw_buffer) { - free_buffer(ss, s->dw_buffer, s->dw_size); + struct socket_sendbuffer tmp; + tmp.buffer = s->dw_buffer; + tmp.sz = s->dw_size; + tmp.id = s->id; + tmp.type = (tmp.sz == USEROBJECT) ? SOCKET_BUFFER_OBJECT : SOCKET_BUFFER_MEMORY; + free_buffer(ss, &tmp); s->dw_buffer = NULL; } socket_unlock(l); @@ -849,12 +913,12 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock if (s->type == SOCKET_TYPE_INVALID || s->id != id || s->type == SOCKET_TYPE_HALFCLOSE || s->type == SOCKET_TYPE_PACCEPT) { - so.free_func(request->buffer); + so.free_func((void *)request->buffer); return -1; } if (s->type == SOCKET_TYPE_PLISTEN || s->type == SOCKET_TYPE_LISTEN) { fprintf(stderr, "socket-server: write to listen fd %d.\n", id); - so.free_func(request->buffer); + so.free_func((void *)request->buffer); return -1; } if (send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED) { @@ -870,7 +934,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock if (sasz == 0) { // udp type mismatch, just drop it. fprintf(stderr, "socket-server: udp socket (%d) type mistach.\n", id); - so.free_func(request->buffer); + so.free_func((void *)request->buffer); return -1; } int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz); @@ -878,7 +942,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock append_sendbuffer_udp(ss,s,priority,request,udp_address); } else { stat_write(ss,s,n); - so.free_func(request->buffer); + so.free_func((void *)request->buffer); return -1; } } @@ -1572,10 +1636,11 @@ can_direct_write(struct socket *s, int id) { // return -1 when error, 0 when success int -socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz) { +socket_server_send(struct socket_server *ss, struct socket_sendbuffer *buf) { + int id = buf->id; struct socket * s = &ss->slot[HASH_ID(id)]; if (s->id != id || s->type == SOCKET_TYPE_INVALID) { - free_buffer(ss, buffer, sz); + free_buffer(ss, buf); return -1; } @@ -1587,7 +1652,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz if (can_direct_write(s,id)) { // send directly struct send_object so; - send_object_init(ss, &so, (void *)buffer, sz); + send_object_init_from_sendbuffer(ss, &so, buf); ssize_t n; if (s->protocol == PROTOCOL_TCP) { n = write(s->fd, so.buffer, so.sz); @@ -1597,7 +1662,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz if (sasz == 0) { fprintf(stderr, "socket-server : set udp (%d) address first.\n", id); socket_unlock(&l); - so.free_func((void *)buffer); + so.free_func((void *)buf->buffer); return -1; } n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz); @@ -1610,12 +1675,11 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz if (n == so.sz) { // write done socket_unlock(&l); - so.free_func((void *)buffer); + so.free_func((void *)buf->buffer); return 0; } // write failed, put buffer into s->dw_* , and let socket thread send it. see send_buffer() - s->dw_buffer = buffer; - s->dw_size = sz; + s->dw_buffer = clone_buffer(buf, &s->dw_size); s->dw_offset = n; sp_write(ss->event_fd, s->fd, s, true); @@ -1630,8 +1694,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz struct request_package request; request.u.send.id = id; - request.u.send.sz = sz; - request.u.send.buffer = (char *)buffer; + request.u.send.buffer = clone_buffer(buf, &request.u.send.sz); send_request(ss, &request, 'D', sizeof(request.u.send)); return 0; @@ -1639,10 +1702,12 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz // return -1 when error, 0 when success int -socket_server_send_lowpriority(struct socket_server *ss, int id, const void * buffer, int sz) { +socket_server_send_lowpriority(struct socket_server *ss, struct socket_sendbuffer *buf) { + int id = buf->id; + struct socket * s = &ss->slot[HASH_ID(id)]; if (s->id != id || s->type == SOCKET_TYPE_INVALID) { - free_buffer(ss, buffer, sz); + free_buffer(ss, buf); return -1; } @@ -1650,8 +1715,7 @@ socket_server_send_lowpriority(struct socket_server *ss, int id, const void * bu struct request_package request; request.u.send.id = id; - request.u.send.sz = sz; - request.u.send.buffer = (char *)buffer; + request.u.send.buffer = clone_buffer(buf, &request.u.send.sz); send_request(ss, &request, 'P', sizeof(request.u.send)); return 0; @@ -1836,10 +1900,11 @@ socket_server_udp(struct socket_server *ss, uintptr_t opaque, const char * addr, } int -socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp_address *addr, const void *buffer, int sz) { +socket_server_udp_send(struct socket_server *ss, const struct socket_udp_address *addr, struct socket_sendbuffer *buf) { + int id = buf->id; struct socket * s = &ss->slot[HASH_ID(id)]; if (s->id != id || s->type == SOCKET_TYPE_INVALID) { - free_buffer(ss, buffer, sz); + free_buffer(ss, buf); return -1; } @@ -1853,7 +1918,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp addrsz = 1+2+16; // 1 type, 2 port, 16 ipv6 break; default: - free_buffer(ss, buffer, sz); + free_buffer(ss, buf); return -1; } @@ -1865,12 +1930,12 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp if (can_direct_write(s,id)) { // send directly struct send_object so; - send_object_init(ss, &so, (void *)buffer, sz); + send_object_init_from_sendbuffer(ss, &so, buf); union sockaddr_all sa; socklen_t sasz = udp_socket_address(s, udp_address, &sa); if (sasz == 0) { socket_unlock(&l); - so.free_func((void *)buffer); + so.free_func((void *)buf->buffer); return -1; } int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz); @@ -1878,7 +1943,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp // sendto succ stat_write(ss,s,n); socket_unlock(&l); - so.free_func((void *)buffer); + so.free_func((void *)buf->buffer); return 0; } } @@ -1888,8 +1953,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp struct request_package request; request.u.send_udp.send.id = id; - request.u.send_udp.send.sz = sz; - request.u.send_udp.send.buffer = (char *)buffer; + request.u.send_udp.send.buffer = clone_buffer(buf, &request.u.send_udp.send.sz); memcpy(request.u.send_udp.address, udp_address, addrsz); diff --git a/skynet-src/socket_server.h b/skynet-src/socket_server.h index 96568502..cbfe9889 100644 --- a/skynet-src/socket_server.h +++ b/skynet-src/socket_server.h @@ -3,6 +3,7 @@ #include #include "socket_info.h" +#include "socket_buffer.h" #define SOCKET_DATA 0 #define SOCKET_CLOSE 1 @@ -33,8 +34,8 @@ void socket_server_shutdown(struct socket_server *, uintptr_t opaque, int id); void socket_server_start(struct socket_server *, uintptr_t opaque, int id); // return -1 when error -int socket_server_send(struct socket_server *, int id, const void * buffer, int sz); -int socket_server_send_lowpriority(struct socket_server *, int id, const void * buffer, int sz); +int socket_server_send(struct socket_server *, struct socket_sendbuffer *buffer); +int socket_server_send_lowpriority(struct socket_server *, struct socket_sendbuffer *buffer); // ctrl command below returns id int socket_server_listen(struct socket_server *, uintptr_t opaque, const char * addr, int port, int backlog); @@ -53,17 +54,17 @@ int socket_server_udp(struct socket_server *, uintptr_t opaque, const char * add int socket_server_udp_connect(struct socket_server *, int id, const char * addr, int port); // If the socket_udp_address is NULL, use last call socket_server_udp_connect address instead // You can also use socket_server_send -int socket_server_udp_send(struct socket_server *, int id, const struct socket_udp_address *, const void *buffer, int sz); +int socket_server_udp_send(struct socket_server *, const struct socket_udp_address *, struct socket_sendbuffer *buffer); // extract the address of the message, struct socket_message * should be SOCKET_UDP const struct socket_udp_address * socket_server_udp_address(struct socket_server *, struct socket_message *, int *addrsz); struct socket_object_interface { - void * (*buffer)(void *); - int (*size)(void *); + const void * (*buffer)(const void *); + size_t (*size)(const void *); void (*free)(void *); }; -// if you send package sz == -1, use soi. +// if you send package with type SOCKET_BUFFER_OBJECT, use soi. void socket_server_userobject(struct socket_server *, struct socket_object_interface *soi); struct socket_info * socket_server_info(struct socket_server *);