diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index b46bb883..8f5a4a18 100644 --- a/lualib-src/lua-socket.c +++ b/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; diff --git a/service-src/service_gate.c b/service-src/service_gate.c index f355932e..53a41dbc 100644 --- a/service-src/service_gate.c +++ b/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); } diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index 3a809481..74dc309f 100644 --- a/service-src/service_harbor.c +++ b/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,7 +245,11 @@ _connect_to(struct harbor *h, const char *ipaddress) { int portid = strtol(port+1, NULL,10); - return skynet_socket_connect(h->ctx, tmp, portid); + if (blocking) { + return skynet_socket_block_connect(h->ctx, tmp, portid); + } else { + return skynet_socket_connect(h->ctx, tmp, portid); + } } static inline void @@ -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 -_connect_master(struct skynet_context * ctx, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { - 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); - +static void +_launch_gate(struct skynet_context * ctx, const char * local_addr) { 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; } diff --git a/skynet-src/skynet_socket.c b/skynet-src/skynet_socket.c index 98654f47..5e1b8111 100644 --- a/skynet-src/skynet_socket.c +++ b/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); diff --git a/skynet-src/skynet_socket.h b/skynet-src/skynet_socket.h index c55df50c..1d6afd45 100644 --- a/skynet-src/skynet_socket.h +++ b/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); diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c index f8dc607e..411d92cc 100644 --- a/skynet-src/socket_server.c +++ b/skynet-src/socket_server.c @@ -78,8 +78,7 @@ struct request_close { struct request_listen { int id; - int port; - int backlog; + int fd; 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; } - sp_nonblocking(sock); + 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 - // 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; - } + int listen_fd = request->fd; 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 -socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * addr, int port) { - struct request_package request; +static int +open_request(struct socket_server *ss, struct request_package *req, uintptr_t opaque, const char *addr, int port) { 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; - request.u.open.id = id; - request.u.open.port = port; - memcpy(request.u.open.host, addr, len); - request.u.open.host[len] = '\0'; + req->u.open.opaque = opaque; + req->u.open.id = id; + req->u.open.port = port; + memcpy(req->u.open.host, addr, len); + 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 len = (addr!=NULL) ? strlen(addr) : 0; - if (len + sizeof(request.u.listen) > 256) { - fprintf(stderr, "socket-server : Invalid listen addr %s.\n",addr); - return 0; + int fd = do_listen(addr, port, backlog); + if (fd < 0) { + return -1; } + 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.backlog = backlog; - 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); + request.u.listen.fd = fd; + send_request(ss, &request, 'L', sizeof(request.u.listen)); return id; } diff --git a/skynet-src/socket_server.h b/skynet-src/socket_server.h index d9e39403..f77dc53c 100644 --- a/skynet-src/socket_server.h +++ b/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