From 300300853b043546f0f86d18fafbf5cfffca297f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Mon, 22 Jul 2013 21:09:38 +0800 Subject: [PATCH] 1. none blocking connect ; 2. socket.readall ; 3.socket.readline will return last line ; 4. lua service report launcher error --- lualib/socket.lua | 120 ++++++++++++++++++++++++++++------- service-src/service_lua.c | 8 +++ service-src/service_socket.c | 103 +++++++++++++++++++----------- service/launcher.lua | 8 +++ 4 files changed, 181 insertions(+), 58 deletions(-) diff --git a/lualib/socket.lua b/lualib/socket.lua index 4d1c012d..0532811d 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -11,6 +11,7 @@ local READREQUEST = {} -- fd:request_size local READSESSION = {} -- fd:session local READLOCK = {} -- fd:queue(session) local READTHREAD= {} -- fd:thread +local CLOSED = {} -- fd:true local selfaddr = skynet.self() @@ -27,24 +28,26 @@ skynet.register_protocol { local qsz = READREQUEST[fd] local buf = READBUF[fd] local bsz - if sz == 0 or buf == true then - buf,bsz = true, qsz + if sz == 0 then + CLOSED[fd] = true else buf,bsz = buffer.push(buf, msg, sz) + READBUF[fd] = buf end - READBUF[fd] = buf local session = READSESSION[fd] if qsz == nil or session == nil then return end - if type(qsz) == "number" then - if qsz > bsz then - return - end - else - -- request readline - if buffer.readline(buf, qsz, true) == nil then - return + 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 @@ -64,7 +67,7 @@ skynet.register_protocol { if t > 0 then -- lock request when t == 0 -- request bytes or readline local buf = READBUF[fd] - if buf == true then + if CLOSED[fd] then skynet.ret() return end @@ -75,7 +78,7 @@ skynet.register_protocol { skynet.ret() return end - else + elseif t == 2 then -- sz is sep if buffer.readline(buf, sz, true) then -- don't real read skynet.ret() @@ -95,7 +98,10 @@ function socket.open(addr, port) if r == "" then error(cmd .. " failed") end - return tonumber(r) + local fd = tonumber(r) + READBUF[fd] = true + CLOSED[fd] = nil + return fd end function socket.stdin() @@ -103,42 +109,112 @@ function socket.stdin() if r == "" then error("stdin bind failed") end - return tonumber(r) + local fd = tonumber(r) + READBUF[fd] = true + CLOSED[fd] = nil + return fd end function socket.close(fd) socket.lock(fd) skynet.call(".socket", "text", "close", fd) - READBUF[fd] = true + READBUF[fd] = nil READLOCK[fd] = nil + CLOSED[fd] = nil end function socket.read(fd, sz) - local str = buffer.pop(READBUF[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 - str = buffer.pop(READBUF[fd],sz) - return str + 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) + 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 str = buffer.readline(READBUF[fd],sep) + local buf = assert(READBUF[fd]) + local 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 + READREQUEST[fd] = sep skynet.call(selfaddr, "system",fd,2,sep) -- singal sep 2 READREQUEST[fd] = nil - str = buffer.readline(READBUF[fd],sep) - return str + 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) diff --git a/service-src/service_lua.c b/service-src/service_lua.c index db472487..5e8f01a3 100644 --- a/service-src/service_lua.c +++ b/service-src/service_lua.c @@ -117,6 +117,12 @@ _report_error(lua_State *L, struct skynet_context *ctx, const char *filename, in lua_pop(L,1); } +static void +_report_launcher_error(struct skynet_context *ctx) { + // sizeof "ERROR" == 5 + skynet_sendname(ctx, ".launcher", PTYPE_TEXT, 0, "ERROR", 5); +} + static int _init(struct snlua *l, struct skynet_context *ctx, const char * args) { lua_State *L = l->L; @@ -144,6 +150,7 @@ _init(struct snlua *l, struct skynet_context *ctx, const char * args) { } else { skynet_error(ctx, "lua parser [%s] error : %s", filename, lua_tostring(L,-1)); } + _report_launcher_error(ctx); return 1; } int n=0; @@ -162,6 +169,7 @@ _init(struct snlua *l, struct skynet_context *ctx, const char * args) { } } _report_error(L, ctx, filename, r); + _report_launcher_error(ctx); return 1; } diff --git a/service-src/service_socket.c b/service-src/service_socket.c index 98c38431..41bc2beb 100644 --- a/service-src/service_socket.c +++ b/service-src/service_socket.c @@ -17,8 +17,9 @@ #define READ_BUFFER 4000 #define STATUS_INVALID 0 -#define STATUS_HALFCLOSE 1 -#define STATUS_SUSPEND 2 +#define STATUS_CONNECT 1 +#define STATUS_HALFCLOSE 2 +#define STATUS_SUSPEND 3 struct write_buffer { struct write_buffer * next; @@ -66,7 +67,7 @@ _set_nonblocking(int fd) } static int -new_socket(struct socket_pool *p, int sock, uint32_t addr) { +new_socket(struct socket_pool *p, int sock, uint32_t addr, int session, bool connecting) { int i; if (p->count >= p->cap) { goto _error; @@ -79,13 +80,18 @@ new_socket(struct socket_pool *p, int sock, uint32_t addr) { if (event_add(p->fd, sock, s)) { goto _error; } - s->status = STATUS_SUSPEND; - _set_nonblocking(sock); + if (connecting) { + event_write(p->fd, sock, s, true); + s->status = STATUS_CONNECT; + } else { + s->status = STATUS_SUSPEND; + } int keepalive = 1; setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepalive , sizeof(keepalive)); s->fd = sock; s->id = id; s->source = addr; + s->session = session; p->count++; p->id = id + 1; if (p->id > MAX_ID) { @@ -101,9 +107,11 @@ _error: } static void -cmd_bind(struct skynet_context * ctx, struct socket_pool *p, int sock, int session, uint32_t source) { - char ret[10]; - int id = new_socket(p, sock, source); +cmd_bind(struct skynet_context * ctx, struct socket_pool *p, int sock, int session, uint32_t source, bool connecting) { + if (!connecting) { + _set_nonblocking(sock); + } + int id = new_socket(p, sock, source, session, connecting); if (id<0) { reply(ctx, source, session, NULL , 0); return; @@ -111,8 +119,6 @@ cmd_bind(struct skynet_context * ctx, struct socket_pool *p, int sock, int sessi if (p->count == 1) { skynet_command(ctx, "TIMEOUT", "0"); } - int sz = sprintf(ret,"%d",id); - reply(ctx, source, session, ret, sz); } static void @@ -142,8 +148,9 @@ cmd_open(struct skynet_context * ctx, struct socket_pool *p, char * cmd, int ses if ( sock < 0 ) { continue; } + _set_nonblocking(sock); status = connect( sock, ai_ptr->ai_addr, ai_ptr->ai_addrlen ); - if ( status != 0 ) { + if ( status != 0 && errno != EINPROGRESS) { close(sock); sock = -1; continue; @@ -157,7 +164,7 @@ cmd_open(struct skynet_context * ctx, struct socket_pool *p, char * cmd, int ses goto _failed; } - cmd_bind(ctx, p, sock, session, source); + cmd_bind(ctx, p, sock, session, source, true); return; _failed: @@ -202,6 +209,13 @@ cmd_close(struct skynet_context * ctx, struct socket_pool *p, int id, int sessio } } +static void +_reply_bind(struct skynet_context * ctx, int sock, int session, uint32_t source) { + char ret[10]; + int sz = sprintf(ret,"%d",sock); + reply(ctx, source, session, ret, sz); +} + static void _ctrl(struct skynet_context * ctx, struct socket_pool *p, char * command, int id, char * arg, int session, uint32_t source) { if (strcmp(command, "open")==0) { @@ -209,7 +223,8 @@ _ctrl(struct skynet_context * ctx, struct socket_pool *p, char * command, int id } else if (strcmp(command, "close")==0) { cmd_close(ctx, p, id, session, source); } else if (strcmp(command, "bind")==0) { - cmd_bind(ctx, p, id, session, source); + cmd_bind(ctx, p, id, session, source, false); + _reply_bind(ctx, id, session, source); } else { skynet_error(ctx, "Unknown command %s", command); reply(ctx, source, session, NULL, 0); @@ -316,7 +331,7 @@ try_send(struct skynet_context *ctx, struct socket_pool *p, uint32_t source, con skynet_error(ctx, "%x try to write socket %d:%x", source, id, s->source); return 0; } - if (s->status != STATUS_SUSPEND) { + if (s->status != STATUS_SUSPEND && s->status != STATUS_CONNECT) { skynet_error(ctx, "%x write to closed socket %d", source, id); return 0; } @@ -335,22 +350,24 @@ try_send(struct skynet_context *ctx, struct socket_pool *p, uint32_t source, con char * ptr = (char *)(msg+1); - for (;;) { - int wt = write(s->fd, ptr, sz); - if (wt < 0) { - switch(errno) { - case EINTR: - continue; + if (s->status != STATUS_CONNECT) { + for (;;) { + int wt = write(s->fd, ptr, sz); + if (wt < 0) { + switch(errno) { + case EINTR: + continue; + } + break; } + if (wt == sz) { + return 0; + } + sz-=wt; + ptr+=wt; + break; } - if (wt == sz) { - return 0; - } - sz-=wt; - ptr+=wt; - - break; } struct write_buffer * buf = malloc(sizeof(*buf)); @@ -384,15 +401,29 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t int i; for (i=0;iev[i]; - if (e->read) { - forward(context, e->s, p); - } - if (e->write) { - struct socket *s = e->s; - sendout(p, s); - if (s->status == STATUS_HALFCLOSE && s->head == NULL) { - force_close(s, p); - reply(context, source, s->session, NULL, 0); + struct socket *s= e->s; + if (s->status == STATUS_CONNECT) { + int error; + socklen_t len = sizeof(error); + int code = getsockopt(s->fd, SOL_SOCKET, SO_ERROR, &error, &len); + if (code < 0 || error) { + force_close(s,p); + reply(context, s->source, s->session, NULL , 0); + } else { + _reply_bind(context, s->id, s->session, s->source); + s->status = STATUS_SUSPEND; + } + } else { + if (e->read) { + forward(context, e->s, p); + } + if (e->write) { + struct socket *s = e->s; + sendout(p, s); + if (s->status == STATUS_HALFCLOSE && s->head == NULL) { + force_close(s, p); + reply(context, source, s->session, NULL, 0); + } } } } diff --git a/service/launcher.lua b/service/launcher.lua index 4fe0d5f7..72ee768e 100644 --- a/service/launcher.lua +++ b/service/launcher.lua @@ -85,6 +85,14 @@ skynet.dispatch("text" , function(session, address , cmd) skynet.redirect(reply.address , 0, "response", reply.session, skynet.address(address)) instance[address] = nil end + elseif cmd == "ERROR" then + -- see serivce-src/service_lua.c + -- init failed + local reply = instance[address] + if reply then + skynet.redirect(reply.address , 0, "response", reply.session, "") + instance[address] = nil + end else -- launch request local service, param = string.match(cmd,"([^ ]+) (.*)")