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

new socket service and socket lib instead of old connection, and rewrite redis lib

Browse Source
This commit is contained in:
云风
2013-06-21 13:02:51 +08:00
parent 0ffd9b8b8e
commit 277b6957cb
16 changed files with 1182 additions and 1248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
Makefile
View File

@@ -21,14 +21,14 @@ all : \
service/logger.so \ service/logger.so \
service/gate.so \ service/gate.so \
service/client.so \ service/client.so \
service/connection.so \
service/master.so \ service/master.so \
service/multicast.so \ service/multicast.so \
service/tunnel.so \ service/tunnel.so \
service/harbor.so \ service/harbor.so \
service/localcast.so \ service/localcast.so \
service/socket.so \
luaclib/skynet.so \ luaclib/skynet.so \
luaclib/socket.so \ luaclib/socketbuffer.so \
luaclib/int64.so \ luaclib/int64.so \
luaclib/mcast.so \ luaclib/mcast.so \
client client
@@ -72,7 +72,7 @@ service/snlua.so : service-src/service_lua.c
gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src
service/gate.so : gate/mread.c gate/ringbuffer.c gate/main.c service/gate.so : gate/mread.c gate/ringbuffer.c gate/main.c
gcc $(CFLAGS) $(SHARED) $^ -o $@ -Igate -Iskynet-src gcc $(CFLAGS) $(SHARED) $^ -o $@ -Igate -Iskynet-src -Iservice-src
service/localcast.so : service-src/service_localcast.c service/localcast.so : service-src/service_localcast.c
gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src
@@ -83,11 +83,11 @@ luaclib/skynet.so : lualib-src/lua-skynet.c lualib-src/lua-seri.c lualib-src/lua
service/client.so : service-src/service_client.c service/client.so : service-src/service_client.c
gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src
service/connection.so : connection/connection.c connection/main.c service/socket.so : service-src/service_socket.c
gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src -Iconnection gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src
luaclib/socket.so : connection/lua-socket.c | luaclib luaclib/socketbuffer.so : lualib-src/lua-socket.c | luaclib
gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src -Iconnection gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src
luaclib/int64.so : lua-int64/int64.c | luaclib luaclib/int64.so : lua-int64/int64.c | luaclib
gcc $(CFLAGS) $(SHARED) -Iluacompat -O2 $^ -o $@ gcc $(CFLAGS) $(SHARED) -Iluacompat -O2 $^ -o $@

145
connection/connection.c
View File

@@ -1,145 +0,0 @@
#include "connection.h"
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <unistd.h>
/* Test for polling API */
#ifdef __linux__
#define HAVE_EPOLL 1
#endif
#if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined (__NetBSD__)
#define HAVE_KQUEUE 1
#endif
#if !defined(HAVE_EPOLL) && !defined(HAVE_KQUEUE)
#error "system does not support epoll or kqueue API"
#endif
/* ! Test for polling API */
#ifdef HAVE_EPOLL
#include <sys/epoll.h>
#elif HAVE_KQUEUE
#include <sys/event.h>
#endif
#define EPOLLQUEUE 32
struct connection_pool {
#ifdef HAVE_EPOLL
int epoll_fd;
#elif HAVE_KQUEUE
int kqueue_fd;
#endif
int queue_len;
int queue_head;
#ifdef HAVE_EPOLL
struct epoll_event ev[EPOLLQUEUE];
#elif HAVE_KQUEUE
struct kevent ev[EPOLLQUEUE];
#endif
};
struct connection_pool *
connection_newpool(int max) {
#ifdef HAVE_EPOLL
int epoll_fd = epoll_create(max);
if (epoll_fd == -1) {
return NULL;
}
#elif HAVE_KQUEUE
int kqueue_fd = kqueue();
if (kqueue_fd == -1) {
return NULL;
}
#endif
struct connection_pool * pool = malloc(sizeof(*pool));
#ifdef HAVE_EPOLL
pool->epoll_fd = epoll_fd;
#elif HAVE_KQUEUE
pool->kqueue_fd = kqueue_fd;
#endif
pool->queue_len = 0;
pool->queue_head = 0;
return pool;
}
void
connection_deletepool(struct connection_pool * pool) {
#if HAVE_EPOLL
close(pool->epoll_fd);
#elif HAVE_KQUEUE
close(pool->kqueue_fd);
#endif
free(pool);
}
int
connection_add(struct connection_pool * pool, int fd, void *ud) {
#if HAVE_EPOLL
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.ptr = ud;
if (epoll_ctl(pool->epoll_fd, EPOLL_CTL_ADD, fd, &ev) == -1) {
return 1;
}
#elif HAVE_KQUEUE
struct kevent ke;
EV_SET(&ke, fd, EVFILT_READ, EV_ADD, 0, 0, ud);
if (kevent(pool->kqueue_fd, &ke, 1, NULL, 0, NULL) == -1) {
return 1;
}
#endif
return 0;
}
void
connection_del(struct connection_pool * pool, int fd) {
#if HAVE_EPOLL
struct epoll_event ev;
epoll_ctl(pool->epoll_fd, EPOLL_CTL_DEL, fd , &ev);
#elif HAVE_KQUEUE
struct kevent ke;
EV_SET(&ke, fd, EVFILT_READ, EV_DELETE, 0, 0, NULL);
kevent(pool->kqueue_fd, &ke, 1, NULL, 0, NULL);
#endif
}
static int
_read_queue(struct connection_pool * pool, int timeout) {
pool->queue_head = 0;
#if HAVE_EPOLL
int n = epoll_wait(pool->epoll_fd , pool->ev, EPOLLQUEUE, timeout);
#elif HAVE_KQUEUE
struct timespec timeoutspec;
timeoutspec.tv_sec = timeout / 1000;
timeoutspec.tv_nsec = (timeout % 1000) * 1000000;
int n = kevent(pool->kqueue_fd, NULL, 0, pool->ev, EPOLLQUEUE, &timeoutspec);
#endif
if (n == -1) {
pool->queue_len = 0;
return -1;
}
pool->queue_len = n;
return n;
}
void *
connection_poll(struct connection_pool * pool, int timeout) {
if (pool->queue_head >= pool->queue_len) {
if (_read_queue(pool, timeout) <= 0) {
return NULL;
}
}
#if HAVE_EPOLL
return pool->ev[pool->queue_head ++].data.ptr;
#elif HAVE_KQUEUE
return pool->ev[pool->queue_head ++].udata;
#endif
}

16
connection/connection.h
View File

@@ -1,16 +0,0 @@
#ifndef SKYNET_CONNECTION_H
#define SKYNET_CONNECTION_H
#include <stddef.h>
struct connection_pool;
struct connection_pool * connection_newpool(int max);
void connection_deletepool(struct connection_pool *);
int connection_add(struct connection_pool *, int fd, void *ud);
void connection_del(struct connection_pool *, int fd);
void * connection_poll(struct connection_pool *, int timeout);
#endif

327
connection/lua-socket.c
View File

@@ -1,327 +0,0 @@
#include "skynet.h"
#include "luacompat52.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <netinet/in.h>
#include <sys/uio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <errno.h>
#include <unistd.h>
#include <lua.h>
#include <lauxlib.h>
static int
_open(lua_State *L) {
const char * ip = luaL_checkstring(L,1);
int port = luaL_checkinteger(L,2);
struct sockaddr_in my_addr;
memset(&my_addr, 0, sizeof(struct sockaddr_in));
my_addr.sin_family = AF_INET;
my_addr.sin_port = htons(port);
my_addr.sin_addr.s_addr=inet_addr(ip);
int fd = socket(AF_INET,SOCK_STREAM,0);
int r = connect(fd,(struct sockaddr *)&my_addr,sizeof(struct sockaddr_in));
if (r == -1) {
close(fd);
return 0;
}
lua_pushinteger(L,fd);
return 1;
}
static int
_blockwrite(int fd, const char * buffer, size_t sz) {
while (sz > 0) {
int bytes = send(fd, buffer, sz, 0);
if (bytes < 0) {
switch (errno) {
case EAGAIN:
case EINTR:
continue;
}
return -1;
}
sz -= bytes;
buffer += bytes;
sleep(0);
}
return 0;
}
static int
_write(lua_State *L) {
int fd = -1;
if (!lua_isnil(L,1)) {
fd = luaL_checkinteger(L,1);
}
int type = lua_type(L,2);
const char * buffer = NULL;
size_t sz;
if (type == LUA_TSTRING) {
buffer = lua_tolstring(L,2,&sz);
} else {
luaL_checktype(L,2, LUA_TLIGHTUSERDATA);
buffer = lua_touserdata(L,2);
sz = luaL_checkinteger(L,3);
}
if (fd >= 0) {
if (_blockwrite(fd, buffer, sz) == 0) {
return 0;
}
}
if (type == LUA_TSTRING) {
lua_settop(L,2);
} else {
lua_pushlstring(L, buffer, sz);
}
return 1;
}
static int
_writeblock(lua_State *L) {
luaL_Buffer b;
int fd = -1;
if (!lua_isnil(L,1)) {
fd = luaL_checkinteger(L,1);
}
int header = luaL_checkinteger(L,2);
int type = lua_type(L,3);
const char * buffer = NULL;
size_t sz;
if (type == LUA_TSTRING) {
buffer = lua_tolstring(L,3,&sz);
} else {
luaL_checktype(L, 3, LUA_TLIGHTUSERDATA);
buffer = lua_touserdata(L,3);
sz = luaL_checkinteger(L,4);
}
if (header == 2) {
if (sz > 65535) {
luaL_error(L, "Too big package %d", (int)sz);
}
} else {
if (header != 4) {
luaL_error(L, "block header must be 2 or 4 bytes");
}
}
uint8_t head[4];
if (header == 2) {
// send big-endian header
head[0] = sz >> 8 & 0xff;
head[1] = sz & 0xff ;
} else {
head[0] = sz >> 24 & 0xff;
head[1] = sz >> 16 & 0xff;
head[2] = sz >> 8 & 0xff;
head[3] = sz & 0xff;
header = 4;
}
if (fd >= 0) {
if (_blockwrite(fd, (const char * )head, header)) {
goto _error;
}
if (_blockwrite(fd , buffer, sz)) {
goto _error;
}
return 0;
}
_error:
luaL_buffinitsize(L,&b, header + sz);
luaL_addlstring(&b, (const char *)head, header);
luaL_addlstring(&b, buffer, sz);
luaL_pushresult(&b);
return 1;
}
struct buffer {
int cap;
int size;
char * buffer;
int read;
};
static int
_new(lua_State *L) {
struct buffer * buffer = lua_newuserdata(L, sizeof(struct buffer));
buffer->cap = 0;
buffer->size = 0;
buffer->buffer = NULL;
buffer->read = 0;
return 1;
}
static int
_delete(lua_State *L) {
struct buffer * buffer = lua_touserdata(L,1);
free(buffer->buffer);
buffer->buffer = NULL;
return 0;
}
static int
_push(lua_State *L) {
luaL_checktype(L,1,LUA_TUSERDATA);
struct buffer * buffer = lua_touserdata(L,1);
int type = lua_type(L,2);
const char * buf;
size_t size;
if (type == LUA_TSTRING) {
buf = lua_tolstring(L,2,&size);
} else {
luaL_checktype(L,2,LUA_TLIGHTUSERDATA);
buf = lua_touserdata(L,2);
size = luaL_checkinteger(L,3);
}
int old_size = buffer->size - buffer->read;
if (size + old_size > buffer->cap) {
if (buffer->cap == 0) {
lua_rawgetp(L, LUA_REGISTRYINDEX, _delete);
lua_setmetatable(L, 1);
}
if (buffer->read == 0) {
buffer->buffer = realloc(buffer->buffer, size+old_size);
buffer->cap = size + old_size;
memcpy(buffer->buffer + old_size , buf , size);
buffer->size += size;
} else {
char * new_buffer = malloc(size + old_size);
memcpy(new_buffer, buffer->buffer + buffer->read, old_size);
memcpy(new_buffer + old_size , buf, size);
free(buffer->buffer);
buffer->buffer = new_buffer;
buffer->read = 0;
buffer->size = old_size + size;
}
} else if (buffer->size + size > buffer->cap) {
memmove(buffer->buffer, buffer->buffer + buffer->read, old_size);
memcpy(buffer->buffer + old_size, buf, size);
buffer->read = 0;
buffer->size = old_size + size;
} else {
memcpy(buffer->buffer + buffer->size, buf, size);
buffer->size += size;
}
return 0;
}
static int
_read(lua_State *L) {
luaL_checktype(L,1,LUA_TUSERDATA);
struct buffer * buffer = lua_touserdata(L,1);
int need = luaL_checkinteger(L,2);
int size = buffer->size - buffer->read;
if (need <= size) {
lua_pushlstring(L, buffer->buffer + buffer->read, need);
buffer->read += need;
return 1;
}
return 0;
}
static int
_readline(lua_State *L) {
luaL_checktype(L,1,LUA_TUSERDATA);
struct buffer * buffer = lua_touserdata(L,1);
size_t sz;
const char * sep = luaL_checklstring(L,2,&sz);
int i;
for (i=buffer->read;i<=buffer->size - (int)sz;i++) {
if (memcmp(buffer->buffer+i,sep,sz) == 0) {
lua_pushlstring(L, buffer->buffer + buffer->read, i-buffer->read);
buffer->read = i+sz;
return 1;
}
}
return 0;
}
/*
userdata buffer
intteger header
function (msg, sz, ...)
...
*/
static int
_readblock(lua_State *L) {
luaL_checktype(L,1,LUA_TUSERDATA);
struct buffer * buffer = lua_touserdata(L,1);
int top = lua_gettop(L);
int size = buffer->size - buffer->read;
if (size < 2) {
return 0;
}
int header_size = luaL_checkinteger(L,2);
uint8_t * buf = (uint8_t *)buffer->buffer + buffer->read;
uint16_t len;
if (header_size == 2) {
len = buf[0] << 8 | buf[1];
} else {
if (header_size != 4) {
return luaL_error(L, "header size must be 2 or 4");
}
len = buf[0] << 24 | buf[1] << 16 | buf[2] << 8 | buf[3];
}
if (size < header_size + len) {
return 0;
}
if (top == 3) {
lua_pushlightuserdata(L, buffer->buffer + buffer->read + header_size);
lua_pushinteger(L, len);
lua_call(L,2,LUA_MULTRET);
buffer->read += len + header_size;
return lua_gettop(L) - 2;
} else {
lua_pushvalue(L,3);
lua_replace(L,1);
lua_pushlightuserdata(L, buffer->buffer + buffer->read + header_size);
lua_replace(L,2);
lua_pushinteger(L, len);
lua_replace(L,3);
lua_call(L,top-1,LUA_MULTRET);
buffer->read += len + header_size;
return lua_gettop(L);
}
}
int
luaopen_socket_c(lua_State *L) {
luaL_Reg l[] = {
{ "open", _open },
{ "write", _write },
{ "new", _new },
{ "push", _push },
{ "read", _read },
{ "readline", _readline },
{ "readblock", _readblock },
{ "writeblock", _writeblock },
{ NULL, NULL },
};
luaL_checkversion(L);
luaL_newlib(L,l);
lua_createtable(L,0,1);
lua_pushcfunction(L, _delete);
lua_setfield(L, -2, "__gc");
lua_rawsetp(L, LUA_REGISTRYINDEX, _delete);
return 1;
}

204
connection/main.c
View File

@@ -1,204 +0,0 @@
#include "connection.h"
#include "skynet.h"
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#define DEFAULT_BUFFER_SIZE 1024
#define DEFAULT_CONNECTION 16
struct connection {
int fd;
uint32_t address;
int close;
};
struct connection_server {
int max_connection;
int current_connection;
struct connection_pool *pool;
struct skynet_context *ctx;
struct connection * conn;
};
struct connection_server *
connection_create(void) {
struct connection_server * server = malloc(sizeof(*server));
memset(server,0,sizeof(*server));
return server;
}
void
connection_release(struct connection_server * server) {
if (server->pool) {
connection_deletepool(server->pool);
}
free(server->conn);
free(server);
}
static void
_expand(struct connection_server * server) {
connection_deletepool(server->pool);
server->pool = connection_newpool(server->max_connection * 2);
int i;
for (i=0;i<server->max_connection;i++) {
struct connection * c = &server->conn[i];
int err = connection_add(server->pool, c->fd , c);
assert(err == 0);
}
server->max_connection *= 2;
}
static void
_add(struct connection_server * server, int fd , uint32_t address) {
++server->current_connection;
if (server->current_connection > server->max_connection) {
_expand(server);
}
int i;
for (i=0;i<server->max_connection;i++) {
struct connection * c = &server->conn[i];
if (c->address == 0) {
c->fd = fd;
c->address = address;
c->close = 0;
int err = connection_add(server->pool, fd , c);
assert(err == 0);
return;
}
}
assert(0);
}
static void
_del(struct connection_server * server, int fd) {
int i;
for (i=0;i<server->max_connection;i++) {
struct connection * c = &server->conn[i];
if (c->fd == fd) {
if (c->close == 0) {
skynet_send(server->ctx, 0, c->address, PTYPE_CLIENT | PTYPE_TAG_DONTCOPY, 0, NULL, 0);
connection_del(server->pool, fd);
}
--server->current_connection;
c->address = 0;
c->fd = 0;
c->close = 0;
close(fd);
return;
}
}
skynet_error(server->ctx, "[connection] Delete invalid handle %d", fd);
}
static void
_poll(struct connection_server * server) {
int timeout = 100;
void * buffer = NULL;
for (;;) {
struct connection * c = connection_poll(server->pool, timeout);
if (c==NULL) {
skynet_command(server->ctx,"TIMEOUT","1");
return;
}
timeout = 0;
if (buffer == NULL) {
buffer = malloc(DEFAULT_BUFFER_SIZE);
}
int size = read(c->fd, buffer, DEFAULT_BUFFER_SIZE);
if (size < 0) {
continue;
}
if (size == 0) {
free(buffer);
buffer = NULL;
if (c->close == 0) {
c->close = 1;
skynet_send(server->ctx, 0, c->address, PTYPE_CLIENT | PTYPE_TAG_DONTCOPY, 0, NULL, 0);
connection_del(server->pool, c->fd);
}
} else {
skynet_send(server->ctx, 0, c->address, PTYPE_CLIENT | PTYPE_TAG_DONTCOPY, 0, buffer, size);
buffer = NULL;
}
}
}
static int
_connection_main(struct skynet_context * ctx, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
if (type == PTYPE_RESPONSE) {
_poll(ud);
return 0;
}
assert(type == PTYPE_TEXT);
const char * param = (const char *)msg + 4;
if (memcmp(msg, "ADD ", 4)==0) {
char * endptr;
int fd = strtol(param, &endptr, 10);
if (endptr == NULL) {
skynet_error(ctx, "[connection] Invalid ADD command from %x (session = %d)", source, session);
return 0;
}
int addr_sz = sz - (endptr - (char *)msg);
if (addr_sz <= 1) {
skynet_error(ctx, "[connection] Invalid ADD command from %x (session = %d)", source, session);
return 0;
}
char addr [addr_sz];
memcpy(addr, endptr+1, addr_sz-1);
addr[addr_sz-1] = '\0';
uint32_t address = strtoul(addr+1, NULL, 16);
if (address == 0) {
skynet_error(ctx, "[connection] Invalid ADD command from %x (session = %d)", source, session);
return 0;
}
_add(ud, fd, address);
} else if (memcmp(msg, "DEL ", 4)==0) {
char * endptr;
int fd = strtol(param, &endptr, 10);
if (endptr == NULL) {
skynet_error(ctx, "[connection] Invalid DEL command from %x (session = %d)", source, session);
return 0;
}
_del(ud, fd);
} else {
skynet_error(ctx, "[connection] Invalid command from %x (session = %d)", source, session);
}
return 0;
}
int
connection_init(struct connection_server * server, struct skynet_context * ctx, char * param) {
server->pool = connection_newpool(DEFAULT_CONNECTION);
if (server->pool == NULL)
return 1;
server->max_connection = strtol(param, NULL, 10);
if (server->max_connection == 0) {
server->max_connection = DEFAULT_CONNECTION;
}
server->current_connection = 0;
server->ctx = ctx;
server->conn = malloc(server->max_connection * sizeof(struct connection));
memset(server->conn, 0, server->max_connection * sizeof(struct connection));
skynet_callback(ctx, server, _connection_main);
skynet_command(ctx,"REG",".connection");
skynet_command(ctx,"TIMEOUT","0");
return 0;
}

149
gate/mread.c
View File

@@ -1,28 +1,6 @@
#include "mread.h" #include "mread.h"
#include "ringbuffer.h" #include "ringbuffer.h"
#include "event.h"
/* Test for polling API */
#ifdef __linux__
#define HAVE_EPOLL 1
#endif
#if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined (__NetBSD__)
#define HAVE_KQUEUE 1
#endif
#if !defined(HAVE_EPOLL) && !defined(HAVE_KQUEUE)
#error "system does not support epoll or kqueue API"
#endif
/* ! Test for polling API */
#include <sys/types.h>
#ifdef HAVE_EPOLL
#include <sys/epoll.h>
#elif HAVE_KQUEUE
#include <sys/event.h>
#endif
#include <netinet/in.h> #include <netinet/in.h>
#include <sys/socket.h> #include <sys/socket.h>
@@ -37,7 +15,6 @@
#include <fcntl.h> #include <fcntl.h>
#define BACKLOG 32 #define BACKLOG 32
#define READQUEUE 32
#define READBLOCKSIZE 2048 #define READBLOCKSIZE 2048
#define RINGBUFFER_DEFAULT 1024 * 1024 #define RINGBUFFER_DEFAULT 1024 * 1024
@@ -75,11 +52,7 @@ struct socket {
struct mread_pool { struct mread_pool {
int listen_fd; int listen_fd;
#ifdef HAVE_EPOLL int efd;
int epoll_fd;
#elif HAVE_KQUEUE
int kqueue_fd;
#endif
int max_connection; int max_connection;
int closed; int closed;
int active; int active;
@@ -88,11 +61,7 @@ struct mread_pool {
struct socket * free_socket; struct socket * free_socket;
int queue_len; int queue_len;
int queue_head; int queue_head;
#ifdef HAVE_EPOLL struct event ev[MAX_EVENT];
struct epoll_event ev[READQUEUE];
#elif HAVE_KQUEUE
struct kevent ev[READQUEUE];
#endif
struct ringbuffer * rb; struct ringbuffer * rb;
}; };
@@ -103,16 +72,7 @@ turn_on(struct mread_pool *self, struct socket * s) {
if (s->status < SOCKET_ALIVE || s->enablewrite) { if (s->status < SOCKET_ALIVE || s->enablewrite) {
return; return;
} }
#ifdef HAVE_EPOLL event_write(self->efd, s->fd, s, true);
struct epoll_event ev;
ev.events = EPOLLIN | EPOLLOUT;
ev.data.ptr = s;
epoll_ctl(self->epoll_fd, EPOLL_CTL_MOD, s->fd, &ev);
#elif HAVE_KQUEUE
struct kevent ke;
EV_SET(&ke, s->fd, EVFILT_WRITE, EV_ENABLE, 0, 0, s);
kevent(self->kqueue_fd, &ke, 1, NULL, 0, NULL);
#endif
s->enablewrite = 1; s->enablewrite = 1;
} }
@@ -122,19 +82,9 @@ turn_off(struct mread_pool *self, struct socket * s) {
return; return;
} }
s->enablewrite = 0; s->enablewrite = 0;
#ifdef HAVE_EPOLL event_write(self->efd, s->fd, s, false);
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.ptr = s;
epoll_ctl(self->epoll_fd, EPOLL_CTL_MOD, s->fd, &ev);
#elif HAVE_KQUEUE
struct kevent ke;
EV_SET(&ke, s->fd, EVFILT_WRITE, EV_DISABLE, 0, 0, s);
kevent(self->kqueue_fd, &ke, 1, NULL, 0, NULL);
#endif
} }
static void static void
free_buffer(struct send_client * sc) { free_buffer(struct send_client * sc) {
struct send_buffer * sb = sc->head; struct send_buffer * sb = sc->head;
@@ -301,46 +251,22 @@ mread_create(uint32_t addr, int port , int max , int buffer_size) {
return NULL; return NULL;
} }
#ifdef HAVE_EPOLL int efd = event_init(max+1);
int epoll_fd = epoll_create(max + 1); if (efd == -1) {
if (epoll_fd == -1) {
close(listen_fd); close(listen_fd);
return NULL; return NULL;
} }
struct epoll_event ev; if (event_add(efd, listen_fd, LISTENSOCKET) == -1) {
ev.events = EPOLLIN;
ev.data.ptr = LISTENSOCKET;
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev) == -1) {
close(listen_fd); close(listen_fd);
close(epoll_fd); close(efd);
return NULL; return NULL;
} }
#elif HAVE_KQUEUE
int kqueue_fd = kqueue();
if (kqueue_fd == -1) {
close(listen_fd);
return NULL;
}
struct kevent ke;
EV_SET(&ke, listen_fd, EVFILT_READ, EV_ADD, 0, 0, LISTENSOCKET);
if (kevent(kqueue_fd, &ke, 1, NULL, 0, NULL) == -1) {
close(listen_fd);
close(kqueue_fd);
return NULL;
}
#endif
struct mread_pool * self = malloc(sizeof(*self)); struct mread_pool * self = malloc(sizeof(*self));
self->listen_fd = listen_fd; self->listen_fd = listen_fd;
#ifdef HAVE_EPOLL self->efd = efd;
self->epoll_fd = epoll_fd;
#elif HAVE_KQUEUE
self->kqueue_fd = kqueue_fd;
#endif
self->max_connection = max; self->max_connection = max;
self->closed = 0; self->closed = 0;
self->active = -1; self->active = -1;
@@ -373,11 +299,7 @@ mread_close(struct mread_pool *self) {
free_buffer(&s->client); free_buffer(&s->client);
free(s); free(s);
mread_close_listen(self); mread_close_listen(self);
#ifdef HAVE_EPOLL close(self->efd);
close(self->epoll_fd);
#elif HAVE_KQUEUE
close(self->kqueue_fd);
#endif
_release_rb(self->rb); _release_rb(self->rb);
free(self); free(self);
} }
@@ -396,14 +318,7 @@ mread_close_listen(struct mread_pool *self) {
static int static int
_read_queue(struct mread_pool * self, int timeout) { _read_queue(struct mread_pool * self, int timeout) {
self->queue_head = 0; self->queue_head = 0;
#ifdef HAVE_EPOLL int n = event_wait(self->efd, self->ev, timeout);
int n = epoll_wait(self->epoll_fd , self->ev, READQUEUE, timeout);
#elif HAVE_KQUEUE
struct timespec timeoutspec;
timeoutspec.tv_sec = timeout / 1000;
timeoutspec.tv_nsec = (timeout % 1000) * 1000000;
int n = kevent(self->kqueue_fd, NULL, 0, self->ev, READQUEUE, &timeoutspec);
#endif
if (n == -1) { if (n == -1) {
self->queue_len = 0; self->queue_len = 0;
return -1; return -1;
@@ -424,13 +339,7 @@ try_close(struct mread_pool * self, struct socket * s) {
s->status = SOCKET_CLOSED; s->status = SOCKET_CLOSED;
s->node = NULL; s->node = NULL;
s->temp = NULL; s->temp = NULL;
#ifdef HAVE_EPOLL event_del(self->efd, s->fd);
epoll_ctl(self->epoll_fd, EPOLL_CTL_DEL, s->fd , NULL);
#elif HAVE_KQUEUE
struct kevent ke;
EV_SET(&ke, s->fd, EVFILT_READ, EV_DELETE, 0, 0, NULL);
kevent(self->kqueue_fd, &ke, 1, NULL, 0, NULL);
#endif
close(s->fd); close(s->fd);
// printf("MREAD close %d (fd=%d)\n",id,s->fd); // printf("MREAD close %d (fd=%d)\n",id,s->fd);
++self->closed; ++self->closed;
@@ -443,20 +352,9 @@ _read_one(struct mread_pool * self) {
if (self->queue_head >= self->queue_len) { if (self->queue_head >= self->queue_len) {
return NULL; return NULL;
} }
struct socket * ret = NULL; struct socket * ret = self->ev[self->queue_head].s;
int writeflag = 0; bool writeflag = self->ev[self->queue_head].write;
int readflag = 0; bool readflag = self->ev[self->queue_head].read;
#ifdef HAVE_EPOLL
ret = self->ev[self->queue_head].data.ptr;
uint32_t flag = self->ev[self->queue_head].events;
writeflag = flag & EPOLLOUT;
readflag = flag & EPOLLIN;
#elif HAVE_KQUEUE
ret = self->ev[self->queue_head].udata;
short flag = self->ev[self->queue_head].filter;
writeflag = flag & EVFILT_WRITE;
readflag = flag & EVFILT_READ;
#endif
++ self->queue_head; ++ self->queue_head;
if (ret == LISTENSOCKET) { if (ret == LISTENSOCKET) {
return ret; return ret;
@@ -493,23 +391,10 @@ _add_client(struct mread_pool * self, int fd) {
close(fd); close(fd);
return NULL; return NULL;
} }
#ifdef HAVE_EPOLL if (event_add(self->efd, fd, s)) {
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.ptr = s;
if (epoll_ctl(self->epoll_fd, EPOLL_CTL_ADD, fd, &ev) == -1) {
close(fd); close(fd);
return NULL; return NULL;
} }
#elif HAVE_KQUEUE
struct kevent ke;
EV_SET(&ke, fd, EVFILT_READ, EV_ADD, 0, 0, s);
if (kevent(self->kqueue_fd, &ke, 1, NULL, 0, NULL) == -1) {
close(fd);
return NULL;
}
#endif
s->fd = fd; s->fd = fd;
s->node = NULL; s->node = NULL;
s->status = SOCKET_SUSPEND; s->status = SOCKET_SUSPEND;

250
lualib-src/lua-socket.c Normal file
View File

@@ -0,0 +1,250 @@
#include <stdlib.h>
#include <string.h>
#include <lua.h>
#include <lauxlib.h>
#include "luacompat52.h"
struct socket_buffer {
int size;
int head;
int tail;
};
static struct socket_buffer *
new_buffer(lua_State *L, int sz) {
struct socket_buffer * buffer = lua_newuserdata(L, sizeof(*buffer) + sz);
buffer->size = sz;
buffer->head = 0;
buffer->tail = 0;
return buffer;
}
static void
copy_buffer(struct socket_buffer * dest, struct socket_buffer * src) {
char * ptr = (char *)(src+1);
if (src->tail >= src->head) {
int sz = src->tail - src->head;
memcpy(dest+1, ptr+ src->head, sz);
dest->tail = sz;
} else {
char * d = (char *)(dest+1);
int part = src->size - src->head;
memcpy(d, ptr + src->head, part);
memcpy(d + part, ptr , src->tail);
dest->tail = src->tail + part;
}
}
static void
append_buffer(struct socket_buffer * buffer, char * msg, int sz) {
char * dst = (char *)(buffer + 1);
if (sz + buffer->tail < buffer->size) {
memcpy(dst + buffer->tail , msg, sz);
buffer->tail += sz;
} else {
int part = buffer->size - buffer->tail;
memcpy(dst + buffer->tail , msg, part);
memcpy(dst, msg + part, sz - part);
buffer->tail = sz - part;
}
}
static int
lpop(lua_State *L) {
struct socket_buffer * buffer = lua_touserdata(L, 1);
if (buffer == NULL) {
return 0;
}
int sz = luaL_checkinteger(L, 2);
int bytes = buffer->tail - buffer->head;
if (bytes < 0) {
bytes += buffer->size;
}
if (sz > bytes || bytes == 0) {
lua_pushnil(L);
lua_pushinteger(L, bytes);
return 2;
}
if (sz == 0) {
sz = bytes;
}
char * ptr = (char *)(buffer+1);
if (buffer->size - buffer->head >=sz) {
lua_pushlstring(L, ptr + buffer->head, sz);
buffer->head+=sz;
} else {
luaL_Buffer b;
luaL_buffinit(L, &b);
luaL_addlstring(&b, ptr + buffer->head, buffer->size - buffer->head);
buffer->head = sz - (buffer->size - buffer->head);
luaL_addlstring(&b, ptr, buffer->head);
luaL_pushresult(&b);
}
bytes -= sz;
lua_pushinteger(L,bytes);
if (bytes == 0) {
buffer->head = buffer->tail = 0;
}
return 2;
}
static inline int
check_sep(struct socket_buffer *buffer, int from, const char * sep, int sz) {
const char * ptr = (const char *)(buffer+1);
int i;
for (i=0;i<sz;i++) {
int index = from + i;
if (index >= buffer->size) {
index = 0;
}
if (ptr[index] != sep[i]) {
return 0;
}
}
return 1;
}
static int
lreadline(lua_State *L) {
struct socket_buffer * buffer = lua_touserdata(L, 1);
if (buffer == NULL) {
return 0;
}
size_t len = 0;
const char *sep = luaL_checklstring(L,2,&len);
int read = !lua_toboolean(L,3);
int bytes = buffer->tail - buffer->head;
if (bytes < 0) {
bytes += buffer->size;
}
int i;
for (i=0;i<=bytes-(int)len;i++) {
int index = buffer->head + i;
if (index >= buffer->size) {
index -= buffer->size;
}
if (check_sep(buffer, index, sep, (int)len)) {
if (read == 0) {
lua_pushboolean(L,1);
} else {
if (i==0) {
lua_pushlstring(L, "", 0);
} else {
const char * ptr = (const char *)(buffer+1);
if (--index < 0) {
index = buffer->size -1;
}
if (index < buffer->head) {
luaL_Buffer b;
luaL_buffinit(L, &b);
luaL_addlstring(&b, ptr + buffer->head, buffer->size-buffer->head);
luaL_addlstring(&b, ptr, index+1);
luaL_pushresult(&b);
} else {
lua_pushlstring(L, ptr + buffer->head, index-buffer->head+1);
}
++index;
}
index+=len;
if (index >= buffer->size) {
index-=buffer->size;
}
buffer->head = index;
}
return 1;
}
}
return 0;
}
static int
lpush(lua_State *L) {
struct socket_buffer * buffer = lua_touserdata(L, 1);
int bytes = 0;
if (buffer) {
bytes = buffer->tail - buffer->head;
if (bytes < 0) {
bytes += buffer->size;
}
}
void * msg = lua_touserdata(L,2);
if (msg == NULL) {
lua_settop(L,1);
lua_pushinteger(L,bytes);
return 2;
}
int sz = luaL_checkinteger(L,3);
if (buffer == NULL) {
struct socket_buffer * nbuf = new_buffer(L, sz * 2);
append_buffer(nbuf, msg, sz);
} else if (sz + bytes >= buffer->size) {
struct socket_buffer * nbuf = new_buffer(L, (sz + bytes) * 2);
copy_buffer(nbuf, buffer);
append_buffer(nbuf, msg, sz);
} else {
lua_settop(L,1);
append_buffer(buffer, msg, sz);
}
lua_pushinteger(L, sz + bytes);
return 2;
}
static int
lunpack(lua_State *L) {
int * msg = lua_touserdata(L,1);
int sz = luaL_checkinteger(L,2);
if (msg == NULL || sz < 4) {
return luaL_error(L, "Invalid socket message");
}
lua_pushinteger(L, *msg);
lua_pushlightuserdata(L, msg+1);
lua_pushinteger(L, sz - 4);
return 3;
}
static int
lpack(lua_State *L) {
int fd = luaL_checkinteger(L, 1);
const void *buffer;
size_t sz;
if (lua_isuserdata(L,2)) {
buffer = lua_touserdata(L,2);
sz = luaL_checkinteger(L, 3);
} else {
buffer = luaL_checklstring(L,2,&sz);
}
int * b = malloc(4 + sz);
*b = fd;
memcpy(b+1, buffer, sz);
lua_pushlightuserdata(L, b);
lua_pushinteger(L, 4+sz);
return 2;
}
int
luaopen_socketbuffer(lua_State *L) {
luaL_checkversion(L);
luaL_Reg l[] = {
{ "push", lpush },
{ "pop", lpop },
{ "readline", lreadline },
{ "unpack", lunpack },
{ "pack", lpack },
{ NULL, NULL },
};
luaL_newlib(L,l);
return 1;
}

175
lualib/redis.lua
View File

@@ -1,25 +1,114 @@
local skynet = require "skynet" local skynet = require "skynet"
local string = string local socket = require "socket"
local table = table local config = require "config"
local unpack = unpack local redis_conf = skynet.getenv "redis"
local assert = assert local name = config (redis_conf)
local redis_manager local readline = socket.readline
local batch = false local readbytes = socket.read
local table = table
local string = string
local redis = {} local redis = {}
local command = {} local command = {}
local meta = {
__index = command,
__gc = function(self)
socket.close(self.__handle)
end,
}
function redis.connect(dbname)
local fd = socket.open(name[dbname])
return setmetatable( { __handle = fd, __mode = false }, meta )
end
function command:disconnect()
socket.close(self.__handle)
setmetatable(self, nil)
end
local function compose_message(msg)
local lines = { "*" .. #msg }
for _,v in ipairs(msg) do
local t = type(v)
if t == "number" then
v = tostring(v)
elseif t == "userdata" then
v = int64.tostring(int64.new(v),10)
end
table.insert(lines,"$"..#v)
table.insert(lines,v)
end
table.insert(lines,"")
local cmd = table.concat(lines,"\r\n")
return cmd
end
local redcmd = {}
redcmd[42] = function(fd, data) -- '*'
local n = tonumber(data)
if n < 0 then
return true, nil
end
local bulk = {}
for i = 1,n do
local line = readline(fd,"\r\n")
local bytes = tonumber(string.sub(line,2))
if bytes < 0 then
table.insert(bulk, nil)
else
local data = readbytes(fd, bytes + 2)
table.insert(bulk, string.sub(data,1,-3))
end
end
return true, bulk
end
redcmd[36] = function(fd, data) -- '$'
local bytes = tonumber(data)
if bytes < 0 then
return true,nil
end
local firstline = readbytes(fd, bytes+2)
return true,string.sub(firstline,1,-3)
end
redcmd[43] = function(fd, data) -- '+'
return true,data
end
redcmd[45] = function(fd, data) -- '-'
return false,data
end
redcmd[58] = function(fd, data) -- ':'
-- todo: return string later
return true, tonumber(data)
end
local function read_response(fd)
local result = readline(fd, "\r\n")
local firstchar = string.byte(result)
local data = string.sub(result,2)
return redcmd[firstchar](fd,data)
end
setmetatable(command, { __index = function(t,k) setmetatable(command, { __index = function(t,k)
local cmd = string.upper(k) local cmd = string.upper(k)
local f = function (self, ...) local f = function (self, ...)
if batch then local fd = self.__handle
skynet.send( self.__handle, "lua", cmd , ...) if self.__mode then
socket.write(fd, compose_message { cmd, ... })
self.__batch = self.__batch + 1
else else
local err, result = skynet.call( self.__handle, "lua", cmd, ...) socket.lock(fd)
assert(err, result) socket.write(fd, compose_message { cmd, ... })
return result local ok, ret = read_response(fd)
socket.unlock(fd)
return ok, ret
end end
end end
t[k] = f t[k] = f
@@ -27,39 +116,53 @@ setmetatable(command, { __index = function(t,k)
end}) end})
function command:exists(key) function command:exists(key)
assert(not batch, "exists can't used in batch mode") assert(not self.__mode, "exists can't used in batch mode")
local result , exists = skynet.call( self.__handle, "lua" , "EXISTS", key) local fd = self.__handle
assert(result, exists) socket.lock(fd)
socket.write(fd, compose_message { "EXISTS", key })
local ok, exists = read_response(fd)
socket.unlock(fd)
assert(ok, exists)
exists = exists ~= 0 exists = exists ~= 0
return exists return exists
end end
function command:batch(mode) function command:batch(mode)
if mode == "end" then if mode == "end" then
assert(batch, "Open batch mode first") local fd = self.__handle
batch = false if self.__mode == "read" then
local err, result = skynet.unpack(skynet.rawcall( self.__handle, "text", mode)) local allok = true
assert(err, result) local allret = {}
return result for i = 1, self.__batch do
local ok, ret = read_response(fd)
allok = allok and ok
allret[i] = ret
end
assert(allok, "batch read failed")
socket.unlock(self.__handle)
self.__mode = false
return allret
else
local allok = true
local allret = true
for i = 1, self.__batch do
local ok, ret = read_response(fd)
allok = allok and ok
if ret~="OK" then
allret = false
end
end
socket.unlock(self.__handle)
self.__mode = false
assert(allret, "reading in batch write")
return allok
end
else else
assert(mode == "read" or mode == "write") assert(mode == "read" or mode == "write")
batch = mode self.__mode = mode
skynet.send( self.__handle, "text", mode) self.__batch = 0
socket.lock(self.__handle)
end end
end end
local meta = {
__index = command
}
function redis.connect(dbname)
local handle = skynet.call(redis_manager, "lua", dbname)
assert(handle ~= nil)
return setmetatable({ __handle = handle } , meta)
end
skynet.init(function()
redis_manager = skynet.uniqueservice("redis-mgr")
end, "redis")
return redis return redis

222
lualib/socket.lua
View File

@@ -1,79 +1,189 @@
local buffer = require "socketbuffer"
local skynet = require "skynet" local skynet = require "skynet"
local c = require "socket.c" local table = table
local next = next
local assert = assert
local coroutine = coroutine
local type = type
local READBUF = {} -- fd:buffer
local READREQUEST = {} -- fd:request_size
local READSESSION = {} -- fd:session
local READLOCK = {} -- fd:queue(session)
local READTHREAD= {} -- fd:thread
local selfaddr = skynet.self()
local function response(session)
skynet.redirect(selfaddr , 0, "response", session, "")
end
skynet.register_protocol {
name = "client",
id = 3, -- PTYPE_CLIENT
pack = buffer.pack,
unpack = buffer.unpack,
dispatch = function (_, _, fd, msg, sz)
local qsz = READREQUEST[fd]
local buf = READBUF[fd]
local bsz
if sz == 0 or buf == true then
buf,bsz = true, qsz
else
buf,bsz = buffer.push(buf, msg, sz)
end
READBUF[fd] = buf
if qsz == nil then
return
end
if type(qsz) == "number" then
if qsz > bsz then
return
end
else
-- request readline
if buffer.readline(buf, qsz, true) == nil then
return
end
end
response(READSESSION[fd])
end
}
skynet.register_protocol {
name = "system",
id = 4, -- PTYPE_SYSTEM
pack = skynet.pack,
unpack = function (...) return ... end,
dispatch = function (session, addr, msg, sz)
fd, t, sz = skynet.unpack(msg,sz)
assert(addr == selfaddr, "PTYPE_SYSTEM message must send by self")
if t == 0 then
-- lock fd
if READTHREAD[fd] == nil then
skynet.ret()
return
end
local q = READLOCK[fd]
if q == nil then
READLOCK[fd] = { session }
else
table.insert(q, session)
end
else
-- request bytes or readline
local buf = READBUF[fd]
if buf == true then
skynet.ret()
return
end
local _,bsz = buffer.push(buf)
if t == 1 then
-- sz is size
if bsz >= sz then
skynet.ret()
return
end
else
-- sz is sep
if buffer.readline(buf, sz, true) then -- don't real read
skynet.ret()
return
end
end
READSESSION[fd] = session
end
end
}
local socket = {} local socket = {}
local fd
local object
local data = {}
local function presend() function socket.open(addr, port)
if next(data) then local cmd = "open" .. " " .. (port and (addr..":"..port) or addr)
local tmp = data local r = skynet.call(".socket", "text", cmd)
data = {} if r == "" then
for _,v in ipairs(tmp) do error(cmd .. " failed")
socket.write(fd, v)
end end
end return tonumber(r)
end
function socket.connect(addr)
local ip, port = string.match(addr,"([^:]+):(.+)")
port = tonumber(port)
socket.close()
fd = c.open(ip,port)
if fd == nil then
return true
end
skynet.send(".connection", "text", "ADD", fd , skynet.address(skynet.self()))
object = c.new()
presend()
end end
function socket.stdin() function socket.stdin()
skynet.send(".connection", "text", "ADD", 1 , skynet.address(skynet.self())) local r = skynet.call(".socket", "text", "bind 1")
object = c.new() if r == "" then
error("stdin bind failed")
end
return tonumber(r)
end end
function socket.push(msg,sz) function socket.close(fd)
if msg then socket.lock(fd)
c.push(object, msg, sz) skynet.call(".socket", "text", "close", fd)
end READBUF[fd] = true
READLOCK[fd] = nil
end end
function socket.read(bytes) function socket.read(fd, sz)
return c.read(object, bytes) assert(coroutine.running() == READTHREAD[fd], "call socket.lock first")
end
function socket.readline(sep) local str = buffer.pop(READBUF[fd],sz)
return c.readline(object, sep)
end
function socket.readblock(...)
return c.readblock(object,...)
end
function socket.write(...)
local str = c.write(fd, ...)
if str then if str then
socket.close() return str
table.insert(data, str)
end
end end
function socket.writeblock(...) READREQUEST[fd] = sz
local str = c.writeblock(fd, ...) skynet.call(selfaddr, "system",fd,1,sz) -- singal size 1
READREQUEST[fd] = nil
str = buffer.pop(READBUF[fd],sz)
return str
end
function socket.readline(fd, sep)
assert(coroutine.running() == READTHREAD[fd], "call socket.lock first")
local str = buffer.readline(READBUF[fd],sep)
if str then if str then
socket.close() return str
table.insert(data, str)
end
end end
function socket.close() READREQUEST[fd] = sep
if fd then skynet.call(selfaddr, "system",fd,2,sep) -- singal sep 2
skynet.send(".connection","text", "DEL", fd) READREQUEST[fd] = nil
fd = nil
str = buffer.readline(READBUF[fd],sep)
return str
end
function socket.write(fd, msg, sz)
assert(coroutine.running() == READTHREAD[fd], "call socket.lock first")
skynet.send(".socket", "client", fd, msg, sz)
end
function socket.lock(fd)
local lt = READTHREAD[fd]
local ct = coroutine.running()
if lt then
assert(lt ~= ct, "already lock")
skynet.call(selfaddr, "system",fd,0)
end
READTHREAD[fd] = ct
end
function socket.unlock(fd)
READTHREAD[fd] = nil
local q = READLOCK[fd]
if q then
if q[1] then
response(q[1])
end
table.remove(q,1)
if q[1] == nil then
READLOCK[fd] = nil
end
end end
end end
return socket return socket

136
service-src/event.h Normal file
View File

@@ -0,0 +1,136 @@
#ifndef skynet_event_h
#define skynet_event_h
#include <stdbool.h>
#include <stdlib.h>
/* Test for polling API */
#ifdef __linux__
#define HAVE_EPOLL 1
#endif
#if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined (__NetBSD__)
#define HAVE_KQUEUE 1
#endif
#if !defined(HAVE_EPOLL) && !defined(HAVE_KQUEUE)
#error "system does not support epoll or kqueue API"
#endif
/* ! Test for polling API */
#include <sys/types.h>
#ifdef HAVE_EPOLL
#include <sys/epoll.h>
#elif HAVE_KQUEUE
#include <sys/event.h>
#endif
#ifdef HAVE_EPOLL
#define MAX_EVENT 32
struct event {
void * s;
bool read;
bool write;
};
static int
event_init(int max) {
return epoll_create(max);
}
static int
event_add(int efd, int sock, void *ud) {
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.ptr = ud;
if (epoll_ctl(efd, EPOLL_CTL_ADD, sock, &ev) == -1) {
return 1;
}
return 0;
}
static void
event_del(int efd, int fd) {
epoll_ctl(efd, EPOLL_CTL_DEL, fd , NULL);
}
static int
event_wait(int efd, struct event * e, int timeout) {
struct epoll_event ev[MAX_EVENT];
int n = epoll_wait(efd , ev, MAX_EVENT, timeout);
int i;
for (i=0;i<n;i++) {
e[i].s = ev[i].data.ptr;
unsigned flag = ev[i].events;
e[i].write = (flag & EPOLLOUT) != 0;
e[i].read = (flag & EPOLLIN) != 0;
}
return n;
}
static void
event_write(int efd, int fd, void *ud, bool enable) {
struct epoll_event ev;
ev.events = EPOLLIN | (enable ? EPOLLOUT : 0);
ev.data.ptr = ud;
epoll_ctl(efd, EPOLL_CTL_MOD, fd, &ev);
}
#elif HAVE_KQUEUE
static int
event_init(int max) {
return kqueue();
}
static int
event_add(int efd, int sock, void *ud) {
struct kevent ke;
EV_SET(&ke, sock, EVFILT_READ, EV_ADD, 0, 0, ud);
if (kevent(efd, &ke, 1, NULL, 0, NULL) == -1) {
return 1;
}
return 0;
}
static void
event_del(int efd, int fd) {
struct kevent ke;
EV_SET(&ke, fd, EVFILT_READ | EVFILT_WRITE, EV_DELETE, 0, 0, NULL);
kevent(efd, &ke, 1, NULL, 0, NULL);
}
static int
event_wait(int kfd, struct event *e, int timeout) {
struct timespec timeoutspec;
timeoutspec.tv_sec = timeout / 1000;
timeoutspec.tv_nsec = (timeout % 1000) * 1000000;
struct kevent ev[MAX_EVENT];
int n = kevent(kfd, NULL, 0, ev, MAX_EVENT, &timeoutspec);
int i;
for (i=0;i<n;i++) {
e[i].s = ev[i].udata;
unsigned flag = ev[i].filter;
e[i].write = (flag & EVFILT_WRITE) != 0;
e[i].read = (flag & EVFILT_READ) != 0;
}
return n;
}
static void
event_write(int efd, int fd, void *ud, bool enable) {
struct kevent ke;
EV_SET(&ke, fd, EVFILT_WRITE, enable ? EV_ENABLE : EV_DISABLE, 0, 0, ud);
kevent(efd, &ke, 1, NULL, 0, NULL);
}
#endif
#endif

439
service-src/service_socket.c Normal file
View File

@@ -0,0 +1,439 @@
#include "skynet.h"
#include "event.h"
#include <string.h>
#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <stdbool.h>
#include <assert.h>
#include <unistd.h>
#include <netdb.h>
#include <errno.h>
#include <fcntl.h>
#define MAX_ID 0x7fffffff
#define MAX_CONNECTION 256
#define READ_BUFFER 4000
#define STATUS_INVALID 0
#define STATUS_HALFCLOSE 1
#define STATUS_SUSPEND 2
struct write_buffer {
struct write_buffer * next;
char *ptr;
size_t sz;
void *buffer;
};
struct socket {
int fd;
int id;
uint32_t source;
int status;
int session;
struct write_buffer * head;
struct write_buffer * tail;
};
struct socket_pool {
int fd;
struct event ev[MAX_EVENT];
int poll;
int id;
int count;
int cap;
struct socket * s;
};
static void
reply(struct skynet_context * ctx, uint32_t source, int session, char * cmd, int sz) {
if (sz < 0) {
sz = strlen(cmd);
}
skynet_send(ctx, 0, source, PTYPE_RESPONSE, session, cmd, sz);
}
static int
_set_nonblocking(int fd)
{
int flag = fcntl(fd, F_GETFL, 0);
if ( -1 == flag ) {
return -1;
}
return fcntl(fd, F_SETFL, flag | O_NONBLOCK);
}
static int
new_socket(struct socket_pool *p, int sock, uint32_t addr) {
int i;
if (p->count >= p->cap) {
goto _error;
}
for (i=0;i<p->cap;i++) {
int id = p->id + i;
int n = id % p->cap;
struct socket * s = &p->s[n];
if (s->status == STATUS_INVALID) {
if (event_add(p->fd, sock, s)) {
goto _error;
}
s->status = STATUS_SUSPEND;
_set_nonblocking(sock);
int keepalive = 1;
setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepalive , sizeof(keepalive));
s->fd = sock;
s->id = id;
s->source = addr;
p->count++;
if (++p->id > MAX_ID) {
p->id = 1;
}
assert(s->head == NULL && s->tail == NULL);
return id;
}
}
_error:
close(sock);
return -1;
}
static void
cmd_bind(struct skynet_context * ctx, struct socket_pool *p, int sock, int session, uint32_t source) {
char ret[10];
int id = new_socket(p, sock, source);
if (id<0) {
reply(ctx, source, session, NULL , 0);
return;
}
if (p->count == 1) {
skynet_command(ctx, "TIMEOUT", "0");
}
int sz = sprintf(ret,"%d",id);
reply(ctx, source, session, ret, sz);
}
static void
cmd_open(struct skynet_context * ctx, struct socket_pool *p, char * cmd, int session, uint32_t source) {
int status;
struct addrinfo ai_hints;
struct addrinfo *ai_list = NULL;
struct addrinfo *ai_ptr = NULL;
char * host = strsep(&cmd, ":");
if (cmd == NULL) {
goto _failed;
}
memset( &ai_hints, 0, sizeof( ai_hints ) );
ai_hints.ai_family = AF_UNSPEC;
ai_hints.ai_socktype = SOCK_STREAM;
ai_hints.ai_protocol = IPPROTO_TCP;
status = getaddrinfo( host, cmd, &ai_hints, &ai_list );
if ( status != 0 ) {
goto _failed;
}
int sock= -1;
for ( ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next ) {
sock = socket( ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol );
if ( sock < 0 ) {
continue;
}
status = connect( sock, ai_ptr->ai_addr, ai_ptr->ai_addrlen );
if ( status != 0 ) {
close(sock);
sock = -1;
continue;
}
break;
}
freeaddrinfo( ai_list );
if (sock < 0) {
goto _failed;
}
cmd_bind(ctx, p, sock, session, source);
return;
_failed:
reply(ctx, source, session, NULL , 0);
}
static void
force_close(struct socket *s, struct socket_pool *p) {
struct write_buffer *wb = s->head;
while (wb) {
struct write_buffer *tmp = wb;
wb = wb->next;
free(tmp->buffer);
free(tmp);
}
s->head = s->tail = NULL;
s->status = STATUS_INVALID;
event_del(p->fd, s->fd);
close(s->fd);
--p->count;
}
static void
cmd_close(struct skynet_context * ctx, struct socket_pool *p, int id, int session, uint32_t source) {
struct socket * s = &p->s[id % p->cap];
if (id != s->id) {
skynet_error(ctx, "%x close invalid socket %d", source, id);
reply(ctx, source, session, "invalid", -1);
return;
}
if (source != s->source) {
skynet_error(ctx, "%x try to close socket %d:%x", source, id, s->source);
reply(ctx, source, session, "permission", -1);
return;
}
if (s->head == NULL) {
force_close(s,p);
reply(ctx, source, session, NULL, 0);
} else {
s->status = STATUS_HALFCLOSE;
s->session = session;
}
}
static void
_ctrl(struct skynet_context * ctx, struct socket_pool *p, char * command, int id, char * arg, int session, uint32_t source) {
if (strcmp(command, "open")==0) {
cmd_open(ctx, p, arg, session, source);
} else if (strcmp(command, "close")==0) {
cmd_close(ctx, p, id, session, source);
} else if (strcmp(command, "bind")==0) {
cmd_bind(ctx, p, id, session, source);
} else {
skynet_error(ctx, "Unknown command %s", command);
reply(ctx, source, session, NULL, 0);
}
}
static char *
parser(const char * msg, int sz, char * buffer, int *id) {
int i;
for (i=0;i<sz && msg[i] && msg[i]!=' ';i++) {
if (msg[i] != ' ') {
buffer[i] = msg[i];
}
}
buffer[i] = '\0';
if (i+1 < sz) {
*id = strtol(msg+i+1, NULL, 10);
memcpy(buffer+i+1,msg+i+1, sz-i-1);
}
buffer[sz] = '\0';
return buffer+i+1;
}
static void
forward(struct skynet_context * context, struct socket *s, struct socket_pool *p) {
int * buffer = malloc(READ_BUFFER + sizeof(int));
*buffer = s->id; // convert endian ?
int r = 0;
for (;;) {
r = read(s->fd, buffer+1, READ_BUFFER);
if (r == -1) {
switch(errno) {
case EWOULDBLOCK:
free(buffer);
return;
case EINTR:
continue;
}
r = 0;
break;
}
break;
}
if (r == 0) {
force_close(s,p);
}
if (s->status == STATUS_HALFCLOSE) {
free(buffer);
} else {
skynet_send(context, 0, s->source, PTYPE_CLIENT, 0, buffer, r + 4);
}
}
static void
sendout(struct socket_pool *p, struct socket *s) {
while (s->head) {
struct write_buffer * tmp = s->head;
for (;;) {
int sz = write(s->fd, tmp->ptr, tmp->sz);
if (sz < 0) {
switch(errno) {
case EINTR:
continue;
case EAGAIN:
return;
}
force_close(s,p);
return;
}
if (sz != tmp->sz) {
tmp->ptr += sz;
tmp->sz -= sz;
return;
}
break;
}
s->head = tmp->next;
free(tmp->buffer);
free(tmp);
}
s->tail = NULL;
event_write(p->fd, s->fd, s, false);
}
static int
try_send(struct skynet_context *ctx, struct socket_pool *p, uint32_t source, const int * msg, size_t sz) {
if (sz < 4) {
skynet_error(ctx, "%x invalid message", source);
return 0;
}
sz-=4;
int id = *msg;
struct socket * s = &p->s[id % p->cap];
if (id != s->id) {
skynet_error(ctx, "%x write to invalid socket %d", source, id);
return 0;
}
if (source != s->source) {
skynet_error(ctx, "%x try to write socket %d:%x", source, id, s->source);
return 0;
}
if (s->status != STATUS_SUSPEND) {
skynet_error(ctx, "%x write to closed socket %d", source, id);
return 0;
}
if (s->head) {
struct write_buffer * buf = malloc(sizeof(*buf));
buf->next = s->tail;
buf->ptr = (char *)(msg+1);
buf->buffer = (void *)msg;
buf->sz = sz;
s->tail = buf;
return 1;
}
char * ptr = (char *)(msg+1);
for (;;) {
int wt = write(s->fd, ptr, sz);
if (wt < 0) {
switch(errno) {
case EINTR:
continue;
}
break;
}
if (wt == sz) {
return 0;
}
sz-=wt;
ptr+=wt;
break;
}
struct write_buffer * buf = malloc(sizeof(*buf));
buf->next = NULL;
buf->ptr = ptr;
buf->sz = sz;
buf->buffer = (void *)msg;
s->head = s->tail = buf;
event_write(p->fd, s->fd, s, true);
return 1;
}
static int
_cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
struct socket_pool *p = ud;
if (type == PTYPE_TEXT) {
char tmp[sz+1];
int id=0;
char * arg = parser(msg, (int)sz, tmp, &id);
_ctrl(context, p , tmp, id, arg, session, source);
return 0;
} else if (type == PTYPE_CLIENT) {
return try_send(context, p, source, msg, sz);
}
if (p->count == 0)
return 0;
assert(type == PTYPE_RESPONSE);
int n = event_wait(p->fd, p->ev, 100); // timeout : 100ms
int i;
for (i=0;i<n;i++) {
struct event *e = &p->ev[i];
if (e->read) {
forward(context, e->s, p);
}
if (e->write) {
struct socket *s = e->s;
sendout(p, s);
if (s->status == STATUS_HALFCLOSE && s->head == NULL) {
force_close(s, p);
reply(context, source, s->session, NULL, 0);
}
}
}
skynet_command(context, "TIMEOUT", n == MAX_EVENT ? "0" : "1");
return 0;
}
int
socket_init(struct socket_pool *pool, struct skynet_context *ctx, const char * args) {
int max = 0;
sscanf(args, "%d",&max);
if (max == 0) {
max = MAX_CONNECTION;
}
pool->cap = max;
int fd = event_init(max);
if (fd < 0) {
return 1;
}
pool->s = malloc(sizeof(struct socket) * max);
memset(pool->s, 0, sizeof(struct socket) * max);
pool->fd = fd;
skynet_callback(ctx, pool, _cb);
skynet_command(ctx,"REG",".socket");
return 0;
}
struct socket_pool *
socket_create(void) {
struct socket_pool *pool = malloc(sizeof(*pool));
memset(pool,0,sizeof(*pool));
pool->id = 1;
pool->fd = -1;
return pool;
}
void
socket_release(struct socket_pool *pool) {
if (pool->fd >= 0) {
close(pool->fd);
}
int i;
for (i=0;i<pool->cap;i++) {
if (pool->s[i].status != STATUS_INVALID && pool->s[i].fd >=0) {
close(pool->s[i].fd);
}
}
free(pool->s);
free(pool);
}

47
service/console.lua
View File

@@ -1,46 +1,15 @@
local skynet = require "skynet" local skynet = require "skynet"
local socket = require "socket" local socket = require "socket"
local function readline(sep)
while true do
local line = socket.readline(sep)
if line then
return line
end
coroutine.yield()
end
end
local function split_package()
while true do
local cmd = readline "\n"
if cmd ~= "" then
skynet.send(skynet.self(), "text", cmd)
end
end
end
local split_co = coroutine.create(split_package)
skynet.register_protocol {
name = "client",
id = 3,
pack = function(...) return ... end,
unpack = function(msg,sz)
assert(msg , "Stdin closed")
socket.push(msg,sz)
assert(coroutine.resume(split_co))
end,
dispatch = function () end
}
skynet.start(function() skynet.start(function()
skynet.dispatch("text", function (session, address, cmd) local stdin = socket.stdin()
local handle = skynet.newservice(cmd) socket.lock(stdin)
while true do
local cmdline = socket.readline(stdin, "\n")
local handle = skynet.newservice(cmdline)
if handle == nil then if handle == nil then
print("Launch error:",cmd) print("Launch error:",cmdline)
end end
end) end
socket.stdin() socket.unlock(stdin)
end) end)

2
service/main.lua
View File

@@ -2,8 +2,8 @@ local skynet = require "skynet"
skynet.start(function() skynet.start(function()
print("Server start") print("Server start")
skynet.launch("socket",128)
local service = skynet.launch("snlua","service_mgr") local service = skynet.launch("snlua","service_mgr")
local connection = skynet.launch("connection","256")
local lualog = skynet.launch("snlua","lualog") local lualog = skynet.launch("snlua","lualog")
local console = skynet.launch("snlua","console") local console = skynet.launch("snlua","console")
local remoteroot = skynet.launch("snlua","remote_root") local remoteroot = skynet.launch("snlua","remote_root")

242
service/redis-cli.lua
View File

@@ -1,242 +0,0 @@
local skynet = require "skynet"
local socket = require "socket"
local int64 = require "int64"
local string = string
local table = table
local tonumber = tonumber
local ipairs = ipairs
local unpack = unpack
local batch_mode = {}
local batch_count = {}
local batch_reply = {}
local redis_server, redis_db = ...
local function compose_message(msg)
local lines = { "*" .. #msg }
for _,v in ipairs(msg) do
local t = type(v)
if t == "number" then
v = tostring(v)
elseif t == "userdata" then
v = int64.tostring(int64.new(v),10)
end
table.insert(lines,"$"..#v)
table.insert(lines,v)
end
table.insert(lines,"")
local cmd = table.concat(lines,"\r\n")
return cmd
end
local function select_db(id)
local result , ok = skynet.call(skynet.self(), "lua", "SELECT", tostring(id))
assert(result and ok == "OK")
end
local request_queue = { head = 1, tail = 1 }
local function push_request_queue(reply)
request_queue[request_queue.tail] = reply
request_queue.tail = request_queue.tail + 1
end
local function pop_request_queue()
assert(request_queue.head < request_queue.tail)
local reply = request_queue[request_queue.head]
request_queue[request_queue.head] = nil
request_queue.head = request_queue.head + 1
return reply
end
local function batch_close(address,session)
local reply = batch_reply[address]
skynet.redirect(address,0, "response", session, skynet.pack(not (type(reply) == "string"), reply))
batch_mode[address] = nil
batch_count[address] = nil
batch_reply[address] = nil
end
local function batch_dec(address)
local bc = batch_count[address]
batch_count[address] = bc - 1
if bc == 1 then
local session = batch_mode[address]
if type(session) == "number" then
batch_close(address, session)
end
end
end
local function response(suc, value)
local reply = pop_request_queue()
local mode = reply.batch
local address = reply.address
if mode == "read" then
if suc then
local br = batch_reply[address]
if type(br) == "table" then
br.n = br.n + 1
br[br.n] = value
end
else
batch_reply[address] = value
end
batch_dec(address)
elseif mode == "write" then
if not suc then
batch_reply[address] = value
end
batch_dec(address)
else
skynet.redirect(address,0, "response", reply.session, skynet.pack(suc, value))
end
end
local function readline(sep)
while true do
local line = socket.readline(sep)
if line then
return line
end
coroutine.yield()
end
end
local function readbytes(bytes)
while true do
local block = socket.read(bytes)
if block then
return block
end
coroutine.yield()
end
end
local redcmd = {}
redcmd[42] = function(data) -- '*'
local n = tonumber(data)
if n < 0 then
response(true, nil)
return
end
local bulk = {}
for i = 1,n do
local line = readline "\r\n"
local bytes = tonumber(string.sub(line,2))
if bytes < 0 then
table.insert(bulk, nil)
else
local data = readbytes(bytes + 2)
table.insert(bulk, string.sub(data,1,-3))
end
end
response(true, bulk)
end
redcmd[36] = function(data) -- '$'
local bytes = tonumber(data)
if bytes < 0 then
response(true,nil)
return
end
local firstline = readbytes(bytes+2)
response(true,string.sub(firstline,1,-3))
end
redcmd[43] = function(data) -- '+'
response(true,data)
end
redcmd[45] = function(data) -- '-'
response(false,data)
end
redcmd[58] = function(data) -- ':'
-- todo: return string later
response(true, tonumber(data))
end
local function split_package()
while true do
local result = readline "\r\n"
local firstchar = string.byte(result)
local data = string.sub(result,2)
local f = redcmd[firstchar]
assert(f)
f(data)
end
end
local function init()
while socket.connect(redis_server) do
skynet.sleep(1000)
end
if redis_db then
select_db(redis_db)
end
end
local split_co = coroutine.create(split_package)
local function reconnect()
init()
for i = request_queue.head, request_queue.tail-1 do
local request = request_queue[i]
socket.write(request.cmd)
end
split_co = coroutine.create(split_package)
end
skynet.register_protocol {
name = "client",
id = 3,
pack = function(...) return ... end,
unpack = function(msg,sz)
if sz == 0 then
skynet.timeout(0, reconnect)
return
end
socket.push(msg,sz)
assert(coroutine.resume(split_co))
end,
dispatch = function () end
}
skynet.start(function()
skynet.dispatch("text", function(session, address, mode)
local last = batch_mode[address]
if mode == "end" then
assert(last == "read" or last == "write" , "Invalid end")
if batch_count[address] == 0 then
batch_close(address, session)
else
batch_mode[address] = session
end
else
assert(last == nil, "Already in batch mode")
if mode == "read" then
batch_reply[address] = { n = 0 }
batch_count[address] = 0
elseif mode == "write" then
batch_count[address] = 0
else
error ("Invalid last batch operation : " .. last)
end
batch_mode[address] = mode
end
end)
skynet.dispatch("lua", function(session, address, ...)
local message = { ... }
local cmd = compose_message(message)
socket.write(cmd)
local mode = batch_mode[address]
if mode == "read" or mode == "write" then
batch_count[address] = batch_count[address] + 1
end
push_request_queue { session = session , address = address, cmd = cmd , batch = mode }
end)
init()
end)

26
service/redis-mgr.lua
View File

@@ -1,26 +0,0 @@
local skynet = require "skynet"
local log = require "log"
local config = require "config"
local redis_conf = skynet.getenv "redis"
local name = config (redis_conf)
local connection = {}
skynet.start(function()
skynet.dispatch("lua", function (session, from, dbname)
if connection[dbname] then
skynet.ret(skynet.pack(connection[dbname]))
return
end
if name[dbname] == nil then
log.Error("Invalid db name : "..dbname)
skynet.ret(skynet.pack(nil))
return
end
local redis_cli = skynet.newservice("redis-cli", name[dbname])
connection[dbname] = redis_cli
skynet.ret(skynet.pack(redis_cli))
end)
end)

4
service/testredis.lua
View File

@@ -3,10 +3,11 @@ local redis = require "redis"
skynet.start(function() skynet.start(function()
local db = redis.connect "main" local db = redis.connect "main"
print(db:select(0))
db:batch "write" -- ignore results db:batch "write" -- ignore results
db:set("A", "hello") db:set("A", "hello")
db:set("B", "world") db:set("B", "world")
db:batch "end" print(db:batch "end")
db:batch "read" db:batch "read"
db:get("A") db:get("A")
@@ -19,6 +20,7 @@ skynet.start(function()
print(db:get "A") print(db:get "A")
print(db:set("A","hello world")) print(db:set("A","hello world"))
print(db:get("A")) print(db:get("A"))
db:disconnect()
skynet.exit() skynet.exit()
end) end)