From 664056977dacd95622d9189e70fef9bd1ed951fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Wed, 5 Sep 2012 10:56:16 +0800 Subject: [PATCH] gate support user defined header size --- connection/lua-socket.c | 35 ++++++++++++++++++++++++----------- gate/main.c | 20 ++++++++++++++++---- service-src/service_harbor.c | 14 +++++++------- service-src/service_master.c | 18 +++++++++--------- service/watchdog.lua | 2 +- 5 files changed, 57 insertions(+), 32 deletions(-) diff --git a/connection/lua-socket.c b/connection/lua-socket.c index 3021b870..ce710e86 100644 --- a/connection/lua-socket.c +++ b/connection/lua-socket.c @@ -75,25 +75,38 @@ _write(lua_State *L) { static int _writeblock(lua_State *L) { int fd = luaL_checkinteger(L,1); - int type = lua_type(L,2); + int header = luaL_checkinteger(L,2); + int type = lua_type(L,3); const char * buffer = NULL; size_t sz; if (type == LUA_TSTRING) { - buffer = lua_tolstring(L,2,&sz); + buffer = lua_tolstring(L,3,&sz); } else if (type == LUA_TLIGHTUSERDATA) { - buffer = lua_touserdata(L,2); - sz = luaL_checkinteger(L,3); + buffer = lua_touserdata(L,3); + sz = luaL_checkinteger(L,4); } - if (sz > 65535) { - luaL_error(L, "Too big package %d", (int)sz); + if (header == 2) { + if (sz > 65535) { + luaL_error(L, "Too big package %d", (int)sz); + } + } else { + if (header != 4) { + luaL_error(L, "block header must be 2 or 4 bytes"); + } } struct iovec buf[2]; - // send big-endian header - uint8_t head[2] = { sz >> 8 & 0xff , sz & 0xff }; - buf[0].iov_base = head; - buf[0].iov_len = 2; + if (header == 2) { + // send big-endian header + uint8_t head[2] = { sz >> 8 & 0xff , sz & 0xff }; + buf[0].iov_base = head; + buf[0].iov_len = 2; + } else { + uint8_t head[4] = { sz >> 24 & 0xff, sz >> 16 & 0xff, sz >> 8 & 0xff , sz & 0xff }; + buf[0].iov_base = head; + buf[0].iov_len = 4; + } buf[1].iov_base = (void *)buffer; buf[1].iov_len = sz; @@ -107,7 +120,7 @@ _writeblock(lua_State *L) { } return 0; } - assert(err == sz +2); + assert(err == sz + header); return 0; } } diff --git a/gate/main.c b/gate/main.c index ff284372..52c5b4ce 100644 --- a/gate/main.c +++ b/gate/main.c @@ -25,6 +25,7 @@ struct gate { int cap; int max_connection; int client_tag; + int header_size; struct connection ** agent; struct connection * map; }; @@ -193,7 +194,7 @@ _cb(struct skynet_context * ctx, void * ud, int type, int session, uint32_t sour getpeername(fd, (struct sockaddr *)&remote_addr, &len); _report(g, ctx, "%d open %d %s:%u",id,fd,inet_ntoa(remote_addr.sin_addr),ntohs(remote_addr.sin_port)); } - uint8_t * plen = mread_pull(m,2); + uint8_t * plen = mread_pull(m,g->header_size); if (plen == NULL) { if (mread_closed(m)) { _remove_id(g,id); @@ -202,7 +203,12 @@ _cb(struct skynet_context * ctx, void * ud, int type, int session, uint32_t sour goto _break; } // big-endian - uint16_t len = plen[0] << 8 | plen[1]; + uint16_t len ; + if (g->header_size == 2) { + len = plen[0] << 8 | plen[1]; + } else { + len = plen[0] << 24 | plen[1] << 16 | plen[2] << 8 | plen[3]; + } void * data = mread_pull(m, len); if (data == NULL) { @@ -230,11 +236,16 @@ gate_init(struct gate *g , struct skynet_context * ctx, char * parm) { char watchdog[sz]; char binding[sz]; int client_tag = 0; - int n = sscanf(parm, "%s %s %d %d %d",watchdog, binding,&client_tag , &max,&buffer); - if (n<3) { + char header; + int n = sscanf(parm, "%c %s %s %d %d %d",&header,watchdog, binding,&client_tag , &max,&buffer); + if (n<4) { skynet_error(ctx, "Invalid gate parm %s",parm); return 1; } + if (header != 'S' && header !='L') { + skynet_error(ctx, "Invalid data header style"); + return 1; + } if (client_tag == 0) { client_tag = PTYPE_CLIENT; } @@ -279,6 +290,7 @@ gate_init(struct gate *g , struct skynet_context * ctx, char * parm) { g->max_connection = max; g->id_index = 0; g->client_tag = client_tag; + g->header_size = header=='S' ? 2 : 4; g->agent = malloc(cap * sizeof(struct connection *)); memset(g->agent, 0, cap * sizeof(struct connection *)); diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index 34cb84b5..3b612e85 100644 --- a/service-src/service_harbor.c +++ b/service-src/service_harbor.c @@ -270,10 +270,10 @@ _message_to_header(const uint32_t *message, struct remote_message_header *header static int _send_package(int fd, const void * buffer, size_t sz) { - uint16_t header = htons(sz); + uint32_t header = htonl(sz); struct iovec part[2]; part[0].iov_base = &header; - part[0].iov_len = 2; + part[0].iov_len = 4; part[1].iov_base = (void*)buffer; part[1].iov_len = sz; @@ -286,7 +286,7 @@ _send_package(int fd, const void * buffer, size_t sz) { continue; } } - if (err != sz+2) { + if (err != sz+4) { return 1; } return 0; @@ -295,11 +295,11 @@ _send_package(int fd, const void * buffer, size_t sz) { static int _send_remote(int fd, const char * buffer, size_t sz, struct remote_message_header * cookie) { - uint16_t sz_header = htons(sz+sizeof(*cookie)); + uint32_t sz_header = htonl(sz+sizeof(*cookie)); struct iovec part[3]; part[0].iov_base = &sz_header; - part[0].iov_len = 2; + part[0].iov_len = 4; part[1].iov_base = (char *)buffer; part[1].iov_len = sz; @@ -318,7 +318,7 @@ _send_remote(int fd, const char * buffer, size_t sz, struct remote_message_heade continue; } } - if (err != sz+sizeof(*cookie)+2) { + if (err != sz+sizeof(*cookie)+4) { return 1; } return 0; @@ -568,7 +568,7 @@ harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) { h->master_fd = master_fd; char tmp[128]; - sprintf(tmp,"gate ! %s %d %d 0",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) { skynet_error(ctx, "Harbor : launch gate failed"); diff --git a/service-src/service_master.c b/service-src/service_master.c index 09dc0749..8498062f 100644 --- a/service-src/service_master.c +++ b/service-src/service_master.c @@ -131,18 +131,18 @@ _connect_to(const char *ipaddress) { static int _send_to(int fd, const void * buf, size_t sz, uint32_t handle) { - char buffer[2 + sz + 12]; - uint16_t header = htons(sz+12); - memcpy(buffer, &header, 2); - memcpy(buffer+2, buf, sz); + char buffer[4 + sz + 12]; + uint32_t header = htonl(sz+12); + memcpy(buffer, &header, 4); + memcpy(buffer+4, buf, sz); uint32_t u32 = 0; - memcpy(buffer+2+sz,&u32,4); + memcpy(buffer+4+sz,&u32,4); u32 = htonl(handle); - memcpy(buffer+2+sz+4,&u32,4); + memcpy(buffer+4+sz+4,&u32,4); u32 = 0; - memcpy(buffer+2+sz+8,&u32,4); + memcpy(buffer+4+sz+8,&u32,4); - sz += 2 + 12; + sz += 4 + 12; for (;;) { int err = send(fd, buffer, sz, 0); @@ -269,7 +269,7 @@ _mainloop(struct skynet_context * context, void * ud, int type, int session, uin int master_init(struct master *m, struct skynet_context *ctx, const char * args) { char tmp[strlen(args) + 32]; - sprintf(tmp,"gate ! %s %d %d 0",args,PTYPE_HARBOR,REMOTE_MAX); + sprintf(tmp,"gate L ! %s %d %d 0",args,PTYPE_HARBOR,REMOTE_MAX); const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp); if (gate_addr == NULL) { skynet_error(ctx, "Master : launch gate failed"); diff --git a/service/watchdog.lua b/service/watchdog.lua index cdc39ae8..674cb981 100644 --- a/service/watchdog.lua +++ b/service/watchdog.lua @@ -52,7 +52,7 @@ skynet.start(function() end end) -- 0 for default client tag - gate = skynet.launch("gate" , skynet.address(skynet.self()), port, 0, max_agent, buffer) + gate = skynet.launch("gate" , "S" , skynet.address(skynet.self()), port, 0, max_agent, buffer) skynet.send(gate,"text", "start") skynet.register(".watchdog") end)