From 8ef978d0eb2874c6cf3ecfbfd4efe150d0e1bae7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Wed, 1 Aug 2012 11:33:10 +0800 Subject: [PATCH] hello world --- Makefile | 26 +++ agent.lua | 7 + client.c | 58 ++++++ console.lua | 14 ++ gate/README | 1 + gate/main.c | 250 +++++++++++++++++++++++ gate/map.c | 139 +++++++++++++ gate/map.h | 13 ++ gate/mread.c | 486 +++++++++++++++++++++++++++++++++++++++++++++ gate/mread.h | 16 ++ gate/ringbuffer.c | 277 ++++++++++++++++++++++++++ gate/ringbuffer.h | 27 +++ lua-skynet.c | 113 +++++++++++ main.lua | 12 ++ rwlock.h | 48 +++++ service_lua.c | 72 +++++++ skynet.h | 15 ++ skynet_blackhole.c | 22 ++ skynet_blackhole.h | 13 ++ skynet_error.c | 41 ++++ skynet_handle.c | 222 +++++++++++++++++++++ skynet_handle.h | 15 ++ skynet_imp.h | 13 ++ skynet_logger.c | 52 +++++ skynet_main.c | 16 ++ skynet_module.c | 138 +++++++++++++ skynet_module.h | 26 +++ skynet_mq.c | 102 ++++++++++ skynet_mq.h | 25 +++ skynet_server.c | 259 ++++++++++++++++++++++++ skynet_server.h | 16 ++ skynet_start.c | 62 ++++++ skynet_timer.c | 206 +++++++++++++++++++ skynet_timer.h | 12 ++ testtimer.lua | 10 + watchdog.lua | 34 ++++ 36 files changed, 2858 insertions(+) create mode 100644 Makefile create mode 100644 agent.lua create mode 100644 client.c create mode 100644 console.lua create mode 100644 gate/README create mode 100644 gate/main.c create mode 100644 gate/map.c create mode 100644 gate/map.h create mode 100644 gate/mread.c create mode 100644 gate/mread.h create mode 100644 gate/ringbuffer.c create mode 100644 gate/ringbuffer.h create mode 100644 lua-skynet.c create mode 100644 main.lua create mode 100644 rwlock.h create mode 100644 service_lua.c create mode 100644 skynet.h create mode 100644 skynet_blackhole.c create mode 100644 skynet_blackhole.h create mode 100644 skynet_error.c create mode 100644 skynet_handle.c create mode 100644 skynet_handle.h create mode 100644 skynet_imp.h create mode 100644 skynet_logger.c create mode 100644 skynet_main.c create mode 100644 skynet_module.c create mode 100644 skynet_module.h create mode 100644 skynet_mq.c create mode 100644 skynet_mq.h create mode 100644 skynet_server.c create mode 100644 skynet_server.h create mode 100644 skynet_start.c create mode 100644 skynet_timer.c create mode 100644 skynet_timer.h create mode 100644 testtimer.lua create mode 100644 watchdog.lua diff --git a/Makefile b/Makefile new file mode 100644 index 00000000..20848d4e --- /dev/null +++ b/Makefile @@ -0,0 +1,26 @@ +all : skynet blackhole.so snlua.so logger.so skynet.so gate.so client + +skynet : skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c skynet_server.c skynet_start.c skynet_timer.c skynet_error.c + gcc -Wall -g -Wl,-E -o $@ $^ -lpthread -ldl -lrt + +blackhole.so : skynet_blackhole.c + gcc -Wall -g -fPIC --shared $^ -o $@ + +logger.so : skynet_logger.c + gcc -Wall -g -fPIC --shared $^ -o $@ + +snlua.so : service_lua.c + gcc -Wall -g -fPIC --shared $^ -o $@ -Wl,-E -llua -lm + +gate.so : gate/mread.c gate/ringbuffer.c gate/map.c gate/main.c + gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Igate + +skynet.so : lua-skynet.c + gcc -Wall -g -fPIC --shared $^ -o $@ + +client : client.c + gcc -Wall -g $^ -o $@ + +clean : + rm skynet client *.so + diff --git a/agent.lua b/agent.lua new file mode 100644 index 00000000..ed42070e --- /dev/null +++ b/agent.lua @@ -0,0 +1,7 @@ +local skynet = require "skynet" + +print("agent",...) + +skynet.callback(function(addr, msg) + print("[agent]",addr,msg) +end) diff --git a/client.c b/client.c new file mode 100644 index 00000000..1ab9108c --- /dev/null +++ b/client.c @@ -0,0 +1,58 @@ +#include +#include +#include +#include + +#include +#include +#include +#include + +static void +test(int fd) { + char tmp[1024]; + while (!feof(stdin)) { + fgets(tmp,sizeof(tmp),stdin); + int n = strlen(tmp) -1; + uint8_t head[2]; + head[0] = n & 0xff; + head[1] = (n >> 8) & 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"); + } + } +} + +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; + } + + test(fd); + + close(fd); + + return 0; +} \ No newline at end of file diff --git a/console.lua b/console.lua new file mode 100644 index 00000000..9931227e --- /dev/null +++ b/console.lua @@ -0,0 +1,14 @@ +local skynet = require "skynet" + +skynet.callback(function() + local cmd = io.read() + local handle = skynet.command("LAUNCH",cmd) + if handle == nil then + print("Launch error:",cmd) + else + print("Lauch:",handle) + end + skynet.command("TIMEOUT","0:0") +end) + +skynet.command("TIMEOUT","0:0") diff --git a/gate/README b/gate/README new file mode 100644 index 00000000..7135d37f --- /dev/null +++ b/gate/README @@ -0,0 +1 @@ +See https://github.com/cloudwu/mread for detail. diff --git a/gate/main.c b/gate/main.c new file mode 100644 index 00000000..41109cf8 --- /dev/null +++ b/gate/main.c @@ -0,0 +1,250 @@ +#include "skynet.h" +#include "mread.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#define WATCHDOG ".watchdog" + +struct connection { + char * agent; + int connection_id; + int uid; +}; + +struct gate { + struct mread_pool * pool; + int id_index; + int cap; + int max_connection; + struct connection ** agent; + struct connection * map; +}; + +struct gate * +gate_create(void) { + struct gate * g = malloc(sizeof(*g)); + g->pool = NULL; + g->max_connection = 0; + g->agent = NULL; + return g; +} + +static inline struct connection * +_id_to_agent(struct gate *g,int uid) { + return g->agent[uid & (g->cap - 1)]; +} + +static void +_parm(char *msg, int sz, int command_sz) { + while (command_sz < sz) { + if (msg[command_sz] != ' ') + break; + ++command_sz; + } + int i; + for (i=command_sz;iagent) { + free(agent->agent); + } + agent->agent = strdup(addr); +} + +static void +_ctrl(struct skynet_context * ctx, struct gate * g, const void * msg, int sz) { + char tmp[sz+1]; + memcpy(tmp, msg, sz); + tmp[sz] = '\0'; + char * command = tmp; + int i; + if (sz == 0) + return; + for (i=0;iconnection_id; + + mread_close_client(g->pool,connection_id); + return; + } + if (memcmp(command,"forward",i)==0) { + _parm(tmp, sz, i); + char * start = tmp; + char * data = strsep(&start, " "); + int id = strtol(data , NULL, 10); + if (start) { + _forward_agent(g, id, start); + } + return; + } + skynet_error(ctx, "[gate] Unkown command : %s", command); +} + +static void +_report(struct skynet_context * ctx, const char * data, ...) { + va_list ap; + va_start(ap, data); + char tmp[1024]; + int n = vsnprintf(tmp, sizeof(tmp), data, ap); + va_end(ap); + if (n>=sizeof(tmp)) { + n = sizeof(tmp) - 1; + tmp[n] = '\0'; + } + + skynet_send(ctx, WATCHDOG, strdup(tmp), n); +} + +static void +_forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_t len) { + struct connection * agent = _id_to_agent(g,uid); + char * tmp = malloc(len + 32); + int n = snprintf(tmp,len+32,"%d data ",uid); + memcpy(tmp+n,data,len); + skynet_send(ctx, agent->agent ? agent->agent : WATCHDOG, tmp, len + n); +} + +static int +_gen_id(struct gate * g, int connection_id) { + int uid = ++g->id_index; + int i; + for (;;) { + for (i=0;icap;i++) { + int hash = (uid + i) & (g->cap - 1); + if (g->agent[hash] == NULL) { + uid = uid + i; + struct connection * conn = &g->map[connection_id]; + conn->uid = uid; + g->agent[hash] = conn; + return uid; + } + } + struct connection ** new_hash = malloc(g->cap * 2 * sizeof(struct connection *)); + memset(new_hash, 0, sizeof(g->cap * 2 * sizeof(struct connection *))); + for (i=0;imax_connection;i++) { + struct connection * conn = &g->map[connection_id]; + assert(conn->uid == 0); + new_hash[conn->uid & (g->cap * 2 -1)] = conn; + } + free(g->agent); + g->agent = new_hash; + } +} + +static void +_remove_id(struct gate *g, int uid) { + struct connection * conn = _id_to_agent(g,uid); + assert(conn->uid == uid); + conn->uid = 0; + if (conn->agent) { + free(conn->agent); + conn->agent = NULL; + } +} + +static void +_cb(struct skynet_context * ctx, void * ud, const char * uid, const void * msg, size_t sz) { + struct gate *g = ud; + if (msg) { + _ctrl(ctx, g , msg , (int)sz); + return; + } + struct mread_pool * m = g->pool; + int connection_id = mread_poll(m,100); // timeout : 100ms + if (connection_id < 0) { + skynet_command(ctx, "TIMEOUT","1:0"); + } else { + int id = g->map[connection_id].uid; + if (id == 0) { + id = _gen_id(g, connection_id); + int fd = mread_socket(m , connection_id); + struct sockaddr_in remote_addr; + socklen_t len = sizeof(struct sockaddr_in); + getpeername(fd, (struct sockaddr *)&remote_addr, &len); + _report(ctx, "%d open %d %s:%u",id,fd,inet_ntoa(remote_addr.sin_addr),ntohs(remote_addr.sin_port)); + } + uint16_t * plen = mread_pull(m,2); + if (plen == NULL) { + if (mread_closed(m)) { + _remove_id(g,id); + _report(ctx, "%d close", id); + } + goto _break; + } + void * data = mread_pull(m, *plen); + if (data == NULL) { + if (mread_closed(m)) { + _remove_id(g,id); + _report(ctx, "%d close", id); + } + goto _break; + } + + _forward(ctx, g, id, data, *plen); + mread_yield(m); +_break: + skynet_command(ctx, "TIMEOUT","0:0"); + } +} + +int +gate_init(struct gate *g , struct skynet_context * ctx, char * parm) { + int port = 0; + int max = 0; + int buffer = 0; + int n = sscanf(parm, "%d %d %d",&port,&max,&buffer); + if (n!=3) { + skynet_error(ctx, "Invalid gate parm %s",parm); + return 1; + } + struct mread_pool * pool = mread_create(port, max, buffer); + if (pool == NULL) { + skynet_error(ctx, "Create gate %s failed",parm); + return 1; + } + + g->pool = pool; + int cap = 1; + while (cap < max) { + cap *= 2; + } + g->cap = cap; + g->max_connection = max; + g->id_index = 0; + + g->agent = malloc(cap * sizeof(struct connection *)); + memset(g->agent, 0, cap * sizeof(struct connection *)); + + g->map = malloc(max * sizeof(struct connection)); + memset(g->map, 0, max * sizeof(struct connection)); + int i; + for (i=0;imap[i].connection_id = i; + } + + skynet_callback(ctx,g,_cb); + skynet_command(ctx,"REG","gate"); + skynet_command(ctx,"TIMEOUT","0:0"); + + return 0; +} diff --git a/gate/map.c b/gate/map.c new file mode 100644 index 00000000..806c41ca --- /dev/null +++ b/gate/map.c @@ -0,0 +1,139 @@ +#include "map.h" + +#include +#include +#include + +struct node { + int fd; + int id; + struct node * next; +}; + +struct map { + int size; + struct node * hash; +}; + +struct map * +map_new(int max) { + int sz = 1; + while (sz <= max) { + sz *= 2; + } + struct map * m = malloc(sizeof(*m)); + m->size = sz; + m->hash = malloc(sizeof(struct node) * sz); + int i; + for (i=0;ihash[i].fd = -1; + m->hash[i].id = 0; + m->hash[i].next = NULL; + } + return m; +} + +void +map_delete(struct map * m) { + free(m->hash); + free(m); +} + +int +map_search(struct map * m, int fd) { + int hash = fd & (m->size-1); + struct node * n = &m->hash[hash]; + do { + if (n->fd == fd) + return n->id; + n = n->next; + } while(n); + return -1; +} + +void +map_insert(struct map * m, int fd, int id) { + int hash = fd & (m->size-1); + struct node * n = &m->hash[hash]; + if (n->fd < 0) { + n->fd = fd; + n->id = id; + return; + } + int ohash = n->fd & (m->size-1); + if (hash != ohash) { + struct node * last = &m->hash[ohash]; + while (last->next != &m->hash[hash]) { + last = last->next; + } + last->next = n->next; + + int ofd = n->fd; + int oid = n->id; + n->fd = fd; + n->id = id; + n->next = NULL; + map_insert(m,ofd, oid); + return; + } + + int last = (n - m->hash) * 2; + int i; + for (i=0;isize;i++) { + int idx = (i + last + 1) & (m->size - 1); + struct node * temp = &m->hash[idx]; + if (temp->fd < 0) { + temp->fd = fd; + temp->id = id; + temp->next = n->next; + n->next = temp; + return; + } + } + assert(0); +} + +void +map_erase(struct map *m , int fd) { + int hash = fd & (m->size-1); + struct node * n = &m->hash[hash]; + if (n->fd == fd) { + if (n->next == NULL) { + n->fd = -1; + return; + } + struct node * next = n->next; + n->fd = next->fd; + n->id = next->id; + n->next = next->next; + next->fd = -1; + next->next = NULL; + return; + } + if (n->next == NULL) { + return; + } + struct node * last = n; + n = n->next; + for(;;) { + if (n->fd == fd) { + n->fd = -1; + last->next = n->next; + n->next = NULL; + return; + } + if (n->next == NULL) + return; + last = n; + n = n->next; + } +} + +void +map_dump(struct map *m) { + int i; + for (i=0;isize;i++) { + struct node * n = &(m->hash[i]); + printf("[%d] fd = %d , id = %d , next = %d\n",i,n->fd,n->id,(int)(n->next - m->hash)); + } +} \ No newline at end of file diff --git a/gate/map.h b/gate/map.h new file mode 100644 index 00000000..1a12c48d --- /dev/null +++ b/gate/map.h @@ -0,0 +1,13 @@ +#ifndef MREAD_MAP_H +#define MREAD_MAP_H + +struct map; + +struct map * map_new(int max); +void map_delete(struct map *); +int map_search(struct map * , int fd); +void map_insert(struct map * , int fd, int id); +void map_erase(struct map *, int fd); +void map_dump(struct map *m); + +#endif diff --git a/gate/mread.c b/gate/mread.c new file mode 100644 index 00000000..7aded37e --- /dev/null +++ b/gate/mread.c @@ -0,0 +1,486 @@ +#include "mread.h" +#include "ringbuffer.h" +#include "map.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define BACKLOG 32 +#define READQUEUE 32 +#define READBLOCKSIZE 2048 +#define RINGBUFFER_DEFAULT 1024 * 1024 + +#define SOCKET_INVALID 0 +#define SOCKET_CLOSED 1 +#define SOCKET_SUSPEND 2 +#define SOCKET_READ 3 +#define SOCKET_POLLIN 4 + +#define SOCKET_ALIVE SOCKET_SUSPEND + +struct socket { + int fd; + struct ringbuffer_block * node; + struct ringbuffer_block * temp; + int status; +}; + +struct mread_pool { + int listen_fd; + int epoll_fd; + int max_connection; + int closed; + int active; + int skip; + struct socket * sockets; + struct socket * free_socket; + struct map * socket_hash; + int queue_len; + int queue_head; + struct epoll_event ev[READQUEUE]; + struct ringbuffer * rb; +}; + +static struct socket * +_create_sockets(int max) { + int i; + struct socket * s = malloc(max * sizeof(struct socket)); + for (i=0;ilisten_fd = listen_fd; + self->epoll_fd = epoll_fd; + self->max_connection = max; + self->closed = 0; + self->active = -1; + self->skip = 0; + self->sockets = _create_sockets(max); + self->free_socket = &self->sockets[0]; + self->socket_hash = map_new(max * 3 / 2); + self->queue_len = 0; + self->queue_head = 0; + if (buffer_size == 0) { + self->rb = _create_rb(RINGBUFFER_DEFAULT); + } else { + self->rb = _create_rb(buffer_size); + } + + return self; +} + +void +mread_close(struct mread_pool *self) { + if (self == NULL) + return; + int i; + struct socket * s = self->sockets; + for (i=0;imax_connection;i++) { + if (s[i].status >= SOCKET_ALIVE) { + close(s[i].fd); + } + } + free(s); + if (self->listen_fd >= 0) { + close(self->listen_fd); + } + close(self->epoll_fd); + _release_rb(self->rb); + map_delete(self->socket_hash); + free(self); +} + +static int +_read_queue(struct mread_pool * self, int timeout) { + self->queue_head = 0; + int n = epoll_wait(self->epoll_fd , self->ev, READQUEUE, timeout); + if (n == -1) { + self->queue_len = 0; + return -1; + } + self->queue_len = n; + return n; +} + +inline static int +_read_one(struct mread_pool * self) { + if (self->queue_head >= self->queue_len) { + return -1; + } + return self->ev[self->queue_head ++].data.fd; +} + +static struct socket * +_alloc_socket(struct mread_pool * self) { + if (self->free_socket == NULL) { + return NULL; + } + struct socket * s = self->free_socket; + int next_free = s->fd; + if (next_free < 0 ) { + self->free_socket = NULL; + } else { + self->free_socket = &self->sockets[next_free]; + } + return s; +} + +static void +_add_client(struct mread_pool * self, int fd) { + struct socket * s = _alloc_socket(self); + if (s == NULL) { + close(fd); + return; + } + struct epoll_event ev; + ev.events = EPOLLIN; + ev.data.fd = fd; + if (epoll_ctl(self->epoll_fd, EPOLL_CTL_ADD, fd, &ev) == -1) { + close(fd); + return; + } + + s->fd = fd; + s->node = NULL; + s->status = SOCKET_SUSPEND; + int id = s - self->sockets; + map_insert(self->socket_hash , fd , id); +} + +static int +_report_closed(struct mread_pool * self) { + int i; + for (i=0;imax_connection;i++) { + if (self->sockets[i].status == SOCKET_CLOSED) { + self->active = i; + return i; + } + } + assert(0); + return -1; +} + +int +mread_poll(struct mread_pool * self , int timeout) { + self->skip = 0; + if (self->active >= 0) { + struct socket * s = &self->sockets[self->active]; + if (s->status == SOCKET_READ) { + return self->active; + } + } + if (self->closed > 0 ) { + return _report_closed(self); + } + if (self->queue_head >= self->queue_len) { + if (_read_queue(self, timeout) == -1) { + self->active = -1; + return -1; + } + } + for (;;) { + int fd = _read_one(self); + if (fd == -1) { + self->active = -1; + return -1; + } + if (fd == self->listen_fd) { + struct sockaddr_in remote_addr; + socklen_t len = sizeof(struct sockaddr_in); + int client_fd = accept(self->listen_fd , (struct sockaddr *)&remote_addr , &len); + if (client_fd >= 0) { +// printf("MREAD connect %s:%u (fd=%d)\n",inet_ntoa(remote_addr.sin_addr),ntohs(remote_addr.sin_port), client_fd); + _add_client(self, client_fd); + } + } else { + int index = map_search(self->socket_hash , fd); + if (index >= 0) { + self->active = index; + struct socket * s = &self->sockets[index]; + s->status = SOCKET_POLLIN; + return index; + } + } + } +} + +int +mread_socket(struct mread_pool * self, int index) { + return self->sockets[index].fd; +} + +static void +_link_node(struct ringbuffer * rb, int id, struct socket * s , struct ringbuffer_block * blk) { + if (s->node) { + ringbuffer_link(rb, s->node , blk); + } else { + blk->id = id; + s->node = blk; + } +} + +void +mread_close_client(struct mread_pool * self, int id) { + struct socket * s = &self->sockets[id]; + s->status = SOCKET_CLOSED; + s->node = NULL; + s->temp = NULL; + close(s->fd); +// printf("MREAD close %d (fd=%d)\n",id,s->fd); + epoll_ctl(self->epoll_fd, EPOLL_CTL_DEL, s->fd , NULL); + + ++self->closed; +} + +static void +_close_active(struct mread_pool * self) { + int id = self->active; + struct socket * s = &self->sockets[id]; + ringbuffer_free(self->rb, s->temp); + ringbuffer_free(self->rb, s->node); + mread_close_client(self, id); +} + +static char * +_ringbuffer_read(struct mread_pool * self, int *size) { + struct socket * s = &self->sockets[self->active]; + if (s->node == NULL) { + *size = 0; + return NULL; + } + int sz = *size; + void * ret; + *size = ringbuffer_data(self->rb, s->node, sz , self->skip, &ret); + return ret; +} + +void * +mread_pull(struct mread_pool * self , int size) { + if (self->active == -1) { + return NULL; + } + struct socket *s = &self->sockets[self->active]; + int rd_size = size; + char * buffer = _ringbuffer_read(self, &rd_size); + if (buffer) { + self->skip += size; + return buffer; + } + switch (s->status) { + case SOCKET_READ: + s->status = SOCKET_SUSPEND; + case SOCKET_CLOSED: + case SOCKET_SUSPEND: + return NULL; + default: + assert(s->status == SOCKET_POLLIN); + break; + } + + int sz = size - rd_size; + int rd = READBLOCKSIZE; + if (rd < sz) { + rd = sz; + } + + int id = self->active; + struct ringbuffer * rb = self->rb; + + struct ringbuffer_block * blk = ringbuffer_alloc(rb , rd); + while (blk == NULL) { + int collect_id = ringbuffer_collect(rb); + mread_close_client(self , collect_id); + if (id == collect_id) { + return NULL; + } + blk = ringbuffer_alloc(rb , rd); + } + + buffer = (char *)(blk + 1); + + for (;;) { + int bytes = recv(s->fd, buffer, rd, MSG_DONTWAIT); + if (bytes > 0) { + ringbuffer_resize(rb, blk , bytes); + if (bytes < sz) { + _link_node(rb, self->active, s , blk); + s->status = SOCKET_SUSPEND; + return NULL; + } + s->status = SOCKET_READ; + break; + } + if (bytes == 0) { + ringbuffer_resize(rb, blk, 0); + _close_active(self); + return NULL; + } + if (bytes == -1) { + switch(errno) { + case EWOULDBLOCK: + ringbuffer_resize(rb, blk, 0); + s->status = SOCKET_SUSPEND; + return NULL; + case EINTR: + continue; + default: + ringbuffer_resize(rb, blk, 0); + _close_active(self); + return NULL; + } + } + } + _link_node(rb, self->active , s , blk); + void * ret; + int real_rd = ringbuffer_data(rb, s->node , size , self->skip, &ret); + if (ret) { + self->skip += size; + return ret; + } + assert(real_rd == size); + struct ringbuffer_block * temp = ringbuffer_alloc(rb, size); + while (temp == NULL) { + int collect_id = ringbuffer_collect(rb); + mread_close_client(self , collect_id); + if (id == collect_id) { + return NULL; + } + temp = ringbuffer_alloc(rb , size); + } + temp->id = id; + if (s->temp) { + ringbuffer_link(rb, temp, s->temp); + } + s->temp = temp; + ret = ringbuffer_copy(rb, s->node, self->skip, temp); + assert(ret); + self->skip += size; + + return ret; +} + +void +mread_yield(struct mread_pool * self) { + if (self->active == -1) { + return; + } + struct socket *s = &self->sockets[self->active]; + ringbuffer_free(self->rb , s->temp); + s->temp = NULL; + if (s->status == SOCKET_CLOSED && s->node == NULL) { + --self->closed; + s->status = SOCKET_INVALID; + map_erase(self->socket_hash , s->fd); + s->fd = self->free_socket - self->sockets; + self->free_socket = s; + self->skip = 0; + self->active = -1; + } else { + if (s->node) { + s->node = ringbuffer_yield(self->rb, s->node, self->skip); + } + self->skip = 0; + if (s->node == NULL) { + self->active = -1; + } + } +} + +int +mread_closed(struct mread_pool * self) { + if (self->active == -1) { + return 0; + } + struct socket * s = &self->sockets[self->active]; + if (s->status == SOCKET_CLOSED && s->node == NULL) { + mread_yield(self); + return 1; + } + return 0; +} diff --git a/gate/mread.h b/gate/mread.h new file mode 100644 index 00000000..c1e271b0 --- /dev/null +++ b/gate/mread.h @@ -0,0 +1,16 @@ +#ifndef MREAD_H +#define MREAD_H + +struct mread_pool; + +struct mread_pool * mread_create(int port , int max , int buffer); +void mread_close(struct mread_pool *m); + +int mread_poll(struct mread_pool *m , int timeout); +void * mread_pull(struct mread_pool *m , int size); +void mread_yield(struct mread_pool *m); +int mread_closed(struct mread_pool *m); +void mread_close_client(struct mread_pool *m, int id); +int mread_socket(struct mread_pool *m , int index); + +#endif diff --git a/gate/ringbuffer.c b/gate/ringbuffer.c new file mode 100644 index 00000000..d645ccde --- /dev/null +++ b/gate/ringbuffer.c @@ -0,0 +1,277 @@ +#include "ringbuffer.h" + +#include +#include +#include +#include + +#define ALIGN(s) (((s) + 3 ) & ~3) + +struct ringbuffer { + int size; + int head; +}; + +static inline int +block_offset(struct ringbuffer * rb, struct ringbuffer_block * blk) { + char * start = (char *)(rb + 1); + return (char *)blk - start; +} + +static inline struct ringbuffer_block * +block_ptr(struct ringbuffer * rb, int offset) { + char * start = (char *)(rb + 1); + return (struct ringbuffer_block *)(start + offset); +} + +static inline struct ringbuffer_block * +block_next(struct ringbuffer * rb, struct ringbuffer_block * blk) { + int align_length = ALIGN(blk->length); + int head = block_offset(rb, blk); + if (align_length + head == rb->size) { + return NULL; + } + assert(align_length + head < rb->size); + return block_ptr(rb, head + align_length); +} + +struct ringbuffer * +ringbuffer_new(int size) { + struct ringbuffer * rb = malloc(sizeof(*rb) + size); + rb->size = size; + rb->head = 0; + struct ringbuffer_block * blk = block_ptr(rb, 0); + blk->length = size; + blk->id = -1; + return rb; +} + +void +ringbuffer_delete(struct ringbuffer * rb) { + free(rb); +} + +void +ringbuffer_link(struct ringbuffer *rb , struct ringbuffer_block * head, struct ringbuffer_block * next) { + while (head->next >=0) { + head = block_ptr(rb, head->next); + } + next->id = head->id; + head->next = block_offset(rb, next); +} + +static struct ringbuffer_block * +_alloc(struct ringbuffer * rb, int total_size , int size) { + struct ringbuffer_block * blk = block_ptr(rb, rb->head); + int align_length = ALIGN(sizeof(struct ringbuffer_block) + size); + blk->length = sizeof(struct ringbuffer_block) + size; + blk->offset = 0; + blk->next = -1; + blk->id = -1; + struct ringbuffer_block * next = block_next(rb, blk); + rb->head = block_offset(rb, next); + if (align_length < total_size) { + next->length = total_size - align_length; + if (next->length >= sizeof(struct ringbuffer_block)) { + next->id = -1; + } + } + return blk; +} + +struct ringbuffer_block * +ringbuffer_alloc(struct ringbuffer * rb, int size) { + int align_length = ALIGN(sizeof(struct ringbuffer_block) + size); + int i; + for (i=0;i<2;i++) { + int free_size = 0; + struct ringbuffer_block * blk = block_ptr(rb, rb->head); + do { + if (blk->length >= sizeof(struct ringbuffer_block) && blk->id >= 0) + return NULL; + free_size += ALIGN(blk->length); + if (free_size >= align_length) { + return _alloc(rb, free_size , size); + } + blk = block_next(rb, blk); + } while(blk); + rb->head = 0; + } + return NULL; +} + +static int +_last_id(struct ringbuffer * rb) { + int i; + for (i=0;i<2;i++) { + struct ringbuffer_block * blk = block_ptr(rb, rb->head); + do { + if (blk->length >= sizeof(struct ringbuffer_block) && blk->id >= 0) + return blk->id; + blk = block_next(rb, blk); + } while(blk); + rb->head = 0; + } + return -1; +} + +int +ringbuffer_collect(struct ringbuffer * rb) { + int id = _last_id(rb); + struct ringbuffer_block *blk = block_ptr(rb, 0); + do { + if (blk->length >= sizeof(struct ringbuffer_block) && blk->id == id) { + blk->id = -1; + } + blk = block_next(rb, blk); + } while(blk); + return id; +} + +void +ringbuffer_resize(struct ringbuffer * rb, struct ringbuffer_block * blk, int size) { + if (size == 0) { + rb->head = block_offset(rb, blk); + return; + } + int align_length = ALIGN(sizeof(struct ringbuffer_block) + size); + int old_length = ALIGN(blk->length); + assert(align_length < old_length); + blk->length = size + sizeof(struct ringbuffer_block); + if (align_length == old_length) { + return; + } + blk = block_next(rb, blk); + blk->length = old_length - align_length; + if (blk->length >= sizeof(struct ringbuffer_block)) { + blk->id = -1; + } + rb->head = block_offset(rb, blk); +} + +static int +_block_id(struct ringbuffer_block * blk) { + assert(blk->length >= sizeof(struct ringbuffer_block)); + int id = blk->id; + assert(id>=0); + return id; +} + +void +ringbuffer_free(struct ringbuffer * rb, struct ringbuffer_block * blk) { + if (blk == NULL) + return; + int id = _block_id(blk); + blk->id = -1; + while (blk->next >= 0) { + blk = block_ptr(rb, blk->next); + assert(_block_id(blk) == id); + blk->id = -1; + } +} + +int +ringbuffer_data(struct ringbuffer * rb, struct ringbuffer_block * blk, int size, int skip, void **ptr) { + int length = blk->length - sizeof(struct ringbuffer_block) - blk->offset; + for (;;) { + if (length > skip) { + if (length - skip >= size) { + char * start = (char *)(blk + 1); + *ptr = (start + blk->offset + skip); + return size; + } + *ptr = NULL; + int ret = length - skip; + while (blk->next >= 0) { + blk = block_ptr(rb, blk->next); + ret += blk->length - sizeof(struct ringbuffer_block); + if (ret >= size) + return size; + } + return ret; + } + if (blk->next < 0) { + assert(length == skip); + *ptr = NULL; + return 0; + } + blk = block_ptr(rb, blk->next); + assert(blk->offset == 0); + skip -= length; + length = blk->length - sizeof(struct ringbuffer_block); + } +} + +void * +ringbuffer_copy(struct ringbuffer * rb, struct ringbuffer_block * from, int skip, struct ringbuffer_block * to) { + int size = to->length - sizeof(struct ringbuffer_block); + int length = from->length - sizeof(struct ringbuffer_block) - from->offset; + char * ptr = (char *)(to+1); + for (;;) { + if (length > skip) { + char * src = (char *)(from + 1); + src += from->offset + skip; + length -= skip; + while (length < size) { + memcpy(ptr, src, length); + assert(from->next >= 0); + from = block_ptr(rb , from->next); + assert(from->offset == 0); + ptr += length; + size -= length; + length = from->length - sizeof(struct ringbuffer_block); + src = (char *)(from + 1); + } + memcpy(ptr, src , size); + to->id = from->id; + return (char *)(to + 1); + } + assert(from->next >= 0); + from = block_ptr(rb, from->next); + assert(from->offset == 0); + skip -= length; + length = from->length - sizeof(struct ringbuffer_block); + } +} + +struct ringbuffer_block * +ringbuffer_yield(struct ringbuffer * rb, struct ringbuffer_block *blk, int skip) { + int length = blk->length - sizeof(struct ringbuffer_block) - blk->offset; + for (;;) { + if (length > skip) { + blk->offset += skip; + return blk; + } + blk->id = -1; + if (blk->next < 0) { + return NULL; + } + blk = block_ptr(rb, blk->next); + assert(blk->offset == 0); + skip -= length; + length = blk->length - sizeof(struct ringbuffer_block); + } +} + +void +ringbuffer_dump(struct ringbuffer * rb) { + struct ringbuffer_block *blk = block_ptr(rb,0); + int i=0; + printf("total size= %d\n",rb->size); + while (blk) { + ++i; + if (i>10) + break; + if (blk->length >= sizeof(*blk)) { + printf("[%u : %d]", (unsigned)(blk->length - sizeof(*blk)), block_offset(rb,blk)); + printf(" id=%d",blk->id); + if (blk->id >=0) { + printf(" offset=%d next=%d",blk->offset, blk->next); + } + } else { + printf("<%u : %d>", blk->length, block_offset(rb,blk)); + } + printf("\n"); + blk = block_next(rb, blk); + } +} diff --git a/gate/ringbuffer.h b/gate/ringbuffer.h new file mode 100644 index 00000000..a083e0cc --- /dev/null +++ b/gate/ringbuffer.h @@ -0,0 +1,27 @@ +#ifndef MREAD_RINGBUFFER_H +#define MREAD_RINGBUFFER_H + +struct ringbuffer; + +struct ringbuffer_block { + int length; + int offset; + int id; + int next; +}; + +struct ringbuffer * ringbuffer_new(int size); +void ringbuffer_delete(struct ringbuffer * rb); +void ringbuffer_link(struct ringbuffer *rb , struct ringbuffer_block * prev, struct ringbuffer_block * next); +struct ringbuffer_block * ringbuffer_alloc(struct ringbuffer * rb, int size); +int ringbuffer_collect(struct ringbuffer * rb); +void ringbuffer_resize(struct ringbuffer * rb, struct ringbuffer_block * blk, int size); +void ringbuffer_free(struct ringbuffer * rb, struct ringbuffer_block * blk); +int ringbuffer_data(struct ringbuffer * rb, struct ringbuffer_block * blk, int size, int skip, void **ptr); +void * ringbuffer_copy(struct ringbuffer * rb, struct ringbuffer_block * from, int skip, struct ringbuffer_block * to); +struct ringbuffer_block * ringbuffer_yield(struct ringbuffer * rb, struct ringbuffer_block *blk, int skip); + +void ringbuffer_dump(struct ringbuffer * rb); + +#endif + diff --git a/lua-skynet.c b/lua-skynet.c new file mode 100644 index 00000000..c7a78dfe --- /dev/null +++ b/lua-skynet.c @@ -0,0 +1,113 @@ +#include "skynet.h" + +#include +#include +#include +#include + +static void +_cb(struct skynet_context * context, void * ud, const char * addr, const void * msg, size_t sz_session) { + lua_State *L = ud; + lua_rawgetp(L, LUA_REGISTRYINDEX, _cb); + int r; + if (msg == NULL) { + lua_pushinteger(L, (int)sz_session); + r = lua_pcall(L, 1, 0 , 0); + } else { + lua_pushstring(L, addr); + lua_pushlstring(L, msg, sz_session); + r = lua_pcall(L, 2, 0 , 0); + } + if (r == LUA_OK) + return; + skynet_error(context, "lua error %s", lua_tostring(L,-1)); +} + +static int +_callback(lua_State *L) { + struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); + if (context == NULL) { + return luaL_error(L, "Init skynet context first"); + } + + luaL_checktype(L,1,LUA_TFUNCTION); + lua_settop(L,1); + lua_rawsetp(L, LUA_REGISTRYINDEX, _cb); + lua_rawgeti(L, LUA_REGISTRYINDEX, LUA_RIDX_MAINTHREAD); + lua_State *gL = lua_tothread(L,-1); + + skynet_callback(context, gL, _cb); + + return 0; +} + +static int +_command(lua_State *L) { + struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); + const char * cmd = luaL_checkstring(L,1); + const char * result; + if (lua_gettop(L) == 2) { + const char * parm = luaL_checkstring(L,2); + result = skynet_command(context, cmd, parm); + } else { + result = skynet_command(context, cmd, NULL); + } + if (result) { + lua_pushstring(L, result); + return 1; + } + return 0; +} + +static int +_send(lua_State *L) { + struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); + const char * dest = luaL_checkstring(L,1); + int type = lua_type(L,2); + if (type == LUA_TSTRING) { + size_t len = 0; + void * msg = (void *)lua_tolstring(L,2,&len); + void * message = malloc(len); + memcpy(message, msg, len); + skynet_send(context, dest, message, len); + } else { + void * msg = lua_touserdata(L,2); + if (msg == NULL) { + return luaL_error(L, "skynet.send need userdata or string (%s)", lua_typename(L,type)); + } + int size = luaL_checkinteger(L,3); + skynet_send(context, dest, msg, size); + } + return 0; +} + +static int +_error(lua_State *L) { + struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); + skynet_error(context, "%s", luaL_checkstring(L,1)); + return 0; +} + +int +luaopen_skynet(lua_State *L) { + luaL_checkversion(L); + luaL_Reg l[] = { + { "send" , _send }, + { "command" , _command }, + { "callback" , _callback }, + { "error", _error }, + { NULL, NULL }, + }; + + luaL_newlibtable(L,l); + + lua_getfield(L, LUA_REGISTRYINDEX, "skynet_context"); + struct skynet_context * ctx = lua_touserdata(L,-1); + if (ctx == NULL) { + return luaL_error(L, "Init skynet context first"); + } + + luaL_setfuncs(L,l,1); + + return 1; +} diff --git a/main.lua b/main.lua new file mode 100644 index 00000000..3982b503 --- /dev/null +++ b/main.lua @@ -0,0 +1,12 @@ +local skynet = require "skynet" + +print("Server start") + +local console = skynet.command("LAUNCH","snlua console.lua") +print("console",console) +local watchdog = skynet.command("LAUNCH","snlua watchdog.lua") +print("watchdog",watchdog) +local gate = skynet.command("LAUNCH","gate 2525 4 0") +print("gate",gate) + +skynet.command("EXIT") diff --git a/rwlock.h b/rwlock.h new file mode 100644 index 00000000..45fe02ac --- /dev/null +++ b/rwlock.h @@ -0,0 +1,48 @@ +#ifndef _RWLOCK_H_ +#define _RWLOCK_H_ + +struct rwlock { + int write; + int read; +}; + +static inline void +rwlock_init(struct rwlock *lock) { + lock->write = 0; + lock->read = 0; +} + +static inline void +rwlock_rlock(struct rwlock *lock) { + for (;;) { + while(lock->write) { + __sync_synchronize(); + } + __sync_add_and_fetch(&lock->read,1); + if (lock->write) { + __sync_sub_and_fetch(&lock->read,1); + } else { + break; + } + } +} + +static inline void +rwlock_wlock(struct rwlock *lock) { + __sync_lock_test_and_set(&lock->write,1); + while(lock->read) { + __sync_synchronize(); + } +} + +static inline void +rwlock_wunlock(struct rwlock *lock) { + __sync_lock_release(&lock->write); +} + +static inline void +rwlock_runlock(struct rwlock *lock) { + __sync_sub_and_fetch(&lock->read,1); +} + +#endif diff --git a/service_lua.c b/service_lua.c new file mode 100644 index 00000000..6423f7b5 --- /dev/null +++ b/service_lua.c @@ -0,0 +1,72 @@ +#include "skynet.h" + +#include +#include +#include + +#include +#include + +lua_State * +snlua_create(void) { + return luaL_newstate(); +} + +static int +_load(lua_State *L, char ** filename) { + const char * name = strsep(filename, " \r\n"); + int r = luaL_loadfile(L,name); + return r != LUA_OK; +} + +int +snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) { + lua_gc(L, LUA_GCSTOP, 0); + luaL_openlibs(L); + lua_pushlightuserdata(L, ctx); + lua_setfield(L, LUA_REGISTRYINDEX, "skynet_context"); + lua_gc(L, LUA_GCRESTART, 0); + + char tmp[strlen(args)+1]; + char *parm = tmp; + strcpy(parm,args); + + const char * filename = parm; + int r = _load(L, &parm); + if (r) { + skynet_error(ctx, "lua parser [%s] error : %s", filename, lua_tostring(L,-1)); + return 1; + } + int n=0; + while(parm) { + const char * arg = strsep(&parm, " \r\n"); + if (arg && arg[0]!='\0') { + lua_pushstring(L, arg); + ++n; + } + } + r = lua_pcall(L,n,0,0); + switch (r) { + case LUA_OK: + return 0; + case LUA_ERRRUN: + skynet_error(ctx, "lua do [%s] error : %s", filename, lua_tostring(L,-1)); + break; + case LUA_ERRMEM: + skynet_error(ctx, "lua memory error : %s",filename); + break; + case LUA_ERRERR: + skynet_error(ctx, "lua message error : %s",filename); + break; + case LUA_ERRGCMM: + skynet_error(ctx, "lua gc error : %s",filename); + break; + }; + + return 1; +} + +void +snlua_release(lua_State *L) { + lua_close(L); +} diff --git a/skynet.h b/skynet.h new file mode 100644 index 00000000..12eb6d2b --- /dev/null +++ b/skynet.h @@ -0,0 +1,15 @@ +#ifndef SKYNET_H +#define SKYNET_H + +#include + +struct skynet_context; + +void skynet_error(struct skynet_context * context, const char *msg, ...); +const char * skynet_command(struct skynet_context * context, const char * cmd , const char * parm); +void skynet_send(struct skynet_context * context, const char * addr , void * msg, size_t sz_session); + +typedef void (*skynet_cb)(struct skynet_context * context, void *ud, const char * uid , const void * msg, size_t sz_session); +void skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb); + +#endif diff --git a/skynet_blackhole.c b/skynet_blackhole.c new file mode 100644 index 00000000..72a97020 --- /dev/null +++ b/skynet_blackhole.c @@ -0,0 +1,22 @@ +#include "skynet.h" +#include "skynet_blackhole.h" + +#include +#include +#include + +static void +_log_message(struct skynet_context *ctx, void *ud, const char * uid, const void * msg, size_t sz) { + const struct blackhole * message = msg; + assert(sz == sizeof(*message)); + + printf("[blackhole] from %d to %s , message_size = %d\n",message->source, message->destination, (int)message->sz); + free(message->data); +} + +int +blackhole_init(void * inst, struct skynet_context *ctx, const char *args) { + skynet_callback(ctx, inst, _log_message); + skynet_command(ctx, "REG", "blackhole"); + return 0; +} diff --git a/skynet_blackhole.h b/skynet_blackhole.h new file mode 100644 index 00000000..3e549839 --- /dev/null +++ b/skynet_blackhole.h @@ -0,0 +1,13 @@ +#ifndef SKYNET_BLACKHOLE_H +#define SKYNET_BLACKHOLE_H + +#include + +struct blackhole { + int source; + char * destination; + void * data; + size_t sz; +}; + +#endif diff --git a/skynet_error.c b/skynet_error.c new file mode 100644 index 00000000..99f5391d --- /dev/null +++ b/skynet_error.c @@ -0,0 +1,41 @@ +#include "skynet.h" +#include "skynet_handle.h" +#include "skynet_mq.h" +#include "skynet_server.h" + +#include +#include +#include + +#define LOG_MESSAGE_SIZE 1024 + +void +skynet_error(struct skynet_context * context, const char *msg, ...) { + static int logger = -1; + if (logger < 0) { + logger = skynet_handle_findname("logger"); + } + if (logger < 0) { + return; + } + + char tmp[LOG_MESSAGE_SIZE]; + + va_list ap; + va_start(ap,msg); + int len = vsnprintf(tmp, LOG_MESSAGE_SIZE, msg, ap); + va_end(ap); + + if (len >= LOG_MESSAGE_SIZE) { + len = LOG_MESSAGE_SIZE - 1; + tmp[len] = '\0'; + } + + struct skynet_message smsg; + smsg.source = skynet_context_handle(context); + smsg.destination = logger; + smsg.data = strdup(tmp); + smsg.sz = len; + skynet_mq_push(&smsg); +} + diff --git a/skynet_handle.c b/skynet_handle.c new file mode 100644 index 00000000..9f26bfee --- /dev/null +++ b/skynet_handle.c @@ -0,0 +1,222 @@ +#include "skynet_handle.h" +#include "skynet_server.h" +#include "rwlock.h" + +#include +#include +#include + +#define DEFAULT_SLOT_SIZE 4 + +struct handle_name { + char * name; + int handle; +}; + +struct handle_storage { + struct rwlock lock; + + int handle_index; + int slot_size; + struct skynet_context ** slot; + + int name_cap; + int name_count; + struct handle_name *name; +}; + +static struct handle_storage *H = NULL; + +int +skynet_handle_register(struct skynet_context *ctx) { + struct handle_storage *s = H; + + rwlock_wlock(&s->lock); + + for (;;) { + int i; + for (i=0;islot_size;i++) { + int hash = (i+s->handle_index) & (s->slot_size-1); + if (s->slot[hash] == NULL) { + s->slot[hash] = ctx; + int handle = s->handle_index + i; + skynet_context_init(ctx, handle); + + rwlock_wunlock(&s->lock); + + s->handle_index = handle + 1; + + return handle; + } + } + struct skynet_context ** new_slot = malloc(s->slot_size * 2 * sizeof(struct skynet_context *)); + memset(new_slot, 0, s->slot_size * 2 * sizeof(struct skynet_context *)); + for (i=0;islot_size;i++) { + int hash = skynet_context_handle(s->slot[i]) & (s->slot_size * 2 - 1); + assert(new_slot[hash] == NULL); + new_slot[hash] = s->slot[i]; + } + free(s->slot); + s->slot = new_slot; + s->slot_size *= 2; + } +} + +void +skynet_handle_retire(int handle) { + struct handle_storage *s = H; + + rwlock_wlock(&s->lock); + + int hash = handle & (s->slot_size-1); + struct skynet_context * ctx = s->slot[hash]; + if (skynet_context_handle(ctx) == handle) { + skynet_context_release(ctx); + s->slot[hash] = NULL; + + int i; + int j=0; + for (i=0;iname_count;i++,j++) { + if (s->name[i].handle == handle) { + free(s->name[i].name); + ++j; + --s->name_count; + } else { + if (i!=j) { + s->name[i] = s->name[j]; + } + } + } + } + + rwlock_wunlock(&s->lock); +} + +struct skynet_context * +skynet_handle_grab(int handle) { + struct handle_storage *s = H; + struct skynet_context * result = NULL; + + rwlock_rlock(&s->lock); + + int hash = handle & (s->slot_size-1); + struct skynet_context * ctx = s->slot[hash]; + if (skynet_context_handle(ctx) == handle) { + result = ctx; + skynet_context_grab(result); + } + + rwlock_runlock(&s->lock); + + return result; +} + +int +skynet_handle_findname(const char * name) { + struct handle_storage *s = H; + + rwlock_rlock(&s->lock); + + int handle = -1; + + int begin = 0; + int end = s->name_count - 1; + while (begin<=end) { + int mid = (begin+end)/2; + struct handle_name *n = &s->name[mid]; + int c = strcmp(n->name, name); + if (c==0) { + handle = n->handle; + break; + } + if (c<0) { + begin = mid + 1; + } else { + end = mid - 1; + } + } + + rwlock_runlock(&s->lock); + + return handle; +} + +static void +_insert_name_before(struct handle_storage *s, char *name, int handle, int before) { + if (s->name_count >= s->name_cap) { + s->name_cap *= 2; + struct handle_name * n = malloc(s->name_cap * sizeof(struct handle_name)); + int i; + for (i=0;iname[i]; + } + for (i=before;iname_count;i++) { + n[i+1] = s->name[i]; + } + free(s->name); + s->name = n; + } else { + int i; + for (i=s->name_count;i>=before;i--) { + s->name[i] = s->name[i-1]; + } + } + s->name[before].name = name; + s->name[before].handle = handle; + s->name_count ++; +} + +static const char * +_insert_name(struct handle_storage *s, const char * name, int handle) { + int begin = 0; + int end = s->name_count - 1; + while (begin<=end) { + int mid = (begin+end)/2; + struct handle_name *n = &s->name[mid]; + int c = strcmp(n->name, name); + if (c==0) { + return NULL; + } + if (c<0) { + begin = mid + 1; + } else { + end = mid - 1; + } + } + char * result = strdup(name); + + _insert_name_before(s, result, handle, begin); + + return result; +} + +const char * +skynet_handle_namehandle(int handle, const char *name) { + rwlock_wlock(&H->lock); + + const char * ret = _insert_name(H, name, handle); + + rwlock_wunlock(&H->lock); + + return ret; +} + +void +skynet_handle_init(void) { + assert(H==NULL); + struct handle_storage * s = malloc(sizeof(*H)); + s->slot_size = DEFAULT_SLOT_SIZE; + s->slot = malloc(s->slot_size * sizeof(struct skynet_context *)); + memset(s->slot, 0, s->slot_size * sizeof(struct handle_slot *)); + + rwlock_init(&s->lock); + s->handle_index = 0; + s->name_cap = 2; + s->name_count = 0; + s->name = malloc(s->name_cap * sizeof(struct handle_name)); + + H = s; + + // Don't need to free H +} + diff --git a/skynet_handle.h b/skynet_handle.h new file mode 100644 index 00000000..6e38cd34 --- /dev/null +++ b/skynet_handle.h @@ -0,0 +1,15 @@ +#ifndef SKYNET_CONTEXT_HANDLE_H +#define SKYNET_CONTEXT_HANDLE_H + +struct skynet_context; + +int skynet_handle_register(struct skynet_context *); +void skynet_handle_retire(int handle); +struct skynet_context * skynet_handle_grab(int handle); + +int skynet_handle_findname(const char * name); +const char * skynet_handle_namehandle(int handle, const char *name); + +void skynet_handle_init(void); + +#endif diff --git a/skynet_imp.h b/skynet_imp.h new file mode 100644 index 00000000..11fdbc6d --- /dev/null +++ b/skynet_imp.h @@ -0,0 +1,13 @@ +#ifndef SKYNET_IMP_H +#define SKYNET_IMP_H + +struct skynet_config { + int thread; + int mqueue_size; + char * logger; + const char * module_path; +}; + +void skynet_start(struct skynet_config * config); + +#endif diff --git a/skynet_logger.c b/skynet_logger.c new file mode 100644 index 00000000..8cad1ea9 --- /dev/null +++ b/skynet_logger.c @@ -0,0 +1,52 @@ +#include "skynet.h" + +#include +#include + +struct logger { + FILE * handle; + int close; +}; + +struct logger * +logger_create(void) { + struct logger * inst = malloc(sizeof(*inst)); + inst->handle = NULL; + inst->close = 0; + return inst; +} + +void +logger_release(struct logger * inst) { + if (inst->close) { + fclose(inst->handle); + } + free(inst); +} + +static void +_logger(struct skynet_context * context, void *ud, const char * uid, const void * msg, size_t sz) { + struct logger * inst = ud; + fprintf(inst->handle, "[%s] ",uid); + fwrite(msg, sz , 1, inst->handle); + fprintf(inst->handle, "\n"); +} + +int +logger_init(struct logger * inst, struct skynet_context *ctx, const char * parm) { + if (parm) { + inst->handle = fopen(parm,"w"); + if (inst->handle == NULL) { + return 1; + } + inst->close = 1; + } else { + inst->handle = stdout; + } + if (inst->handle) { + skynet_callback(ctx, inst, _logger); + skynet_command(ctx, "REG", "logger"); + return 0; + } + return 1; +} diff --git a/skynet_main.c b/skynet_main.c new file mode 100644 index 00000000..33a2b8c0 --- /dev/null +++ b/skynet_main.c @@ -0,0 +1,16 @@ +#include "skynet_imp.h" +#include + +int +main(void) { + struct skynet_config config; + + config.thread = 8; + config.mqueue_size = 256; + config.module_path = "./"; + config.logger = NULL; + + skynet_start(&config); + + return 0; +} diff --git a/skynet_module.c b/skynet_module.c new file mode 100644 index 00000000..acbac17e --- /dev/null +++ b/skynet_module.c @@ -0,0 +1,138 @@ +#include "skynet_module.h" + +#include +#include +#include +#include +#include +#include + +#define MAX_MODULE_TYPE 32 +#define SO ".so" + +struct modules { + int count; + int lock; + const char * path; + struct skynet_module m[MAX_MODULE_TYPE]; +}; + +static struct modules * M = NULL; + +static void * +_try_open(struct modules *m, const char * name) { + size_t path_size = strlen(m->path); + size_t name_size = strlen(name); + char tmp[path_size + name_size + sizeof(SO)]; + memcpy(tmp, m->path, path_size); + memcpy(tmp + path_size , name, name_size); + memcpy(tmp + path_size + name_size , SO, sizeof(SO)); + + void * dl = dlopen(tmp, RTLD_NOW | RTLD_GLOBAL); + + if (dl == NULL) { + printf("try open %s failed : %s\n",tmp,dlerror()); + } + + return dl; +} + + +static struct skynet_module * +_query(const char * name) { + int i; + for (i=0;icount;i++) { + if (strcmp(M->m[i].name,name)==0) { + return &M->m[i]; + } + } + return NULL; +} + +static int +_open_sym(struct skynet_module *mod) { + size_t name_size = strlen(mod->name); + char tmp[name_size + 9]; // create/init/release , longest name is release (7) + memcpy(tmp, mod->name, name_size); + strcpy(tmp+name_size, "_create"); + mod->create = dlsym(mod->module, tmp); + strcpy(tmp+name_size, "_init"); + mod->init = dlsym(mod->module, tmp); + strcpy(tmp+name_size, "_release"); + mod->release = dlsym(mod->module, tmp); + + return mod->init == NULL; +} + +struct skynet_module * +skynet_module_query(const char * name) { + struct skynet_module * result = _query(name); + if (result) + return result; + + while(__sync_lock_test_and_set(&M->lock,1)) {} + + result = _query(name); // double check + + if (result == NULL && M->count < MAX_MODULE_TYPE) { + int index = M->count; + void * dl = _try_open(M,name); + if (dl) { + M->m[index].name = name; + M->m[index].module = dl; + + if (_open_sym(&M->m[index]) == 0) { + M->m[index].name = strdup(name); + M->count ++; + result = &M->m[index]; + } + } + } + + __sync_lock_release(&M->lock); + + return result; +} + +void +skynet_module_insert(struct skynet_module *mod) { + while(__sync_lock_test_and_set(&M->lock,1)) {} + + struct skynet_module * m = _query(mod->name); + assert(m == NULL && M->count < MAX_MODULE_TYPE); + int index = M->count; + M->m[index] = *mod; + ++M->count; + __sync_lock_release(&M->lock); +} + +void * +skynet_module_instance_create(struct skynet_module *m) { + if (m->create) { + return m->create(); + } else { + return (void *)(intptr_t)(~0); + } +} + +int +skynet_module_instance_init(struct skynet_module *m, void * inst, struct skynet_context *ctx, const char * parm) { + return m->init(inst, ctx, parm); +} + +void +skynet_module_instance_release(struct skynet_module *m, void *inst) { + if (m->release) { + m->release(inst); + } +} + +void +skynet_module_init(const char *path) { + struct modules *m = malloc(sizeof(*m)); + m->count = 0; + m->path = strdup(path); + m->lock = 0; + + M = m; +} diff --git a/skynet_module.h b/skynet_module.h new file mode 100644 index 00000000..a2d96306 --- /dev/null +++ b/skynet_module.h @@ -0,0 +1,26 @@ +#ifndef SKYNET_MODULE_H +#define SKYNET_MODULE_H + +struct skynet_context; + +typedef void * (*skynet_dl_create)(void); +typedef int (*skynet_dl_init)(void * inst, struct skynet_context *, const char * parm); +typedef void (*skynet_dl_release)(void * inst); + +struct skynet_module { + const char * name; + void * module; + skynet_dl_create create; + skynet_dl_init init; + skynet_dl_release release; +}; + +void skynet_module_insert(struct skynet_module *mod); +struct skynet_module * skynet_module_query(const char * name); +void * skynet_module_instance_create(struct skynet_module *); +int skynet_module_instance_init(struct skynet_module *, void * inst, struct skynet_context *ctx, const char * parm); +void skynet_module_instance_release(struct skynet_module *, void *inst); + +void skynet_module_init(const char *path); + +#endif diff --git a/skynet_mq.c b/skynet_mq.c new file mode 100644 index 00000000..a7041a8c --- /dev/null +++ b/skynet_mq.c @@ -0,0 +1,102 @@ +#include "skynet_mq.h" + +#include +#include +#include + +struct message_queue { + int cap; + int head; + int tail; + int lock; + struct skynet_message *queue; +}; + +static struct message_queue *Q = NULL; + +struct message_queue * +skynet_mq_create(int cap) { + struct message_queue *q = malloc(sizeof(*q)); + q->cap = cap; + q->head = 0; + q->tail = 0; + q->lock = 0; + q->queue = malloc(sizeof(struct skynet_message) * cap); + + return q; +} + +void +skynet_mq_release(struct message_queue *q) { + free(q->queue); + free(q); +} + +static inline void +_lock_queue(struct message_queue *q) { + while (__sync_lock_test_and_set(&q->lock,1)) {} +} + +static inline void +_unlock_queue(struct message_queue *q) { + __sync_lock_release(&q->lock); +} + +int +skynet_mq_leave(struct message_queue *q, struct skynet_message *message) { + int ret = -1; + _lock_queue(q); + + if (q->head != q->tail) { + *message = q->queue[q->head]; + ret = message->destination; + if ( ++ q->head >= q->cap) { + q->head = 0; + } + } + + _unlock_queue(q); + + return ret; +} + +void +skynet_mq_enter(struct message_queue *q, struct skynet_message *message) { + _lock_queue(q); + + q->queue[q->tail] = *message; + if (++ q->tail >= q->cap) { + q->tail = 0; + } + + if (q->head == q->tail) { + struct skynet_message *new_queue = malloc(sizeof(struct skynet_message) * q->cap * 2); + int i; + for (i=0;icap;i++) { + new_queue[i] = q->queue[(q->head + i) % q->cap]; + } + q->head = 0; + q->tail = q->cap; + q->cap *= 2; + + free(q->queue); + q->queue = new_queue; + } + + _unlock_queue(q); +} + +int +skynet_mq_pop(struct skynet_message *message) { + return skynet_mq_leave(Q,message); +} + +void +skynet_mq_push(struct skynet_message *message) { + skynet_mq_enter(Q,message); +} + +void +skynet_mq_init(int cap) { + Q = skynet_mq_create(cap); +} diff --git a/skynet_mq.h b/skynet_mq.h new file mode 100644 index 00000000..b77a12fa --- /dev/null +++ b/skynet_mq.h @@ -0,0 +1,25 @@ +#ifndef SKYNET_MESSAGE_QUEUE_H +#define SKYNET_MESSAGE_QUEUE_H + +#include + +struct skynet_message { + int source; + int destination; + void * data; + size_t sz; +}; + +struct message_queue; + +int skynet_mq_pop(struct skynet_message *message); +void skynet_mq_push(struct skynet_message *message); + +struct message_queue * skynet_mq_create(int cap); +void skynet_mq_release(struct message_queue *q); +int skynet_mq_leave(struct message_queue *q, struct skynet_message *message); +void skynet_mq_enter(struct message_queue *q, struct skynet_message *message); + +void skynet_mq_init(int cap); + +#endif diff --git a/skynet_server.c b/skynet_server.c new file mode 100644 index 00000000..98b4d0a7 --- /dev/null +++ b/skynet_server.c @@ -0,0 +1,259 @@ +#include "skynet_server.h" +#include "skynet_module.h" +#include "skynet_handle.h" +#include "skynet_mq.h" +#include "skynet_blackhole.h" +#include "skynet_timer.h" +#include "skynet.h" + +#include +#include +#include +#include + +#define BLACKHOLE "blackhole" +#define DEFAULT_MESSAGE_QUEUE 16 + +struct skynet_context { + void * instance; + struct skynet_module * mod; + int handle; + int calling; + int ref; + char handle_name[10]; + char result[32]; + void * cb_ud; + skynet_cb cb; + struct message_queue *queue; +}; + +static void +_id_to_hex(char * str, int id) { + int i; + static char hex[16] = { '0','1','2','3','4','5','6','7','8','9','A','B','C','D','E','F' }; + for (i=0;i<8;i++) { + str[i] = hex[(id >> ((7-i) * 4))&0xf]; + } + str[8] = '\0'; +} + +struct skynet_context * +skynet_context_new(const char * name, char *parm) { + struct skynet_module * mod = skynet_module_query(name); + + if (mod == NULL) + return NULL; + + void *inst = skynet_module_instance_create(mod); + if (inst == NULL) + return NULL; + struct skynet_context * ctx = malloc(sizeof(*ctx)); + ctx->mod = mod; + ctx->instance = inst; + ctx->ref = 2; + ctx->cb = NULL; + ctx->cb_ud = NULL; + char * uid = ctx->handle_name; + uid[0] = ':'; + _id_to_hex(uid+1, ctx->handle); + + ctx->handle = skynet_handle_register(ctx); + ctx->queue = skynet_mq_create(DEFAULT_MESSAGE_QUEUE); + ctx->calling = 1; + // init function maybe use ctx->handle, so it must init at last + + int r = skynet_module_instance_init(mod, inst, ctx, parm); + if (r == 0) { + __sync_synchronize(); + ctx->calling = 0; + return skynet_context_release(ctx); + } else { + skynet_context_release(ctx); + skynet_handle_retire(ctx->handle); + return NULL; + } +} + +void +skynet_context_grab(struct skynet_context *ctx) { + __sync_add_and_fetch(&ctx->ref,1); +} + +static void +_delete_context(struct skynet_context *ctx) { + skynet_module_instance_release(ctx->mod, ctx->instance); + skynet_mq_release(ctx->queue); + free(ctx); +} + +struct skynet_context * +skynet_context_release(struct skynet_context *ctx) { + if (__sync_sub_and_fetch(&ctx->ref,1) == 0) { + _delete_context(ctx); + return NULL; + } + return ctx; +} + +static void +_drop_message(int source, const char * addr , void * data, size_t sz) { + struct blackhole * b = malloc(sizeof(*b)); + b->source = source; + b->destination = strdup(addr); + b->data = data; + b->sz = sz; + + int des = skynet_handle_findname(BLACKHOLE); + if (des<0) + return; + + struct skynet_message msg; + msg.source = source; + msg.destination = des; + msg.data = b; + msg.sz = sizeof(*b); + skynet_mq_push(&msg); +} + +static void +_dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { + if (msg->source == -1) { + ctx->cb(ctx, ctx->cb_ud, NULL, msg->data, msg->sz); + } else { + char tmp[10]; + tmp[0] = ':'; + _id_to_hex(tmp+1, msg->source); + ctx->cb(ctx, ctx->cb_ud, tmp, msg->data, msg->sz); + + free(msg->data); + } +} + +int +skynet_context_message_dispatch(void) { + struct skynet_message msg; + int handle = skynet_mq_pop(&msg); + if (handle < 0) { + return 1; + } + struct skynet_context * ctx = skynet_handle_grab(handle); + if (ctx == NULL) { + char tmp[10]; + tmp[0] = ':'; + _id_to_hex(tmp+1, msg.destination); + _drop_message(msg.source, tmp, msg.data, msg.sz); + return 0; + } + if (__sync_lock_test_and_set(&ctx->calling, 1)) { + // When calling, push to context's message queue + skynet_mq_enter(ctx->queue, &msg); + } else { + if (ctx->cb == NULL) { + char tmp[10]; + tmp[0] = ':'; + _id_to_hex(tmp+1, msg.destination); + _drop_message(msg.source, tmp, msg.data, msg.sz); + } else { + _dispatch_message(ctx, &msg); + while(skynet_mq_leave(ctx->queue,&msg) >=0) { + _dispatch_message(ctx,&msg); + } + } + __sync_lock_release(&ctx->calling); + } + + skynet_context_release(ctx); + + return 0; +} + +const char * +skynet_command(struct skynet_context * context, const char * cmd , const char * parm) { + if (strcmp(cmd,"TIMEOUT") == 0) { + char * session_ptr = NULL; + int ti = strtol(parm, &session_ptr, 10); + char sep = session_ptr[0]; + assert(sep == ':'); + int session = strtol(session_ptr+1, NULL, 10); + skynet_timeout(context->handle, ti, session); + return NULL; + } + + if (strcmp(cmd,"NOW") == 0) { + uint32_t ti = skynet_gettime(); + sprintf(context->result,"%u",ti); + return context->result; + } + + if (strcmp(cmd,"REG") == 0) { + if (parm == NULL || parm[0] == '\0') { + return context->handle_name; + } else { + return skynet_handle_namehandle(context->handle, parm); + } + } + + if (strcmp(cmd,"EXIT") == 0) { + skynet_handle_retire(context->handle); + return NULL; + } + + if (strcmp(cmd,"LAUNCH") == 0) { + size_t sz = strlen(parm); + char tmp[sz+1]; + strcpy(tmp,parm); + char * parm = tmp; + char * mod = strsep(&parm, " \t\r\n"); + parm = strsep(&parm, "\r\n"); + struct skynet_context * inst = skynet_context_new(mod,parm); + if (inst == NULL) { + return NULL; + } else { + context->result[0] = ':'; + _id_to_hex(context->result+1, inst->handle); + return context->result; + } + } + + return NULL; +} + +void +skynet_send(struct skynet_context * context, const char * addr , void * msg, size_t sz) { + int des = -1; + if (addr[0] == ':') { + des = strtol(addr+1, NULL, 16); + } else if (addr[0] == '.') { + des = skynet_handle_findname(addr + 1); + if (des < 0) { + _drop_message(context->handle, addr, (void *)msg, sz); + return; + } + } + + assert(des >= 0); + struct skynet_message smsg; + smsg.source = context->handle; + smsg.destination = des; + smsg.data = msg; + smsg.sz = sz; + skynet_mq_push(&smsg); +} + +int +skynet_context_handle(struct skynet_context *ctx) { + return ctx->handle; +} + +void +skynet_context_init(struct skynet_context *ctx, int handle) { + ctx->handle = handle; +} + +void +skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb) { + assert(context->cb == NULL); + context->cb = cb; + context->cb_ud = ud; +} + diff --git a/skynet_server.h b/skynet_server.h new file mode 100644 index 00000000..37ce3db5 --- /dev/null +++ b/skynet_server.h @@ -0,0 +1,16 @@ +#ifndef SKYNET_SERVER_H +#define SKYNET_SERVER_H + +struct skynet_context; +struct skynet_message; + +struct skynet_context * skynet_context_new(const char * name, char * parm); +void skynet_context_grab(struct skynet_context *); +struct skynet_context * skynet_context_release(struct skynet_context *); +int skynet_context_handle(struct skynet_context *); +void skynet_context_init(struct skynet_context *, int handle); +void skynet_context_push(struct skynet_context *, struct skynet_message *message); +int skynet_context_pop(struct skynet_context *, struct skynet_message *message); +int skynet_context_message_dispatch(void); // return 1 when block + +#endif diff --git a/skynet_start.c b/skynet_start.c new file mode 100644 index 00000000..b1fa9cab --- /dev/null +++ b/skynet_start.c @@ -0,0 +1,62 @@ +#include "skynet_server.h" +#include "skynet_imp.h" +#include "skynet_mq.h" +#include "skynet_handle.h" +#include "skynet_module.h" +#include "skynet_timer.h" + +#include +#include +#include + +static void * +_timer(void *p) { + for (;;) { + skynet_updatetime(); + usleep(2500); + } + return NULL; +} + +static void * +_worker(void *p) { + for (;;) { + if (skynet_context_message_dispatch()) { + usleep(1000); + } + } + return NULL; +} + +static void +_start(int thread) { + pthread_t pid[thread+1]; + + pthread_create(&pid[0], NULL, _timer, NULL); + int i; + for (i=1;imqueue_size); + skynet_module_init(config->module_path); + skynet_handle_init(); + skynet_timer_init(); + struct skynet_context *ctx; + ctx = skynet_context_new("logger", config->logger); + assert(ctx); + ctx = skynet_context_new("blackhole", NULL); + assert(ctx); + char parm[] = "main.lua"; + ctx = skynet_context_new("snlua", parm); + + _start(config->thread); +} + diff --git a/skynet_timer.c b/skynet_timer.c new file mode 100644 index 00000000..d62c8b52 --- /dev/null +++ b/skynet_timer.c @@ -0,0 +1,206 @@ + #include "skynet_timer.h" +#include "skynet_mq.h" + +#include +#include +#include +#include + +typedef void (*timer_execute_func)(void *ud,void *arg); + +#define TIME_NEAR_SHIFT 8 +#define TIME_NEAR (1 << TIME_NEAR_SHIFT) +#define TIME_LEVEL_SHIFT 6 +#define TIME_LEVEL (1 << TIME_LEVEL_SHIFT) +#define TIME_NEAR_MASK (TIME_NEAR-1) +#define TIME_LEVEL_MASK (TIME_LEVEL-1) + +struct timer_node { + struct timer_node *next; + int expire; +}; + +struct link_list { + struct timer_node head; + struct timer_node *tail; +}; + +struct timer { + struct link_list near[TIME_NEAR]; + struct link_list t[4][TIME_LEVEL-1]; + int lock; + int time; + uint32_t current; +}; + +static struct timer * TI = NULL; + +static inline struct timer_node * +link_clear(struct link_list *list) +{ + struct timer_node * ret = list->head.next; + list->head.next = 0; + list->tail = &(list->head); + + return ret; +} + +static inline void +link(struct link_list *list,struct timer_node *node) +{ + list->tail->next = node; + list->tail = node; + node->next=0; +} + +static void +add_node(struct timer *T,struct timer_node *node) +{ + int time=node->expire; + int current_time=T->time; + + if ((time|TIME_NEAR_MASK)==(current_time|TIME_NEAR_MASK)) { + link(&T->near[time&TIME_NEAR_MASK],node); + } + else { + int i; + int mask=TIME_NEAR << TIME_LEVEL_SHIFT; + for (i=0;i<3;i++) { + if ((time|(mask-1))==(current_time|(mask-1))) { + break; + } + mask <<= TIME_LEVEL_SHIFT; + } + link(&T->t[i][((time>>(TIME_NEAR_SHIFT + i*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)-1],node); + } +} + +static void +timer_add(struct timer *T,void *arg,size_t sz,int time) +{ + struct timer_node *node = (struct timer_node *)malloc(sizeof(*node)+sz); + memcpy(node+1,arg,sz); + + while (__sync_lock_test_and_set(&T->lock,1)) {}; + + node->expire=time+T->time; + add_node(T,node); + + __sync_lock_release(&T->lock); +} + +static void +timer_execute(struct timer *T) +{ + while (__sync_lock_test_and_set(&T->lock,1)) {}; + int idx=T->time & TIME_NEAR_MASK; + struct timer_node *current; + int mask,i,time; + + while (T->near[idx].head.next) { + current=link_clear(&T->near[idx]); + + do { + struct timer_node *temp=current; + skynet_mq_push((struct skynet_message *)(temp+1)); + current=current->next; + free(temp); + } while (current); + } + + ++T->time; + + mask = TIME_NEAR; + time = T->time >> TIME_NEAR_SHIFT; + i=0; + + while ((T->time & (mask-1))==0) { + idx=time & TIME_LEVEL_MASK; + if (idx!=0) { + --idx; + current=link_clear(&T->t[i][idx]); + while (current) { + struct timer_node *temp=current->next; + add_node(T,current); + current=temp; + } + break; + } + mask <<= TIME_LEVEL_SHIFT; + time >>= TIME_LEVEL_SHIFT; + ++i; + } + __sync_lock_release(&T->lock); +} + +static struct timer * +timer_create_timer() +{ + struct timer *r=(struct timer *)malloc(sizeof(struct timer)); + memset(r,0,sizeof(*r)); + + int i,j; + + for (i=0;inear[i]); + } + + for (i=0;i<4;i++) { + for (j=0;jt[i][j]); + } + } + + r->lock = 0; + r->current = 0; + + return r; +} + +void +skynet_timeout(int handle, int time, int session) { + struct skynet_message message; + message.source = -1; + message.destination = handle; + message.data = NULL; + message.sz = (size_t) session; + if (time == 0) { + skynet_mq_push(&message); + } else { + timer_add(TI, &message, sizeof(message), time); + } +} + +static uint32_t +_gettime(void) { + struct timespec ti; + clock_gettime(CLOCK_MONOTONIC, &ti); + uint32_t t = (uint32_t)(ti.tv_sec & 0xffffff) * 100; + t += ti.tv_nsec / 10000000; + + return t; +} + +void +skynet_updatetime(void) { + uint32_t ct = _gettime(); + if (ct > TI->current) { + int diff = ct-TI->current; + TI->current = ct; + int i; + for (i=0;icurrent; +} + +void +skynet_timer_init(void) { + TI = timer_create_timer(); + TI->current = _gettime(); +} diff --git a/skynet_timer.h b/skynet_timer.h new file mode 100644 index 00000000..65ce817b --- /dev/null +++ b/skynet_timer.h @@ -0,0 +1,12 @@ +#ifndef SKYNET_TIMER_H +#define SKYNET_TIMER_H + +#include + +void skynet_timeout(int handle, int time, int session); +void skynet_updatetime(void); +uint32_t skynet_gettime(void); + +void skynet_timer_init(void); + +#endif diff --git a/testtimer.lua b/testtimer.lua new file mode 100644 index 00000000..e95ee527 --- /dev/null +++ b/testtimer.lua @@ -0,0 +1,10 @@ +local skynet = require "skynet" + +skynet.callback(function(addr,content) + if content == nil then + print("sn:",addr) + skynet.command("TIMEOUT","100:"..tostring(addr+1)) + end +end) + +skynet.command("TIMEOUT","0:0") \ No newline at end of file diff --git a/watchdog.lua b/watchdog.lua new file mode 100644 index 00000000..8fa6874d --- /dev/null +++ b/watchdog.lua @@ -0,0 +1,34 @@ +local skynet = require "skynet" + +local command = {} + +function command:open(parm) + local fd,addr = string.match(parm,"(%d+) ([^%s]+)") + fd = tonumber(fd) + print("[watchdog] open",self,fd,addr) + local agent = skynet.command("LAUNCH","snlua agent.lua ".. self) + if agent then + skynet.send(".gate","forward ".. self .. " " .. agent) + end +end + +function command:close() + print("[watchdog] close",self) +end + +function command:data(data) + print("[watchdog] data",self,#data,data) +end + +skynet.callback(function(from , message) + local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)") + id = tonumber(id) + local f = command[cmd] + if f then + f(id,parm) + else + skynet.error(string.format("[watchdog] Unknown command : %s %d %s",cmd,id,parm)) + end +end) + +skynet.command("REG","watchdog") \ No newline at end of file