diff --git a/Makefile b/Makefile index c694a03d..b0c5f84c 100644 --- a/Makefile +++ b/Makefile @@ -28,7 +28,7 @@ all : \ service/localcast.so \ service/socket.so \ luaclib/skynet.so \ - luaclib/socketbuffer.so \ + luaclib/socketdriver.so \ luaclib/int64.so \ luaclib/mcast.so \ luaclib/bson.so \ @@ -90,8 +90,8 @@ service/client.so : service-src/service_client.c service/socket.so : service-src/service_socket.c gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src -luaclib/socketbuffer.so : lualib-src/lua-socket.c | luaclib - gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src +luaclib/socketdriver.so : lualib-src/lua-socket.c | luaclib + gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src -Iservice-src luaclib/int64.so : lua-int64/int64.c | luaclib gcc $(CFLAGS) $(SHARED) -Iluacompat -O2 $^ -o $@ diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index 08fc7a1a..ab4ec6b8 100644 --- a/lualib-src/lua-socket.c +++ b/lualib-src/lua-socket.c @@ -1,250 +1,465 @@ #include #include +#include +#include +#include #include #include #include "luacompat52.h" +#include "skynet_socket.h" +#include "service_lua.h" + +#define BACKLOG 32 +// 2 ** 12 == 4096 +#define LARGE_PAGE_NODE 12 + +struct buffer_node { + char * msg; + int sz; + struct buffer_node *next; +}; struct socket_buffer { int size; - int head; - int tail; + int offset; + struct buffer_node *head; + struct buffer_node *tail; }; -static struct socket_buffer * -new_buffer(lua_State *L, int sz) { - struct socket_buffer * buffer = lua_newuserdata(L, sizeof(*buffer) + sz); - buffer->size = sz; - buffer->head = 0; - buffer->tail = 0; - - return buffer; -} - -static void -copy_buffer(struct socket_buffer * dest, struct socket_buffer * src) { - char * ptr = (char *)(src+1); - if (src->tail >= src->head) { - int sz = src->tail - src->head; - memcpy(dest+1, ptr+ src->head, sz); - dest->tail = sz; - } else { - char * d = (char *)(dest+1); - int part = src->size - src->head; - memcpy(d, ptr + src->head, part); - memcpy(d + part, ptr , src->tail); - dest->tail = src->tail + part; - } -} - -static void -append_buffer(struct socket_buffer * buffer, char * msg, int sz) { - char * dst = (char *)(buffer + 1); - if (sz + buffer->tail < buffer->size) { - memcpy(dst + buffer->tail , msg, sz); - buffer->tail += sz; - } else { - int part = buffer->size - buffer->tail; - memcpy(dst + buffer->tail , msg, part); - memcpy(dst, msg + part, sz - part); - buffer->tail = sz - part; - } -} - static int -lpop(lua_State *L) { - struct socket_buffer * buffer = lua_touserdata(L, 1); - if (buffer == NULL) { - return 0; - } - int sz = luaL_checkinteger(L, 2); - int bytes = buffer->tail - buffer->head; - if (bytes < 0) { - bytes += buffer->size; - } - - if (sz > bytes || bytes == 0) { - lua_pushnil(L); - lua_pushinteger(L, bytes); - return 2; - } - - if (sz == 0) { - sz = bytes; - } - - char * ptr = (char *)(buffer+1); - if (buffer->size - buffer->head >=sz) { - lua_pushlstring(L, ptr + buffer->head, sz); - buffer->head+=sz; - } else { - luaL_Buffer b; - luaL_buffinit(L, &b); - luaL_addlstring(&b, ptr + buffer->head, buffer->size - buffer->head); - buffer->head = sz - (buffer->size - buffer->head); - luaL_addlstring(&b, ptr, buffer->head); - luaL_pushresult(&b); - } - - bytes -= sz; - - lua_pushinteger(L,bytes); - if (bytes == 0) { - buffer->head = buffer->tail = 0; - } - - return 2; -} - -static inline int -check_sep(struct socket_buffer *buffer, int from, const char * sep, int sz) { - const char * ptr = (const char *)(buffer+1); +lfreepool(lua_State *L) { + struct buffer_node * pool = lua_touserdata(L, 1); + int sz = lua_rawlen(L,1) / sizeof(*pool); int i; for (i=0;i= buffer->size) { - index = 0; - } - if (ptr[index] != sep[i]) { - return 0; - } - } - return 1; -} - -static int -lreadline(lua_State *L) { - struct socket_buffer * buffer = lua_touserdata(L, 1); - if (buffer == NULL) { - return 0; - } - size_t len = 0; - const char *sep = luaL_checklstring(L,2,&len); - int read = !lua_toboolean(L,3); - int bytes = buffer->tail - buffer->head; - - if (bytes < 0) { - bytes += buffer->size; - } - - int i; - for (i=0;i<=bytes-(int)len;i++) { - int index = buffer->head + i; - if (index >= buffer->size) { - index -= buffer->size; - } - if (check_sep(buffer, index, sep, (int)len)) { - if (read == 0) { - lua_pushboolean(L,1); - } else { - if (i==0) { - lua_pushlstring(L, "", 0); - } else { - const char * ptr = (const char *)(buffer+1); - if (--index < 0) { - index = buffer->size -1; - } - if (index < buffer->head) { - luaL_Buffer b; - luaL_buffinit(L, &b); - luaL_addlstring(&b, ptr + buffer->head, buffer->size-buffer->head); - luaL_addlstring(&b, ptr, index+1); - luaL_pushresult(&b); - } else { - lua_pushlstring(L, ptr + buffer->head, index-buffer->head+1); - } - ++index; - } - index+=len; - if (index >= buffer->size) { - index-=buffer->size; - } - buffer->head = index; - } - return 1; + struct buffer_node *node = &pool[i]; + if (node->msg) { + free(node->msg); + node->msg = NULL; } } return 0; } static int -lpush(lua_State *L) { - struct socket_buffer * buffer = lua_touserdata(L, 1); - int bytes = 0; - if (buffer) { - bytes = buffer->tail - buffer->head; - if (bytes < 0) { - bytes += buffer->size; +lnewpool(lua_State *L, int sz) { + struct buffer_node * pool = lua_newuserdata(L, sizeof(struct buffer_node) * sz); + int i; + for (i=0;isize = 0; + sb->offset = 0; + sb->head = NULL; + sb->tail = NULL; + + return 1; +} + +/* + userdata send_buffer + table pool + lightuserdata msg + int size + + return size + */ +static int +lpushbuffer(lua_State *L) { + struct socket_buffer *sb = lua_touserdata(L,1); + if (sb == NULL) { + return luaL_error(L, "need buffer object at param 1"); + } + char * msg = lua_touserdata(L,3); + if (msg == NULL) { + return luaL_error(L, "need message block at param 3"); + } + int pool_index = 2; + luaL_checktype(L,pool_index,LUA_TTABLE); + int sz = luaL_checkinteger(L,4); + lua_rawgeti(L,pool_index,1); + struct buffer_node * free_node = lua_touserdata(L,-1); // sb poolt msg size free_node + lua_pop(L,1); + if (free_node == NULL) { + int tsz = lua_rawlen(L,pool_index); + if (tsz == 0) + tsz++; + int size = 8; + if (tsz <= LARGE_PAGE_NODE-3) { + size <<= tsz; + } else { + size <<= LARGE_PAGE_NODE-3; + } + lnewpool(L, size); + free_node = lua_touserdata(L,-1); + lua_rawseti(L, pool_index, tsz+1); + } + lua_pushlightuserdata(L, free_node->next); + lua_rawseti(L, pool_index, 1); // sb poolt msg size + free_node->msg = msg; + free_node->sz = sz; + free_node->next = NULL; + + if (sb->head == NULL) { + assert(sb->tail == NULL); + sb->head = sb->tail = free_node; + } else { + sb->tail->next = free_node; + sb->tail = free_node; + } + sb->size += sz; + + lua_pushinteger(L, sb->size); + + return 1; +} + +static void +return_free_node(lua_State *L, int pool, struct socket_buffer *sb) { + struct buffer_node *free_node = sb->head; + sb->head = free_node->next; + if (sb->head == NULL) { + sb->tail = NULL; + } + lua_rawgeti(L,pool,1); + free_node->next = lua_touserdata(L,-1); + lua_pop(L,1); + free(free_node->msg); + free_node->msg = NULL; + + free_node->sz = 0; + lua_pushlightuserdata(L, free_node); + lua_rawseti(L, pool, 1); +} + +static void +pop_lstring(lua_State *L, struct socket_buffer *sb, int sz, int skip) { + struct buffer_node * current = sb->head; + if (sz < current->sz - sb->offset) { + lua_pushlstring(L, current->msg + sb->offset, sz-skip); + sb->offset+=sz; + return; + } + if (sz == current->sz - sb->offset) { + lua_pushlstring(L, current->msg + sb->offset, sz-skip); + sb->offset = 0; + return_free_node(L,2,sb); + return; + } + + luaL_Buffer b; + luaL_buffinit(L, &b); + for (;;) { + int bytes = current->sz - sb->offset; + if (bytes >= sz) { + if (sz > skip) { + luaL_addlstring(&b, current->msg + sb->offset, sz - skip); + } + sb->offset += sz; + if (bytes == sz) { + return_free_node(L,2,sb); + sb->offset = 0; + } + break; + } + luaL_addlstring(&b, current->msg + sb->offset, (sz - skip < bytes) ? sz - skip : bytes); + return_free_node(L,2,sb); + sb->offset = 0; + sz-=bytes; + if (sz==0) + break; + current = sb->head; + assert(current); + } + luaL_pushresult(&b); +} + +/* + userdata send_buffer + table pool + integer sz + */ +static int +lpopbuffer(lua_State *L) { + struct socket_buffer * sb = lua_touserdata(L, 1); + if (sb == NULL) { + return luaL_error(L, "Need buffer object at param 1"); + } + luaL_checktype(L,2,LUA_TTABLE); + int sz = luaL_checkinteger(L,3); + if (sb->size < sz || sz == 0) { + lua_pushnil(L); + } else { + pop_lstring(L,sb,sz,0); + sb->size -= sz; + } + lua_pushinteger(L, sb->size); + + return 2; +} + +/* + userdata send_buffer + table pool + */ +static int +lclearbuffer(lua_State *L) { + struct socket_buffer * sb = lua_touserdata(L, 1); + if (sb == NULL) { + return luaL_error(L, "Need buffer object at param 1"); + } + luaL_checktype(L,2,LUA_TTABLE); + while(sb->head) { + return_free_node(L,2,sb); + } + return 0; +} + +static int +lreadall(lua_State *L) { + struct socket_buffer * sb = lua_touserdata(L, 1); + if (sb == NULL) { + return luaL_error(L, "Need buffer object at param 1"); + } + luaL_checktype(L,2,LUA_TTABLE); + luaL_Buffer b; + luaL_buffinit(L, &b); + while(sb->head) { + struct buffer_node *current = sb->head; + luaL_addlstring(&b, current->msg + sb->offset, current->sz - sb->offset); + return_free_node(L,2,sb); + } + luaL_pushresult(&b); + return 1; +} + +static int +ldrop(lua_State *L) { + void * msg = lua_touserdata(L,1); + luaL_checkinteger(L,2); + free(msg); + return 0; +} + +static bool +check_sep(struct buffer_node * node, int from, const char *sep, int seplen) { + for (;;) { + int sz = node->sz - from; + if (sz >= seplen) { + return memcmp(node->msg+from,sep,seplen) == 0; + } + if (sz > 0) { + if (memcmp(node->msg + from, sep, sz)) { + return false; + } + } + node = node->next; + sep += sz; + seplen -= sz; + from = 0; + } +} + +/* + userdata send_buffer + table pool , nil for check + string sep + */ +static int +lreadline(lua_State *L) { + struct socket_buffer * sb = lua_touserdata(L, 1); + if (sb == NULL) { + return luaL_error(L, "Need buffer object at param 1"); + } + // only check + bool check = !lua_istable(L, 2); + size_t seplen = 0; + const char *sep = luaL_checklstring(L,3,&seplen); + int i; + struct buffer_node *current = sb->head; + if (current == NULL) + return 0; + int from = sb->offset; + int bytes = current->sz - from; + for (i=0;i<=sb->size - (int)seplen;i++) { + if (check_sep(current, from, sep, seplen)) { + if (check) { + lua_pushboolean(L,true); + } else { + pop_lstring(L, sb, i+seplen, seplen); + sb->size -= i+seplen; + } + return 1; + } + ++from; + --bytes; + if (bytes == 0) { + current = current->next; + from = 0; + if (current == NULL) + break; + bytes = current->sz; } } - void * msg = lua_touserdata(L,2); - if (msg == NULL) { - lua_settop(L,1); - lua_pushinteger(L,bytes); - return 2; - } + return 0; +} - int sz = luaL_checkinteger(L,3); - - if (buffer == NULL) { - struct socket_buffer * nbuf = new_buffer(L, sz * 2); - append_buffer(nbuf, msg, sz); - } else if (sz + bytes >= buffer->size) { - struct socket_buffer * nbuf = new_buffer(L, (sz + bytes) * 2); - copy_buffer(nbuf, buffer); - append_buffer(nbuf, msg, sz); - } else { - lua_settop(L,1); - append_buffer(buffer, msg, sz); - } - lua_pushinteger(L, sz + bytes); +static int +lstr2p(lua_State *L) { + size_t sz = 0; + const char * str = luaL_checklstring(L,1,&sz); + void *ptr = malloc(sz); + memcpy(ptr, str, sz); + lua_pushlightuserdata(L, ptr); + lua_pushinteger(L, (int)sz); return 2; } +// for skynet socket + +/* + lightuserdata msg + integer size + + return type n1 n2 ptr_or_string +*/ static int lunpack(lua_State *L) { - int * msg = lua_touserdata(L,1); - int sz = luaL_checkinteger(L,2); - if (msg == NULL || sz < 4) { - return luaL_error(L, "Invalid socket message"); + struct skynet_socket_message *message = lua_touserdata(L,1); + int size = luaL_checkinteger(L,2); + + lua_pushinteger(L, message->type); + lua_pushinteger(L, message->id); + lua_pushinteger(L, message->ud); + if (message->buffer == NULL) { + lua_pushlstring(L, (char *)(message+1),size - sizeof(*message)); + } else { + lua_pushlightuserdata(L, message->buffer); } - lua_pushinteger(L, *msg); - lua_pushlightuserdata(L, msg+1); - lua_pushinteger(L, sz - 4); - return 3; + return 4; } static int -lpack(lua_State *L) { - int fd = luaL_checkinteger(L, 1); - const void *buffer; - size_t sz; +lconnect(lua_State *L) { + const char * host = luaL_checkstring(L,1); + int port = luaL_checkinteger(L,2); + struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); + int id = skynet_socket_connect(ctx, host, port); + lua_pushinteger(L, id); + + return 1; +} + +static int +lclose(lua_State *L) { + int id = luaL_checkinteger(L,1); + struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); + skynet_socket_close(ctx, id); + return 0; +} + +static int +llisten(lua_State *L) { + const char * host = luaL_checkstring(L,1); + int port = luaL_checkinteger(L,2); + int backlog = luaL_optinteger(L,3,BACKLOG); + struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); + int id = skynet_socket_listen(ctx, host,port,backlog); + + lua_pushinteger(L,id); + return 1; +} + +static int +lsend(lua_State *L) { + struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); + int id = luaL_checkinteger(L, 1); + void *buffer; + int sz; if (lua_isuserdata(L,2)) { buffer = lua_touserdata(L,2); - sz = luaL_checkinteger(L, 3); + sz = luaL_checkinteger(L,3); } else { - buffer = luaL_checklstring(L,2,&sz); + size_t len = 0; + const char * str = luaL_checklstring(L, 2, &len); + buffer = malloc(len); + memcpy(buffer, str, len); + sz = (int)len; } - int * b = malloc(4 + sz); - *b = fd; - memcpy(b+1, buffer, sz); - lua_pushlightuserdata(L, b); - lua_pushinteger(L, 4+sz); - return 2; + int err = skynet_socket_send(ctx, id, buffer, sz); + lua_pushboolean(L, err); + return 1; +} + +static int +lbind(lua_State *L) { + struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); + int fd = luaL_checkinteger(L, 1); + int id = skynet_socket_bind(ctx,fd); + lua_pushinteger(L,id); + return 1; +} + +static int +laccept(lua_State *L) { + struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1)); + int id = luaL_checkinteger(L, 1); + skynet_socket_accept(ctx,id); + return 0; } int -luaopen_socketbuffer(lua_State *L) { +luaopen_socketdriver(lua_State *L) { luaL_checkversion(L); luaL_Reg l[] = { - { "push", lpush }, - { "pop", lpop }, + { "buffer", lnewbuffer }, + { "push", lpushbuffer }, + { "pop", lpopbuffer }, + { "drop", ldrop }, + { "readall", lreadall }, + { "clear", lclearbuffer }, { "readline", lreadline }, + { "str2p", lstr2p }, + { "unpack", lunpack }, - { "pack", lpack }, { NULL, NULL }, }; luaL_newlib(L,l); + luaL_Reg l2[] = { + { "connect", lconnect }, + { "close", lclose }, + { "listen", llisten }, + { "send", lsend }, + { "bind", lbind }, + { "accept", laccept }, + { NULL, NULL }, + }; + lua_getfield(L, LUA_REGISTRYINDEX, "skynet_lua"); + struct snlua *lua = lua_touserdata(L,-1); + if (lua == NULL || lua->ctx == NULL) { + return luaL_error(L, "Init skynet context first"); + } + assert(lua->L == L); + lua_pop(L,1); + + lua_pushlightuserdata(L, lua->ctx); + luaL_setfuncs(L,l2,1); + return 1; } diff --git a/lualib/socket.lua b/lualib/socket.lua index 65065bee..8937dd85 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -1,273 +1,240 @@ -local buffer = require "socketbuffer" +local driver = require "socketdriver" local skynet = require "skynet" -local table = table -local next = next local assert = assert -local coroutine = coroutine -local type = type -local READBUF = {} -- fd:buffer -local READREQUEST = {} -- fd:request_size -local READSESSION = {} -- fd:session -local READLOCK = {} -- fd:queue(session) -local READTHREAD= {} -- fd:thread -local CLOSED = {} -- fd:true +local socket = {} -- api +local buffer_pool = {} -- store all message buffer object +local socket_pool = setmetatable( -- store all socket object + {}, + { __gc = function(p) + for id,v in pairs(p) do + driver.close(id) + -- don't need clear v.buffer, because buffer pool will be free at the end + p[id] = nil + end + end + } +) -local selfaddr = skynet.self() -local sockets = assert(skynet.localname ".socket") +local socket_message = {} -local function response(session) - skynet.redirect(selfaddr , 0, "response", session, "") +local function wakeup(s) + local co = s.co + if co then + s.co = nil + skynet.wakeup(co) + end +end + +local function suspend(s) + assert(not s.co) + s.co = coroutine.running() + skynet.wait() +end + +-- read skynet_socket.h for these macro +-- SKYNET_SOCKET_TYPE_DATA = 1 +socket_message[1] = function(id, size, data) + local s = socket_pool[id] + if s == nil then + print("socket: drop package from " .. id) + driver.drop(data) + return + end + + local sz = driver.push(s.buffer, buffer_pool, data, size) + local rr = s.read_required + local rrt = type(rr) + if rrt == "number" then + -- read size + if sz >= rr then + s.read_required = nil + wakeup(s) + end + elseif rrt == "string" then + -- read line + if driver.readline(s.buffer,nil,rr) then + s.read_required = nil + wakeup(s) + end + end +end + +-- SKYNET_SOCKET_TYPE_CONNECT = 2 +socket_message[2] = function(id, _ , addr) + local s = socket_pool[id] + if s == nil then + return + end + -- log remote addr + s.connected = true + wakeup(s) +end + +-- SKYNET_SOCKET_TYPE_CLOSE = 3 +socket_message[3] = function(id) + local s = socket_pool[id] + if s == nil then + return + end + s.connected = false + wakeup(s) +end + +-- SKYNET_SOCKET_TYPE_ACCEPT = 4 +socket_message[4] = function(id, newid, addr) + local s = socket_pool[id] + if s == nil then + driver.close(newid) + return + end + skynet.fork(s.callback, newid, addr) +end + +-- SKYNET_SOCKET_TYPE_ERROR = 5 +socket_message[5] = function(id) + print("error on ", id) + local s = socket_pool[id] + if s == nil then + return + end + s.connected = false + wakeup(s) end skynet.register_protocol { - name = "client", - id = 3, -- PTYPE_CLIENT - pack = buffer.pack, - unpack = buffer.unpack, - dispatch = function (_, _, fd, msg, sz) - local qsz = READREQUEST[fd] - local buf = READBUF[fd] - local bsz - if sz == 0 then - CLOSED[fd] = true - else - buf,bsz = buffer.push(buf, msg, sz) - READBUF[fd] = buf - end - local session = READSESSION[fd] - if qsz == nil or session == nil then - return - end - if sz > 0 then - if type(qsz) == "number" then - if qsz > bsz then - return - end - else - -- request readline - if buffer.readline(buf, qsz, true) == nil then - return - end - end - end - - response(session) - READSESSION[fd] = nil + name = "socket", + id = 6, -- PTYPE_SOCKET + unpack = driver.unpack, + dispatch = function (_, _, t, n1, n2, data) + socket_message[t](n1,n2,data) end } -skynet.register_protocol { - name = "system", - id = 4, -- PTYPE_SYSTEM - pack = skynet.pack, - unpack = function (...) return ... end, - dispatch = function (session, addr, msg, sz) - fd, t, sz = skynet.unpack(msg,sz) - assert(addr == selfaddr, "PTYPE_SYSTEM message must send by self") - if t > 0 then -- lock request when t == 0 - -- request bytes or readline - local buf = READBUF[fd] - if CLOSED[fd] then - skynet.ret() - return - end - local _,bsz = buffer.push(buf) - if t == 1 then - -- sz is size - if bsz >= sz then - skynet.ret() - return - end - elseif t == 2 then - -- sz is sep - if buffer.readline(buf, sz, true) then -- don't real read - skynet.ret() - return - end - end - READSESSION[fd] = session - end +local function connect(id) + local s = { + id = id, + buffer = driver.buffer(), + connected = false, + read_require = false, + co = false, + } + socket_pool[id] = s + suspend(s) + if s.connected then + return id end -} - -local socket = {} +end function socket.open(addr, port) - local cmd = "open" .. " " .. (port and (addr..":"..port) or addr) - local r = skynet.call(sockets, "text", cmd) - if r == "" then - return nil, cmd .. " failed" - end - local fd = tonumber(r) - READBUF[fd] = true - CLOSED[fd] = nil - return fd -end - -function socket.bind(sock) - local r = skynet.call(sockets, "text", "bind " .. tonumber(sock)) - if r == "" then - error("stdin bind failed") - end - local fd = tonumber(r) - READBUF[fd] = true - CLOSED[fd] = nil - return fd + local id = driver.connect(addr,port) + return connect(id) end function socket.stdin() - return socket.bind(1) + local id = driver.bind(1) + return connect(id) +end + +function socket.accept(id) + driver.accept(id) + return connect(id) end function socket.close(fd) - socket.lock(fd) - skynet.call(sockets, "text", "close", fd) - READBUF[fd] = nil - READLOCK[fd] = nil - CLOSED[fd] = nil +-- socket.lock(fd) + local s = socket_pool[id] + if s == nil then + return + end + if s.connected then + driver.close(s.id) + suspend(s) + end + if s.buffer then + driver.clear(s.buffer) + end + socket_pool[id] = nil end -function socket.read(fd, sz) - local buf = assert(READBUF[fd]) - local str, bytes = buffer.pop(buf,sz) - if str then - return str - end - - if CLOSED[fd] then - READBUF[fd] = nil - CLOSED[fd] = nil - str = buffer.pop(buf, bytes) - return nil, str - end - - READREQUEST[fd] = sz - skynet.call(selfaddr, "system",fd,1,sz) -- singal size 1 - READREQUEST[fd] = nil - - buf = READBUF[fd] - - str, bytes = buffer.pop(buf,sz) - if str then - return str - end - - if CLOSED[fd] then - READBUF[fd] = nil - CLOSED[fd] = nil - str = buffer.pop(buf, bytes) - return nil, str - end -end - -function socket.readall(fd) - local buf = assert(READBUF[fd]) - if CLOSED[fd] then - CLOSED[fd] = nil - READBUF[fd] = nil - if buf == nil then - return "" - end - local _, bytes = buffer.push(buf) - local ret = buffer.pop(buf, bytes) +function socket.read(id, sz) + local s = socket_pool[id] + assert(s) + local ret = driver.pop(s.buffer, buffer_pool, sz) + if ret then return ret end - READREQUEST[fd] = math.huge - skynet.call(selfaddr, "system",fd,3) -- singal readall - READREQUEST[fd] = nil - assert(CLOSED[fd]) - buf = READBUF[fd] - READBUF[fd] = nil - CLOSED[fd] = nil - if buf == nil then - return "" - end - local _, bytes = buffer.push(buf) - local ret = buffer.pop(buf, bytes) - return ret -end - -function socket.readline(fd, sep) - local buf = assert(READBUF[fd]) - local str = buffer.readline(buf,sep) - if str then - return str + if not s.connected then + return false, driver.readall(s.buffer, buffer_pool) end - if CLOSED[fd] then - READBUF[fd] = nil - CLOSED[fd] = nil - local _, bytes = buffer.push(buf) - str = buffer.pop(buf, bytes) - return nil, str - end - - READREQUEST[fd] = sep - skynet.call(selfaddr, "system",fd,2,sep) -- singal sep 2 - READREQUEST[fd] = nil - - buf = READBUF[fd] - str = buffer.readline(buf,sep) - if str then - return str - end - - if CLOSED[fd] then - READBUF[fd] = nil - CLOSED[fd] = nil - local _, bytes = buffer.push(buf) - str = buffer.pop(buf, bytes) - return nil, str - end -end - -function socket.write(fd, msg, sz) - if CLOSED[fd] or not READBUF[fd] then - return - end - skynet.send(sockets, "client", fd, msg, sz) - return true -end - -function socket.invalid(fd) - return CLOSED[fd] or not READBUF[fd] -end - -function socket.lock(fd) - if CLOSED[fd] or not READBUF[fd] then - return - end - local locked = READTHREAD[fd] - if locked then - -- lock fd - local session = skynet.genid() - local q = READLOCK[fd] - if q == nil then - READLOCK[fd] = { session } - else - table.insert(q, session) - end - - skynet.redirect(selfaddr , 0, "system", session, skynet.pack(fd,0)) - coroutine.yield("CALL",session) + assert(not s.read_required) + s.read_required = sz + suspend(s) + ret = driver.pop(s.buffer, buffer_pool, sz) + if ret then + return ret else - READTHREAD[fd] = true + return false, driver.readall(s.buffer, buffer_pool) end end +function socket.readall(id) + local s = socket_pool[id] + assert(s) + if not s.connected then + return driver.readall(s.buffer, buffer_pool) + end + assert(not s.read_required) + s.read_required = true + suspend(s) + assert(s.connected == false) + return driver.readall(s.buffer, buffer_pool) +end + +function socket.readline(id, sep) + sep = sep or "\n" + local s = socket_pool[id] + assert(s) + local ret = driver.readline(s.buffer, buffer_pool, sep) + if ret then + return ret + end + if not s.connected then + return false, driver.readall(s.buffer, buffer_pool) + end + assert(not s.read_required) + s.read_required = sep + suspend(s) + if s.connected then + return driver.readline(s.buffer, buffer_pool, sep) + else + return false, driver.readall(s.buffer, buffer_pool) + end +end + +socket.write = assert(driver.send) + +function socket.invalid(id) + return socket_pool[id] == nil +end + +function socket.listen(host,port,func) + local id = driver.listen(host,port) + local s = { + id = id, + connected = true, + callback = func + } + socket_pool[id] = s + return id +end + +function socket.lock(id) +end + function socket.unlock(fd) - READTHREAD[fd] = nil - local q = READLOCK[fd] - if q then - if q[1] then - READTHREAD[fd] = true - response(q[1]) - table.remove(q,1) - else - READLOCK[fd] = nil - end - end + end return socket diff --git a/service/testsocket.lua b/service/testsocket.lua new file mode 100644 index 00000000..8c452e2e --- /dev/null +++ b/service/testsocket.lua @@ -0,0 +1,22 @@ +local skynet = require "skynet" +local socket = require "socket" + +local function accepter(id, addr) + print("connect from " .. addr .. " " .. id) + socket.accept(id) + socket.write(id, "Hello Skynet\n") + while true do + local str = socket.readline(id,"\n") + if str then + socket.write(id, str .. "\n") + else + socket.close(id) + print("closed " .. id) + return + end + end +end + +skynet.start(function() + socket.listen("127.0.0.1", 8000, accepter) +end) diff --git a/skynet-src/skynet_socket.c b/skynet-src/skynet_socket.c index e1b4dd19..177f124d 100644 --- a/skynet-src/skynet_socket.c +++ b/skynet-src/skynet_socket.c @@ -128,3 +128,9 @@ skynet_socket_close(struct skynet_context *ctx, int id) { uint32_t source = skynet_context_handle(ctx); socket_server_close(SOCKET_SERVER, source, id); } + +void +skynet_socket_accept(struct skynet_context *ctx, int id) { + uint32_t source = skynet_context_handle(ctx); + socket_server_accept(SOCKET_SERVER, source, id); +} diff --git a/skynet-src/skynet_socket.h b/skynet-src/skynet_socket.h index 6fa9c746..97bf219a 100644 --- a/skynet-src/skynet_socket.h +++ b/skynet-src/skynet_socket.h @@ -3,11 +3,11 @@ struct skynet_context; -#define SKYNET_SOCKET_TYPE_DATA 0 -#define SKYNET_SOCKET_TYPE_CONNECT 1 -#define SKYNET_SOCKET_TYPE_CLOSE 2 -#define SKYNET_SOCKET_TYPE_ACCEPT 3 -#define SKYNET_SOCKET_TYPE_ERROR 4 +#define SKYNET_SOCKET_TYPE_DATA 1 +#define SKYNET_SOCKET_TYPE_CONNECT 2 +#define SKYNET_SOCKET_TYPE_CLOSE 3 +#define SKYNET_SOCKET_TYPE_ACCEPT 4 +#define SKYNET_SOCKET_TYPE_ERROR 5 struct skynet_socket_message { int type; @@ -26,5 +26,6 @@ int skynet_socket_listen(struct skynet_context *ctx, const char *host, int port, int skynet_socket_connect(struct skynet_context *ctx, const char *host, int port); int skynet_socket_bind(struct skynet_context *ctx, int fd); void skynet_socket_close(struct skynet_context *ctx, int id); +void skynet_socket_accept(struct skynet_context *ctx, int id); #endif diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c index 302f1baf..0e02ffa0 100644 --- a/skynet-src/socket_server.c +++ b/skynet-src/socket_server.c @@ -25,6 +25,7 @@ #define SOCKET_TYPE_CONNECTED 4 #define SOCKET_TYPE_HALFCLOSE 5 #define SOCKET_TYPE_BIND 6 +#define SOCKET_TYPE_NOTACCEPT 7 #define MAX_SOCKET (1<head = s->tail = NULL; - sp_del(ss->event_fd, s->fd); + if (s->type != SOCKET_TYPE_NOTACCEPT) { + sp_del(ss->event_fd, s->fd); + } if (s->type != SOCKET_TYPE_BIND) { close(s->fd); } @@ -214,13 +223,15 @@ socket_server_release(struct socket_server *ss) { } static struct socket * -new_fd(struct socket_server *ss, int id, int fd, uintptr_t opaque) { +new_fd(struct socket_server *ss, int id, int fd, uintptr_t opaque, bool add) { struct socket * s = &ss->slot[id % MAX_SOCKET]; assert(s->type == SOCKET_TYPE_RESERVE); - if (sp_add(ss->event_fd, fd, s)) { - s->type = SOCKET_TYPE_INVALID; - return NULL; + if (add) { + if (sp_add(ss->event_fd, fd, s)) { + s->type = SOCKET_TYPE_INVALID; + return NULL; + } } s->id = id; @@ -276,7 +287,7 @@ open_socket(struct socket_server *ss, struct request_open * request, struct sock goto _failed; } - ns = new_fd(ss, id, sock, request->opaque); + ns = new_fd(ss, id, sock, request->opaque, true); if (ns == NULL) { close(sock); goto _failed; @@ -341,7 +352,9 @@ static int send_socket(struct socket_server *ss, struct request_send * request, struct socket_message *result) { int id = request->id; struct socket * s = &ss->slot[id % MAX_SOCKET]; - if (s->type == SOCKET_TYPE_INVALID || s->id != id || s->type == SOCKET_TYPE_HALFCLOSE) { + if (s->type == SOCKET_TYPE_INVALID || s->id != id + || s->type == SOCKET_TYPE_HALFCLOSE + || s->type == SOCKET_TYPE_NOTACCEPT) { FREE(request->buffer); return -1; } @@ -415,7 +428,7 @@ listen_socket(struct socket_server *ss, struct request_listen * request, struct if (listen(listen_fd, request->backlog) == -1) { goto _failed; } - struct socket *s = new_fd(ss, id, listen_fd, request->opaque); + struct socket *s = new_fd(ss, id, listen_fd, request->opaque, true); if (s == NULL) { goto _failed; } @@ -466,7 +479,7 @@ bind_socket(struct socket_server *ss, struct request_bind *request, struct socke result->id = id; result->opaque = request->opaque; result->ud = 0; - struct socket *s = new_fd(ss, id, request->fd, request->opaque); + struct socket *s = new_fd(ss, id, request->fd, request->opaque, true); if (s == NULL) { result->data = NULL; return SOCKET_ERROR; @@ -477,6 +490,27 @@ bind_socket(struct socket_server *ss, struct request_bind *request, struct socke return SOCKET_OPEN; } +static int +accept_socket(struct socket_server *ss, struct request_accept *request, struct socket_message *result) { + int id = request->id; + result->id = id; + result->opaque = request->opaque; + result->ud = 0; + result->data = NULL; + struct socket *s = &ss->slot[id % MAX_SOCKET]; + if (s->type != SOCKET_TYPE_NOTACCEPT || s->id !=id) { + return SOCKET_ERROR; + } + if (sp_add(ss->event_fd, s->fd, s)) { + s->type = SOCKET_TYPE_INVALID; + return SOCKET_ERROR; + } + s->type = SOCKET_TYPE_CONNECTED; + s->opaque = request->opaque; + result->data = "accept"; + return SOCKET_OPEN; +} + static void block_readpipe(int pipefd, void *buffer, int sz) { for (;;) { @@ -506,6 +540,8 @@ ctrl_cmd(struct socket_server *ss, struct socket_message *result) { block_readpipe(fd, buffer, len); // ctrl command only exist in local fd, so don't worry about endian. switch (type) { + case 'A': + return accept_socket(ss,(struct request_accept *)buffer, result); case 'B': return bind_socket(ss,(struct request_bind *)buffer, result); case 'L': @@ -618,15 +654,15 @@ report_accept(struct socket_server *ss, struct socket *s, struct socket_message return 0; } sp_nonblocking(client_fd); - struct socket *ns = new_fd(ss, id, client_fd, s->opaque); + struct socket *ns = new_fd(ss, id, client_fd, s->opaque, false); if (ns == NULL) { close(client_fd); return 0; } - ns->type = SOCKET_TYPE_CONNECTED; + ns->type = SOCKET_TYPE_NOTACCEPT; result->opaque = s->opaque; - result->id = id; - result->ud = s->id; + result->id = s->id; + result->ud = id; result->data = NULL; void * sin_addr = (u.s.sa_family == AF_INET) ? (void*)&u.v4.sin_addr : (void *)&u.v6.sin6_addr; @@ -785,3 +821,12 @@ socket_server_bind(struct socket_server *ss, uintptr_t opaque, int fd) { return id; } +void +socket_server_accept(struct socket_server *ss, uintptr_t opaque, int id) { + struct request_package request; + request.u.accept.id = id; + request.u.accept.opaque = opaque; + send_request(ss, &request, 'A', sizeof(request.u.accept)); +} + + diff --git a/skynet-src/socket_server.h b/skynet-src/socket_server.h index 6d4ac10c..f1cb7d28 100644 --- a/skynet-src/socket_server.h +++ b/skynet-src/socket_server.h @@ -25,6 +25,7 @@ int socket_server_poll(struct socket_server *, struct socket_message *result); void socket_server_exit(struct socket_server *); void socket_server_close(struct socket_server *, uintptr_t opaque, int id); +void socket_server_accept(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);