Compare commits

...

29 Commits

Author SHA1 Message Date
Cloud Wu
ee61e60631 ready for 0.9.0 2014-11-17 14:55:21 +08:00
云风
c8c95f1690 Merge pull request #203 from cloudwu/dev
Ready for 0.9.0
2014-11-17 14:52:16 +08:00
Cloud Wu
3be0f50722 use snax.newservice would be better 2014-11-15 21:33:25 +08:00
Cloud Wu
6ee014dd99 bugfix: init should be out 2014-11-15 21:28:03 +08:00
云风
2a4bee5bdd Merge pull request #202 from niuys/patch-1
ecsape char '\Z' wrong
2014-11-14 20:06:44 +08:00
snail
4f21696387 ecsape char '\Z' wrong
In mysql-connector, it's '\Z', not '\z'.Now, this bug  will change oringinal data
2014-11-14 18:11:01 +08:00
Cloud Wu
6df72e5842 remove snax queue mode 2014-11-14 16:13:48 +08:00
Cloud Wu
b9e06dc065 add snax.self and snax.exit 2014-11-14 15:11:05 +08:00
Cloud Wu
8e4a175155 use fork to dispatch more socket message 2014-11-13 11:23:15 +08:00
Cloud Wu
fa78623b1c support user defined send object 2014-11-11 13:54:43 +08:00
Cloud Wu
e06a9e3701 dispatch read before write, and try to dispath both 2014-11-11 11:33:57 +08:00
Cloud Wu
c87dea3d9b fix issue #200 2014-11-10 20:58:10 +08:00
Cloud Wu
929ce385a3 fprintf need cr 2014-11-10 10:53:29 +08:00
Cloud Wu
765749f608 accept may not start fd, so it can't close at last 2014-11-04 18:25:15 +08:00
Cloud Wu
5fad5d63ae socket close fd when auth failed 2014-11-04 18:12:25 +08:00
Cloud Wu
af10c59eaf release v0.8.1 2014-11-03 15:18:00 +08:00
云风
a77b71530d Merge pull request #195 from cloudwu/dev
release 0.8.1
2014-11-03 15:13:55 +08:00
Cloud Wu
803a79059a update sproto lib 2014-10-31 16:49:54 +08:00
Cloud Wu
6c9ad16077 bugfix: socket open address, and httpd bodylimit 2014-10-29 16:51:39 +08:00
Cloud Wu
2ab689f7c1 remove sha1 from lua_mysqlaux, use lua-crypt instead 2014-10-29 11:18:47 +08:00
Cloud Wu
f8a50929a8 bugfix: break when write all 2014-10-29 11:04:35 +08:00
Cloud Wu
af3ca3bb2f merge lua bugfix, read http://www.lua.org/bugs.html 2014-10-28 19:40:27 +08:00
Cloud Wu
aaf8617dd5 skynet_free(NULL) is ok 2014-10-28 17:14:30 +08:00
Cloud Wu
1d3e393fd4 Merge branch 'czlc-master' into dev 2014-10-28 17:13:35 +08:00
Cloud Wu
c3df405e33 Merge branch 'master' of github.com:czlc/skynet-1 into czlc-master 2014-10-28 17:12:16 +08:00
Cloud Wu
b1ef8a33d9 harbor can post error message back when the destination is not exist 2014-10-28 17:00:24 +08:00
Cloud Wu
954625a530 fix issue #190 2014-10-28 16:48:35 +08:00
czlc
9850ca1b94 free buffer 2014-10-28 16:43:29 +08:00
Cloud Wu
01e1a6aafa fix issue #189 2014-10-28 16:04:01 +08:00
23 changed files with 498 additions and 622 deletions

View File

@@ -1,372 +0,0 @@
//
// lua_mysqlaux.c
//
// Created by changfeng on 6/17/14.
// Copyright (c) 2014 changfeng. All rights reserved.
//
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <lua.h>
#include <lauxlib.h>
#define SHA1SIZE 20
#define ROTL(bits,word) (((word) << (bits)) | ((word) >> (32-(bits))))
typedef unsigned int uint32_t;
struct sha
{
uint32_t digest[5];
uint32_t w[80];
uint32_t a,b,c,d,e,f;
int err;
};
static uint32_t padded_length_in_bits(uint32_t len)
{
if(len%64 == 56)
{
len++;
}
while((len%64)!=56)
{
len++;
}
return len*8;
}
static int calculate_sha1(struct sha *sha1, const unsigned char *text, uint32_t length)
{
unsigned int i,j;
unsigned char *buffer=NULL, *pbuffer=NULL;
uint32_t bits=0;
uint32_t temp=0,k=0;
uint32_t lb = length*8;
if (!sha1)
{
return 0;
}
// initialize the default digest values
sha1->digest[0] = 0x67452301;
sha1->digest[1] = 0xEFCDAB89;
sha1->digest[2] = 0x98BADCFE;
sha1->digest[3] = 0x10325476;
sha1->digest[4] = 0xC3D2E1F0;
sha1->a=sha1->b=sha1->c=sha1->d=sha1->e=sha1->f=0;
if (!text || !length)
{
return 0;
}
bits = padded_length_in_bits(length);
buffer = (unsigned char *) malloc((bits/8)+8);
memset(buffer,0,(bits/8)+8);
if(buffer == NULL)
{
return 1;
}
pbuffer = buffer;
memcpy(buffer, text, length);
//add 1 on the last of the message..
*(buffer+length) = 0x80;
for(i=length+1; i<(bits/8); i++)
{
*(buffer+i) = 0x00;
}
*(buffer +(bits/8)+4+0) = (lb>>24) & 0xFF;
*(buffer +(bits/8)+4+1) = (lb>>16) & 0xFF;
*(buffer +(bits/8)+4+2) = (lb>>8) & 0xFF;
*(buffer +(bits/8)+4+3) = (lb>>0) & 0xFF;
//main loop
for(i=0; i<((bits+64)/512); i++)
{
//first empty the block for each pass..
for(j=0; j<80; j++)
{
sha1->w[j] = 0x00;
}
//fill the first 16 words with the characters read directly from the buffer.
for(j=0; j<16; j++)
{
sha1->w[j] =buffer[j*4+0];
sha1->w[j] = sha1->w[j]<<8;
sha1->w[j] |= buffer[j*4+1];
sha1->w[j] = sha1->w[j]<<8;
sha1->w[j] |= buffer[j*4+2];
sha1->w[j] = sha1->w[j]<<8;
sha1->w[j] |= buffer[j*4+3];
}
//fill the rest 64 words using the formula
for(j=16; j<80; j++)
{
sha1->w[j] = (ROTL(1,(sha1->w[j-3] ^ sha1->w[j-8] ^ sha1->w[j-14] ^ sha1->w[j-16])));
}
//initialize hash for this chunck reading that has been stored in the structure digest
sha1->a = sha1->digest[0];
sha1->b = sha1->digest[1];
sha1->c = sha1->digest[2];
sha1->d = sha1->digest[3];
sha1->e = sha1->digest[4];
//for all the 80 32bit blocks calculate f and use k accordingly per specification.
for(j=0; j<80; j++)
{
if((j>=0) && (j<20))
{
sha1->f = ((sha1->b)&(sha1->c)) | ((~(sha1->b))&(sha1->d));
k = 0x5A827999;
}
else if((j>=20) && (j<40))
{
sha1->f = (sha1->b)^(sha1->c)^(sha1->d);
k = 0x6ED9EBA1;
}
else if((j>=40) && (j<60))
{
sha1->f = ((sha1->b)&(sha1->c)) | ((sha1->b)&(sha1->d)) | ((sha1->c)&(sha1->d));
k = 0x8F1BBCDC;
}
else if((j>=60) && (j<80))
{
sha1->f = (sha1->b)^(sha1->c)^(sha1->d);
k = 0xCA62C1D6;
}
temp = ROTL(5,(sha1->a)) + (sha1->f) + (sha1->e) + k + sha1->w[j];
sha1->e = (sha1->d);
sha1->d = (sha1->c);
sha1->c = ROTL(30,(sha1->b));
sha1->b = (sha1->a);
sha1->a = temp;
//reset temp to 0 to be in safe side only, not mandatory.
temp =0x00;
}
// append to total hash.
sha1->digest[0] += sha1->a;
sha1->digest[1] += sha1->b;
sha1->digest[2] += sha1->c;
sha1->digest[3] += sha1->d;
sha1->digest[4] += sha1->e;
//since we used 512bit size block per each pass, let us update the buffer pointer accordingly.
buffer = buffer+64;
}
free(pbuffer);
return 0;
}
static void int2ch4(int intVal,unsigned char *result)
{
result[0]= (unsigned char)((intVal>>24) & 0x000000ff);
result[1]= (unsigned char)((intVal>>16) & 0x000000ff);
result[2]= (unsigned char)((intVal>> 8) & 0x000000ff);
result[3]= (unsigned char)((intVal>> 0) & 0x000000ff);
}
static int sha1_bin (lua_State *L) {
const void * msg = NULL;
size_t len =0;
if( lua_gettop(L) != 1 ){
return 0;
}
if( lua_isnil(L,1) ) {
msg = NULL;
len =0;
}else{
msg=luaL_checklstring(L,1,&len);
}
struct sha tmpsha;
calculate_sha1( &tmpsha, msg, (uint32_t)len);
unsigned char tmpret[SHA1SIZE+8];
memset(tmpret,0,SHA1SIZE+8);
int i=0;
for ( i=0; i<5; i++)
{
int2ch4(tmpsha.digest[i], tmpret+i*4);
}
lua_pushlstring(L, (char *)tmpret, SHA1SIZE);
return 1;
}
static unsigned int num_escape_sql_str(unsigned char *dst, unsigned char *src, size_t size)
{
unsigned int n =0;
while (size) {
/* the highest bit of all the UTF-8 chars
* is always 1 */
if ((*src & 0x80) == 0) {
switch (*src) {
case '\0':
case '\b':
case '\n':
case '\r':
case '\t':
case 26: /* \z */
case '\\':
case '\'':
case '"':
n++;
break;
default:
break;
}
}
src++;
size--;
}
return n;
}
static unsigned char*
escape_sql_str(unsigned char *dst, unsigned char *src, size_t size)
{
while (size) {
if ((*src & 0x80) == 0) {
switch (*src) {
case '\0':
*dst++ = '\\';
*dst++ = '0';
break;
case '\b':
*dst++ = '\\';
*dst++ = 'b';
break;
case '\n':
*dst++ = '\\';
*dst++ = 'n';
break;
case '\r':
*dst++ = '\\';
*dst++ = 'r';
break;
case '\t':
*dst++ = '\\';
*dst++ = 't';
break;
case 26:
*dst++ = '\\';
*dst++ = 'z';
break;
case '\\':
*dst++ = '\\';
*dst++ = '\\';
break;
case '\'':
*dst++ = '\\';
*dst++ = '\'';
break;
case '"':
*dst++ = '\\';
*dst++ = '"';
break;
default:
*dst++ = *src;
break;
}
} else {
*dst++ = *src;
}
src++;
size--;
} /* while (size) */
return dst;
}
static int
quote_sql_str(lua_State *L)
{
size_t len, dlen, escape;
unsigned char *p;
unsigned char *src, *dst;
if (lua_gettop(L) != 1) {
return luaL_error(L, "expecting one argument");
}
src = (unsigned char *) luaL_checklstring(L, 1, &len);
if (len == 0) {
dst = (unsigned char *) "''";
dlen = sizeof("''") - 1;
lua_pushlstring(L, (char *) dst, dlen);
return 1;
}
escape = num_escape_sql_str(NULL, src, len);
dlen = sizeof("''") - 1 + len + escape;
p = lua_newuserdata(L, dlen);
dst = p;
*p++ = '\'';
if (escape == 0) {
memcpy(p, src, len);
p+=len;
} else {
p = (unsigned char *) escape_sql_str(p, src, len);
}
*p++ = '\'';
if (p != dst + dlen) {
return luaL_error(L, "quote sql string error");
}
lua_pushlstring(L, (char *) dst, p - dst);
return 1;
}
static struct luaL_Reg mysqlauxlib[] = {
{"sha1_bin", sha1_bin},
{"quote_sql_str",quote_sql_str},
{NULL, NULL}
};
int luaopen_mysqlaux_c (lua_State *L) {
lua_newtable(L);
luaL_setfuncs(L, mysqlauxlib, 0);
return 1;
}

View File

@@ -403,7 +403,7 @@ static int traverseephemeron (global_State *g, Table *h) {
reallymarkobject(g, gcvalue(gval(n))); /* mark it now */ reallymarkobject(g, gcvalue(gval(n))); /* mark it now */
} }
} }
if (prop) if (g->gcstate != GCSatomic || prop)
linktable(h, &g->ephemeron); /* have to propagate again */ linktable(h, &g->ephemeron); /* have to propagate again */
else if (hasclears) /* does table have white keys? */ else if (hasclears) /* does table have white keys? */
linktable(h, &g->allweak); /* may have to clean white keys */ linktable(h, &g->allweak); /* may have to clean white keys */

View File

@@ -1,3 +1,21 @@
v0.9.0 (2014-11-17)
-----------
* Add UDP support
* Add IPv6 support
* socket send package can define a release method
* dispatch read before write in epoll
* remove snax queue mode
* Fix a bug in big-endian architecture
v0.8.1 (2014-11-3)
-----------
* Send to an invalid remote service will raise an error
* Bugifx: socket open address string
* Remove sha1 from mysqlaux
* merge lua and sproto bugfix , use crypt lib instead
* Fix a memory leak in socket
* minor bugfix in http module
v0.8.0 (2014-10-27) v0.8.0 (2014-10-27)
----------- -----------
* Add mysql client driver * Add mysql client driver

View File

@@ -123,7 +123,7 @@ $(LUA_CLIB_PATH)/sproto.so : lualib-src/sproto/sproto.c lualib-src/sproto/lsprot
$(LUA_CLIB_PATH)/lpeg.so : 3rd/lpeg/lpcap.c 3rd/lpeg/lpcode.c 3rd/lpeg/lpprint.c 3rd/lpeg/lptree.c 3rd/lpeg/lpvm.c | $(LUA_CLIB_PATH) $(LUA_CLIB_PATH)/lpeg.so : 3rd/lpeg/lpcap.c 3rd/lpeg/lpcode.c 3rd/lpeg/lpprint.c 3rd/lpeg/lptree.c 3rd/lpeg/lpvm.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) -I3rd/lpeg $^ -o $@ $(CC) $(CFLAGS) $(SHARED) -I3rd/lpeg $^ -o $@
$(LUA_CLIB_PATH)/mysqlaux.so : 3rd/lua-mysqlaux/lua_mysqlaux.c | $(LUA_CLIB_PATH) $(LUA_CLIB_PATH)/mysqlaux.so : lualib-src/lua_mysqlaux.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) $^ -o $@ $(CC) $(CFLAGS) $(SHARED) $^ -o $@
clean : clean :

View File

@@ -66,6 +66,7 @@ skynet.start(function()
end end
local balance = 1 local balance = 1
local id = socket.listen("0.0.0.0", 8001) local id = socket.listen("0.0.0.0", 8001)
skynet.error("Listen web port 8001")
socket.start(id , function(id, addr) socket.start(id , function(id, addr)
skynet.error(string.format("%s connected, pass it to agent :%08x", addr, agent[balance])) skynet.error(string.format("%s connected, pass it to agent :%08x", addr, agent[balance]))
skynet.send(agent[balance], "lua", id) skynet.send(agent[balance], "lua", id)

View File

@@ -240,19 +240,12 @@ filter_data_(lua_State *L, int fd, uint8_t * buffer, int size) {
buffer += need; buffer += need;
size -= need; size -= need;
if (size == 0) { if (size == 0) {
if (q == NULL || q->head == q->tail ) { lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
lua_pushvalue(L, lua_upvalueindex(TYPE_DATA)); lua_pushinteger(L, fd);
lua_pushinteger(L, fd); lua_pushlightuserdata(L, uc->pack.buffer);
lua_pushlightuserdata(L, uc->pack.buffer); lua_pushinteger(L, uc->pack.size);
lua_pushinteger(L, uc->pack.size); skynet_free(uc);
skynet_free(uc); return 5;
return 5;
}
else{
push_data(L, fd, uc->pack.buffer, uc->pack.size, 0);
skynet_free(uc);
return 1;
}
} }
// more data // more data
push_data(L, fd, uc->pack.buffer, uc->pack.size, 0); push_data(L, fd, uc->pack.buffer, uc->pack.size, 0);
@@ -281,21 +274,13 @@ filter_data_(lua_State *L, int fd, uint8_t * buffer, int size) {
} }
if (size == pack_size) { if (size == pack_size) {
// just one package // just one package
if ( q == NULL || q->head == q->tail) { lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
lua_pushvalue(L, lua_upvalueindex(TYPE_DATA)); lua_pushinteger(L, fd);
lua_pushinteger(L, fd); void * result = skynet_malloc(pack_size);
void * result = skynet_malloc(pack_size); memcpy(result, buffer, size);
memcpy(result, buffer, size); lua_pushlightuserdata(L, result);
lua_pushlightuserdata(L, result); lua_pushinteger(L, size);
lua_pushinteger(L, size); return 5;
return 5;
}
else{
push_data(L, fd, buffer, pack_size, 1);
buffer += pack_size;
size -= pack_size;
return 1;
}
} }
// more data // more data
push_data(L, fd, buffer, pack_size, 1); push_data(L, fd, buffer, pack_size, 1);
@@ -317,9 +302,9 @@ filter_data(lua_State *L, int fd, uint8_t * buffer, int size) {
} }
static void static void
pushstring(lua_State *L, const char * msg) { pushstring(lua_State *L, const char * msg, int size) {
if (msg) { if (msg) {
lua_pushstring(L, msg); lua_pushlstring(L, msg, size);
} else { } else {
lua_pushliteral(L, ""); lua_pushliteral(L, "");
} }
@@ -365,12 +350,12 @@ lfilter(lua_State *L) {
lua_pushvalue(L, lua_upvalueindex(TYPE_OPEN)); lua_pushvalue(L, lua_upvalueindex(TYPE_OPEN));
// ignore listen id (message->id); // ignore listen id (message->id);
lua_pushinteger(L, message->ud); lua_pushinteger(L, message->ud);
pushstring(L, buffer); pushstring(L, buffer, size);
return 4; return 4;
case SKYNET_SOCKET_TYPE_ERROR: case SKYNET_SOCKET_TYPE_ERROR:
lua_pushvalue(L, lua_upvalueindex(TYPE_ERROR)); lua_pushvalue(L, lua_upvalueindex(TYPE_ERROR));
lua_pushinteger(L, message->id); lua_pushinteger(L, message->id);
pushstring(L, buffer); pushstring(L, buffer, size);
return 4; return 4;
default: default:
// never get here // never get here

View File

@@ -179,7 +179,7 @@ wb_string(struct write_block *wb, const char *str, int len) {
wb_push(wb, str, len); wb_push(wb, str, len);
} }
} else { } else {
int n; uint8_t n;
if (len < 0x10000) { if (len < 0x10000) {
n = COMBINE_TYPE(TYPE_LONG_STRING, 2); n = COMBINE_TYPE(TYPE_LONG_STRING, 2);
wb_push(wb, &n, 1); wb_push(wb, &n, 1);

170
lualib-src/lua_mysqlaux.c Executable file
View File

@@ -0,0 +1,170 @@
//
// lua_mysqlaux.c
//
// Created by changfeng on 6/17/14.
// Copyright (c) 2014 changfeng. All rights reserved.
//
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <lua.h>
#include <lauxlib.h>
static unsigned int num_escape_sql_str(unsigned char *dst, unsigned char *src, size_t size)
{
unsigned int n =0;
while (size) {
/* the highest bit of all the UTF-8 chars
* is always 1 */
if ((*src & 0x80) == 0) {
switch (*src) {
case '\0':
case '\b':
case '\n':
case '\r':
case '\t':
case 26: /* \Z */
case '\\':
case '\'':
case '"':
n++;
break;
default:
break;
}
}
src++;
size--;
}
return n;
}
static unsigned char*
escape_sql_str(unsigned char *dst, unsigned char *src, size_t size)
{
while (size) {
if ((*src & 0x80) == 0) {
switch (*src) {
case '\0':
*dst++ = '\\';
*dst++ = '0';
break;
case '\b':
*dst++ = '\\';
*dst++ = 'b';
break;
case '\n':
*dst++ = '\\';
*dst++ = 'n';
break;
case '\r':
*dst++ = '\\';
*dst++ = 'r';
break;
case '\t':
*dst++ = '\\';
*dst++ = 't';
break;
case 26:
*dst++ = '\\';
*dst++ = 'Z';
break;
case '\\':
*dst++ = '\\';
*dst++ = '\\';
break;
case '\'':
*dst++ = '\\';
*dst++ = '\'';
break;
case '"':
*dst++ = '\\';
*dst++ = '"';
break;
default:
*dst++ = *src;
break;
}
} else {
*dst++ = *src;
}
src++;
size--;
} /* while (size) */
return dst;
}
static int
quote_sql_str(lua_State *L)
{
size_t len, dlen, escape;
unsigned char *p;
unsigned char *src, *dst;
if (lua_gettop(L) != 1) {
return luaL_error(L, "expecting one argument");
}
src = (unsigned char *) luaL_checklstring(L, 1, &len);
if (len == 0) {
dst = (unsigned char *) "''";
dlen = sizeof("''") - 1;
lua_pushlstring(L, (char *) dst, dlen);
return 1;
}
escape = num_escape_sql_str(NULL, src, len);
dlen = sizeof("''") - 1 + len + escape;
p = lua_newuserdata(L, dlen);
dst = p;
*p++ = '\'';
if (escape == 0) {
memcpy(p, src, len);
p+=len;
} else {
p = (unsigned char *) escape_sql_str(p, src, len);
}
*p++ = '\'';
if (p != dst + dlen) {
return luaL_error(L, "quote sql string error");
}
lua_pushlstring(L, (char *) dst, p - dst);
return 1;
}
static struct luaL_Reg mysqlauxlib[] = {
{"quote_sql_str",quote_sql_str},
{NULL, NULL}
};
int luaopen_mysqlaux_c (lua_State *L) {
lua_newtable(L);
luaL_setfuncs(L, mysqlauxlib, 0);
return 1;
}

View File

@@ -19,17 +19,17 @@
*/ */
LUALIB_API void luaL_setfuncs (lua_State *L, const luaL_Reg *l, int nup) { LUALIB_API void luaL_setfuncs (lua_State *L, const luaL_Reg *l, int nup) {
#ifdef luaL_checkversion #ifdef luaL_checkversion
luaL_checkversion(L); luaL_checkversion(L);
#endif #endif
luaL_checkstack(L, nup, "too many upvalues"); luaL_checkstack(L, nup, "too many upvalues");
for (; l->name != NULL; l++) { /* fill the table with given functions */ for (; l->name != NULL; l++) { /* fill the table with given functions */
int i; int i;
for (i = 0; i < nup; i++) /* copy upvalues to the top */ for (i = 0; i < nup; i++) /* copy upvalues to the top */
lua_pushvalue(L, -nup); lua_pushvalue(L, -nup);
lua_pushcclosure(L, l->func, nup); /* closure with those upvalues */ lua_pushcclosure(L, l->func, nup); /* closure with those upvalues */
lua_setfield(L, -(nup + 2), l->name); lua_setfield(L, -(nup + 2), l->name);
} }
lua_pop(L, nup); /* remove upvalues */ lua_pop(L, nup); /* remove upvalues */
} }
#define luaL_newlibtable(L,l) \ #define luaL_newlibtable(L,l) \
@@ -62,18 +62,20 @@ ldeleteproto(lua_State *L) {
static int static int
lquerytype(lua_State *L) { lquerytype(lua_State *L) {
const char * type_name;
struct sproto *sp = lua_touserdata(L,1); struct sproto *sp = lua_touserdata(L,1);
struct sproto_type *st;
if (sp == NULL) { if (sp == NULL) {
return luaL_argerror(L, 1, "Need a sproto object"); return luaL_argerror(L, 1, "Need a sproto object");
} }
const char * typename = luaL_checkstring(L,2); type_name = luaL_checkstring(L,2);
struct sproto_type *st = sproto_type(sp, typename); st = sproto_type(sp, type_name);
if (st) { if (st) {
lua_pushlightuserdata(L, st); lua_pushlightuserdata(L, st);
return 1; return 1;
} }
return luaL_error(L, "type %s not found", typename); return luaL_error(L, "type %s not found", type_name);
} }
struct encode_ud { struct encode_ud {
@@ -119,9 +121,10 @@ encode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
switch (type) { switch (type) {
case SPROTO_TINTEGER: { case SPROTO_TINTEGER: {
lua_Integer v = luaL_checkinteger(L, -1); lua_Integer v = luaL_checkinteger(L, -1);
lua_Integer vh;
lua_pop(L,1); lua_pop(L,1);
// notice: in lua 5.2, lua_Integer maybe 52bit // notice: in lua 5.2, lua_Integer maybe 52bit
lua_Integer vh = v >> 31; vh = v >> 31;
if (vh == 0 || vh == -1) { if (vh == 0 || vh == -1) {
*(uint32_t *)value = (uint32_t)v; *(uint32_t *)value = (uint32_t)v;
return 4; return 4;
@@ -148,13 +151,14 @@ encode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
} }
case SPROTO_TSTRUCT: { case SPROTO_TSTRUCT: {
struct encode_ud sub; struct encode_ud sub;
int r;
sub.L = L; sub.L = L;
sub.st = st; sub.st = st;
sub.tbl_index = lua_gettop(L); sub.tbl_index = lua_gettop(L);
sub.array_tag = NULL; sub.array_tag = NULL;
sub.array_index = 0; sub.array_index = 0;
sub.deep = self->deep + 1; sub.deep = self->deep + 1;
int r = sproto_encode(st, value, length, encode, &sub); r = sproto_encode(st, value, length, encode, &sub);
lua_pop(L,1); lua_pop(L,1);
return r; return r;
} }
@@ -165,6 +169,7 @@ encode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
static void * static void *
expand_buffer(lua_State *L, int osz, int nsz) { expand_buffer(lua_State *L, int osz, int nsz) {
void *output;
do { do {
osz *= 2; osz *= 2;
} while (osz < nsz); } while (osz < nsz);
@@ -172,7 +177,7 @@ expand_buffer(lua_State *L, int osz, int nsz) {
luaL_error(L, "object is too large (>%d)", ENCODE_MAXSIZE); luaL_error(L, "object is too large (>%d)", ENCODE_MAXSIZE);
return NULL; return NULL;
} }
void *output = lua_newuserdata(L, osz); output = lua_newuserdata(L, osz);
lua_replace(L, lua_upvalueindex(1)); lua_replace(L, lua_upvalueindex(1));
lua_pushinteger(L, osz); lua_pushinteger(L, osz);
lua_replace(L, lua_upvalueindex(2)); lua_replace(L, lua_upvalueindex(2));
@@ -188,6 +193,7 @@ expand_buffer(lua_State *L, int osz, int nsz) {
*/ */
static int static int
lencode(lua_State *L) { lencode(lua_State *L) {
struct encode_ud self;
void * buffer = lua_touserdata(L, lua_upvalueindex(1)); void * buffer = lua_touserdata(L, lua_upvalueindex(1));
int sz = lua_tointeger(L, lua_upvalueindex(2)); int sz = lua_tointeger(L, lua_upvalueindex(2));
@@ -197,7 +203,6 @@ lencode(lua_State *L) {
} }
luaL_checktype(L, 2, LUA_TTABLE); luaL_checktype(L, 2, LUA_TTABLE);
luaL_checkstack(L, ENCODE_DEEPLEVEL + 8, NULL); luaL_checkstack(L, ENCODE_DEEPLEVEL + 8, NULL);
struct encode_ud self;
self.L = L; self.L = L;
self.st = st; self.st = st;
self.tbl_index = 2; self.tbl_index = 2;
@@ -261,15 +266,16 @@ decode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
break; break;
} }
case SPROTO_TSTRUCT: { case SPROTO_TSTRUCT: {
lua_newtable(L);
struct decode_ud sub; struct decode_ud sub;
int r;
lua_newtable(L);
sub.L = L; sub.L = L;
sub.result_index = lua_gettop(L); sub.result_index = lua_gettop(L);
sub.deep = self->deep + 1; sub.deep = self->deep + 1;
sub.array_index = 0; sub.array_index = 0;
sub.array_tag = NULL; sub.array_tag = NULL;
int r = sproto_decode(st, value, length, decode, &sub); r = sproto_decode(st, value, length, decode, &sub);
if (r < 0 || r != length) if (r < 0 || r != length)
return r; return r;
lua_settop(L, sub.result_index); lua_settop(L, sub.result_index);
@@ -312,22 +318,25 @@ getbuffer(lua_State *L, int index, size_t *sz) {
static int static int
ldecode(lua_State *L) { ldecode(lua_State *L) {
struct sproto_type * st = lua_touserdata(L, 1); struct sproto_type * st = lua_touserdata(L, 1);
const void * buffer;
struct decode_ud self;
size_t sz;
int r;
if (st == NULL) { if (st == NULL) {
return luaL_argerror(L, 1, "Need a sproto_type object"); return luaL_argerror(L, 1, "Need a sproto_type object");
} }
size_t sz=0; sz = 0;
const void * buffer = getbuffer(L, 2, &sz); buffer = getbuffer(L, 2, &sz);
if (!lua_istable(L, -1)) { if (!lua_istable(L, -1)) {
lua_newtable(L); lua_newtable(L);
} }
luaL_checkstack(L, ENCODE_DEEPLEVEL*2 + 8, NULL); luaL_checkstack(L, ENCODE_DEEPLEVEL*2 + 8, NULL);
struct decode_ud self;
self.L = L; self.L = L;
self.result_index = lua_gettop(L); self.result_index = lua_gettop(L);
self.array_index = 0; self.array_index = 0;
self.array_tag = NULL; self.array_tag = NULL;
self.deep = 0; self.deep = 0;
int r = sproto_decode(st, buffer, (int)sz, decode, &self); r = sproto_decode(st, buffer, (int)sz, decode, &self);
if (r < 0) { if (r < 0) {
return luaL_error(L, "decode error"); return luaL_error(L, "decode error");
} }
@@ -359,11 +368,12 @@ lpack(lua_State *L) {
// the worst-case space overhead of packing is 2 bytes per 2 KiB of input (256 words = 2KiB). // the worst-case space overhead of packing is 2 bytes per 2 KiB of input (256 words = 2KiB).
size_t maxsz = (sz + 2047) / 2048 * 2 + sz; size_t maxsz = (sz + 2047) / 2048 * 2 + sz;
void * output = lua_touserdata(L, lua_upvalueindex(1)); void * output = lua_touserdata(L, lua_upvalueindex(1));
int bytes;
int osz = lua_tointeger(L, lua_upvalueindex(2)); int osz = lua_tointeger(L, lua_upvalueindex(2));
if (osz < maxsz) { if (osz < maxsz) {
output = expand_buffer(L, osz, maxsz); output = expand_buffer(L, osz, maxsz);
} }
int bytes = sproto_pack(buffer, sz, output, maxsz); bytes = sproto_pack(buffer, sz, output, maxsz);
if (bytes > maxsz) { if (bytes > maxsz) {
return luaL_error(L, "packing error, return size = %d", bytes); return luaL_error(L, "packing error, return size = %d", bytes);
} }
@@ -402,14 +412,18 @@ pushfunction_withbuffer(lua_State *L, const char * name, lua_CFunction func) {
static int static int
lprotocol(lua_State *L) { lprotocol(lua_State *L) {
struct sproto * sp = lua_touserdata(L, 1); struct sproto * sp = lua_touserdata(L, 1);
struct sproto_type * request;
struct sproto_type * response;
int t;
int tag;
if (sp == NULL) { if (sp == NULL) {
return luaL_argerror(L, 1, "Need a sproto_type object"); return luaL_argerror(L, 1, "Need a sproto_type object");
} }
int t = lua_type(L,2); t = lua_type(L,2);
int tag;
if (t == LUA_TNUMBER) { if (t == LUA_TNUMBER) {
const char * name;
tag = lua_tointeger(L, 2); tag = lua_tointeger(L, 2);
const char * name = sproto_protoname(sp, tag); name = sproto_protoname(sp, tag);
if (name == NULL) if (name == NULL)
return 0; return 0;
lua_pushstring(L, name); lua_pushstring(L, name);
@@ -420,13 +434,13 @@ lprotocol(lua_State *L) {
return 0; return 0;
lua_pushinteger(L, tag); lua_pushinteger(L, tag);
} }
struct sproto_type * request = sproto_protoquery(sp, tag, SPROTO_REQUEST); request = sproto_protoquery(sp, tag, SPROTO_REQUEST);
if (request == NULL) { if (request == NULL) {
lua_pushnil(L); lua_pushnil(L);
} else { } else {
lua_pushlightuserdata(L, request); lua_pushlightuserdata(L, request);
} }
struct sproto_type * response = sproto_protoquery(sp, tag, SPROTO_RESPONSE); response = sproto_protoquery(sp, tag, SPROTO_RESPONSE);
if (response == NULL) { if (response == NULL) {
lua_pushnil(L); lua_pushnil(L);
} else { } else {

View File

@@ -119,8 +119,8 @@ todword(const uint8_t *p) {
static int static int
count_array(const uint8_t * stream) { count_array(const uint8_t * stream) {
uint32_t length = todword(stream); uint32_t length = todword(stream);
stream += SIZEOF_LENGTH;
int n = 0; int n = 0;
stream += SIZEOF_LENGTH;
while (length > 0) { while (length > 0) {
uint32_t nsz; uint32_t nsz;
if (length < SIZEOF_LENGTH) if (length < SIZEOF_LENGTH)
@@ -180,6 +180,10 @@ static const uint8_t *
import_field(struct sproto *s, struct field *f, const uint8_t * stream) { import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
uint32_t sz; uint32_t sz;
const uint8_t * result; const uint8_t * result;
int fn;
int i;
int array = 0;
int tag = -1;
f->tag = -1; f->tag = -1;
f->type = -1; f->type = -1;
f->name = NULL; f->name = NULL;
@@ -188,13 +192,10 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
sz = todword(stream); sz = todword(stream);
stream += SIZEOF_LENGTH; stream += SIZEOF_LENGTH;
result = stream + sz; result = stream + sz;
int fn = struct_field(stream, sz); fn = struct_field(stream, sz);
if (fn < 0) if (fn < 0)
return NULL; return NULL;
stream += SIZEOF_HEADER; stream += SIZEOF_HEADER;
int i;
int array = 0;
int tag = -1;
for (i=0;i<fn;i++) { for (i=0;i<fn;i++) {
int value; int value;
++tag; ++tag;
@@ -203,7 +204,7 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
tag+= value/2; tag+= value/2;
continue; continue;
} }
if (tag == 0) { // name if (tag == 0) { // name
if (value != 0) if (value != 0)
return NULL; return NULL;
f->name = import_string(s, stream + fn * SIZEOF_FIELD); f->name = import_string(s, stream + fn * SIZEOF_FIELD);
@@ -213,12 +214,12 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
return NULL; return NULL;
value = value/2 - 1; value = value/2 - 1;
switch(tag) { switch(tag) {
case 1: // buildin case 1: // buildin
if (value >= SPROTO_TSTRUCT) if (value >= SPROTO_TSTRUCT)
return NULL; // invalid buildin type return NULL; // invalid buildin type
f->type = value; f->type = value;
break; break;
case 2: // type index case 2: // type index
if (value >= s->type_n) if (value >= s->type_n)
return NULL; // invalid type index return NULL; // invalid type index
if (f->type >= 0) if (f->type >= 0)
@@ -226,10 +227,10 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
f->type = SPROTO_TSTRUCT; f->type = SPROTO_TSTRUCT;
f->st = &s->type[value]; f->st = &s->type[value];
break; break;
case 3: // tag case 3: // tag
f->tag = value; f->tag = value;
break; break;
case 4: // array case 4: // array
if (value) if (value)
array = SPROTO_TARRAY; array = SPROTO_TARRAY;
break; break;
@@ -248,10 +249,10 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
.type { .type {
.field { .field {
name 0 : string name 0 : string
buildin 1 : integer buildin 1 : integer
type 2 : integer type 2 : integer
tag 3 : integer tag 3 : integer
array 4 : boolean array 4 : boolean
} }
name 0 : string name 0 : string
fields 1 : *field fields 1 : *field
@@ -259,11 +260,16 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
*/ */
static const uint8_t * static const uint8_t *
import_type(struct sproto *s, struct sproto_type *t, const uint8_t * stream) { import_type(struct sproto *s, struct sproto_type *t, const uint8_t * stream) {
const uint8_t * result;
uint32_t sz = todword(stream); uint32_t sz = todword(stream);
int i; int i;
int fn;
int n;
int maxn;
int last;
stream += SIZEOF_LENGTH; stream += SIZEOF_LENGTH;
const uint8_t * result = stream + sz; result = stream + sz;
int fn = struct_field(stream, sz); fn = struct_field(stream, sz);
if (fn <= 0 || fn > 2) if (fn <= 0 || fn > 2)
return NULL; return NULL;
for (i=0;i<fn*SIZEOF_FIELD;i+=SIZEOF_FIELD) { for (i=0;i<fn*SIZEOF_FIELD;i+=SIZEOF_FIELD) {
@@ -279,20 +285,21 @@ import_type(struct sproto *s, struct sproto_type *t, const uint8_t * stream) {
return result; return result;
} }
stream += todword(stream)+SIZEOF_LENGTH; // second data stream += todword(stream)+SIZEOF_LENGTH; // second data
int n = count_array(stream); n = count_array(stream);
if (n<0) if (n<0)
return NULL; return NULL;
stream += SIZEOF_LENGTH; stream += SIZEOF_LENGTH;
int maxn = n; maxn = n;
int last = -1; last = -1;
t->n = n; t->n = n;
t->f = pool_alloc(&s->memory, sizeof(struct field) * n); t->f = pool_alloc(&s->memory, sizeof(struct field) * n);
for (i=0;i<n;i++) { for (i=0;i<n;i++) {
int tag;
struct field *f = &t->f[i]; struct field *f = &t->f[i];
stream = import_field(s, f, stream); stream = import_field(s, f, stream);
if (stream == NULL) if (stream == NULL)
return NULL; return NULL;
int tag = f->tag; tag = f->tag;
if (tag < last) if (tag < last)
return NULL; // tag must in ascending order return NULL; // tag must in ascending order
if (tag > last+1) { if (tag > last+1) {
@@ -312,24 +319,27 @@ import_type(struct sproto *s, struct sproto_type *t, const uint8_t * stream) {
/* /*
.protocol { .protocol {
name 0 : string name 0 : string
tag 1 : integer tag 1 : integer
request 2 : integer request 2 : integer
response 3 : integer response 3 : integer
} }
*/ */
static const uint8_t * static const uint8_t *
import_protocol(struct sproto *s, struct protocol *p, const uint8_t * stream) { import_protocol(struct sproto *s, struct protocol *p, const uint8_t * stream) {
const uint8_t * result;
uint32_t sz = todword(stream); uint32_t sz = todword(stream);
int fn;
int i;
int tag;
stream += SIZEOF_LENGTH; stream += SIZEOF_LENGTH;
const uint8_t * result = stream + sz; result = stream + sz;
int fn = struct_field(stream, sz); fn = struct_field(stream, sz);
stream += SIZEOF_HEADER; stream += SIZEOF_HEADER;
p->name = NULL; p->name = NULL;
p->tag = -1; p->tag = -1;
p->p[SPROTO_REQUEST] = NULL; p->p[SPROTO_REQUEST] = NULL;
p->p[SPROTO_RESPONSE] = NULL; p->p[SPROTO_RESPONSE] = NULL;
int i; tag = 0;
int tag = 0;
for (i=0;i<fn;i++,tag++) { for (i=0;i<fn;i++,tag++) {
int value = toword(stream + SIZEOF_FIELD * i); int value = toword(stream + SIZEOF_FIELD * i);
if (value & 1) { if (value & 1) {
@@ -338,24 +348,24 @@ import_protocol(struct sproto *s, struct protocol *p, const uint8_t * stream) {
} }
value = value/2 - 1; value = value/2 - 1;
switch (i) { switch (i) {
case 0: // name case 0: // name
if (value != -1) { if (value != -1) {
return NULL; return NULL;
} }
p->name = import_string(s, stream + SIZEOF_FIELD *fn); p->name = import_string(s, stream + SIZEOF_FIELD *fn);
break; break;
case 1: // tag case 1: // tag
if (value < 0) { if (value < 0) {
return NULL; return NULL;
} }
p->tag = value; p->tag = value;
break; break;
case 2: // request case 2: // request
if (value < 0 || value>=s->type_n) if (value < 0 || value>=s->type_n)
return NULL; return NULL;
p->p[SPROTO_REQUEST] = &s->type[value]; p->p[SPROTO_REQUEST] = &s->type[value];
break; break;
case 3: // response case 3: // response
if (value < 0 || value>=s->type_n) if (value < 0 || value>=s->type_n)
return NULL; return NULL;
p->p[SPROTO_RESPONSE] = &s->type[value]; p->p[SPROTO_RESPONSE] = &s->type[value];
@@ -374,22 +384,23 @@ import_protocol(struct sproto *s, struct protocol *p, const uint8_t * stream) {
static struct sproto * static struct sproto *
create_from_bundle(struct sproto *s, const uint8_t * stream, size_t sz) { create_from_bundle(struct sproto *s, const uint8_t * stream, size_t sz) {
const uint8_t * content;
const uint8_t * typedata = NULL;
const uint8_t * protocoldata = NULL;
int fn = struct_field(stream, sz); int fn = struct_field(stream, sz);
int i;
if (fn < 0) if (fn < 0)
return NULL; return NULL;
stream += SIZEOF_HEADER; stream += SIZEOF_HEADER;
content = stream + fn*SIZEOF_FIELD;
const uint8_t * content = stream + fn*SIZEOF_FIELD;
const uint8_t * typedata = NULL;
const uint8_t * protocoldata = NULL;
int i;
for (i=0;i<fn;i++) { for (i=0;i<fn;i++) {
int value = toword(stream + i*SIZEOF_FIELD); int value = toword(stream + i*SIZEOF_FIELD);
int n;
if (value != 0) if (value != 0)
return NULL; return NULL;
int n = count_array(content); n = count_array(content);
if (n<0) if (n<0)
return NULL; return NULL;
if (i == 0) { if (i == 0) {
@@ -423,8 +434,9 @@ create_from_bundle(struct sproto *s, const uint8_t * stream, size_t sz) {
struct sproto * struct sproto *
sproto_create(const void * proto, size_t sz) { sproto_create(const void * proto, size_t sz) {
struct pool mem; struct pool mem;
struct sproto * s;
pool_init(&mem); pool_init(&mem);
struct sproto * s = pool_alloc(&mem, sizeof(*s)); s = pool_alloc(&mem, sizeof(*s));
if (s == NULL) if (s == NULL)
return NULL; return NULL;
memset(s, 0, sizeof(*s)); memset(s, 0, sizeof(*s));
@@ -445,34 +457,35 @@ sproto_release(struct sproto * s) {
void void
sproto_dump(struct sproto *s) { sproto_dump(struct sproto *s) {
int i,j;
static const char * buildin[] = { static const char * buildin[] = {
"integer", "integer",
"boolean", "boolean",
"string", "string",
}; };
printf("=== %d types ===\n", s->type_n); printf("=== %d types ===\n", s->type_n);
int i,j;
for (i=0;i<s->type_n;i++) { for (i=0;i<s->type_n;i++) {
struct sproto_type *t = &s->type[i]; struct sproto_type *t = &s->type[i];
printf("%s\n", t->name); printf("%s\n", t->name);
for (j=0;j<t->n;j++) { for (j=0;j<t->n;j++) {
char array[2] = { 0, 0 }; char array[2] = { 0, 0 };
const char * typename = NULL; const char * type_name = NULL;
struct field *f = &t->f[j]; struct field *f = &t->f[j];
if (f->type & SPROTO_TARRAY) { if (f->type & SPROTO_TARRAY) {
array[0] = '*'; array[0] = '*';
} else { } else {
array[0] = 0; array[0] = 0;
} }
int t = f->type & ~SPROTO_TARRAY; {
if (t == SPROTO_TSTRUCT) { int t = f->type & ~SPROTO_TARRAY;
typename = f->st->name; if (t == SPROTO_TSTRUCT) {
} else { type_name = f->st->name;
assert(t<SPROTO_TSTRUCT); } else {
typename = buildin[t]; assert(t<SPROTO_TSTRUCT);
type_name = buildin[t];
}
} }
printf("\t%s (%d) %s%s\n", f->name, f->tag, array, type_name);
printf("\t%s (%d) %s%s\n", f->name, f->tag, array, typename);
} }
} }
printf("=== %d protocol ===\n", s->protocol_n); printf("=== %d protocol ===\n", s->protocol_n);
@@ -518,10 +531,11 @@ query_proto(struct sproto *sp, int tag) {
struct sproto_type * struct sproto_type *
sproto_protoquery(struct sproto *sp, int proto, int what) { sproto_protoquery(struct sproto *sp, int proto, int what) {
struct protocol * p;
if (what <0 || what >1) { if (what <0 || what >1) {
return NULL; return NULL;
} }
struct protocol * p = query_proto(sp, proto); p = query_proto(sp, proto);
if (p) { if (p) {
return p->p[what]; return p->p[what];
} }
@@ -555,13 +569,15 @@ sproto_name(struct sproto_type * st) {
static struct field * static struct field *
findtag(struct sproto_type *st, int tag) { findtag(struct sproto_type *st, int tag) {
int begin, end;
if (st->base >=0 ) { if (st->base >=0 ) {
tag -= st->base; tag -= st->base;
if (tag < 0 || tag >= st->n) if (tag < 0 || tag >= st->n)
return NULL; return NULL;
return &st->f[tag]; return &st->f[tag];
} }
int begin = 0, end = st->n; begin = 0;
end = st->n;
while (begin < end) { while (begin < end) {
int mid = (begin+end)/2; int mid = (begin+end)/2;
struct field *f = &st->f[mid]; struct field *f = &st->f[mid];
@@ -580,7 +596,7 @@ findtag(struct sproto_type *st, int tag) {
// encode & decode // encode & decode
// sproto_callback(void *ud, int tag, int type, struct sproto_type *, void *value, int length) // sproto_callback(void *ud, int tag, int type, struct sproto_type *, void *value, int length)
// return size, -1 means error // return size, -1 means error
static inline int static inline int
fill_size(uint8_t * data, int sz) { fill_size(uint8_t * data, int sz) {
@@ -623,18 +639,20 @@ encode_uint64(uint64_t v, uint8_t * data, int size) {
static int static int
encode_string(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) { encode_string(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) {
int sz;
if (size < SIZEOF_LENGTH) if (size < SIZEOF_LENGTH)
return -1; return -1;
int sz = cb(ud, f->name, SPROTO_TSTRING, 0, NULL, data+SIZEOF_LENGTH, size-SIZEOF_LENGTH); sz = cb(ud, f->name, SPROTO_TSTRING, 0, NULL, data+SIZEOF_LENGTH, size-SIZEOF_LENGTH);
return fill_size(data, sz); return fill_size(data, sz);
} }
static int static int
encode_struct(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) { encode_struct(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) {
int sz;
if (size < SIZEOF_LENGTH) { if (size < SIZEOF_LENGTH) {
return -1; return -1;
} }
int sz = cb(ud, f->name, SPROTO_TSTRUCT, 0, f->st, data+SIZEOF_LENGTH, size-SIZEOF_LENGTH); sz = cb(ud, f->name, SPROTO_TSTRUCT, 0, f->st, data+SIZEOF_LENGTH, size-SIZEOF_LENGTH);
return fill_size(data, sz); return fill_size(data, sz);
} }
@@ -656,18 +674,21 @@ uint32_to_uint64(int negative, uint8_t *buffer) {
static uint8_t * static uint8_t *
encode_integer_array(sproto_callback cb, void *ud, struct field *f, uint8_t *buffer, int size) { encode_integer_array(sproto_callback cb, void *ud, struct field *f, uint8_t *buffer, int size) {
uint8_t * header = buffer; uint8_t * header = buffer;
int intlen;
int index;
if (size < 1) if (size < 1)
return NULL; return NULL;
buffer++; buffer++;
size--; size--;
int intlen = sizeof(uint32_t); intlen = sizeof(uint32_t);
int index = 1; index = 1;
for (;;) { for (;;) {
int sz;
union { union {
uint64_t u64; uint64_t u64;
uint32_t u32; uint32_t u32;
} u; } u;
int sz = cb(ud, f->name, SPROTO_TINTEGER, index, f->st, &u, sizeof(u)); sz = cb(ud, f->name, SPROTO_TINTEGER, index, f->st, &u, sizeof(u));
if (sz < 0) if (sz < 0)
return NULL; return NULL;
if (sz == 0) if (sz == 0)
@@ -685,24 +706,26 @@ encode_integer_array(sproto_callback cb, void *ud, struct field *f, uint8_t *buf
uint32_to_uint64(v & 0x80000000, buffer); uint32_to_uint64(v & 0x80000000, buffer);
} }
} else { } else {
uint64_t v;
if (sz != sizeof(uint64_t)) if (sz != sizeof(uint64_t))
return NULL; return NULL;
if (intlen == sizeof(uint32_t)) { if (intlen == sizeof(uint32_t)) {
int i;
// rearrange // rearrange
size -= (index-1) * sizeof(uint32_t); size -= (index-1) * sizeof(uint32_t);
if (size < sizeof(uint64_t)) if (size < sizeof(uint64_t))
return NULL; return NULL;
buffer += (index-1) * sizeof(uint32_t); buffer += (index-1) * sizeof(uint32_t);
int i;
for (i=index-2;i>=0;i--) { for (i=index-2;i>=0;i--) {
int negative;
memcpy(header+1+i*sizeof(uint64_t), header+1+i*sizeof(uint32_t), sizeof(uint32_t)); memcpy(header+1+i*sizeof(uint64_t), header+1+i*sizeof(uint32_t), sizeof(uint32_t));
int negative = header[1+i*sizeof(uint64_t)+3] & 0x80; negative = header[1+i*sizeof(uint64_t)+3] & 0x80;
uint32_to_uint64(negative, header+1+i*sizeof(uint64_t)); uint32_to_uint64(negative, header+1+i*sizeof(uint64_t));
} }
intlen = sizeof(uint64_t); intlen = sizeof(uint64_t);
} }
uint64_t v = u.u64; v = u.u64;
buffer[0] = v & 0xff; buffer[0] = v & 0xff;
buffer[1] = (v >> 8) & 0xff; buffer[1] = (v >> 8) & 0xff;
buffer[2] = (v >> 16) & 0xff; buffer[2] = (v >> 16) & 0xff;
@@ -727,12 +750,16 @@ encode_integer_array(sproto_callback cb, void *ud, struct field *f, uint8_t *buf
static int static int
encode_array(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) { encode_array(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) {
uint8_t * buffer;
int index;
int type;
int sz;
if (size < SIZEOF_LENGTH) if (size < SIZEOF_LENGTH)
return -1; return -1;
size -= SIZEOF_LENGTH; size -= SIZEOF_LENGTH;
int index = 1; index = 1;
uint8_t * buffer = data + SIZEOF_LENGTH; buffer = data + SIZEOF_LENGTH;
int type = f->type & ~SPROTO_TARRAY; type = f->type & ~SPROTO_TARRAY;
switch (type) { switch (type) {
case SPROTO_TINTEGER: case SPROTO_TINTEGER:
buffer = encode_integer_array(cb,ud,f,buffer,size); buffer = encode_integer_array(cb,ud,f,buffer,size);
@@ -757,10 +784,11 @@ encode_array(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int s
break; break;
default: default:
for (;;) { for (;;) {
int sz;
if (size < SIZEOF_LENGTH) if (size < SIZEOF_LENGTH)
return -1; return -1;
size -= SIZEOF_LENGTH; size -= SIZEOF_LENGTH;
int sz = cb(ud, f->name, type, index, f->st, buffer+SIZEOF_LENGTH, size); sz = cb(ud, f->name, type, index, f->st, buffer+SIZEOF_LENGTH, size);
if (sz < 0) if (sz < 0)
return -1; return -1;
if (sz == 0) if (sz == 0)
@@ -772,7 +800,7 @@ encode_array(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int s
} }
break; break;
} }
int sz = buffer - (data + SIZEOF_LENGTH); sz = buffer - (data + SIZEOF_LENGTH);
if (sz == 0) if (sz == 0)
return 0; return 0;
return fill_size(data, sz); return fill_size(data, sz);
@@ -783,13 +811,16 @@ sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback c
uint8_t * header = buffer; uint8_t * header = buffer;
uint8_t * data; uint8_t * data;
int header_sz = SIZEOF_HEADER + st->maxn * SIZEOF_FIELD; int header_sz = SIZEOF_HEADER + st->maxn * SIZEOF_FIELD;
int i;
int index;
int lasttag;
int datasz;
if (size < header_sz) if (size < header_sz)
return -1; return -1;
data = header + header_sz; data = header + header_sz;
size -= header_sz; size -= header_sz;
int i; index = 0;
int index = 0; lasttag = -1;
int lasttag = -1;
for (i=0;i<st->n;i++) { for (i=0;i<st->n;i++) {
struct field *f = &st->f[i]; struct field *f = &st->f[i];
int type = f->type; int type = f->type;
@@ -813,7 +844,7 @@ sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback c
if (sz == sizeof(uint32_t)) { if (sz == sizeof(uint32_t)) {
if (u.u32 < 0x7fff) { if (u.u32 < 0x7fff) {
value = (u.u32+1) * 2; value = (u.u32+1) * 2;
sz = 2; // sz can be any number > 0 sz = 2; // sz can be any number > 0
} else { } else {
sz = encode_integer(u.u32, data, size); sz = encode_integer(u.u32, data, size);
} }
@@ -837,12 +868,14 @@ sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback c
if (sz < 0) if (sz < 0)
return -1; return -1;
if (sz > 0) { if (sz > 0) {
uint8_t * record;
int tag;
if (value == 0) { if (value == 0) {
data += sz; data += sz;
size -= sz; size -= sz;
} }
uint8_t * record = header+SIZEOF_HEADER+SIZEOF_FIELD*index; record = header+SIZEOF_HEADER+SIZEOF_FIELD*index;
int tag = f->tag - lasttag - 1; tag = f->tag - lasttag - 1;
if (tag > 0) { if (tag > 0) {
// skip tag // skip tag
tag = (tag - 1) * 2 + 1; tag = (tag - 1) * 2 + 1;
@@ -862,7 +895,7 @@ sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback c
header[0] = index & 0xff; header[0] = index & 0xff;
header[1] = (index >> 8) & 0xff; header[1] = (index >> 8) & 0xff;
int datasz = data - (header + header_sz); datasz = data - (header + header_sz);
data = header + header_sz; data = header + header_sz;
if (index != st->maxn) { if (index != st->maxn) {
memmove(header + SIZEOF_HEADER + index * SIZEOF_FIELD, data, datasz); memmove(header + SIZEOF_HEADER + index * SIZEOF_FIELD, data, datasz);
@@ -909,9 +942,10 @@ decode_array(sproto_callback cb, void *ud, struct field *f, uint8_t * stream) {
stream += SIZEOF_LENGTH; stream += SIZEOF_LENGTH;
switch (type) { switch (type) {
case SPROTO_TINTEGER: { case SPROTO_TINTEGER: {
int len;
if (sz < 1) if (sz < 1)
return -1; return -1;
int len = *stream; len = *stream;
++stream; ++stream;
--sz; --sz;
if (len == sizeof(uint32_t)) { if (len == sizeof(uint32_t)) {
@@ -956,6 +990,8 @@ sproto_decode(struct sproto_type *st, const void * data, int size, sproto_callba
uint8_t * stream; uint8_t * stream;
uint8_t * datastream; uint8_t * datastream;
int fn; int fn;
int i;
int tag;
if (size < SIZEOF_HEADER) if (size < SIZEOF_HEADER)
return -1; return -1;
stream = (void *)data; stream = (void *)data;
@@ -967,17 +1003,18 @@ sproto_decode(struct sproto_type *st, const void * data, int size, sproto_callba
datastream = stream + fn * SIZEOF_FIELD; datastream = stream + fn * SIZEOF_FIELD;
size -= fn * SIZEOF_FIELD ; size -= fn * SIZEOF_FIELD ;
int i; tag = -1;
int tag = -1;
for (i=0;i<fn;i++) { for (i=0;i<fn;i++) {
++ tag; uint8_t * currentdata;
struct field * f;
int value = toword(stream + i * SIZEOF_FIELD); int value = toword(stream + i * SIZEOF_FIELD);
++ tag;
if (value & 1) { if (value & 1) {
tag += value/2; tag += value/2;
continue; continue;
} }
value = value/2 - 1; value = value/2 - 1;
uint8_t * currentdata = datastream; currentdata = datastream;
if (value < 0) { if (value < 0) {
uint32_t sz; uint32_t sz;
if (size < SIZEOF_LENGTH) if (size < SIZEOF_LENGTH)
@@ -988,7 +1025,7 @@ sproto_decode(struct sproto_type *st, const void * data, int size, sproto_callba
datastream += sz+SIZEOF_LENGTH; datastream += sz+SIZEOF_LENGTH;
size -= sz+SIZEOF_LENGTH; size -= sz+SIZEOF_LENGTH;
} }
struct field * f= findtag(st, tag); f = findtag(st, tag);
if (f == NULL) if (f == NULL)
continue; continue;
if (value < 0) { if (value < 0) {

View File

@@ -83,7 +83,7 @@ local function readall(readbytes, bodylimit)
else else
-- identity mode -- identity mode
if length then if length then
if length > bodylimit then if bodylimit and length > bodylimit then
return 413 return 413
end end
if #body >= length then if #body >= length then
@@ -130,6 +130,7 @@ local function writeall(writefunc, statuscode, bodyfunc, header)
end end
else else
writefunc("\r\n0\r\n\r\n") writefunc("\r\n0\r\n\r\n")
break
end end
end end
else else

View File

@@ -6,7 +6,7 @@
local socketchannel = require "socketchannel" local socketchannel = require "socketchannel"
local bit = require "bit32" local bit = require "bit32"
local mysqlaux = require "mysqlaux.c" local mysqlaux = require "mysqlaux.c"
local crypt = require "crypt"
local sub = string.sub local sub = string.sub
@@ -20,7 +20,7 @@ local bxor = bit.bxor
local bor = bit.bor local bor = bit.bor
local lshift = bit.lshift local lshift = bit.lshift
local rshift = bit.rshift local rshift = bit.rshift
local sha1= mysqlaux.sha1_bin local sha1= crypt.sha1
local concat = table.concat local concat = table.concat
local unpack = unpack local unpack = unpack
local setmetatable = setmetatable local setmetatable = setmetatable

View File

@@ -132,6 +132,14 @@ function snax.kill(obj, ...)
skynet_call(obj.handle, "snax", t.system.exit, ...) skynet_call(obj.handle, "snax", t.system.exit, ...)
end end
function snax.self()
return snax.bind(skynet.self(), SERVICE_NAME)
end
function snax.exit(...)
snax.kill(snax.self(), ...)
end
local function test_result(ok, ...) local function test_result(ok, ...)
if ok then if ok then
return ... return ...

View File

@@ -53,20 +53,32 @@ function gateserver.start(handler)
local MSG = {} local MSG = {}
function MSG.data(fd, msg, sz) local function dispatch_msg(fd, msg, sz)
if connection[fd] then if connection[fd] then
handler.message(fd, msg, sz) handler.message(fd, msg, sz)
else
skynet.error(string.format("Drop message from fd (%d) : %s", fd, netpack.tostring(msg,sz)))
end end
end end
function MSG.more() MSG.data = dispatch_msg
for fd, msg, sz in netpack.pop, queue do
if connection[fd] then local function dispatch_queue()
handler.message(fd, msg, sz) local fd, msg, sz = netpack.pop(queue)
if fd then
-- may dispatch even the handler.message blocked
-- If the handler.message never block, the queue should be empty, so only fork once and then exit.
skynet.fork(dispatch_queue)
dispatch_msg(fd, msg, sz)
for fd, msg, sz in netpack.pop, queue do
dispatch_msg(fd, msg, sz)
end end
end end
end end
MSG.more = dispatch_queue
function MSG.open(fd, msg) function MSG.open(fd, msg)
if client_number >= maxclient then if client_number >= maxclient then
socketdriver.close(fd) socketdriver.close(fd)

View File

@@ -90,7 +90,7 @@ local function launch_slave(auth_handler)
local function ret_pack(ok, err, ...) local function ret_pack(ok, err, ...)
if ok then if ok then
skynet.ret(skynet.pack(err, ...)) skynet.ret(skynet.pack(err, ...))
elseif err ~= socket_error then else
error(err) error(err)
end end
end end
@@ -163,6 +163,7 @@ local function launch_master(conf)
if err ~= socket_error then if err ~= socket_error then
skynet.error(string.format("invalid client (fd = %d) error = %s", fd, err)) skynet.error(string.format("invalid client (fd = %d) error = %s", fd, err))
end end
socket.start(fd)
end end
socket.close(fd) socket.close(fd)
end) end)

View File

@@ -316,6 +316,7 @@ forward_local_messsage(struct harbor *h, void *msg, int sz) {
destination = (destination & HANDLE_MASK) | ((uint32_t)h->id << HANDLE_REMOTE_SHIFT); destination = (destination & HANDLE_MASK) | ((uint32_t)h->id << HANDLE_REMOTE_SHIFT);
if (skynet_send(h->ctx, header.source, destination, type, (int)header.session, (void *)msg, sz-HEADER_COOKIE_LENGTH) < 0) { if (skynet_send(h->ctx, header.source, destination, type, (int)header.session, (void *)msg, sz-HEADER_COOKIE_LENGTH) < 0) {
skynet_send(h->ctx, destination, header.source , PTYPE_ERROR, (int)header.session, NULL, 0);
skynet_error(h->ctx, "Unknown destination :%x from :%x", destination, header.source); skynet_error(h->ctx, "Unknown destination :%x from :%x", destination, header.source);
} }
} }

View File

@@ -12,10 +12,6 @@ package.path = snax_path .. "?.lua;" .. package.path
SERVICE_NAME = snax_name SERVICE_NAME = snax_name
SERVICE_PATH = snax_path SERVICE_PATH = snax_path
local mode
local thread_id
local message_queue = {}
local init = false
local profile_table = {} local profile_table = {}
local function update_stat(name, ti) local function update_stat(name, ti)
@@ -38,57 +34,6 @@ local function dispatch(f, ...)
return skynet.pack(f(...)) return skynet.pack(f(...))
end end
local function message_dispatch()
while true do
if #message_queue==0 then
thread_id = coroutine.running()
skynet.wait()
else
local msg = table.remove(message_queue,1)
local method = msg.method
local f = method[4]
if f then
if method[2] == "accept" then
-- no return
profile.start()
local ok, data = xpcall(f, traceback, table.unpack(msg))
local ti = profile.stop()
update_stat(method[3], ti)
if not ok then
print(string.format("Error on [:%x] to [:%x] %s", msg.source, skynet.self(), tostring(data)))
end
else
profile.start()
local ok, data, size = xpcall(dispatch, traceback, f, table.unpack(msg))
local ti = profile.stop()
update_stat(method[3], ti)
if ok then
-- skynet.PTYPE_RESPONSE == 1
c.send(msg.source, 1, msg.session, data, size)
if method[2] == "system" then
init = false
skynet.exit()
break
end
else
-- Can't throw error, so print it directly
print(string.format("Error on [:%x] to [:%x] %s", msg.source, skynet.self(), tostring(data)))
c.send(msg.source, skynet.PTYPE_ERROR, msg.session, "")
end
end
end
end
end
end
local function queue( session, source, method, ...)
table.insert(message_queue, {session = session, source = source, method = method, ... })
if thread_id then
skynet.wakeup(thread_id)
thread_id = nil
end
end
local function return_f(f, ...) local function return_f(f, ...)
return skynet.ret(skynet.pack(f(...))) return skynet.ret(skynet.pack(f(...)))
end end
@@ -108,6 +53,7 @@ local function timing( method, ... )
end end
skynet.start(function() skynet.start(function()
local init = false
skynet.dispatch("snax", function ( session , source , id, ...) skynet.dispatch("snax", function ( session , source , id, ...)
local method = func[id] local method = func[id]
@@ -119,17 +65,12 @@ skynet.start(function()
elseif command == "init" then elseif command == "init" then
assert(not init, "Already init") assert(not init, "Already init")
local initfunc = method[4] or function() end local initfunc = method[4] or function() end
mode = initfunc(...) initfunc(...)
if mode == "queue" then
skynet.fork(message_dispatch)
end
skynet.ret() skynet.ret()
skynet.info_func(function() skynet.info_func(function()
return profile_table return profile_table
end) end)
init = true init = true
elseif mode == "queue" then
queue( session, source, method , ...)
else else
assert(init, "Never init") assert(init, "Never init")
assert(command == "exit") assert(command == "exit")
@@ -141,11 +82,7 @@ skynet.start(function()
end end
else else
assert(init, "Init first") assert(init, "Init first")
if mode == "queue" then timing(method, ...)
queue(session, source, method , ...)
else
timing(method, ...)
end
end end
end) end)
end) end)

View File

@@ -169,7 +169,11 @@ skynet_context_new(const char * name, const char *param) {
int int
skynet_context_newsession(struct skynet_context *ctx) { skynet_context_newsession(struct skynet_context *ctx) {
// session always be a positive number // session always be a positive number
int session = (++ctx->session_id) & 0x7fffffff; int session = ++ctx->session_id;
if (session <= 0) {
ctx->session_id = 1;
return 1;
}
return session; return session;
} }
@@ -691,7 +695,7 @@ skynet_sendname(struct skynet_context * context, uint32_t source, const char * a
if (type & PTYPE_TAG_DONTCOPY) { if (type & PTYPE_TAG_DONTCOPY) {
skynet_free(data); skynet_free(data);
} }
return session; return -1;
} }
} else { } else {
_filter_args(context, type, &session, (void **)&data, &sz); _filter_args(context, type, &session, (void **)&data, &sz);

View File

@@ -61,6 +61,7 @@ forward_message(int type, bool padding, struct socket_message * result) {
if (skynet_context_push((uint32_t)result->opaque, &message)) { if (skynet_context_push((uint32_t)result->opaque, &message)) {
// todo: report somewhere to close socket // todo: report somewhere to close socket
// don't call skynet_socket_close here (It will block mainloop) // don't call skynet_socket_close here (It will block mainloop)
skynet_free(sm->buffer);
skynet_free(sm); skynet_free(sm);
} }
} }

View File

@@ -40,8 +40,9 @@
struct write_buffer { struct write_buffer {
struct write_buffer * next; struct write_buffer * next;
char *ptr; char *ptr;
int sz;
void *buffer; void *buffer;
int sz;
bool userobject;
}; };
struct wb_list { struct wb_list {
@@ -68,6 +69,7 @@ struct socket_server {
int alloc_id; int alloc_id;
int event_n; int event_n;
int event_index; int event_index;
struct socket_object_interface soi;
struct event ev[MAX_EVENT]; struct event ev[MAX_EVENT];
struct socket slot[MAX_SOCKET]; struct socket slot[MAX_SOCKET];
char buffer[MAX_INFO]; char buffer[MAX_INFO];
@@ -137,9 +139,40 @@ union sockaddr_all {
struct sockaddr_in6 v6; struct sockaddr_in6 v6;
}; };
struct send_object {
void * buffer;
int sz;
void (*free_func)(void *);
};
#define MALLOC skynet_malloc #define MALLOC skynet_malloc
#define FREE skynet_free #define FREE skynet_free
static inline bool
send_object_init(struct socket_server *ss, struct send_object *so, void *object, int sz) {
if (sz < 0) {
so->buffer = ss->soi.buffer(object);
so->sz = ss->soi.size(object);
so->free_func = ss->soi.free;
return true;
} else {
so->buffer = object;
so->sz = sz;
so->free_func = FREE;
return false;
}
}
static inline void
write_buffer_free(struct socket_server *ss, struct write_buffer *wb) {
if (wb->userobject) {
ss->soi.free(wb->buffer);
} else {
FREE(wb->buffer);
}
FREE(wb);
}
static void static void
socket_keepalive(int fd) { socket_keepalive(int fd) {
int keepalive = 1; int keepalive = 1;
@@ -213,6 +246,7 @@ socket_server_create() {
ss->alloc_id = 0; ss->alloc_id = 0;
ss->event_n = 0; ss->event_n = 0;
ss->event_index = 0; ss->event_index = 0;
memset(&ss->soi, 0, sizeof(ss->soi));
FD_ZERO(&ss->rfds); FD_ZERO(&ss->rfds);
assert(ss->recvctrl_fd < FD_SETSIZE); assert(ss->recvctrl_fd < FD_SETSIZE);
@@ -220,13 +254,12 @@ socket_server_create() {
} }
static void static void
free_wb_list(struct wb_list *list) { free_wb_list(struct socket_server *ss, struct wb_list *list) {
struct write_buffer *wb = list->head; struct write_buffer *wb = list->head;
while (wb) { while (wb) {
struct write_buffer *tmp = wb; struct write_buffer *tmp = wb;
wb = wb->next; wb = wb->next;
FREE(tmp->buffer); write_buffer_free(ss, tmp);
FREE(tmp);
} }
list->head = NULL; list->head = NULL;
list->tail = NULL; list->tail = NULL;
@@ -242,8 +275,8 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
return; return;
} }
assert(s->type != SOCKET_TYPE_RESERVE); assert(s->type != SOCKET_TYPE_RESERVE);
free_wb_list(&s->high); free_wb_list(ss,&s->high);
free_wb_list(&s->low); free_wb_list(ss,&s->low);
if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) { if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
sp_del(ss->event_fd, s->fd); sp_del(ss->event_fd, s->fd);
} }
@@ -395,8 +428,7 @@ send_list(struct socket_server *ss, struct socket *s, struct wb_list *list, stru
break; break;
} }
list->head = tmp->next; list->head = tmp->next;
FREE(tmp->buffer); write_buffer_free(ss,tmp);
FREE(tmp);
} }
list->tail = NULL; list->tail = NULL;
@@ -469,10 +501,12 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
} }
static int static int
append_sendbuffer_(struct wb_list *s, struct request_send * request, int n) { append_sendbuffer_(struct socket_server *ss, struct wb_list *s, struct request_send * request, int n) {
struct write_buffer * buf = MALLOC(sizeof(*buf)); struct write_buffer * buf = MALLOC(sizeof(*buf));
buf->ptr = request->buffer+n; struct send_object so;
buf->sz = request->sz - n; buf->userobject = send_object_init(ss, &so, request->buffer, request->sz);
buf->ptr = so.buffer+n;
buf->sz = so.sz - n;
buf->buffer = request->buffer; buf->buffer = request->buffer;
buf->next = NULL; buf->next = NULL;
if (s->head == NULL) { if (s->head == NULL) {
@@ -487,13 +521,13 @@ append_sendbuffer_(struct wb_list *s, struct request_send * request, int n) {
} }
static inline void static inline void
append_sendbuffer(struct socket *s, struct request_send * request, int n) { append_sendbuffer(struct socket_server *ss, struct socket *s, struct request_send * request, int n) {
s->wb_size += append_sendbuffer_(&s->high, request, n); s->wb_size += append_sendbuffer_(ss, &s->high, request, n);
} }
static inline void static inline void
append_sendbuffer_low(struct socket *s, struct request_send * request) { append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_send * request) {
s->wb_size += append_sendbuffer_(&s->low, request, 0); s->wb_size += append_sendbuffer_(ss, &s->low, request, 0);
} }
static inline int static inline int
@@ -512,15 +546,17 @@ static int
send_socket(struct socket_server *ss, struct request_send * request, struct socket_message *result, int priority) { send_socket(struct socket_server *ss, struct request_send * request, struct socket_message *result, int priority) {
int id = request->id; int id = request->id;
struct socket * s = &ss->slot[HASH_ID(id)]; struct socket * s = &ss->slot[HASH_ID(id)];
struct send_object so;
send_object_init(ss, &so, request->buffer, request->sz);
if (s->type == SOCKET_TYPE_INVALID || s->id != id if (s->type == SOCKET_TYPE_INVALID || s->id != id
|| s->type == SOCKET_TYPE_HALFCLOSE || s->type == SOCKET_TYPE_HALFCLOSE
|| s->type == SOCKET_TYPE_PACCEPT) { || s->type == SOCKET_TYPE_PACCEPT) {
FREE(request->buffer); so.free_func(request->buffer);
return -1; return -1;
} }
assert(s->type != SOCKET_TYPE_PLISTEN && s->type != SOCKET_TYPE_LISTEN); assert(s->type != SOCKET_TYPE_PLISTEN && s->type != SOCKET_TYPE_LISTEN);
if (send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED) { if (send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED) {
int n = write(s->fd, request->buffer, request->sz); int n = write(s->fd, so.buffer, so.sz);
if (n<0) { if (n<0) {
switch(errno) { switch(errno) {
case EINTR: case EINTR:
@@ -528,22 +564,22 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
n = 0; n = 0;
break; break;
default: default:
fprintf(stderr, "socket-server: write to %d (fd=%d) error.",id,s->fd); fprintf(stderr, "socket-server: write to %d (fd=%d) error :%s.\n",id,s->fd,strerror(errno));
force_close(ss,s,result); force_close(ss,s,result);
return SOCKET_CLOSE; return SOCKET_CLOSE;
} }
} }
if (n == request->sz) { if (n == so.sz) {
FREE(request->buffer); so.free_func(request->buffer);
return -1; return -1;
} }
append_sendbuffer(s, request, n); // add to high priority list, even priority == PRIORITY_LOW append_sendbuffer(ss, s, request, n); // add to high priority list, even priority == PRIORITY_LOW
sp_write(ss->event_fd, s->fd, s, true); sp_write(ss->event_fd, s->fd, s, true);
} else { } else {
if (priority == PRIORITY_LOW) { if (priority == PRIORITY_LOW) {
append_sendbuffer_low(s, request); append_sendbuffer_low(ss, s, request);
} else { } else {
append_sendbuffer(s, request, 0); append_sendbuffer(ss, s, request, 0);
} }
} }
return -1; return -1;
@@ -660,7 +696,7 @@ block_readpipe(int pipefd, void *buffer, int sz) {
if (n<0) { if (n<0) {
if (errno == EINTR) if (errno == EINTR)
continue; continue;
fprintf(stderr, "socket-server : read pipe error %s.",strerror(errno)); fprintf(stderr, "socket-server : read pipe error %s.\n",strerror(errno));
return; return;
} }
// must atomic read from a pipe // must atomic read from a pipe
@@ -904,15 +940,20 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
fprintf(stderr, "socket-server: invalid socket\n"); fprintf(stderr, "socket-server: invalid socket\n");
break; break;
default: default:
if (e->write) { if (e->read) {
int type = send_buffer(ss, s, result); int type = forward_message(ss, s, result);
if (e->write) {
// Try to dispatch write message next step if write flag set.
e->read = false;
--ss->event_index;
}
if (type == -1) if (type == -1)
break; break;
clear_closed_event(ss, result, type); clear_closed_event(ss, result, type);
return type; return type;
} }
if (e->read) { if (e->write) {
int type = forward_message(ss, s, result); int type = send_buffer(ss, s, result);
if (type == -1) if (type == -1)
break; break;
clear_closed_event(ss, result, type); clear_closed_event(ss, result, type);
@@ -1087,3 +1128,9 @@ socket_server_nodelay(struct socket_server *ss, int id) {
request.u.setopt.value = 1; request.u.setopt.value = 1;
send_request(ss, &request, 'T', sizeof(request.u.setopt)); send_request(ss, &request, 'T', sizeof(request.u.setopt));
} }
void
socket_server_userobject(struct socket_server *ss, struct socket_object_interface *soi) {
ss->soi = *soi;
}

View File

@@ -38,4 +38,13 @@ int socket_server_bind(struct socket_server *, uintptr_t opaque, int fd);
void socket_server_nodelay(struct socket_server *, int id); void socket_server_nodelay(struct socket_server *, int id);
struct socket_object_interface {
void * (*buffer)(void *);
int (*size)(void *);
void (*free)(void *);
};
// if you send package sz == -1, use soi.
void socket_server_userobject(struct socket_server *, struct socket_object_interface *soi);
#endif #endif

View File

@@ -1,5 +1,6 @@
local skynet = require "skynet" local skynet = require "skynet"
local queue = require "skynet.queue" local queue = require "skynet.queue"
local snax = require "snax"
local i = 0 local i = 0
local hello = "hello" local hello = "hello"
@@ -32,6 +33,10 @@ function accept.hello()
end) end)
end end
function accept.exit(...)
snax.exit(...)
end
function response.error() function response.error()
error "throw an error" error "throw an error"
end end
@@ -40,9 +45,6 @@ function init( ... )
print ("ping server start:", ...) print ("ping server start:", ...)
-- init queue -- init queue
lock = queue() lock = queue()
-- You can return "queue" for queue service mode
-- return "queue"
end end
function exit(...) function exit(...)

View File

@@ -2,7 +2,7 @@ local skynet = require "skynet"
local snax = require "snax" local snax = require "snax"
skynet.start(function() skynet.start(function()
local ps = snax.uniqueservice ("pingserver", "hello world") local ps = snax.newservice ("pingserver", "hello world")
print(ps.req.ping("foobar")) print(ps.req.ping("foobar"))
print(ps.post.hello()) print(ps.post.hello())
print(pcall(ps.req.error)) print(pcall(ps.req.error))
@@ -31,6 +31,6 @@ end
print(string.format("%s\tcount:%d time:%f", name, v.count, v.time)) print(string.format("%s\tcount:%d time:%f", name, v.count, v.time))
end end
print(snax.kill(ps,"exit")) print(ps.post.exit("exit")) -- == snax.kill(ps, "exit")
skynet.exit() skynet.exit()
end) end)