1. none blocking connect ; 2. socket.readall ; 3.socket.readline will return last line ; 4. lua service report launcher error

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)

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;
 }
 

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;i<n;i++) {
 		struct event *e = &p->ev[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);
+				}
 			}
 		}
 	}

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,"([^ ]+) (.*)")