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 */
}
}
if (prop)
if (g->gcstate != GCSatomic || prop)
linktable(h, &g->ephemeron); /* have to propagate again */
else if (hasclears) /* does table have 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)
-----------
* 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)
$(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 $@
clean :

View File

@@ -66,6 +66,7 @@ skynet.start(function()
end
local balance = 1
local id = socket.listen("0.0.0.0", 8001)
skynet.error("Listen web port 8001")
socket.start(id , function(id, addr)
skynet.error(string.format("%s connected, pass it to agent :%08x", addr, agent[balance]))
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;
size -= need;
if (size == 0) {
if (q == NULL || q->head == q->tail ) {
lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
lua_pushinteger(L, fd);
lua_pushlightuserdata(L, uc->pack.buffer);
lua_pushinteger(L, uc->pack.size);
skynet_free(uc);
return 5;
}
else{
push_data(L, fd, uc->pack.buffer, uc->pack.size, 0);
skynet_free(uc);
return 1;
}
lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
lua_pushinteger(L, fd);
lua_pushlightuserdata(L, uc->pack.buffer);
lua_pushinteger(L, uc->pack.size);
skynet_free(uc);
return 5;
}
// more data
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) {
// just one package
if ( q == NULL || q->head == q->tail) {
lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
lua_pushinteger(L, fd);
void * result = skynet_malloc(pack_size);
memcpy(result, buffer, size);
lua_pushlightuserdata(L, result);
lua_pushinteger(L, size);
return 5;
}
else{
push_data(L, fd, buffer, pack_size, 1);
buffer += pack_size;
size -= pack_size;
return 1;
}
lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
lua_pushinteger(L, fd);
void * result = skynet_malloc(pack_size);
memcpy(result, buffer, size);
lua_pushlightuserdata(L, result);
lua_pushinteger(L, size);
return 5;
}
// more data
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
pushstring(lua_State *L, const char * msg) {
pushstring(lua_State *L, const char * msg, int size) {
if (msg) {
lua_pushstring(L, msg);
lua_pushlstring(L, msg, size);
} else {
lua_pushliteral(L, "");
}
@@ -365,12 +350,12 @@ lfilter(lua_State *L) {
lua_pushvalue(L, lua_upvalueindex(TYPE_OPEN));
// ignore listen id (message->id);
lua_pushinteger(L, message->ud);
pushstring(L, buffer);
pushstring(L, buffer, size);
return 4;
case SKYNET_SOCKET_TYPE_ERROR:
lua_pushvalue(L, lua_upvalueindex(TYPE_ERROR));
lua_pushinteger(L, message->id);
pushstring(L, buffer);
pushstring(L, buffer, size);
return 4;
default:
// 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);
}
} else {
int n;
uint8_t n;
if (len < 0x10000) {
n = COMBINE_TYPE(TYPE_LONG_STRING, 2);
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) {
#ifdef luaL_checkversion
luaL_checkversion(L);
luaL_checkversion(L);
#endif
luaL_checkstack(L, nup, "too many upvalues");
for (; l->name != NULL; l++) { /* fill the table with given functions */
int i;
for (i = 0; i < nup; i++) /* copy upvalues to the top */
lua_pushvalue(L, -nup);
lua_pushcclosure(L, l->func, nup); /* closure with those upvalues */
lua_setfield(L, -(nup + 2), l->name);
}
lua_pop(L, nup); /* remove upvalues */
luaL_checkstack(L, nup, "too many upvalues");
for (; l->name != NULL; l++) { /* fill the table with given functions */
int i;
for (i = 0; i < nup; i++) /* copy upvalues to the top */
lua_pushvalue(L, -nup);
lua_pushcclosure(L, l->func, nup); /* closure with those upvalues */
lua_setfield(L, -(nup + 2), l->name);
}
lua_pop(L, nup); /* remove upvalues */
}
#define luaL_newlibtable(L,l) \
@@ -62,18 +62,20 @@ ldeleteproto(lua_State *L) {
static int
lquerytype(lua_State *L) {
const char * type_name;
struct sproto *sp = lua_touserdata(L,1);
struct sproto_type *st;
if (sp == NULL) {
return luaL_argerror(L, 1, "Need a sproto object");
}
const char * typename = luaL_checkstring(L,2);
struct sproto_type *st = sproto_type(sp, typename);
type_name = luaL_checkstring(L,2);
st = sproto_type(sp, type_name);
if (st) {
lua_pushlightuserdata(L, st);
return 1;
}
return luaL_error(L, "type %s not found", typename);
return luaL_error(L, "type %s not found", type_name);
}
struct encode_ud {
@@ -119,9 +121,10 @@ encode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
switch (type) {
case SPROTO_TINTEGER: {
lua_Integer v = luaL_checkinteger(L, -1);
lua_Integer vh;
lua_pop(L,1);
// notice: in lua 5.2, lua_Integer maybe 52bit
lua_Integer vh = v >> 31;
vh = v >> 31;
if (vh == 0 || vh == -1) {
*(uint32_t *)value = (uint32_t)v;
return 4;
@@ -148,13 +151,14 @@ encode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
}
case SPROTO_TSTRUCT: {
struct encode_ud sub;
int r;
sub.L = L;
sub.st = st;
sub.tbl_index = lua_gettop(L);
sub.array_tag = NULL;
sub.array_index = 0;
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);
return r;
}
@@ -165,6 +169,7 @@ encode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
static void *
expand_buffer(lua_State *L, int osz, int nsz) {
void *output;
do {
osz *= 2;
} 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);
return NULL;
}
void *output = lua_newuserdata(L, osz);
output = lua_newuserdata(L, osz);
lua_replace(L, lua_upvalueindex(1));
lua_pushinteger(L, osz);
lua_replace(L, lua_upvalueindex(2));
@@ -188,6 +193,7 @@ expand_buffer(lua_State *L, int osz, int nsz) {
*/
static int
lencode(lua_State *L) {
struct encode_ud self;
void * buffer = lua_touserdata(L, lua_upvalueindex(1));
int sz = lua_tointeger(L, lua_upvalueindex(2));
@@ -197,7 +203,6 @@ lencode(lua_State *L) {
}
luaL_checktype(L, 2, LUA_TTABLE);
luaL_checkstack(L, ENCODE_DEEPLEVEL + 8, NULL);
struct encode_ud self;
self.L = L;
self.st = st;
self.tbl_index = 2;
@@ -261,15 +266,16 @@ decode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
break;
}
case SPROTO_TSTRUCT: {
lua_newtable(L);
struct decode_ud sub;
int r;
lua_newtable(L);
sub.L = L;
sub.result_index = lua_gettop(L);
sub.deep = self->deep + 1;
sub.array_index = 0;
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)
return r;
lua_settop(L, sub.result_index);
@@ -312,22 +318,25 @@ getbuffer(lua_State *L, int index, size_t *sz) {
static int
ldecode(lua_State *L) {
struct sproto_type * st = lua_touserdata(L, 1);
const void * buffer;
struct decode_ud self;
size_t sz;
int r;
if (st == NULL) {
return luaL_argerror(L, 1, "Need a sproto_type object");
}
size_t sz=0;
const void * buffer = getbuffer(L, 2, &sz);
sz = 0;
buffer = getbuffer(L, 2, &sz);
if (!lua_istable(L, -1)) {
lua_newtable(L);
}
luaL_checkstack(L, ENCODE_DEEPLEVEL*2 + 8, NULL);
struct decode_ud self;
self.L = L;
self.result_index = lua_gettop(L);
self.array_index = 0;
self.array_tag = NULL;
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) {
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).
size_t maxsz = (sz + 2047) / 2048 * 2 + sz;
void * output = lua_touserdata(L, lua_upvalueindex(1));
int bytes;
int osz = lua_tointeger(L, lua_upvalueindex(2));
if (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) {
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
lprotocol(lua_State *L) {
struct sproto * sp = lua_touserdata(L, 1);
struct sproto_type * request;
struct sproto_type * response;
int t;
int tag;
if (sp == NULL) {
return luaL_argerror(L, 1, "Need a sproto_type object");
}
int t = lua_type(L,2);
int tag;
t = lua_type(L,2);
if (t == LUA_TNUMBER) {
const char * name;
tag = lua_tointeger(L, 2);
const char * name = sproto_protoname(sp, tag);
name = sproto_protoname(sp, tag);
if (name == NULL)
return 0;
lua_pushstring(L, name);
@@ -420,13 +434,13 @@ lprotocol(lua_State *L) {
return 0;
lua_pushinteger(L, tag);
}
struct sproto_type * request = sproto_protoquery(sp, tag, SPROTO_REQUEST);
request = sproto_protoquery(sp, tag, SPROTO_REQUEST);
if (request == NULL) {
lua_pushnil(L);
} else {
lua_pushlightuserdata(L, request);
}
struct sproto_type * response = sproto_protoquery(sp, tag, SPROTO_RESPONSE);
response = sproto_protoquery(sp, tag, SPROTO_RESPONSE);
if (response == NULL) {
lua_pushnil(L);
} else {

View File

@@ -119,8 +119,8 @@ todword(const uint8_t *p) {
static int
count_array(const uint8_t * stream) {
uint32_t length = todword(stream);
stream += SIZEOF_LENGTH;
int n = 0;
stream += SIZEOF_LENGTH;
while (length > 0) {
uint32_t nsz;
if (length < SIZEOF_LENGTH)
@@ -180,6 +180,10 @@ static const uint8_t *
import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
uint32_t sz;
const uint8_t * result;
int fn;
int i;
int array = 0;
int tag = -1;
f->tag = -1;
f->type = -1;
f->name = NULL;
@@ -188,13 +192,10 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
sz = todword(stream);
stream += SIZEOF_LENGTH;
result = stream + sz;
int fn = struct_field(stream, sz);
fn = struct_field(stream, sz);
if (fn < 0)
return NULL;
stream += SIZEOF_HEADER;
int i;
int array = 0;
int tag = -1;
for (i=0;i<fn;i++) {
int value;
++tag;
@@ -203,7 +204,7 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
tag+= value/2;
continue;
}
if (tag == 0) { // name
if (tag == 0) { // name
if (value != 0)
return NULL;
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;
value = value/2 - 1;
switch(tag) {
case 1: // buildin
case 1: // buildin
if (value >= SPROTO_TSTRUCT)
return NULL; // invalid buildin type
f->type = value;
break;
case 2: // type index
case 2: // type index
if (value >= s->type_n)
return NULL; // invalid type index
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->st = &s->type[value];
break;
case 3: // tag
case 3: // tag
f->tag = value;
break;
case 4: // array
case 4: // array
if (value)
array = SPROTO_TARRAY;
break;
@@ -248,10 +249,10 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
.type {
.field {
name 0 : string
buildin 1 : integer
buildin 1 : integer
type 2 : integer
tag 3 : integer
array 4 : boolean
tag 3 : integer
array 4 : boolean
}
name 0 : string
fields 1 : *field
@@ -259,11 +260,16 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
*/
static const uint8_t *
import_type(struct sproto *s, struct sproto_type *t, const uint8_t * stream) {
const uint8_t * result;
uint32_t sz = todword(stream);
int i;
int fn;
int n;
int maxn;
int last;
stream += SIZEOF_LENGTH;
const uint8_t * result = stream + sz;
int fn = struct_field(stream, sz);
result = stream + sz;
fn = struct_field(stream, sz);
if (fn <= 0 || fn > 2)
return NULL;
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;
}
stream += todword(stream)+SIZEOF_LENGTH; // second data
int n = count_array(stream);
n = count_array(stream);
if (n<0)
return NULL;
stream += SIZEOF_LENGTH;
int maxn = n;
int last = -1;
maxn = n;
last = -1;
t->n = n;
t->f = pool_alloc(&s->memory, sizeof(struct field) * n);
for (i=0;i<n;i++) {
int tag;
struct field *f = &t->f[i];
stream = import_field(s, f, stream);
if (stream == NULL)
return NULL;
int tag = f->tag;
tag = f->tag;
if (tag < last)
return NULL; // tag must in ascending order
if (tag > last+1) {
@@ -312,24 +319,27 @@ import_type(struct sproto *s, struct sproto_type *t, const uint8_t * stream) {
/*
.protocol {
name 0 : string
tag 1 : integer
request 2 : integer
tag 1 : integer
request 2 : integer
response 3 : integer
}
*/
static const uint8_t *
import_protocol(struct sproto *s, struct protocol *p, const uint8_t * stream) {
const uint8_t * result;
uint32_t sz = todword(stream);
int fn;
int i;
int tag;
stream += SIZEOF_LENGTH;
const uint8_t * result = stream + sz;
int fn = struct_field(stream, sz);
result = stream + sz;
fn = struct_field(stream, sz);
stream += SIZEOF_HEADER;
p->name = NULL;
p->tag = -1;
p->p[SPROTO_REQUEST] = NULL;
p->p[SPROTO_RESPONSE] = NULL;
int i;
int tag = 0;
tag = 0;
for (i=0;i<fn;i++,tag++) {
int value = toword(stream + SIZEOF_FIELD * i);
if (value & 1) {
@@ -338,24 +348,24 @@ import_protocol(struct sproto *s, struct protocol *p, const uint8_t * stream) {
}
value = value/2 - 1;
switch (i) {
case 0: // name
case 0: // name
if (value != -1) {
return NULL;
}
p->name = import_string(s, stream + SIZEOF_FIELD *fn);
break;
case 1: // tag
case 1: // tag
if (value < 0) {
return NULL;
}
p->tag = value;
break;
case 2: // request
case 2: // request
if (value < 0 || value>=s->type_n)
return NULL;
p->p[SPROTO_REQUEST] = &s->type[value];
break;
case 3: // response
case 3: // response
if (value < 0 || value>=s->type_n)
return NULL;
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 *
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 i;
if (fn < 0)
return NULL;
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++) {
int value = toword(stream + i*SIZEOF_FIELD);
int n;
if (value != 0)
return NULL;
int n = count_array(content);
n = count_array(content);
if (n<0)
return NULL;
if (i == 0) {
@@ -423,8 +434,9 @@ create_from_bundle(struct sproto *s, const uint8_t * stream, size_t sz) {
struct sproto *
sproto_create(const void * proto, size_t sz) {
struct pool mem;
struct sproto * s;
pool_init(&mem);
struct sproto * s = pool_alloc(&mem, sizeof(*s));
s = pool_alloc(&mem, sizeof(*s));
if (s == NULL)
return NULL;
memset(s, 0, sizeof(*s));
@@ -445,34 +457,35 @@ sproto_release(struct sproto * s) {
void
sproto_dump(struct sproto *s) {
int i,j;
static const char * buildin[] = {
"integer",
"boolean",
"string",
};
printf("=== %d types ===\n", s->type_n);
int i,j;
for (i=0;i<s->type_n;i++) {
struct sproto_type *t = &s->type[i];
printf("%s\n", t->name);
for (j=0;j<t->n;j++) {
char array[2] = { 0, 0 };
const char * typename = NULL;
const char * type_name = NULL;
struct field *f = &t->f[j];
if (f->type & SPROTO_TARRAY) {
array[0] = '*';
} else {
array[0] = 0;
}
int t = f->type & ~SPROTO_TARRAY;
if (t == SPROTO_TSTRUCT) {
typename = f->st->name;
} else {
assert(t<SPROTO_TSTRUCT);
typename = buildin[t];
{
int t = f->type & ~SPROTO_TARRAY;
if (t == SPROTO_TSTRUCT) {
type_name = f->st->name;
} else {
assert(t<SPROTO_TSTRUCT);
type_name = buildin[t];
}
}
printf("\t%s (%d) %s%s\n", f->name, f->tag, array, typename);
printf("\t%s (%d) %s%s\n", f->name, f->tag, array, type_name);
}
}
printf("=== %d protocol ===\n", s->protocol_n);
@@ -518,10 +531,11 @@ query_proto(struct sproto *sp, int tag) {
struct sproto_type *
sproto_protoquery(struct sproto *sp, int proto, int what) {
struct protocol * p;
if (what <0 || what >1) {
return NULL;
}
struct protocol * p = query_proto(sp, proto);
p = query_proto(sp, proto);
if (p) {
return p->p[what];
}
@@ -555,13 +569,15 @@ sproto_name(struct sproto_type * st) {
static struct field *
findtag(struct sproto_type *st, int tag) {
int begin, end;
if (st->base >=0 ) {
tag -= st->base;
if (tag < 0 || tag >= st->n)
return NULL;
return &st->f[tag];
}
int begin = 0, end = st->n;
begin = 0;
end = st->n;
while (begin < end) {
int mid = (begin+end)/2;
struct field *f = &st->f[mid];
@@ -580,7 +596,7 @@ findtag(struct sproto_type *st, int tag) {
// encode & decode
// 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
fill_size(uint8_t * data, int sz) {
@@ -623,18 +639,20 @@ encode_uint64(uint64_t v, uint8_t * data, int size) {
static int
encode_string(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) {
int sz;
if (size < SIZEOF_LENGTH)
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);
}
static int
encode_struct(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) {
int sz;
if (size < SIZEOF_LENGTH) {
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);
}
@@ -656,18 +674,21 @@ uint32_to_uint64(int negative, uint8_t *buffer) {
static uint8_t *
encode_integer_array(sproto_callback cb, void *ud, struct field *f, uint8_t *buffer, int size) {
uint8_t * header = buffer;
int intlen;
int index;
if (size < 1)
return NULL;
buffer++;
size--;
int intlen = sizeof(uint32_t);
int index = 1;
intlen = sizeof(uint32_t);
index = 1;
for (;;) {
int sz;
union {
uint64_t u64;
uint32_t u32;
} 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)
return NULL;
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);
}
} else {
uint64_t v;
if (sz != sizeof(uint64_t))
return NULL;
if (intlen == sizeof(uint32_t)) {
int i;
// rearrange
size -= (index-1) * sizeof(uint32_t);
if (size < sizeof(uint64_t))
return NULL;
buffer += (index-1) * sizeof(uint32_t);
int 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));
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));
}
intlen = sizeof(uint64_t);
}
uint64_t v = u.u64;
v = u.u64;
buffer[0] = v & 0xff;
buffer[1] = (v >> 8) & 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
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)
return -1;
size -= SIZEOF_LENGTH;
int index = 1;
uint8_t * buffer = data + SIZEOF_LENGTH;
int type = f->type & ~SPROTO_TARRAY;
index = 1;
buffer = data + SIZEOF_LENGTH;
type = f->type & ~SPROTO_TARRAY;
switch (type) {
case SPROTO_TINTEGER:
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;
default:
for (;;) {
int sz;
if (size < SIZEOF_LENGTH)
return -1;
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)
return -1;
if (sz == 0)
@@ -772,7 +800,7 @@ encode_array(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int s
}
break;
}
int sz = buffer - (data + SIZEOF_LENGTH);
sz = buffer - (data + SIZEOF_LENGTH);
if (sz == 0)
return 0;
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 * data;
int header_sz = SIZEOF_HEADER + st->maxn * SIZEOF_FIELD;
int i;
int index;
int lasttag;
int datasz;
if (size < header_sz)
return -1;
data = header + header_sz;
size -= header_sz;
int i;
int index = 0;
int lasttag = -1;
index = 0;
lasttag = -1;
for (i=0;i<st->n;i++) {
struct field *f = &st->f[i];
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 (u.u32 < 0x7fff) {
value = (u.u32+1) * 2;
sz = 2; // sz can be any number > 0
sz = 2; // sz can be any number > 0
} else {
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)
return -1;
if (sz > 0) {
uint8_t * record;
int tag;
if (value == 0) {
data += sz;
size -= sz;
}
uint8_t * record = header+SIZEOF_HEADER+SIZEOF_FIELD*index;
int tag = f->tag - lasttag - 1;
record = header+SIZEOF_HEADER+SIZEOF_FIELD*index;
tag = f->tag - lasttag - 1;
if (tag > 0) {
// skip tag
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[1] = (index >> 8) & 0xff;
int datasz = data - (header + header_sz);
datasz = data - (header + header_sz);
data = header + header_sz;
if (index != st->maxn) {
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;
switch (type) {
case SPROTO_TINTEGER: {
int len;
if (sz < 1)
return -1;
int len = *stream;
len = *stream;
++stream;
--sz;
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 * datastream;
int fn;
int i;
int tag;
if (size < SIZEOF_HEADER)
return -1;
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;
size -= fn * SIZEOF_FIELD ;
int i;
int tag = -1;
tag = -1;
for (i=0;i<fn;i++) {
++ tag;
uint8_t * currentdata;
struct field * f;
int value = toword(stream + i * SIZEOF_FIELD);
++ tag;
if (value & 1) {
tag += value/2;
continue;
}
value = value/2 - 1;
uint8_t * currentdata = datastream;
currentdata = datastream;
if (value < 0) {
uint32_t sz;
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;
size -= sz+SIZEOF_LENGTH;
}
struct field * f= findtag(st, tag);
f = findtag(st, tag);
if (f == NULL)
continue;
if (value < 0) {

View File

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

View File

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

View File

@@ -132,6 +132,14 @@ function snax.kill(obj, ...)
skynet_call(obj.handle, "snax", t.system.exit, ...)
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, ...)
if ok then
return ...

View File

@@ -53,20 +53,32 @@ function gateserver.start(handler)
local MSG = {}
function MSG.data(fd, msg, sz)
local function dispatch_msg(fd, msg, sz)
if connection[fd] then
handler.message(fd, msg, sz)
else
skynet.error(string.format("Drop message from fd (%d) : %s", fd, netpack.tostring(msg,sz)))
end
end
function MSG.more()
for fd, msg, sz in netpack.pop, queue do
if connection[fd] then
handler.message(fd, msg, sz)
MSG.data = dispatch_msg
local function dispatch_queue()
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
MSG.more = dispatch_queue
function MSG.open(fd, msg)
if client_number >= maxclient then
socketdriver.close(fd)
@@ -128,4 +140,4 @@ function gateserver.start(handler)
end)
end
return gateserver
return gateserver

View File

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

View File

@@ -12,10 +12,6 @@ package.path = snax_path .. "?.lua;" .. package.path
SERVICE_NAME = snax_name
SERVICE_PATH = snax_path
local mode
local thread_id
local message_queue = {}
local init = false
local profile_table = {}
local function update_stat(name, ti)
@@ -38,57 +34,6 @@ local function dispatch(f, ...)
return skynet.pack(f(...))
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, ...)
return skynet.ret(skynet.pack(f(...)))
end
@@ -108,6 +53,7 @@ local function timing( method, ... )
end
skynet.start(function()
local init = false
skynet.dispatch("snax", function ( session , source , id, ...)
local method = func[id]
@@ -119,17 +65,12 @@ skynet.start(function()
elseif command == "init" then
assert(not init, "Already init")
local initfunc = method[4] or function() end
mode = initfunc(...)
if mode == "queue" then
skynet.fork(message_dispatch)
end
initfunc(...)
skynet.ret()
skynet.info_func(function()
return profile_table
end)
init = true
elseif mode == "queue" then
queue( session, source, method , ...)
else
assert(init, "Never init")
assert(command == "exit")
@@ -141,11 +82,7 @@ skynet.start(function()
end
else
assert(init, "Init first")
if mode == "queue" then
queue(session, source, method , ...)
else
timing(method, ...)
end
timing(method, ...)
end
end)
end)

View File

@@ -169,7 +169,11 @@ skynet_context_new(const char * name, const char *param) {
int
skynet_context_newsession(struct skynet_context *ctx) {
// 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;
}
@@ -691,7 +695,7 @@ skynet_sendname(struct skynet_context * context, uint32_t source, const char * a
if (type & PTYPE_TAG_DONTCOPY) {
skynet_free(data);
}
return session;
return -1;
}
} else {
_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)) {
// todo: report somewhere to close socket
// don't call skynet_socket_close here (It will block mainloop)
skynet_free(sm->buffer);
skynet_free(sm);
}
}

View File

@@ -40,8 +40,9 @@
struct write_buffer {
struct write_buffer * next;
char *ptr;
int sz;
void *buffer;
int sz;
bool userobject;
};
struct wb_list {
@@ -68,6 +69,7 @@ struct socket_server {
int alloc_id;
int event_n;
int event_index;
struct socket_object_interface soi;
struct event ev[MAX_EVENT];
struct socket slot[MAX_SOCKET];
char buffer[MAX_INFO];
@@ -137,9 +139,40 @@ union sockaddr_all {
struct sockaddr_in6 v6;
};
struct send_object {
void * buffer;
int sz;
void (*free_func)(void *);
};
#define MALLOC skynet_malloc
#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
socket_keepalive(int fd) {
int keepalive = 1;
@@ -213,6 +246,7 @@ socket_server_create() {
ss->alloc_id = 0;
ss->event_n = 0;
ss->event_index = 0;
memset(&ss->soi, 0, sizeof(ss->soi));
FD_ZERO(&ss->rfds);
assert(ss->recvctrl_fd < FD_SETSIZE);
@@ -220,13 +254,12 @@ socket_server_create() {
}
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;
while (wb) {
struct write_buffer *tmp = wb;
wb = wb->next;
FREE(tmp->buffer);
FREE(tmp);
write_buffer_free(ss, tmp);
}
list->head = NULL;
list->tail = NULL;
@@ -242,8 +275,8 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
return;
}
assert(s->type != SOCKET_TYPE_RESERVE);
free_wb_list(&s->high);
free_wb_list(&s->low);
free_wb_list(ss,&s->high);
free_wb_list(ss,&s->low);
if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
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;
}
list->head = tmp->next;
FREE(tmp->buffer);
FREE(tmp);
write_buffer_free(ss,tmp);
}
list->tail = NULL;
@@ -469,10 +501,12 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
}
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));
buf->ptr = request->buffer+n;
buf->sz = request->sz - n;
struct send_object so;
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->next = NULL;
if (s->head == NULL) {
@@ -487,13 +521,13 @@ append_sendbuffer_(struct wb_list *s, struct request_send * request, int n) {
}
static inline void
append_sendbuffer(struct socket *s, struct request_send * request, int n) {
s->wb_size += append_sendbuffer_(&s->high, request, n);
append_sendbuffer(struct socket_server *ss, struct socket *s, struct request_send * request, int n) {
s->wb_size += append_sendbuffer_(ss, &s->high, request, n);
}
static inline void
append_sendbuffer_low(struct socket *s, struct request_send * request) {
s->wb_size += append_sendbuffer_(&s->low, request, 0);
append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_send * request) {
s->wb_size += append_sendbuffer_(ss, &s->low, request, 0);
}
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) {
int id = request->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
|| s->type == SOCKET_TYPE_HALFCLOSE
|| s->type == SOCKET_TYPE_PACCEPT) {
FREE(request->buffer);
so.free_func(request->buffer);
return -1;
}
assert(s->type != SOCKET_TYPE_PLISTEN && s->type != SOCKET_TYPE_LISTEN);
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) {
switch(errno) {
case EINTR:
@@ -528,22 +564,22 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
n = 0;
break;
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);
return SOCKET_CLOSE;
}
}
if (n == request->sz) {
FREE(request->buffer);
if (n == so.sz) {
so.free_func(request->buffer);
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);
} else {
if (priority == PRIORITY_LOW) {
append_sendbuffer_low(s, request);
append_sendbuffer_low(ss, s, request);
} else {
append_sendbuffer(s, request, 0);
append_sendbuffer(ss, s, request, 0);
}
}
return -1;
@@ -660,7 +696,7 @@ block_readpipe(int pipefd, void *buffer, int sz) {
if (n<0) {
if (errno == EINTR)
continue;
fprintf(stderr, "socket-server : read pipe error %s.",strerror(errno));
fprintf(stderr, "socket-server : read pipe error %s.\n",strerror(errno));
return;
}
// 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");
break;
default:
if (e->write) {
int type = send_buffer(ss, s, result);
if (e->read) {
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)
break;
clear_closed_event(ss, result, type);
return type;
}
if (e->read) {
int type = forward_message(ss, s, result);
if (e->write) {
int type = send_buffer(ss, s, result);
if (type == -1)
break;
clear_closed_event(ss, result, type);
@@ -1087,3 +1128,9 @@ socket_server_nodelay(struct socket_server *ss, int id) {
request.u.setopt.value = 1;
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);
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

View File

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

View File

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