gate support TCP_NODELAY

examples/main.lua

@@ -12,6 +12,7 @@ skynet.start(function()
 		port = 8888,
 		maxclient = max_client,
 	})
+	print("Watchdog listen on ", 8888)
 
 	skynet.exit()
 end)

examples/watchdog.lua

@@ -34,6 +34,7 @@ end
 
 function CMD.start(conf)
 	skynet.call(gate, "lua", "open" , conf)
+	skynet.call(gate, "lua", "nodelay", true)
 end
 
 skynet.start(function()

lualib-src/lua-socket.c

@@ -476,6 +476,14 @@ lstart(lua_State *L) {
 	return 0;
 }
 
+static int
+lnodelay(lua_State *L) {
+	struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1));
+	int id = luaL_checkinteger(L, 1);
+	skynet_socket_nodelay(ctx,id);
+	return 0;
+}
+
 int
 luaopen_socketdriver(lua_State *L) {
 	luaL_checkversion(L);
@@ -502,6 +510,7 @@ luaopen_socketdriver(lua_State *L) {
 		{ "lsend", lsendlow },
 		{ "bind", lbind },
 		{ "start", lstart },
+		{ "nodelay", lnodelay },
 		{ NULL, NULL },
 	};
 	lua_getfield(L, LUA_REGISTRYINDEX, "skynet_context");

service/gate.lua

@@ -8,6 +8,7 @@ local watchdog
 local maxclient
 local client_number = 0
 local CMD = setmetatable({}, { __gc = function() netpack.clear(queue) end })
+local nodelay = false
 
 local connection = {}	-- fd -> connection : { fd , client, agent , ip, mode }
 local forwarding = {}	-- agent -> connection
@@ -22,6 +23,13 @@ function CMD.open( source , conf )
 	socketdriver.start(socket)
 end
 
+function CMD.nodelay(source, v)
+	if v ~= false then
+		v = true
+	end
+	nodelay = v
+end
+
 function CMD.close()
 	assert(socket)
 	socketdriver.close(socket)
@@ -106,6 +114,9 @@ function MSG.open(fd, msg)
 	}
 	connection[fd] = c
 	client_number = client_number + 1
+	if nodelay then
+		socketdriver.nodelay(fd)
+	end
 	skynet.send(watchdog, "lua", "socket", "open", fd, msg)
 end
 

skynet-src/skynet_socket.c

@@ -150,3 +150,8 @@ skynet_socket_start(struct skynet_context *ctx, int id) {
 	uint32_t source = skynet_context_handle(ctx);
 	socket_server_start(SOCKET_SERVER, source, id);
 }
+
+void
+skynet_socket_nodelay(struct skynet_context *ctx, int id) {
+	socket_server_nodelay(SOCKET_SERVER, id);
+}

skynet-src/skynet_socket.h

@@ -28,5 +28,6 @@ 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_start(struct skynet_context *ctx, int id);
+void skynet_socket_nodelay(struct skynet_context *ctx, int id);
 
 #endif

skynet-src/socket_server.c

@@ -5,6 +5,7 @@
 
 #include <sys/types.h>
 #include <sys/socket.h>
+#include <netinet/tcp.h>
 #include <unistd.h>
 #include <errno.h>
 #include <stdlib.h>
@@ -34,6 +35,8 @@
 #define PRIORITY_HIGH 0
 #define PRIORITY_LOW 1
 
+#define HASH_ID(id) (((unsigned)id) % MAX_SOCKET)
+
 struct write_buffer {
 	struct write_buffer * next;
 	char *ptr;
@@ -107,6 +110,12 @@ struct request_start {
 	uintptr_t opaque;
 };
 
+struct request_setopt {
+	int id;
+	int what;
+	int value;
+};
+
 struct request_package {
 	uint8_t header[8];	// 6 bytes dummy
 	union {
@@ -117,6 +126,7 @@ struct request_package {
 		struct request_listen listen;
 		struct request_bind bind;
 		struct request_start start;
+		struct request_setopt setopt;
 	} u;
 	uint8_t dummy[256];
 };
@@ -144,7 +154,7 @@ reserve_id(struct socket_server *ss) {
 		if (id < 0) {
 			id = __sync_and_and_fetch(&(ss->alloc_id), 0x7fffffff);
 		}
-		struct socket *s = &ss->slot[id % MAX_SOCKET];
+		struct socket *s = &ss->slot[HASH_ID(id)];
 		if (s->type == SOCKET_TYPE_INVALID) {
 			if (__sync_bool_compare_and_swap(&s->type, SOCKET_TYPE_INVALID, SOCKET_TYPE_RESERVE)) {
 				s->id = id;
@@ -267,7 +277,7 @@ check_wb_list(struct wb_list *s) {
 
 static struct socket *
 new_fd(struct socket_server *ss, int id, int fd, uintptr_t opaque, bool add) {
-	struct socket * s = &ss->slot[id % MAX_SOCKET];
+	struct socket * s = &ss->slot[HASH_ID(id)];
 	assert(s->type == SOCKET_TYPE_RESERVE);
 
 	if (add) {
@@ -357,7 +367,7 @@ open_socket(struct socket_server *ss, struct request_open * request, struct sock
 	return -1;
 _failed:
 	freeaddrinfo( ai_list );
-	ss->slot[id % MAX_SOCKET].type = SOCKET_TYPE_INVALID;
+	ss->slot[HASH_ID(id)].type = SOCKET_TYPE_INVALID;
 	return SOCKET_ERROR;
 }
 
@@ -502,7 +512,7 @@ send_buffer_empty(struct socket *s) {
 static int
 send_socket(struct socket_server *ss, struct request_send * request, struct socket_message *result, int priority) {
 	int id = request->id;
-	struct socket * s = &ss->slot[id % MAX_SOCKET];
+	struct socket * s = &ss->slot[HASH_ID(id)];
 	if (s->type == SOCKET_TYPE_INVALID || s->id != id 
 		|| s->type == SOCKET_TYPE_HALFCLOSE
 		|| s->type == SOCKET_TYPE_PACCEPT) {
@@ -556,7 +566,7 @@ _failed:
 	result->id = id;
 	result->ud = 0;
 	result->data = NULL;
-	ss->slot[id % MAX_SOCKET].type = SOCKET_TYPE_INVALID;
+	ss->slot[HASH_ID(id)].type = SOCKET_TYPE_INVALID;
 
 	return SOCKET_ERROR;
 }
@@ -564,7 +574,7 @@ _failed:
 static int
 close_socket(struct socket_server *ss, struct request_close *request, struct socket_message *result) {
 	int id = request->id;
-	struct socket * s = &ss->slot[id % MAX_SOCKET];
+	struct socket * s = &ss->slot[HASH_ID(id)];
 	if (s->type == SOCKET_TYPE_INVALID || s->id != id) {
 		result->id = id;
 		result->opaque = request->opaque;
@@ -612,7 +622,7 @@ start_socket(struct socket_server *ss, struct request_start *request, struct soc
 	result->opaque = request->opaque;
 	result->ud = 0;
 	result->data = NULL;
-	struct socket *s = &ss->slot[id % MAX_SOCKET];
+	struct socket *s = &ss->slot[HASH_ID(id)];
 	if (s->type == SOCKET_TYPE_INVALID || s->id !=id) {
 		return SOCKET_ERROR;
 	}
@@ -633,6 +643,17 @@ start_socket(struct socket_server *ss, struct request_start *request, struct soc
 	return -1;
 }
 
+static void
+setopt_socket(struct socket_server *ss, struct request_setopt *request) {
+	int id = request->id;
+	struct socket *s = &ss->slot[HASH_ID(id)];
+	if (s->type == SOCKET_TYPE_INVALID || s->id !=id) {
+		return;
+	}
+	int v = request->value;
+	setsockopt(s->fd, IPPROTO_TCP, request->what, &v, sizeof(v));
+}
+
 static void
 block_readpipe(int pipefd, void *buffer, int sz) {
 	for (;;) {
@@ -696,6 +717,9 @@ ctrl_cmd(struct socket_server *ss, struct socket_message *result) {
 		return send_socket(ss, (struct request_send *)buffer, result, PRIORITY_HIGH);
 	case 'P':
 		return send_socket(ss, (struct request_send *)buffer, result, PRIORITY_LOW);
+	case 'T':
+		setopt_socket(ss, (struct request_setopt *)buffer);
+		return -1;
 	default:
 		fprintf(stderr, "socket-server: Unknown ctrl %c.\n",type);
 		return -1;
@@ -942,9 +966,7 @@ socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * a
 // return -1 when error
 int64_t 
 socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz) {
-	if (id < 0)
-		return -1;
-	struct socket * s = &ss->slot[id % MAX_SOCKET];
+	struct socket * s = &ss->slot[HASH_ID(id)];
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
 		return -1;
 	}
@@ -960,9 +982,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
 
 void 
 socket_server_send_lowpriority(struct socket_server *ss, int id, const void * buffer, int sz) {
-	if (id < 0)
-		return;
-	struct socket * s = &ss->slot[id % MAX_SOCKET];
+	struct socket * s = &ss->slot[HASH_ID(id)];
 	if (s->id != id || s->type == SOCKET_TYPE_INVALID) {
 		return;
 	}
@@ -1057,4 +1077,11 @@ socket_server_start(struct socket_server *ss, uintptr_t opaque, int id) {
 	send_request(ss, &request, 'S', sizeof(request.u.start));
 }
 
-
+void
+socket_server_nodelay(struct socket_server *ss, int id) {
+	struct request_package request;
+	request.u.setopt.id = id;
+	request.u.setopt.what = TCP_NODELAY;
+	request.u.setopt.value = 1;
+	send_request(ss, &request, 'T', sizeof(request.u.setopt));
+}

skynet-src/socket_server.h

@@ -36,4 +36,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);
 
+void socket_server_nodelay(struct socket_server *, int id);
+
 #endif