add block connect for harbor

lualib-src/lua-socket.c

@@ -397,6 +397,9 @@ llisten(lua_State *L) {
 	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);
+	if (id < 0) {
+		return luaL_error(L, "Listen error");
+	}
 
 	lua_pushinteger(L,id);
 	return 1;

service-src/service_gate.c

@@ -292,7 +292,7 @@ _cb(struct skynet_context * ctx, void * ud, int type, int session, uint32_t sour
 	return 0;
 }
 
-static void
+static int
 start_listen(struct gate *g, char * listen_addr) {
 	struct skynet_context * ctx = g->ctx;
 	char * portstr = strchr(listen_addr,':');
@@ -302,18 +302,22 @@ start_listen(struct gate *g, char * listen_addr) {
 		port = strtol(listen_addr, NULL, 10);
 		if (port <= 0) {
 			skynet_error(ctx, "Invalid gate address %s",listen_addr);
-			return;
+			return 1;
 		}
 	} else {
 		port = strtol(portstr + 1, NULL, 10);
 		if (port <= 0) {
 			skynet_error(ctx, "Invalid gate address %s",listen_addr);
-			return;
+			return 1;
 		}
 		portstr[0] = '\0';
 		host = listen_addr;
 	}
 	g->listen_id = skynet_socket_listen(ctx, host, port, BACKLOG);
+	if (g->listen_id < 0) {
+		return 1;
+	}
+	return 0;
 }
 
 int
@@ -370,8 +374,7 @@ gate_init(struct gate *g , struct skynet_context * ctx, char * parm) {
 	g->client_tag = client_tag;
 	g->header_size = header=='S' ? 2 : 4;
 
-	start_listen(g,binding);
 	skynet_callback(ctx,g,_cb);
 
-	return 0;
+	return start_listen(g,binding);
 }

service-src/service_harbor.c

@@ -49,7 +49,6 @@ struct remote_message_header {
 
 struct harbor {
 	struct skynet_context *ctx;
-	const void * starting_msg;
 	char * local_addr;
 	int id;
 	struct hashmap * map;
@@ -211,7 +210,6 @@ harbor_create(void) {
 		h->remote_addr[i] = NULL;
 	}
 	h->map = _hash_new();
-	h->starting_msg = NULL;
 	return h;
 }
 
@@ -235,7 +233,7 @@ harbor_release(struct harbor *h) {
 }
 
 static int
-_connect_to(struct harbor *h, const char *ipaddress) {
+_connect_to(struct harbor *h, const char *ipaddress, bool blocking) {
 	char * port = strchr(ipaddress,':');
 	if (port==NULL) {
 		return -1;
@@ -247,8 +245,12 @@ _connect_to(struct harbor *h, const char *ipaddress) {
 
 	int portid = strtol(port+1, NULL,10);
 
+	if (blocking) {
+		return skynet_socket_block_connect(h->ctx, tmp, portid);
+	} else {
 		return skynet_socket_connect(h->ctx, tmp, portid);
 	}
+}
 
 static inline void
 to_bigendian(uint8_t *buffer, uint32_t n) {
@@ -318,7 +320,7 @@ _update_remote_address(struct harbor *h, int harbor_id, const char * ipaddr) {
 		free(h->remote_addr[harbor_id]);
 		h->remote_addr[harbor_id] = NULL;
 	}
-	h->remote_fd[harbor_id] = _connect_to(h, ipaddr);
+	h->remote_fd[harbor_id] = _connect_to(h, ipaddr, false);
 	h->connected[harbor_id] = false;
 }
 
@@ -397,7 +399,6 @@ _remote_send_handle(struct harbor *h, uint32_t source, uint32_t destination, int
 		cookie.session = (uint32_t)session;
 		_send_remote(context, fd, msg,sz,&cookie);
 	} else {
-		_request_master(h, NULL, 0, harbor_id);
 		skynet_error(context, "Drop message to harbor %d from %x to %x (session = %d, msgsz = %d)",harbor_id, source, destination,session,(int)sz);
 	}
 	return 0;
@@ -551,38 +552,10 @@ _mainloop(struct skynet_context * context, void * ud, int type, int session, uin
 	}
 }
 
-static int
+static void
-_connect_master(struct skynet_context * ctx, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
+_launch_gate(struct skynet_context * ctx, const char * local_addr) {
-	struct harbor *h = ud;
-	if (type != PTYPE_SOCKET) {
-		// when message before connected, forward to self.
-		// If recv the same message, sleep a while.
-		if (h->starting_msg == msg) {
-			usleep(1000);
-		} else {
-			if (h->starting_msg == NULL) {
-				h->starting_msg = msg;
-			}
-		}
-		skynet_forward(ctx, 0);
-		return 0;
-	}
-	const struct skynet_socket_message * message = msg;
-	if (message->id != h->master_fd) {
-		if (message->type == SKYNET_SOCKET_TYPE_DATA) {
-			free(message->buffer);
-		}
-		skynet_error(ctx, "Invalid socket message incoming before connecting to master");
-		return 0;
-	}
-	if (message->type == SKYNET_SOCKET_TYPE_ERROR) {
-		fprintf(stderr, "Harbor: Connect to master failed\n");
-		exit(1);
-	}
-	assert(message->type == SKYNET_SOCKET_TYPE_CONNECT);
-
 	char tmp[128];
-	sprintf(tmp,"gate L ! %s %d %d 0",h->local_addr, PTYPE_HARBOR, REMOTE_MAX);
+	sprintf(tmp,"gate L ! %s %d %d 0",local_addr, PTYPE_HARBOR, REMOTE_MAX);
 	const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp);
 	if (gate_addr == NULL) {
 		fprintf(stderr, "Harbor : launch gate failed\n");
@@ -597,12 +570,6 @@ _connect_master(struct skynet_context * ctx, void * ud, int type, int session, u
 	int n = sprintf(tmp,"broker %s",self_addr);
 	skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, tmp, n);
 	skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, "start", 5);
-
-	skynet_callback(ctx, h, _mainloop);
-
-	_request_master(h, h->local_addr, strlen(h->local_addr), h->id);
-
-	return 0;
 }
 
 int
@@ -614,12 +581,17 @@ harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) {
 	int harbor_id = 0;
 	sscanf(args,"%s %s %d",master_addr, local_addr, &harbor_id);
 	h->master_addr = strdup(master_addr);
-	h->master_fd = _connect_to(h, master_addr);
+	h->master_fd = _connect_to(h, master_addr, true);
+	if (h->master_fd == -1) {
+		fprintf(stderr, "Harbor: Connect to master failed\n");
+		exit(1);
+	}
 	h->local_addr = strdup(local_addr);
 	h->id = harbor_id;
-	h->starting_msg = NULL;
 
-	skynet_callback(ctx, h, _connect_master);
+	_launch_gate(ctx, local_addr);
+	skynet_callback(ctx, h, _mainloop);
+	_request_master(h, local_addr, strlen(local_addr), harbor_id);
 
 	return 0;
 }

skynet-src/skynet_socket.c

@@ -120,6 +120,12 @@ skynet_socket_connect(struct skynet_context *ctx, const char *host, int port) {
 	return socket_server_connect(SOCKET_SERVER, source, host, port);
 }
 
+int 
+skynet_socket_block_connect(struct skynet_context *ctx, const char *host, int port) {
+	uint32_t source = skynet_context_handle(ctx);
+	return socket_server_block_connect(SOCKET_SERVER, source, host, port);
+}
+
 int 
 skynet_socket_bind(struct skynet_context *ctx, int fd) {
 	uint32_t source = skynet_context_handle(ctx);

skynet-src/skynet_socket.h

@@ -24,6 +24,7 @@ int skynet_socket_poll();
 int skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz);
 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_block_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_start(struct skynet_context *ctx, int id);

skynet-src/socket_server.c

@@ -78,8 +78,7 @@ struct request_close {
 
 struct request_listen {
 	int id;
-	int port;
+	int fd;
-	int backlog;
 	uintptr_t opaque;
 	char host[1];
 };
@@ -246,7 +245,7 @@ new_fd(struct socket_server *ss, int id, int fd, uintptr_t opaque, bool add) {
 
 // return -1 when connecting
 static int
-open_socket(struct socket_server *ss, struct request_open * request, struct socket_message *result) {
+open_socket(struct socket_server *ss, struct request_open * request, struct socket_message *result, bool blocking) {
 	int id = request->id;
 	result->opaque = request->opaque;
 	result->id = id;
@@ -274,7 +273,9 @@ open_socket(struct socket_server *ss, struct request_open * request, struct sock
 		if ( sock < 0 ) {
 			continue;
 		}
+		if (!blocking) {
 			sp_nonblocking(sock);
+		}
 		status = connect( sock,	ai_ptr->ai_addr, ai_ptr->ai_addrlen	);
 		if ( status	!= 0 && errno != EINPROGRESS) {
 			close(sock);
@@ -404,32 +405,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
 static int
 listen_socket(struct socket_server *ss, struct request_listen * request, struct socket_message *result) {
 	int id = request->id;
-	// only support ipv4
+	int listen_fd = request->fd;
-	// todo: support ipv6 by getaddrinfo
-	uint32_t addr = INADDR_ANY;
-	if (request->host[0]) {
-		addr=inet_addr(request->host);
-	}
-	int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
-	if (listen_fd < 0) {
-		goto _failed_noclose;
-	}
-	int reuse = 1;
-	if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(int))==-1) {
-		goto _failed;
-	}
-
-	struct sockaddr_in my_addr;
-	memset(&my_addr, 0, sizeof(struct sockaddr_in));
-	my_addr.sin_family = AF_INET;
-	my_addr.sin_port = htons(request->port);
-	my_addr.sin_addr.s_addr = addr;
-	if (bind(listen_fd, (struct sockaddr *)&my_addr, sizeof(struct sockaddr)) == -1) {
-		goto _failed;
-	}
-	if (listen(listen_fd, request->backlog) == -1) {
-		goto _failed;
-	}
 	struct socket *s = new_fd(ss, id, listen_fd, request->opaque, false);
 	if (s == NULL) {
 		goto _failed;
@@ -438,7 +414,6 @@ listen_socket(struct socket_server *ss, struct request_listen * request, struct
 	return -1;
 _failed:
 	close(listen_fd);
-_failed_noclose:
 	result->opaque = request->opaque;
 	result->id = id;
 	result->ud = 0;
@@ -554,7 +529,7 @@ ctrl_cmd(struct socket_server *ss, struct socket_message *result) {
 	case 'K':
 		return close_socket(ss,(struct request_close *)buffer, result);
 	case 'O':
-		return open_socket(ss, (struct request_open *)buffer, result);
+		return open_socket(ss, (struct request_open *)buffer, result, false);
 	case 'X':
 		result->opaque = 0;
 		result->id = 0;
@@ -748,22 +723,42 @@ send_request(struct socket_server *ss, struct request_package *request, char typ
 	}
 }
 
-int 
+static int
-socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * addr, int port) {
+open_request(struct socket_server *ss, struct request_package *req, uintptr_t opaque, const char *addr, int port) {
-	struct request_package request;
 	int len = strlen(addr);
-	if (len + sizeof(request.u.open) > 256) {
+	if (len + sizeof(req->u.open) > 256) {
 		fprintf(stderr, "socket-server : Invalid addr %s.\n",addr);
 		return 0;
 	}
 	int id = reverve_id(ss);
-	request.u.open.opaque = opaque;
+	req->u.open.opaque = opaque;
-	request.u.open.id = id;
+	req->u.open.id = id;
-	request.u.open.port = port;
+	req->u.open.port = port;
-	memcpy(request.u.open.host, addr, len);
+	memcpy(req->u.open.host, addr, len);
-	request.u.open.host[len] = '\0';
+	req->u.open.host[len] = '\0';
+
+	return len;
+}
+
+int 
+socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * addr, int port) {
+	struct request_package request;
+	int len = open_request(ss, &request, opaque, addr, port);
 	send_request(ss, &request, 'O', sizeof(request.u.open) + len);
-	return id;
+	return request.u.open.id;
+}
+
+int 
+socket_server_block_connect(struct socket_server *ss, uintptr_t opaque, const char * addr, int port) {
+	struct request_package request;
+	struct socket_message result;
+	open_request(ss, &request, opaque, addr, port);
+	int ret = open_socket(ss, &request.u.open, &result, true);
+	if (ret == SOCKET_OPEN) {
+		return result.id;
+	} else {
+		return -1;
+	}
 }
 
 // return -1 when error
@@ -798,27 +793,52 @@ socket_server_close(struct socket_server *ss, uintptr_t opaque, int id) {
 	send_request(ss, &request, 'K', sizeof(request.u.close));
 }
 
+static int
+do_listen(const char * host, int port, int backlog) {
+	// only support ipv4
+	// todo: support ipv6 by getaddrinfo
+	uint32_t addr = INADDR_ANY;
+	if (host[0]) {
+		addr=inet_addr(host);
+	}
+	int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
+	if (listen_fd < 0) {
+		return -1;
+	}
+	int reuse = 1;
+	if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(int))==-1) {
+		goto _failed;
+	}
+
+	struct sockaddr_in my_addr;
+	memset(&my_addr, 0, sizeof(struct sockaddr_in));
+	my_addr.sin_family = AF_INET;
+	my_addr.sin_port = htons(port);
+	my_addr.sin_addr.s_addr = addr;
+	if (bind(listen_fd, (struct sockaddr *)&my_addr, sizeof(struct sockaddr)) == -1) {
+		goto _failed;
+	}
+	if (listen(listen_fd, backlog) == -1) {
+		goto _failed;
+	}
+	return listen_fd;
+_failed:
+	close(listen_fd);
+	return -1;
+}
+
 int 
 socket_server_listen(struct socket_server *ss, uintptr_t opaque, const char * addr, int port, int backlog) {
-	struct request_package request;
+	int fd = do_listen(addr, port, backlog);
-	int len = (addr!=NULL) ? strlen(addr) : 0;
+	if (fd < 0) {
-	if (len + sizeof(request.u.listen) > 256) {
+		return -1;
-		fprintf(stderr, "socket-server : Invalid listen addr %s.\n",addr);
-		return 0;
 	}
+	struct request_package request;
 	int id = reverve_id(ss);
 	request.u.listen.opaque = opaque;
 	request.u.listen.id = id;
-	request.u.listen.port = port;
+	request.u.listen.fd = fd;
-	request.u.listen.backlog = backlog;
+	send_request(ss, &request, 'L', sizeof(request.u.listen));
-	if (len == 0) {
-		request.u.listen.host[0] = '\0';
-	} else {
-		int len = strlen(addr);
-		memcpy(request.u.listen.host, addr, len);
-		request.u.listen.host[len] = '\0';
-	}
-	send_request(ss, &request, 'L', sizeof(request.u.listen) + len);
 	return id;
 }
 

skynet-src/socket_server.h

@@ -35,4 +35,6 @@ int socket_server_listen(struct socket_server *, uintptr_t opaque, const char *
 int socket_server_connect(struct socket_server *, uintptr_t opaque, const char * addr, int port);
 int socket_server_bind(struct socket_server *, uintptr_t opaque, int fd);
 
+int socket_server_block_connect(struct socket_server *, uintptr_t opaque, const char * addr, int port);
+
 #endif