lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-22 02:53:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

delete unused feature : group/localcast/forward, etc

Browse Source
This commit is contained in:
Cloud Wu
2014-04-18 14:09:07 +08:00
parent 2fc7ceda44
commit d568e54c63
37 changed files with 27 additions and 2017 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
Makefile
View File

@@ -7,17 +7,17 @@ LUA_CLIB_PATH ?= luaclib
CSERVICE_PATH ?= cservice
SKYNET_BUILD_PATH ?= .
CFLAGS = -g -Wall -I$(LUA_INC) $(MYCFLAGS)
CFLAGS = -g -O2 -Wall -I$(LUA_INC) $(MYCFLAGS)
$(LUA_STATICLIB) :
cd 3rd/lua && $(MAKE) CC=$(CC) $(PLAT)
CSERVICE = snlua logger gate master multicast tunnel harbor localcast
LUA_CLIB = skynet socketdriver int64 mcast bson mongo md5 netpack cjson clientsocket
CSERVICE = snlua logger gate master harbor
LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket
SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \
skynet_server.c skynet_start.c skynet_timer.c skynet_error.c \
skynet_harbor.c skynet_multicast.c skynet_group.c skynet_env.c \
skynet_harbor.c skynet_env.c \
skynet_monitor.c skynet_socket.c socket_server.c
all : \
@@ -41,17 +41,14 @@ endef
$(foreach v, $(CSERVICE), $(eval $(call CSERVICE_TEMP,$(v))))
$(LUA_CLIB_PATH)/skynet.so : lualib-src/lua-skynet.c lualib-src/lua-seri.c lualib-src/trace_service.c lualib-src/timingqueue.c | $(LUA_CLIB_PATH)
$(LUA_CLIB_PATH)/skynet.so : lualib-src/lua-skynet.c lualib-src/lua-seri.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src -Iservice-src -Ilualib-src
$(LUA_CLIB_PATH)/socketdriver.so : lualib-src/lua-socket.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src -Iservice-src
$(LUA_CLIB_PATH)/int64.so : 3rd/lua-int64/int64.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) -O2 $^ -o $@
$(LUA_CLIB_PATH)/mcast.so : lualib-src/lua-localcast.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src -Iservice-src
$(CC) $(CFLAGS) $(SHARED) $^ -o $@
$(LUA_CLIB_PATH)/bson.so : lualib-src/lua-bson.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) $^ -o $@
@@ -60,7 +57,7 @@ $(LUA_CLIB_PATH)/mongo.so : lualib-src/lua-mongo.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) $^ -o $@
$(LUA_CLIB_PATH)/md5.so : 3rd/lua-md5/md5.c 3rd/lua-md5/md5lib.c 3rd/lua-md5/compat-5.2.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) -O2 -I3rd/lua-md5 $^ -o $@
$(CC) $(CFLAGS) $(SHARED) -I3rd/lua-md5 $^ -o $@
$(LUA_CLIB_PATH)/netpack.so : lualib-src/lua-netpack.c | $(LUA_CLIB_PATH)
$(CC) $(CFLAGS) $(SHARED) $^ -Iskynet-src -o $@

111
client-src/client.c
View File

@@ -1,111 +0,0 @@
#include <pthread.h>
#include <netinet/in.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <errno.h>
struct args {
int fd;
};
static int
readall(int fd, void * buffer, size_t sz) {
for (;;) {
int err = recv(fd , buffer, sz, MSG_WAITALL);
if (err < 0) {
if (errno == EAGAIN || errno == EINTR)
continue;
break;
}
return err;
}
perror("Socket error");
exit(1);
}
static void *
_read(void *ud) {
struct args *p = ud;
int fd = p->fd;
fflush(stdout);
for (;;) {
uint8_t header[2];
fflush(stdout);
if (readall(fd, header, 2) == 0)
break;
size_t len = header[0] << 8 | header[1];
if (len>0) {
char tmp[len+1];
readall(fd, tmp, len);
tmp[len]='\0';
printf("%s\n",tmp);
}
}
return NULL;
}
static void *
test(void *ud) {
struct args *p = ud;
int fd = p->fd;
char tmp[1024];
while (!feof(stdin)) {
fgets(tmp,sizeof(tmp),stdin);
int n = strlen(tmp) -1;
uint8_t head[2];
head[0] = (n >> 8) & 0xff;
head[1] = n & 0xff;
int r;
r = send(fd, head, 2, 0);
if (r<0) {
perror("send head");
}
r = send(fd, tmp , n, 0);
if (r<0) {
perror("send data");
}
}
return NULL;
}
int
main(int argc, char * argv[]) {
if (argc < 3) {
printf("connect address port\n");
return 1;
}
int fd = socket(AF_INET,SOCK_STREAM,0);
struct sockaddr_in my_addr;
my_addr.sin_addr.s_addr=inet_addr(argv[1]);
my_addr.sin_family=AF_INET;
my_addr.sin_port=htons(strtol(argv[2],NULL,10));
int r = connect(fd,(struct sockaddr *)&my_addr,sizeof(struct sockaddr_in));
if (r == -1) {
perror("Connect failed:");
return 1;
}
struct args arg = { fd };
pthread_t pid ;
pthread_create(&pid, NULL, _read, &arg);
pthread_t pid_stdin;
pthread_create(&pid_stdin, NULL, test, &arg);
pthread_join(pid, NULL);
close(fd);
return 0;
}

5
lualib-src/lua-clientsocket.c
View File

@@ -193,7 +193,10 @@ readline_stdin(void * arg) {
struct queue * q = arg;
char tmp[1024];
while (!feof(stdin)) {
fgets(tmp,sizeof(tmp),stdin);
if (fgets(tmp,sizeof(tmp),stdin) == NULL) {
// read stdin failed
exit(1);
}
int n = strlen(tmp) -1;
char * str = malloc(n+1);

59
lualib-src/lua-localcast.c
View File

@@ -1,59 +0,0 @@
#include <lua.h>
#include <lauxlib.h>
#include "localcast.h"
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
/*
table handles
string msg
lightuserdata ptr
integer sz
*/
static int
_pack_message(lua_State *L) {
luaL_checktype(L,1,LUA_TTABLE);
int type = lua_type(L,2);
void * msg = NULL;
size_t sz = 0;
switch(type) {
case LUA_TSTRING: {
const char * str = lua_tolstring(L,2,&sz);
msg = malloc(sz);
memcpy(msg, str, sz);
break;
}
case LUA_TLIGHTUSERDATA:
msg = lua_touserdata(L,2);
sz = luaL_checkinteger(L,3);
break;
default:
luaL_error(L, "type error : %s", lua_typename(L,type));
break;
}
struct localcast *lc = malloc(sizeof(struct localcast));
lc->n = lua_rawlen(L,1);
uint32_t *group = malloc(lc->n * sizeof(uint32_t));
int i;
for (i=0;i<lc->n;i++) {
lua_rawgeti(L,1,i+1);
group[i] = lua_tounsigned(L,-1);
lua_pop(L,1);
}
lc->msg = msg;
lc->sz = sz;
lc->group = group;
lua_pushlightuserdata(L,lc);
lua_pushinteger(L, sizeof(*lc));
return 2;
};
int
luaopen_mcast_c(lua_State *L) {
luaL_checkversion(L);
lua_pushcfunction(L, _pack_message);
return 1;
}

279
lualib-src/lua-skynet.c
View File

@@ -1,8 +1,6 @@
#include "skynet.h"
#include "trace_service.h"
#include "lua-seri.h"
#include "service_lua.h"
#include "timingqueue.h"
#define KNRM "\x1B[0m"
#define KRED "\x1B[31m"
@@ -12,89 +10,10 @@
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <time.h>
#if defined(__APPLE__)
#include <mach/task.h>
#include <mach/mach.h>
#endif
struct stat {
lua_State *L;
int count;
uint32_t ti_sec;
uint32_t ti_nsec;
struct trace_pool *trace;
struct tqueue * tq;
struct snlua *lua;
};
static void
_stat_begin(struct stat *S, struct timespec *ti) {
S->count++;
current_time(ti);
}
inline static void
_stat_end(struct stat *S, struct timespec *ti) {
diff_time(ti, &S->ti_sec, &S->ti_nsec);
}
static int
_stat(lua_State *L) {
lua_rawgetp(L, LUA_REGISTRYINDEX, _stat);
struct stat *S = lua_touserdata(L,-1);
if (S==NULL) {
luaL_error(L, "set callback first");
}
const char * what = luaL_checkstring(L,1);
if (strcmp(what,"count")==0) {
lua_pushinteger(L, S->count);
return 1;
}
if (strcmp(what,"time")==0) {
double t = (double)S->ti_sec + (double)S->ti_nsec / NANOSEC;
lua_pushnumber(L, t);
return 1;
}
if (strcmp(what,"trace")==0) {
lua_pushlightuserdata(L, S->trace);
return 1;
}
return 0;
}
static inline double
current_time_tick(struct stat *S) {
return (double)S->ti_sec + (double)S->ti_nsec / NANOSEC;
}
static struct stat *
get_stat(lua_State *L) {
struct stat * S = lua_touserdata(L, lua_upvalueindex(2));
if (S == NULL) {
lua_rawgetp(L, LUA_REGISTRYINDEX, _stat);
S = lua_touserdata(L, -1);
lua_replace(L, lua_upvalueindex(2));
}
return S;
}
static inline void
save_session(lua_State *L, int type, int session) {
if (session > 0 && (type & 0xff) != PTYPE_RESPONSE) {
struct stat * S = get_stat(L);
tqueue_push(S->tq, session, current_time_tick(S));
}
}
static int
_cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
struct stat *S = ud;
lua_State *L = S->L;
struct timespec ti;
_stat_begin(S, &ti);
lua_State *L = ud;
int trace = 1;
int r;
int top = lua_gettop(L);
@@ -111,45 +30,9 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t
lua_pushinteger(L, session);
lua_pushnumber(L, source);
if (type == PTYPE_RESPONSE && session > 0) {
double t = tqueue_pop(S->tq, session);
if (t != 0) {
t = current_time_tick(S) - t;
lua_pushnumber(L, t);
r = lua_pcall(L, 6, 0 , trace);
} else {
r = lua_pcall(L, 5, 0 , trace);
}
} else {
r = lua_pcall(L, 5, 0 , trace);
}
_stat_end(S, &ti);
struct trace_info *tti = trace_yield(S->trace);
if (tti) {
skynet_error(context, "Untraced time %f", trace_delete(S->trace, tti));
}
r = lua_pcall(L, 5, 0 , trace);
if (r == LUA_OK) {
if (S->lua->reload) {
skynet_callback(context, NULL, 0);
struct snlua * lua = S->lua;
assert(lua->L == L);
const char * cmd = lua->reload;
lua->reload = NULL;
lua->L = luaL_newstate();
int err = lua->init(lua, context, cmd);
if (err) {
skynet_callback(context, S, _cb);
skynet_error(context, "lua reload failed : %s", cmd);
lua_close(lua->L);
lua->L = L;
} else {
skynet_error(context, "lua reload %s", cmd);
lua_close(L);
}
}
return 0;
}
const char * self = skynet_command(context, "REG", NULL);
@@ -173,34 +56,9 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t
return 0;
}
static int
_timing(lua_State *L) {
int session = luaL_checkinteger(L,1);
struct stat * S = get_stat(L);
double t = tqueue_pop(S->tq, session);
if (t != 0) {
t = current_time_tick(S) - t;
}
lua_pushnumber(L, t);
return 1;
}
static int
_delete_stat(lua_State *L) {
struct stat * S = lua_touserdata(L,1);
trace_release(S->trace);
tqueue_delete(S->tq);
return 0;
}
static int
_callback(lua_State *L) {
struct snlua *lua = lua_touserdata(L, lua_upvalueindex(1));
if (lua == NULL || lua->ctx == NULL) {
return luaL_error(L, "Init skynet context first");
}
struct skynet_context * context = lua->ctx;
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
luaL_checktype(L,1,LUA_TFUNCTION);
lua_settop(L,1);
@@ -209,21 +67,7 @@ _callback(lua_State *L) {
lua_rawgeti(L, LUA_REGISTRYINDEX, LUA_RIDX_MAINTHREAD);
lua_State *gL = lua_tothread(L,-1);
struct stat * S = lua_newuserdata(L, sizeof(*S));
memset(S, 0, sizeof(*S));
S->L = gL;
S->trace = trace_create();
S->tq = tqueue_new();
S->lua = lua;
lua_createtable(L,0,1);
lua_pushcfunction(L, _delete_stat);
lua_setfield(L,-2,"__gc");
lua_setmetatable(L, -2);
lua_rawsetp(L, LUA_REGISTRYINDEX, _stat);
skynet_callback(context, S, _cb);
skynet_callback(context, gL, _cb);
return 0;
}
@@ -272,19 +116,16 @@ _sendname(lua_State *L, struct skynet_context * context, const char * dest) {
size_t len = 0;
void * msg = (void *)lua_tolstring(L,4,&len);
session = skynet_sendname(context, dest, type, session , msg, len);
save_session(L, type, session);
break;
}
case LUA_TNIL :
session = skynet_sendname(context, dest, type, session , NULL, 0);
save_session(L, type, session);
break;
case LUA_TLIGHTUSERDATA: {
luaL_checktype(L, 4, LUA_TLIGHTUSERDATA);
void * msg = lua_touserdata(L,4);
int size = luaL_checkinteger(L,5);
session = skynet_sendname(context, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
save_session(L, type, session);
break;
}
default:
@@ -350,14 +191,12 @@ _send(lua_State *L) {
msg = NULL;
}
session = skynet_send(context, 0, dest, type, session , msg, len);
save_session(L, type, session);
break;
}
case LUA_TLIGHTUSERDATA: {
void * msg = lua_touserdata(L,4);
int size = luaL_checkinteger(L,5);
session = skynet_send(context, 0, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
save_session(L, type, session);
break;
}
default:
@@ -389,14 +228,12 @@ _redirect(lua_State *L) {
msg = NULL;
}
session = skynet_send(context, source, dest, type, session , msg, len);
save_session(L, type, session);
break;
}
case LUA_TLIGHTUSERDATA: {
void * msg = lua_touserdata(L,5);
int size = luaL_checkinteger(L,6);
session = skynet_send(context, source, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
save_session(L, type, session);
break;
}
default:
@@ -405,15 +242,6 @@ _redirect(lua_State *L) {
return 0;
}
static int
_forward(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
uint32_t dest = luaL_checkunsigned(L,1);
skynet_forward(context, dest);
return 0;
}
static int
_error(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
@@ -444,75 +272,6 @@ _harbor(lua_State *L) {
return 2;
}
static int
_context(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
lua_pushlightuserdata(L, context);
return 1;
}
// trace api
static int
_trace_new(lua_State *L) {
struct trace_pool *p = lua_touserdata(L,1);
struct trace_info *t = trace_new(p);
if (t==NULL) {
return luaL_error(L, "Last trace didn't close");
}
lua_pushlightuserdata(L,t);
return 1;
}
static int
_trace_delete(lua_State *L) {
struct trace_pool *p = lua_touserdata(L,1);
struct trace_info *t = lua_touserdata(L,2);
double ti = trace_delete(p,t);
lua_pushnumber(L, ti);
return 1;
}
static int
_trace_switch(lua_State *L) {
int session = luaL_checkinteger(L,2);
if (session <=0)
return 0;
struct trace_pool *p = lua_touserdata(L,1);
trace_switch(p, session);
return 0;
}
static int
_trace_yield(lua_State *L) {
struct trace_pool *p = lua_touserdata(L,1);
struct trace_info * t = trace_yield(p);
if (t) {
lua_pushlightuserdata(L,t);
return 1;
}
return 0;
}
static int
_trace_register(lua_State *L) {
int session = luaL_checkinteger(L,2);
if (session <=0)
return 0;
struct trace_pool *p = lua_touserdata(L,1);
trace_register(p, session);
return 0;
}
static int
_reload(lua_State *L) {
struct snlua *lua = lua_touserdata(L,lua_upvalueindex(1));
lua->reload = luaL_checkstring(L,1);
lua_settop(L,1);
lua_replace(L,lua_upvalueindex(2));
return 0;
}
int
luaopen_skynet_c(lua_State *L) {
luaL_checkversion(L);
@@ -520,31 +279,17 @@ luaopen_skynet_c(lua_State *L) {
luaL_Reg l[] = {
{ "send" , _send },
{ "genid", _genid },
{ "timing", _timing },
{ "redirect", _redirect },
{ "forward", _forward },
{ "command" , _command },
{ "error", _error },
{ "tostring", _tostring },
{ "harbor", _harbor },
{ "context", _context },
{ "pack", _luaseri_pack },
{ "unpack", _luaseri_unpack },
{ "callback", _callback },
{ NULL, NULL },
};
luaL_Reg l2[] = {
{ "stat", _stat },
{ "trace_new", _trace_new },
{ "trace_delete", _trace_delete },
{ "trace_switch", _trace_switch },
{ "trace_yield", _trace_yield },
{ "trace_register", _trace_register },
{ NULL, NULL },
};
lua_createtable(L, 0, (sizeof(l) + sizeof(l2))/sizeof(luaL_Reg)-2);
lua_getfield(L, LUA_REGISTRYINDEX, "skynet_lua");
struct snlua *lua = lua_touserdata(L,-1);
if (lua == NULL || lua->ctx == NULL) {
@@ -552,19 +297,9 @@ luaopen_skynet_c(lua_State *L) {
}
assert(lua->L == L);
lua_pushvalue(L,-1);
lua_pushcclosure(L,_callback,1);
lua_setfield(L, -3, "callback");
lua_pushnil(L);
lua_pushcclosure(L,_reload,2);
lua_setfield(L, -2, "reload");
luaL_newlibtable(L, l);
lua_pushlightuserdata(L, lua->ctx);
lua_pushnil(L);
luaL_setfuncs(L,l,2);
luaL_setfuncs(L,l2,0);
luaL_setfuncs(L,l,1);
return 1;
}

197
lualib-src/timingqueue.c
View File

@@ -1,197 +0,0 @@
#include "timingqueue.h"
#include <stdlib.h>
#include <assert.h>
#include <stdio.h>
#define INIT_CAP 8
struct session_time {
int session;
double time;
};
struct tqueue {
int cap;
int n;
int head;
int tail;
struct session_time * q;
};
struct tqueue *
tqueue_new() {
struct tqueue * tq = malloc(sizeof(*tq));
tq->cap = INIT_CAP;
tq->head = 0;
tq->tail = 0;
tq->n = 0;
tq->q = malloc(sizeof(struct session_time) * tq->cap);
return tq;
}
void
tqueue_delete(struct tqueue *tq) {
free(tq->q);
free(tq);
}
static struct session_time *
expand_queue(struct tqueue *tq) {
if (tq->n == tq->cap) {
struct session_time * q = malloc(sizeof(*q) * tq->cap * 2);
int i;
for (i=0;i<tq->cap;i++) {
int index = tq->head + i;
if (index >= tq->cap) {
index -= tq->cap;
}
q[i] = tq->q[index];
}
tq->head = 0;
tq->tail = tq->cap;
tq->cap *= 2;
free(tq->q);
tq->q = q;
return &q[tq->tail-1];
} else {
int i;
for (i=0;i<tq->cap-1;i++) {
int index = tq->head + i;
if (index >= tq->cap) {
index -= tq->cap;
}
if (tq->q[index].session == 0)
break;
}
int p = i + tq->head;
if (p >= tq->cap)
p -= tq->cap;
for (++i;i<tq->cap-1;i++) {
int index = tq->head + i;
if (index >= tq->cap) {
index -= tq->cap;
}
if (tq->q[index].session == 0) {
continue;
}
tq->q[p] = tq->q[index];
++ p;
if (p >= tq->cap) {
p -= tq->cap;
}
}
tq->tail = p + 1;
if (tq->tail >= tq->cap) {
tq->tail -= tq->cap;
}
return &tq->q[p];
}
}
static inline struct session_time *
last_one(struct tqueue * tq, struct session_time *current) {
--current;
if (current < tq->q) {
return &tq->q[tq->cap-1];
}
return current;
}
void
tqueue_push(struct tqueue *tq, int session, double time) {
assert(session !=0);
++ tq->n;
if (tq->head == tq->tail) {
// queue is empty;
tq->head = 0;
tq->tail = 1;
tq->q[0].session = session;
tq->q[0].time = time;
return;
}
struct session_time * st = &tq->q[tq->tail++];
if (tq->tail >= tq->cap) {
tq->tail = 0;
}
if (tq->tail == tq->head) {
st = expand_queue(tq);
}
st->session = session;
st->time = time;
// session must great than last one
int i;
for (i=1;i<tq->n;i++) {
struct session_time *last = last_one(tq, st);
if (session > last->session)
return;
// swap st, last
struct session_time temp = *last;
*last = *st;
*st = temp;
st = last;
}
}
double
tqueue_pop(struct tqueue *tq, int session) {
if (tq->head == tq->tail) {
return 0;
}
if (session == tq->q[tq->head].session) {
--tq->n;
double ret = tq->q[tq->head].time;
do {
++tq->head;
if (tq->head == tq->cap) {
tq->head = 0;
}
} while (tq->head != tq->tail && tq->q[tq->head].session == 0);
return ret;
}
// binary search session
int n = tq->tail - tq->head;
if (n < 0) {
n += tq->cap;
}
int begin = 1;
int end = n;
while (begin < end) {
int mid = (begin + end)/2;
int index = mid + tq->head;
if (index >= tq->cap) {
index -= tq->cap;
}
struct session_time *st = &tq->q[index];
if (st->session == session) {
--tq->n;
st->session = 0;
return st->time;
}
if (session > st->session) {
begin = mid + 1;
} else {
end = mid;
}
}
// not found
return 0;
}
void
tqueue_dump(struct tqueue *tq) {
printf("cap = %d, head = %d, tail = %d, n = %d\n",
tq->cap, tq->head, tq->tail, tq->n);
int head = tq->head;
while (head!=tq->tail) {
struct session_time *st = &tq->q[head];
if (st->session != 0) {
printf("%d : %f\n", st->session, st->time);
}
++ head;
if (head >= tq->cap)
head -= tq->cap;
}
}

13
lualib-src/timingqueue.h
View File

@@ -1,13 +0,0 @@
#ifndef skynet_timing_queue_h
#define skynet_timing_queue_h
struct tqueue * tqueue_new();
void tqueue_delete(struct tqueue *);
void tqueue_push(struct tqueue *, int session, double time);
double tqueue_pop(struct tqueue *, int session);
// for debug
void tqueue_dump(struct tqueue *tq);
#endif

172
lualib-src/trace_service.c
View File

@@ -1,172 +0,0 @@
#include "trace_service.h"
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <time.h>
#include <stdint.h>
#define HASH_SIZE 32
#if defined(__APPLE__)
#include <mach/task.h>
#include <mach/mach.h>
#endif
void
current_time(struct timespec *ti) {
#if !defined(__APPLE__)
clock_gettime(CLOCK_THREAD_CPUTIME_ID, ti);
#else
struct task_thread_times_info aTaskInfo;
mach_msg_type_number_t aTaskInfoCount = TASK_THREAD_TIMES_INFO_COUNT;
assert(KERN_SUCCESS == task_info(mach_task_self(), TASK_THREAD_TIMES_INFO, (task_info_t )&aTaskInfo, &aTaskInfoCount));
ti->tv_sec = aTaskInfo.user_time.seconds;
ti->tv_nsec = aTaskInfo.user_time.microseconds * 1000;
#endif
}
void
diff_time(struct timespec *ti, uint32_t *sec, uint32_t *nsec) {
struct timespec end;
current_time(&end);
int diffsec = end.tv_sec - ti->tv_sec;
assert(diffsec>=0);
int diffnsec = end.tv_nsec - ti->tv_nsec;
if (diffnsec < 0) {
--diffsec;
diffnsec += NANOSEC;
}
*nsec += diffnsec;
if (*nsec > NANOSEC) {
++*sec;
*nsec -= NANOSEC;
}
*sec += diffsec;
}
struct trace_info {
int session;
struct trace_info * prev;
struct trace_info * next;
struct timespec ti;
uint32_t ti_sec;
uint32_t ti_nsec;
};
struct trace_pool {
struct trace_info * current;
struct trace_info * slot[HASH_SIZE];
};
struct trace_pool *
trace_create() {
struct trace_pool * p = malloc(sizeof(*p));
memset(p, 0, sizeof(*p));
return p;
}
static void
_free_slot(struct trace_info *t) {
while (t) {
struct trace_info *next = t->next;
free(t);
t = next;
}
}
void
trace_release(struct trace_pool *p) {
int i;
for (i=0;i<HASH_SIZE;i++) {
_free_slot(p->slot[i]);
}
free(p->current);
}
struct trace_info *
trace_new(struct trace_pool *p) {
if (p->current) {
return NULL;
}
struct trace_info *t = malloc(sizeof(*t));
p->current = t;
t->session = 0;
t->prev = NULL;
t->next = NULL;
t->ti_sec = 0;
t->ti_nsec = 0;
current_time(&t->ti);
return t;
}
void
trace_register(struct trace_pool *p, int session) {
struct trace_info *t = p->current;
if (t == NULL) {
return;
}
int hash = session % HASH_SIZE;
assert(t->session == 0 && session > 0);
t->session = session;
t->prev = NULL;
t->next = p->slot[hash];
if (p->slot[hash]) {
p->slot[hash]->prev = t;
}
p->slot[hash] = t;
}
struct trace_info *
trace_yield(struct trace_pool *p) {
struct trace_info *t = p->current;
if (t == NULL)
return NULL;
diff_time(&t->ti,&t->ti_sec,&t->ti_nsec);
if (t->session == 0) {
return t;
} else {
p->current = NULL;
return NULL;
}
}
void
trace_switch(struct trace_pool *p, int session) {
assert(p->current == NULL && session > 0);
struct trace_info *t = p->slot[session % HASH_SIZE];
struct trace_info *prev = NULL;
while (t) {
if (t->session == session) {
p->current = t;
t->session = 0;
if (prev == NULL) {
p->slot[session % HASH_SIZE] = t->next;
} else {
prev->next = t->next;
}
if (t->next) {
t->next->prev = prev;
}
current_time(&t->ti);
return;
}
prev = t;
t=t->next;
}
}
double
trace_delete(struct trace_pool *p, struct trace_info *t) {
assert(p->current == t);
p->current = NULL;
if (t) {
double ti = (double)t->ti_sec + (double)t->ti_nsec / NANOSEC;
free(t);
return ti;
} else {
return 0;
}
}

23
lualib-src/trace_service.h
View File

@@ -1,23 +0,0 @@
#ifndef TRACE_SERVICE_H
#define TRACE_SERVICE_H
#include <time.h>
#include <stdint.h>
#define NANOSEC 1000000000
void current_time(struct timespec *ti);
void diff_time(struct timespec *ti, uint32_t *sec, uint32_t *nsec);
struct trace_pool;
struct trace_info;
struct trace_pool * trace_create();
void trace_release(struct trace_pool *);
struct trace_info * trace_new(struct trace_pool *);
void trace_register(struct trace_pool *, int session);
void trace_switch(struct trace_pool *, int session);
struct trace_info * trace_yield(struct trace_pool *);
double trace_delete(struct trace_pool *, struct trace_info *);
#endif

4
lualib/config.lua
View File

@@ -1,5 +1,3 @@
require "loadx"
local function config(path , pre)
assert(path)
local env = pre or {}
@@ -8,4 +6,4 @@ local function config(path , pre)
return env
end
return config
return config

40
lualib/loadx.lua
View File

@@ -1,40 +0,0 @@
if _VERSION == "Lua 5.2" then
return
end
local assert = assert
assert(_VERSION == "Lua 5.1")
local setfenv = setfenv
local loadstring = loadstring
local load = load
local io = io
function _G.load(ld, chunkname, mode, env)
local f,err
if type(ld) == "string" then
f,err = loadstring(ld, chunkname)
else
f,err = load(ld, chunkname)
end
if f == nil then
assert(f, err)
return f, err
end
if env then
setfenv(f, env)
end
return f,err
end
function _G.loadfile(filename, mode, env)
local f = io.open(filename, "rb")
assert(f, filename)
local source = f:read "*a"
f:close()
return _G.load(source, "@" .. filename, mode, env)
end
_G.loadstring = _G.load

24
lualib/localgroup.lua
View File

@@ -1,24 +0,0 @@
local skynet = require "skynet"
local lg
local group = {}
function group.create()
return skynet.call(lg, "lua", "NEW" , skynet.self())
end
group.address = assert(skynet.query_group)
group.enter = assert(skynet.enter_group)
group.leave = assert(skynet.leave_group)
function group.release(id)
skynet.send(lg, "lua" , "DELETE", id)
end
skynet.init(function()
lg = skynet.uniqueservice("group_local")
assert(lg)
end, "localgroup")
return group

1
lualib/logger.lua
View File

@@ -22,6 +22,7 @@ logger.CRITICAL = 50
logger.FATAL = 60
local function log_to_disk(...)
-- implement your .lualog service
Skynet.send(".lualog" , "lua" , ...)
end

35
lualib/mcgroup.lua
View File

@@ -1,35 +0,0 @@
local skynet = require "skynet"
local SERVICE
local group = {}
function group.create()
return skynet.call(SERVICE, "lua", "NEW" , skynet.self())
end
function group.address(id)
local send_id = id * 2 + 1
return skynet.query_group(send_id)
end
function group.enter(id, handle)
handle = handle or skynet.self()
skynet.send(SERVICE, "lua" , "ENTER", handle, id)
end
function group.leave(id, handle)
handle = handle or skynet.self()
skynet.send(SERVICE, "lua" , "LEAVE", handle, id)
end
function group.release(id)
skynet.send(SERVICE, "lua" , "DELETE", skynet.self(), id)
end
skynet.init(function()
SERVICE = skynet.uniqueservice(true, "group_mgr")
skynet.uniqueservice("group_agent", SERVICE)
end, "mcgroup")
return group

133
lualib/skynet.lua
View File

@@ -1,5 +1,4 @@
local c = require "skynet.c"
local mc = require "mcast.c"
local tostring = tostring
local tonumber = tonumber
local coroutine = coroutine
@@ -12,7 +11,7 @@ local skynet = {
-- read skynet.h
PTYPE_TEXT = 0,
PTYPE_RESPONSE = 1,
PTYPE_MULTICAST = 2,
-- PTYPE_MULTICAST = 2, -- DEPRECATED
PTYPE_CLIENT = 3,
PTYPE_SYSTEM = 4,
PTYPE_HARBOR = 5,
@@ -46,15 +45,9 @@ local watching_service = {}
local watching_session = {}
local error_queue = {}
-- timing remote call, turn off by default. Use skynet.timing_call() to turn on.
local timing_call = nil
-- suspend is function
local suspend
local trace_handle
local trace_func = function() end
local function string_to_handle(str)
return tonumber("0x" .. string.sub(str , 2))
end
@@ -66,7 +59,6 @@ local function dispatch_error_queue()
if session then
local co = session_id_coroutine[session]
session_id_coroutine[session] = nil
c.trace_switch(trace_handle, session)
return suspend(co, coroutine.resume(co, false))
end
end
@@ -139,18 +131,9 @@ local function dispatch_wakeup()
end
end
local function trace_count()
local info = c.trace_yield(trace_handle)
if info then
local ti = c.trace_delete(trace_handle, info)
trace_func(info, ti)
end
end
-- suspend is local function
function suspend(co, result, command, param, size)
if not result then
trace_count()
local session = session_coroutine_id[co]
local addr = session_coroutine_address[co]
if session and session ~= 0 then
@@ -161,22 +144,16 @@ function suspend(co, result, command, param, size)
error(debug.traceback(co,tostring(command)))
end
if command == "CALL" then
c.trace_register(trace_handle, param)
session_id_coroutine[param] = co
elseif command == "SLEEP" then
c.trace_register(trace_handle, param)
session_id_coroutine[param] = co
sleep_session[co] = param
elseif command == "RETURN" then
local co_session = session_coroutine_id[co]
local co_address = session_coroutine_address[co]
if param == nil then
trace_count()
error(debug.traceback(co))
end
-- c.send maybe throw a error, so call trace_count first.
-- The coroutine execute time after skynet.ret() will not be trace.
trace_count()
c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size)
return suspend(co, coroutine.resume(co))
elseif command == "EXIT" then
@@ -184,10 +161,8 @@ function suspend(co, result, command, param, size)
session_coroutine_id[co] = nil
session_coroutine_address[co] = nil
else
trace_count()
error("Unknown command : " .. command .. "\n" .. debug.traceback(co))
end
trace_count()
dispatch_wakeup()
dispatch_error_queue()
end
@@ -208,7 +183,6 @@ function skynet.sleep(ti)
local succ, ret = coroutine_yield("SLEEP", session)
sleep_session[coroutine.running()] = nil
if ret == true then
c.trace_switch(trace_handle, session)
return "BREAK"
end
end
@@ -220,7 +194,6 @@ end
function skynet.wait()
local session = c.genid()
coroutine_yield("SLEEP", session)
c.trace_switch(trace_handle, session)
local co = coroutine.running()
sleep_session[co] = nil
session_id_coroutine[session] = nil
@@ -291,15 +264,7 @@ function skynet.send(addr, typename, ...)
return c.send(addr, p.id, 0 , p.pack(...))
end
function skynet.cast(group, typename, ...)
local p = proto[typename]
if #group > 0 then
return c.send(".cast", p.id, 0, mc(group, p.pack(...)))
end
end
skynet.genid = assert(c.genid)
skynet.forward = assert(c.forward)
skynet.redirect = function(dest,source,typename,...)
return c.redirect(dest, source, proto[typename].id, ...)
@@ -387,33 +352,15 @@ function skynet.fork(func,...)
table.insert(fork_queue, co)
end
local function timing(session, source, ti)
if ti == nil then
return
end
local t = timing_call[source]
if t == nil then
t = { n = 1, ti = ti }
timing_call[source] = t
else
t.n = t.n + 1
t.ti = t.ti + ti
end
end
local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
local function raw_dispatch_message(prototype, msg, sz, session, source)
-- skynet.PTYPE_RESPONSE = 1, read skynet.h
if prototype == 1 then
if timing_call then
timing(session, source, ...)
end
local co = session_id_coroutine[session]
if co == "BREAK" then
session_id_coroutine[session] = nil
elseif co == nil then
unknown_response(session, source, msg, sz)
else
c.trace_switch(trace_handle, session)
session_id_coroutine[session] = nil
suspend(co, coroutine.resume(co, true, msg, sz))
end
@@ -424,7 +371,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
local co = co_create(f)
session_coroutine_id[co] = session
session_coroutine_address[co] = source
suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz,...)))
suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz)))
else
print("Unknown request :" , p.unpack(msg,sz))
error(string.format("Can't dispatch type %s : ", p.name))
@@ -493,22 +440,6 @@ local function group_command(cmd, handle, address)
end
end
function skynet.enter_group(handle , address)
c.command("GROUP", group_command("ENTER", handle, address))
end
function skynet.leave_group(handle , address)
c.command("GROUP", group_command("LEAVE", handle, address))
end
function skynet.clear_group(handle)
c.command("GROUP", "CLEAR " .. tostring(handle))
end
function skynet.query_group(handle)
return string_to_handle(c.command("GROUP","QUERY " .. tostring(handle)))
end
function skynet.address(addr)
if type(addr) == "number" then
return string.format(":%x",addr)
@@ -531,12 +462,6 @@ end
----- debug
local internal_info_func
function skynet.info_func(func)
internal_info_func = func
end
local dbgcmd = {}
function dbgcmd.MEM()
@@ -549,15 +474,8 @@ function dbgcmd.GC()
collectgarbage "collect"
end
local function query_state(stat, what)
stat[what] = c.stat(what)
end
function dbgcmd.STAT()
local stat = {}
query_state(stat, "count")
query_state(stat, "time")
stat.boottime = debug.getregistry().skynet_boottime
stat.mqlen = skynet.mqlen()
skynet.ret(skynet.pack(stat))
end
@@ -570,23 +488,6 @@ function dbgcmd.INFO()
end
end
function dbgcmd.TIMING()
if timing_call then
skynet.ret(skynet.pack(timing_call))
-- turn off timing
timing_call = nil
else
-- turn on timing
timing_call = {}
skynet.ret(skynet.pack(timing_call))
end
end
function dbgcmd.RELOAD(...)
local cmd = table.concat({...}, " ")
c.reload(cmd)
end
local function _debug_dispatch(session, address, cmd, ...)
local f = dbgcmd[cmd]
assert(f, cmd)
@@ -686,7 +587,6 @@ end
function skynet.start(start_func)
c.callback(dispatch_message)
trace_handle = assert(c.stat "trace")
skynet.timeout(0, function()
init_service(start_func)
end)
@@ -696,24 +596,11 @@ function skynet.filter(f ,start_func)
c.callback(function(...)
dispatch_message(f(...))
end)
trace_handle = assert(c.stat "trace")
skynet.timeout(0, function()
init_service(start_func)
end)
end
function skynet.trace()
return c.trace_new(trace_handle)
end
function skynet.trace_session(session)
return c.trace_register(trace_handle, session)
end
function skynet.trace_callback(func)
trace_func = func
end
function skynet.endless()
return c.command("ENDLESS")~=nil
end
@@ -722,10 +609,6 @@ function skynet.abort()
c.command("ABORT")
end
function skynet.context_ptr()
return c.context()
end
function skynet.monitor(service, query)
local monitor
if query then
@@ -741,14 +624,4 @@ function skynet.mqlen()
return tonumber(c.command "MQLEN")
end
function skynet.timing_call(tc)
local ret = timing_call
timing_call = tc or {}
return ret
end
function skynet.timing_session(session)
return c.timing(session)
end
return skynet

13
service-src/localcast.h
View File

@@ -1,13 +0,0 @@
#ifndef SKYNET_LOCAL_CAST_H
#define SKYNET_LOCAL_CAST_H
#include <stdint.h>
struct localcast {
int n;
const uint32_t * group;
void *msg;
size_t sz;
};
#endif

29
service-src/service_localcast.c
View File

@@ -1,29 +0,0 @@
#include "skynet.h"
#include "skynet_handle.h"
#include "skynet_multicast.h"
#include "localcast.h"
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
static int
_maincb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
const struct localcast *lc = msg;
size_t s = lc->sz | type << HANDLE_REMOTE_SHIFT;
struct skynet_multicast_message * mc = skynet_multicast_create(lc->msg, s, source);
skynet_multicast_cast(context, mc, lc->group, lc->n);
free((void *)lc->group);
return 0;
}
int
localcast_init(void * ud, struct skynet_context *ctx, const char * args) {
skynet_callback(ctx, ud, _maincb);
skynet_command(ctx, "REG", ".cast");
return 0;
}

2
service-src/service_lua.h
View File

@@ -3,9 +3,7 @@
struct snlua {
lua_State * L;
const char * reload;
struct skynet_context * ctx;
struct tqueue * tq;
int (*init)(struct snlua *l, struct skynet_context *ctx, const char * args);
};

61
service-src/service_multicast.c
View File

@@ -1,61 +0,0 @@
#include "skynet.h"
#include "skynet_handle.h"
#include "skynet_multicast.h"
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
struct skynet_multicast_group *
multicast_create() {
return skynet_multicast_newgroup();
}
void
multicast_release(struct skynet_multicast_group *g) {
skynet_multicast_deletegroup(g);
}
static int
_maincb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
struct skynet_multicast_group *g = ud;
if (type == PTYPE_SYSTEM) {
char cmd = '\0';
uint32_t handle = 0;
sscanf(msg,"%c %x",&cmd,&handle);
if (handle == 0) {
skynet_error(context, "Invalid handle %s",msg);
return 0;
}
switch (cmd) {
case 'E':
skynet_multicast_entergroup(g, handle);
break;
case 'L':
skynet_multicast_leavegroup(g, handle);
break;
case 'C':
skynet_command(context, "EXIT", NULL);
break;
default:
skynet_error(context, "Invalid command %s",msg);
break;
}
return 0;
} else {
sz |= type << HANDLE_REMOTE_SHIFT;
struct skynet_multicast_message * mc = skynet_multicast_create(msg, sz, source);
skynet_multicast_castgroup(context, g, mc);
return 1;
}
}
int
multicast_init(struct skynet_multicast_group *g, struct skynet_context *ctx, const char * args) {
skynet_callback(ctx, g, _maincb);
return 0;
}

22
service-src/service_tunnel.c
View File

@@ -1,22 +0,0 @@
#include "skynet.h"
#include <stdio.h>
static int
_cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
uint32_t dest = (uint32_t)(uintptr_t)ud;
skynet_forward(context, dest);
return 0;
}
int
tunnel_init(void * dummy, struct skynet_context *ctx, const char * args) {
uint32_t dest = skynet_queryname(ctx, args);
if (dest == 0) {
skynet_error(ctx, "Can't create tunnel to %s",args);
return 1;
}
skynet_callback(ctx, (void*)(intptr_t)dest, _cb);
return 0;
}

2
service/debug_console.lua
View File

@@ -105,11 +105,9 @@ function COMMAND.help()
list = "List all the service",
stat = "Dump all stats",
info = "Info address : get service infomation",
timing = "timing address : get service timing infomation",
kill = "kill address : kill service",
mem = "mem : show memory status",
gc = "gc : force every lua service do garbage collect",
reload = "reload address : reload a lua service",
start = "lanuch a new lua service",
clearcache = "clear lua code cache",
}

60
service/group_agent.lua
View File

@@ -1,60 +0,0 @@
local skynet = require "skynet"
local group_mgr = ...
group_mgr = tonumber(group_mgr)
local command = {}
local group = {}
function command.CREATE(id)
assert(group[id] == nil)
group[id] = {}
local recv_id = id * 2
local addr = skynet.query_group(recv_id)
skynet.ret(skynet.pack(addr))
end
function command.AGENT(id, handle)
local tunnel = skynet.launch("tunnel", skynet.address(handle))
assert(tunnel)
local send_id = id * 2 + 1
skynet.enter_group(send_id, tunnel)
table.insert(group[id] , tunnel)
end
function command.ENTER(id, handle)
local recv_id = id * 2
local send_id = id * 2 + 1
skynet.enter_group(recv_id, handle)
skynet.enter_group(send_id, handle)
end
function command.LEAVE(id, handle)
local recv_id = id * 2
local send_id = id * 2 + 1
skynet.leave_group(recv_id, handle)
skynet.leave_group(send_id, handle)
end
function command.CLEAR(id)
local g = group[id]
assert(g)
local recv_id = id * 2
local send_id = id * 2 + 1
skynet.clear_group(recv_id)
skynet.clear_group(send_id)
group[id] = nil
for _,v in ipairs(g) do
skynet.kill(v)
end
end
skynet.start(function()
skynet.dispatch("lua", function(session, address, cmd , id , handle)
local f = command[cmd]
assert(f, cmd)
f(id,handle)
end)
skynet.send(group_mgr , "lua", "MASTER", skynet.self())
end)

32
service/group_local.lua
View File

@@ -1,32 +0,0 @@
local skynet = require "skynet"
local local_group = {}
for i=1,10000 do
local_group[i] = true
end
local command = {}
function command.NEW()
local k = next(local_group)
if k then
local_group[k] = nil
skynet.ret(skynet.pack(k))
else
skynet.ret(skynet.pack(nil))
end
end
function command.DELETE(id)
assert(handle_group[id] == nil)
skynet.clear_group(id)
local_group[id] = true
end
skynet.start(function()
skynet.dispatch("lua", function(_, _, cmd, id)
local f = command[cmd]
assert(f, cmd, id)
f(id)
end)
end)

71
service/group_mgr.lua
View File

@@ -1,71 +0,0 @@
local skynet = require "skynet"
-- read group_local for id 10000
local id = 10000
local harbor_ctrl = {}
local multicast = {}
local command = {}
function command.MASTER(address, harbor)
harbor_ctrl[harbor] = address
end
local function create_group_in(harbor, id)
local local_ctrl = assert(harbor_ctrl[harbor])
local g = multicast[id]
g[harbor] = false
local local_mc = skynet.call(local_ctrl, "lua", "CREATE", id)
for _,mc in pairs(g) do
if mc then
skynet.send(local_ctrl, "lua", "AGENT", id, mc)
end
end
for harbor_id,ctrl in pairs(harbor_ctrl) do
if harbor_id ~= harbor then
skynet.send(ctrl, "lua", "AGENT", id, local_mc)
end
end
g[harbor] = local_mc
end
function command.NEW()
id = id + 1
multicast[id] = {}
skynet.ret(skynet.pack(id))
end
function command.ENTER(address, harbor, id)
local g = multicast[id]
assert(g,id)
if g[harbor] == nil then
create_group_in(harbor, id)
end
local local_ctrl = assert(harbor_ctrl[harbor])
skynet.send(local_ctrl, "lua", "ENTER", id, address)
end
function command.LEAVE(address, harbor, id)
assert(multicast[id], id)
local local_ctrl = assert(harbor_ctrl[harbor])
skynet.send(local_ctrl, "lua", "LEAVE", id, address)
end
function command.DELETE(_,_, id)
assert(multicast[id], id)
multicast[id] = nil
for harbor_id,_ in pairs(g) do
skynet.send(harbor_ctrl[harbor_id], "lua", "CLEAR", id)
end
end
skynet.start(function()
skynet.dispatch("lua", function(_, _, cmd, address, param)
local f = command[cmd]
assert(f, cmd, param)
local harbor = skynet.harbor(address)
f(address, harbor, param)
end)
end)

3
skynet-src/skynet.h
View File

@@ -6,7 +6,7 @@
#define PTYPE_TEXT 0
#define PTYPE_RESPONSE 1
#define PTYPE_MULTICAST 2
#define PTYPE_MULTICAST_DEPRECATED 2
#define PTYPE_CLIENT 3
#define PTYPE_SYSTEM 4
#define PTYPE_HARBOR 5
@@ -29,7 +29,6 @@ uint32_t skynet_queryname(struct skynet_context * context, const char * name);
int skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int type, int session, void * msg, size_t sz);
int skynet_sendname(struct skynet_context * context, const char * destination , int type, int session, void * msg, size_t sz);
void skynet_forward(struct skynet_context *, uint32_t destination);
int skynet_isremote(struct skynet_context *, uint32_t handle, int * harbor);
typedef int (*skynet_cb)(struct skynet_context * context, void *ud, int type, int session, uint32_t source , const void * msg, size_t sz);

149
skynet-src/skynet_group.c
View File

@@ -1,149 +0,0 @@
#include "skynet_group.h"
#include "skynet_multicast.h"
#include "skynet_server.h"
#include "skynet.h"
#include <string.h>
#include <stdio.h>
#include <assert.h>
#define HASH_SIZE 1024
struct group_node {
int handle;
struct skynet_context *ctx;
struct group_node * next;
};
struct group {
int lock;
struct group_node * node[HASH_SIZE];
};
struct group * _G = NULL;
inline static void
_lock(struct group *g) {
while (__sync_lock_test_and_set(&g->lock,1)) {}
}
inline static void
_unlock(struct group *g) {
__sync_lock_release(&g->lock);
}
static struct skynet_context *
_create_group(struct group * g, int handle) {
int hash = handle % HASH_SIZE;
struct skynet_context * inst = skynet_context_new("multicast",NULL);
assert(inst);
struct group_node * new_node = malloc(sizeof(struct group_node));
new_node->handle = handle;
new_node->ctx = inst;
new_node->next = g->node[hash];
g->node[hash] = new_node;
return inst;
}
uint32_t
skynet_group_query(int handle) {
struct group *g = _G;
_lock(g);
int hash = handle % HASH_SIZE;
struct group_node * node = g->node[hash];
while (node) {
if (node->handle == handle) {
struct skynet_context * ctx = node->ctx;
uint32_t addr = skynet_context_handle(ctx);
_unlock(g);
return addr;
}
node = node->next;
}
struct skynet_context * ctx = _create_group(g, handle);
uint32_t addr = skynet_context_handle(ctx);
_unlock(g);
return addr;
}
static void
send_command(struct skynet_context *ctx, const char * cmd, uint32_t node) {
char * tmp = malloc(16);
int n = sprintf(tmp, "%s %x", cmd, node);
skynet_context_send(ctx, tmp, n+1 , 0, PTYPE_SYSTEM, 0);
}
void
skynet_group_enter(int handle, uint32_t n) {
struct group *g = _G;
_lock(g);
int hash = handle % HASH_SIZE;
struct group_node * node = g->node[hash];
while (node) {
if (node->handle == handle) {
send_command(node->ctx, "E", n);
_unlock(g);
return;
}
node = node->next;
}
struct skynet_context * inst = _create_group(g, handle);
send_command(inst, "E", n);
_unlock(g);
}
void
skynet_group_leave(int handle, uint32_t n) {
struct group *g = _G;
_lock(g);
int hash = handle % HASH_SIZE;
struct group_node * node = g->node[hash];
while (node) {
if (node->handle == handle) {
send_command(node->ctx, "L", n);
break;
}
node = node->next;
}
_unlock(g);
}
void
skynet_group_clear(int handle) {
struct group *g = _G;
_lock(g);
int hash = handle % HASH_SIZE;
struct group_node ** pnode = &g->node[hash];
while (*pnode) {
struct group_node * node = *pnode;
if (node->handle == handle) {
struct skynet_context * ctx = node->ctx;
char * cmd = malloc(8);
int n = sprintf(cmd, "C");
skynet_context_send(ctx, cmd, n+1, 0 , PTYPE_SYSTEM, 0);
*pnode = node->next;
free(node);
break;
}
pnode = &node->next;
}
_unlock(g);
}
void
skynet_group_init() {
struct group * g = malloc(sizeof(*g));
memset(g,0,sizeof(*g));
_G = g;
}

13
skynet-src/skynet_group.h
View File

@@ -1,13 +0,0 @@
#ifndef SKYNET_GROUP_H
#define SKYNET_GROUP_H
#include <stdint.h>
uint32_t skynet_group_query(int handle);
void skynet_group_enter(int handle, uint32_t node);
void skynet_group_leave(int handle, uint32_t node);
void skynet_group_clear(int handle);
void skynet_group_init();
#endif

9
skynet-src/skynet_mq.c
View File

@@ -1,7 +1,6 @@
#include "skynet.h"
#include "skynet_mq.h"
#include "skynet_handle.h"
#include "skynet_multicast.h"
#include <stdio.h>
#include <stdlib.h>
@@ -287,13 +286,7 @@ _drop_queue(struct message_queue *q) {
int s = 0;
while(!skynet_mq_pop(q, &msg)) {
++s;
int type = msg.sz >> HANDLE_REMOTE_SHIFT;
if (type == PTYPE_MULTICAST) {
assert((msg.sz & HANDLE_MASK) == 0);
skynet_multicast_dispatch((struct skynet_multicast_message *)msg.data, NULL, NULL);
} else {
free(msg.data);
}
free(msg.data);
}
_release(q);
return s;

237
skynet-src/skynet_multicast.c
View File

@@ -1,237 +0,0 @@
#include "skynet.h"
#include "skynet_multicast.h"
#include "skynet_server.h"
#include "skynet_handle.h"
#include <stdlib.h>
#include <string.h>
struct skynet_multicast_message {
int ref;
const void * msg;
size_t sz;
uint32_t source;
};
struct skynet_multicast_message *
skynet_multicast_create(const void * msg, size_t sz, uint32_t source) {
struct skynet_multicast_message * mc = malloc(sizeof(*mc));
mc->ref = 0;
mc->msg = msg;
mc->sz = sz;
mc->source = source;
return mc;
}
void
skynet_multicast_copy(struct skynet_multicast_message *mc, int copy) {
int r = __sync_add_and_fetch(&mc->ref, copy);
if (r == 0) {
free((void *)mc->msg);
free(mc);
}
}
void
skynet_multicast_dispatch(struct skynet_multicast_message * msg, void * ud, skynet_multicast_func func) {
if (func) {
func(ud, msg->source, msg->msg, msg->sz);
}
int ref = __sync_sub_and_fetch(&msg->ref, 1);
if (ref == 0) {
free((void *)msg->msg);
free(msg);
}
}
struct array {
int cap;
int number;
uint32_t *data;
};
struct skynet_multicast_group {
struct array enter_queue;
struct array leave_queue;
int cap;
int number;
uint32_t * data;
};
struct skynet_multicast_group *
skynet_multicast_newgroup() {
struct skynet_multicast_group * g = malloc(sizeof(*g));
memset(g,0,sizeof(*g));
return g;
}
void
skynet_multicast_deletegroup(struct skynet_multicast_group * g) {
free(g->data);
free(g->enter_queue.data);
free(g->leave_queue.data);
free(g);
}
static void
push_array(struct array * a, uint32_t v) {
if (a->number >= a->cap) {
a->cap *= 2;
if (a->cap == 0) {
a->cap = 4;
}
a->data = realloc(a->data, a->cap * sizeof(uint32_t));
}
a->data[a->number++] = v;
}
void
skynet_multicast_entergroup(struct skynet_multicast_group * group, uint32_t handle) {
push_array(&group->enter_queue, handle);
}
void
skynet_multicast_leavegroup(struct skynet_multicast_group * group, uint32_t handle) {
push_array(&group->leave_queue, handle);
}
static int
compar_uint(const void *a, const void *b) {
const uint32_t * aa = a;
const uint32_t * bb = b;
return (int)(*aa - *bb);
}
static void
combine_queue(struct skynet_context * from, struct skynet_multicast_group * group) {
qsort(group->enter_queue.data, group->enter_queue.number, sizeof(uint32_t), compar_uint);
qsort(group->leave_queue.data, group->leave_queue.number, sizeof(uint32_t), compar_uint);
int i;
int enter = group->enter_queue.number;
uint32_t last = 0;
int new_size = group->number + enter;
if (new_size > group->cap) {
group->data = realloc(group->data, new_size * sizeof(uint32_t));
group->cap = new_size;
}
// combine enter queue
int old_index = group->number - 1;
int new_index = new_size - 1;
for (i= enter - 1;i >=0 ; i--) {
uint32_t handle = group->enter_queue.data[i];
if (handle == last)
continue;
last = handle;
if (old_index < 0) {
group->data[new_index] = handle;
} else {
uint32_t p = group->data[old_index];
if (handle == p)
continue;
if (handle > p) {
group->data[new_index] = handle;
} else {
group->data[new_index] = group->data[old_index];
--old_index;
last = 0;
++i;
}
}
--new_index;
}
while (old_index >= 0) {
group->data[new_index] = group->data[old_index];
--old_index;
--new_index;
}
group->enter_queue.number = 0;
// remove leave queue
old_index = new_index + 1;
new_index = 0;
int count = new_size - old_index;
int leave = group->leave_queue.number;
for (i=0;i<leave;i++) {
if (old_index >= new_size) {
count = 0;
break;
}
uint32_t handle = group->leave_queue.data[i];
uint32_t p = group->data[old_index];
if (handle == p) {
--count;
++old_index;
} else if ( handle > p) {
group->data[new_index] = group->data[old_index];
++new_index;
++old_index;
--i;
} else {
skynet_error(from, "Try to remove a none exist handle : %x", handle);
}
}
while (new_index < count) {
group->data[new_index] = group->data[old_index];
++new_index;
++old_index;
}
group->leave_queue.number = 0;
group->number = new_index;
}
int
skynet_multicast_castgroup(struct skynet_context * from, struct skynet_multicast_group * group, struct skynet_multicast_message *msg) {
combine_queue(from, group);
int release = 0;
if (group->number > 0) {
uint32_t source = skynet_context_handle(from);
skynet_multicast_copy(msg, group->number);
int i;
for (i=0;i<group->number;i++) {
uint32_t p = group->data[i];
struct skynet_context * ctx = skynet_handle_grab(p);
if (ctx) {
skynet_context_send(ctx, msg, 0 , source, PTYPE_MULTICAST , 0);
skynet_context_release(ctx);
} else {
skynet_multicast_leavegroup(group, p);
++release;
}
}
}
skynet_multicast_copy(msg, -release);
return group->number - release;
}
void
skynet_multicast_cast(struct skynet_context * from, struct skynet_multicast_message *msg, const uint32_t *dests, int n) {
uint32_t source = skynet_context_handle(from);
skynet_multicast_copy(msg, n);
if (n == 0)
return;
int i;
int release = 0;
for (i=0;i<n;i++) {
uint32_t p = dests[i];
struct skynet_context * ctx = skynet_handle_grab(p);
if (ctx) {
skynet_context_send(ctx, msg, 0 , source, PTYPE_MULTICAST , 0);
skynet_context_release(ctx);
} else {
++release;
}
}
if (release != 0) {
skynet_multicast_copy(msg, -release);
}
}

24
skynet-src/skynet_multicast.h
View File

@@ -1,24 +0,0 @@
#ifndef SKYNET_MULTICAST_H
#define SKYNET_MULTICAST_H
#include <stddef.h>
#include <stdint.h>
struct skynet_multicast_message;
struct skynet_multicast_group;
struct skynet_context;
typedef void (*skynet_multicast_func)(void *ud, uint32_t source, const void * msg, size_t sz);
struct skynet_multicast_message * skynet_multicast_create(const void * msg, size_t sz, uint32_t source);
void skynet_multicast_copy(struct skynet_multicast_message *, int copy);
void skynet_multicast_dispatch(struct skynet_multicast_message * msg, void * ud, skynet_multicast_func func);
void skynet_multicast_cast(struct skynet_context * from, struct skynet_multicast_message *msg, const uint32_t *dests, int n);
struct skynet_multicast_group * skynet_multicast_newgroup();
void skynet_multicast_deletegroup(struct skynet_multicast_group * group);
void skynet_multicast_entergroup(struct skynet_multicast_group * group, uint32_t handle);
void skynet_multicast_leavegroup(struct skynet_multicast_group * group, uint32_t handle);
int skynet_multicast_castgroup(struct skynet_context * from, struct skynet_multicast_group * group, struct skynet_multicast_message *msg);
#endif

118
skynet-src/skynet_server.c
View File

@@ -6,8 +6,6 @@
#include "skynet_harbor.h"
#include "skynet_env.h"
#include "skynet.h"
#include "skynet_multicast.h"
#include "skynet_group.h"
#include "skynet_monitor.h"
#include <string.h>
@@ -41,7 +39,6 @@ struct skynet_context {
void * cb_ud;
skynet_cb cb;
int session_id;
uint32_t forward;
struct message_queue *queue;
bool init;
bool endless;
@@ -102,7 +99,6 @@ skynet_context_new(const char * name, const char *param) {
ctx->cb_ud = NULL;
ctx->session_id = 0;
ctx->forward = 0;
ctx->init = false;
ctx->endless = false;
ctx->handle = skynet_handle_register(ctx);
@@ -192,67 +188,14 @@ skynet_isremote(struct skynet_context * ctx, uint32_t handle, int * harbor) {
return ret;
}
static void
_send_message(uint32_t des, struct skynet_message *msg) {
if (skynet_harbor_message_isremote(des)) {
struct remote_message * rmsg = malloc(sizeof(*rmsg));
rmsg->destination.handle = des;
rmsg->message = msg->data;
rmsg->sz = msg->sz;
skynet_harbor_send(rmsg, msg->source, msg->session);
} else {
if (skynet_context_push(des, msg)) {
free(msg->data);
skynet_error(NULL, "Drop message from %x forward to %x (size=%d)", msg->source, des, (int)msg->sz);
}
}
}
static int
_forwarding(struct skynet_context *ctx, struct skynet_message *msg) {
if (ctx->forward) {
uint32_t des = ctx->forward;
ctx->forward = 0;
_send_message(des, msg);
return 1;
}
return 0;
}
static void
_mc(void *ud, uint32_t source, const void * msg, size_t sz) {
struct skynet_context * ctx = ud;
int type = sz >> HANDLE_REMOTE_SHIFT;
sz &= HANDLE_MASK;
ctx->cb(ctx, ctx->cb_ud, type, 0, source, msg, sz);
if (ctx->forward) {
uint32_t des = ctx->forward;
ctx->forward = 0;
struct skynet_message message;
message.source = source;
message.session = 0;
message.data = malloc(sz);
memcpy(message.data, msg, sz);
message.sz = sz | (type << HANDLE_REMOTE_SHIFT);
_send_message(des, &message);
}
}
static void
_dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
assert(ctx->init);
CHECKCALLING_BEGIN(ctx)
int type = msg->sz >> HANDLE_REMOTE_SHIFT;
size_t sz = msg->sz & HANDLE_MASK;
if (type == PTYPE_MULTICAST) {
skynet_multicast_dispatch((struct skynet_multicast_message *)msg->data, ctx, _mc);
} else {
int reserve = ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz);
reserve |= _forwarding(ctx, msg);
if (!reserve) {
free(msg->data);
}
}
ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz);
free(msg->data);
CHECKCALLING_END(ctx)
}
@@ -308,41 +251,6 @@ _copy_name(char name[GLOBALNAME_LENGTH], const char * addr) {
}
}
static const char *
_group_command(struct skynet_context * ctx, const char * cmd, int handle, uint32_t v) {
uint32_t self;
if (v != 0) {
if (skynet_harbor_message_isremote(v)) {
skynet_error(ctx, "Can't add remote handle %x",v);
return NULL;
}
self = v;
} else {
self = ctx->handle;
}
if (strcmp(cmd, "ENTER") == 0) {
skynet_group_enter(handle, self);
return NULL;
}
if (strcmp(cmd, "LEAVE") == 0) {
skynet_group_leave(handle, self);
return NULL;
}
if (strcmp(cmd, "QUERY") == 0) {
uint32_t addr = skynet_group_query(handle);
if (addr == 0) {
return NULL;
}
_id_to_hex(ctx->result, addr);
return ctx->result;
}
if (strcmp(cmd, "CLEAR") == 0) {
skynet_group_clear(handle);
return NULL;
}
return NULL;
}
uint32_t
skynet_queryname(struct skynet_context * context, const char * name) {
switch(name[0]) {
@@ -514,18 +422,6 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
return context->result;
}
if (strcmp(cmd,"GROUP") == 0) {
int sz = strlen(param);
char tmp[sz+1];
strcpy(tmp,param);
tmp[sz] = '\0';
char cmd[sz+1];
int handle=0;
uint32_t addr=0;
sscanf(tmp, "%s %d :%x",cmd,&handle,&addr);
return _group_command(context, cmd, handle,addr);
}
if (strcmp(cmd,"ENDLESS") == 0) {
if (context->endless) {
strcpy(context->result, "1");
@@ -572,16 +468,6 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
return NULL;
}
void
skynet_forward(struct skynet_context * context, uint32_t destination) {
assert(context->forward == 0);
if (destination == 0) {
context->forward = context->handle;
} else {
context->forward = destination;
}
}
static void
_filter_args(struct skynet_context * context, int type, int *session, void ** data, size_t * sz) {
int needcopy = !(type & PTYPE_TAG_DONTCOPY);

7
skynet-src/skynet_start.c
View File

@@ -6,7 +6,6 @@
#include "skynet_module.h"
#include "skynet_timer.h"
#include "skynet_harbor.h"
#include "skynet_group.h"
#include "skynet_monitor.h"
#include "skynet_socket.h"
@@ -188,7 +187,6 @@ _start_master(const char * master) {
void
skynet_start(struct skynet_config * config) {
skynet_group_init();
skynet_harbor_init(config->harbor);
skynet_handle_init(config->harbor);
skynet_mq_init();
@@ -215,11 +213,6 @@ skynet_start(struct skynet_config * config) {
return;
}
ctx = skynet_context_new("localcast", NULL);
if (ctx == NULL) {
fprintf(stderr,"launch local cast error");
exit(1);
}
ctx = skynet_context_new("snlua", "launcher");
if (ctx) {
skynet_command(ctx, "REG", ".launcher");

9
test/testdb.lua
View File

@@ -1,9 +0,0 @@
local skynet = require "skynet"
skynet.start(function()
local result = skynet.call("SIMPLEDB","text","SET","foobar","hello")
print(result)
result = skynet.call("SIMPLEDB","text","GET","foobar")
print(result)
skynet.exit()
end)

19
test/testgroup.lua
View File

@@ -1,19 +0,0 @@
local skynet = require "skynet"
local group = require "mcgroup"
--local group = require "localgroup"
skynet.start(function()
local gid = group.create()
local gaddr = group.address(gid)
local g = {}
print("=== Create Group ===",gid,skynet.address(gaddr))
for i=1,10 do
local address = skynet.newservice("testgroup_c", tostring(i))
table.insert(g, address)
group.enter(gid , address)
end
skynet.cast(g,"text","Cast")
skynet.sleep(1000)
skynet.send(gaddr,"text","Hello World")
skynet.exit()
end)

13
test/testgroup_c.lua
View File

@@ -1,13 +0,0 @@
local skynet = require "skynet"
local group = require "mcgroup"
--local group = require "localgroup"
local id, g = ...
skynet.start(function()
skynet.dispatch("text",function(session,address,text)
print("===>",id, text)
end)
if g then
group.enter(tonumber(g))
end
end)

10
test/testqueue.lua
View File

@@ -1,10 +0,0 @@
local skynet = require "skynet"
local mqueue = require "mqueue"
skynet.start(function()
local pingqueue = skynet.newservice "pingqueue"
print(mqueue.call(pingqueue, "A"))
print(mqueue.call(pingqueue, "B"))
print(mqueue.call(pingqueue, "C"))
skynet.exit()
end)

28
test/testtrace.lua
View File

@@ -1,28 +0,0 @@
local skynet = require "skynet"
local trace_cache = {}
skynet.trace_callback(function(handle, ti)
-- skynet will call this function when the thread which create handle(traceid) end, and pass the time (ti)
print("TRACE",trace_cache[handle],ti)
trace_cache[handle] = nil
end)
local function f (i)
-- create a trace object
local traceid = skynet.trace()
-- name traceid in trace_cache[]
trace_cache[traceid] = "thread " .. i
for i=1,i do
skynet.sleep(i)
end
end
skynet.start(function()
for i = 1 , 100 do
skynet.fork(f, i)
end
end)