From 277b6957cbc3fc816cbca14644f9eaa730d09a20 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Fri, 21 Jun 2013 13:02:51 +0800 Subject: [PATCH] new socket service and socket lib instead of old connection, and rewrite redis lib --- Makefile | 14 +- connection/connection.c | 145 ------------ connection/connection.h | 16 -- connection/lua-socket.c | 327 -------------------------- connection/main.c | 204 ---------------- gate/mread.c | 149 ++---------- lualib-src/lua-socket.c | 250 ++++++++++++++++++++ lualib/redis.lua | 179 +++++++++++--- lualib/socket.lua | 250 ++++++++++++++------ service-src/event.h | 136 +++++++++++ service-src/service_socket.c | 439 +++++++++++++++++++++++++++++++++++ service/console.lua | 47 +--- service/main.lua | 2 +- service/redis-cli.lua | 242 ------------------- service/redis-mgr.lua | 26 --- service/testredis.lua | 4 +- 16 files changed, 1182 insertions(+), 1248 deletions(-) delete mode 100644 connection/connection.c delete mode 100644 connection/connection.h delete mode 100644 connection/lua-socket.c delete mode 100644 connection/main.c create mode 100644 lualib-src/lua-socket.c create mode 100644 service-src/event.h create mode 100644 service-src/service_socket.c delete mode 100644 service/redis-cli.lua delete mode 100644 service/redis-mgr.lua diff --git a/Makefile b/Makefile index 7538f6c8..f50cb8dc 100644 --- a/Makefile +++ b/Makefile @@ -21,14 +21,14 @@ all : \ service/logger.so \ service/gate.so \ service/client.so \ - service/connection.so \ service/master.so \ service/multicast.so \ service/tunnel.so \ service/harbor.so \ service/localcast.so \ + service/socket.so \ luaclib/skynet.so \ - luaclib/socket.so \ + luaclib/socketbuffer.so \ luaclib/int64.so \ luaclib/mcast.so \ client @@ -72,7 +72,7 @@ service/snlua.so : service-src/service_lua.c gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src 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 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 gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src -service/connection.so : connection/connection.c connection/main.c - gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src -Iconnection +service/socket.so : service-src/service_socket.c + gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src -luaclib/socket.so : connection/lua-socket.c | luaclib - gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src -Iconnection +luaclib/socketbuffer.so : lualib-src/lua-socket.c | luaclib + gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src luaclib/int64.so : lua-int64/int64.c | luaclib gcc $(CFLAGS) $(SHARED) -Iluacompat -O2 $^ -o $@ diff --git a/connection/connection.c b/connection/connection.c deleted file mode 100644 index 318ca373..00000000 --- a/connection/connection.c +++ /dev/null @@ -1,145 +0,0 @@ -#include "connection.h" - -#include -#include -#include -#include - -/* 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 -#elif HAVE_KQUEUE -#include -#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 -} - diff --git a/connection/connection.h b/connection/connection.h deleted file mode 100644 index d9236174..00000000 --- a/connection/connection.h +++ /dev/null @@ -1,16 +0,0 @@ -#ifndef SKYNET_CONNECTION_H -#define SKYNET_CONNECTION_H - -#include - -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 diff --git a/connection/lua-socket.c b/connection/lua-socket.c deleted file mode 100644 index 47b052ae..00000000 --- a/connection/lua-socket.c +++ /dev/null @@ -1,327 +0,0 @@ -#include "skynet.h" - -#include "luacompat52.h" - -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include - -#include -#include - -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; -} - diff --git a/connection/main.c b/connection/main.c deleted file mode 100644 index 717687ac..00000000 --- a/connection/main.c +++ /dev/null @@ -1,204 +0,0 @@ -#include "connection.h" -#include "skynet.h" - - -#include -#include - -#include -#include -#include -#include -#include -#include - -#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;imax_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;imax_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;imax_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; -} - - - diff --git a/gate/mread.c b/gate/mread.c index 993c8512..79b9044e 100644 --- a/gate/mread.c +++ b/gate/mread.c @@ -1,28 +1,6 @@ #include "mread.h" #include "ringbuffer.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 - -#ifdef HAVE_EPOLL -#include -#elif HAVE_KQUEUE -#include -#endif +#include "event.h" #include #include @@ -37,7 +15,6 @@ #include #define BACKLOG 32 -#define READQUEUE 32 #define READBLOCKSIZE 2048 #define RINGBUFFER_DEFAULT 1024 * 1024 @@ -75,11 +52,7 @@ struct socket { struct mread_pool { int listen_fd; -#ifdef HAVE_EPOLL - int epoll_fd; -#elif HAVE_KQUEUE - int kqueue_fd; -#endif + int efd; int max_connection; int closed; int active; @@ -88,11 +61,7 @@ struct mread_pool { struct socket * free_socket; int queue_len; int queue_head; -#ifdef HAVE_EPOLL - struct epoll_event ev[READQUEUE]; -#elif HAVE_KQUEUE - struct kevent ev[READQUEUE]; -#endif + struct event ev[MAX_EVENT]; struct ringbuffer * rb; }; @@ -103,16 +72,7 @@ turn_on(struct mread_pool *self, struct socket * s) { if (s->status < SOCKET_ALIVE || s->enablewrite) { return; } -#ifdef HAVE_EPOLL - 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 + event_write(self->efd, s->fd, s, true); s->enablewrite = 1; } @@ -122,19 +82,9 @@ turn_off(struct mread_pool *self, struct socket * s) { return; } s->enablewrite = 0; -#ifdef HAVE_EPOLL - 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 + event_write(self->efd, s->fd, s, false); } - static void free_buffer(struct send_client * sc) { struct send_buffer * sb = sc->head; @@ -301,46 +251,22 @@ mread_create(uint32_t addr, int port , int max , int buffer_size) { return NULL; } -#ifdef HAVE_EPOLL - int epoll_fd = epoll_create(max + 1); - if (epoll_fd == -1) { + int efd = event_init(max+1); + if (efd == -1) { close(listen_fd); return NULL; } - struct epoll_event ev; - ev.events = EPOLLIN; - ev.data.ptr = LISTENSOCKET; - - if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev) == -1) { + if (event_add(efd, listen_fd, LISTENSOCKET) == -1) { close(listen_fd); - close(epoll_fd); + close(efd); 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)); self->listen_fd = listen_fd; -#ifdef HAVE_EPOLL - self->epoll_fd = epoll_fd; -#elif HAVE_KQUEUE - self->kqueue_fd = kqueue_fd; -#endif + self->efd = efd; self->max_connection = max; self->closed = 0; self->active = -1; @@ -373,11 +299,7 @@ mread_close(struct mread_pool *self) { free_buffer(&s->client); free(s); mread_close_listen(self); -#ifdef HAVE_EPOLL - close(self->epoll_fd); -#elif HAVE_KQUEUE - close(self->kqueue_fd); -#endif + close(self->efd); _release_rb(self->rb); free(self); } @@ -396,14 +318,7 @@ mread_close_listen(struct mread_pool *self) { static int _read_queue(struct mread_pool * self, int timeout) { self->queue_head = 0; -#ifdef HAVE_EPOLL - 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 + int n = event_wait(self->efd, self->ev, timeout); if (n == -1) { self->queue_len = 0; return -1; @@ -424,13 +339,7 @@ try_close(struct mread_pool * self, struct socket * s) { s->status = SOCKET_CLOSED; s->node = NULL; s->temp = NULL; -#ifdef HAVE_EPOLL - 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 + event_del(self->efd, s->fd); close(s->fd); // printf("MREAD close %d (fd=%d)\n",id,s->fd); ++self->closed; @@ -443,20 +352,9 @@ _read_one(struct mread_pool * self) { if (self->queue_head >= self->queue_len) { return NULL; } - struct socket * ret = NULL; - int writeflag = 0; - int readflag = 0; -#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 + struct socket * ret = self->ev[self->queue_head].s; + bool writeflag = self->ev[self->queue_head].write; + bool readflag = self->ev[self->queue_head].read; ++ self->queue_head; if (ret == LISTENSOCKET) { return ret; @@ -493,23 +391,10 @@ _add_client(struct mread_pool * self, int fd) { close(fd); return NULL; } -#ifdef HAVE_EPOLL - struct epoll_event ev; - ev.events = EPOLLIN; - ev.data.ptr = s; - if (epoll_ctl(self->epoll_fd, EPOLL_CTL_ADD, fd, &ev) == -1) { + if (event_add(self->efd, fd, s)) { close(fd); 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->node = NULL; s->status = SOCKET_SUSPEND; diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c new file mode 100644 index 00000000..08fc7a1a --- /dev/null +++ b/lualib-src/lua-socket.c @@ -0,0 +1,250 @@ +#include +#include + +#include +#include + +#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= 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; +} diff --git a/lualib/redis.lua b/lualib/redis.lua index 11809619..4697c24d 100644 --- a/lualib/redis.lua +++ b/lualib/redis.lua @@ -1,65 +1,168 @@ local skynet = require "skynet" -local string = string -local table = table -local unpack = unpack -local assert = assert +local socket = require "socket" +local config = require "config" +local redis_conf = skynet.getenv "redis" +local name = config (redis_conf) -local redis_manager -local batch = false +local readline = socket.readline +local readbytes = socket.read +local table = table +local string = string local redis = {} - 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) local cmd = string.upper(k) - local f = function(self, ...) - if batch then - skynet.send( self.__handle, "lua", cmd , ...) + local f = function (self, ...) + local fd = self.__handle + if self.__mode then + socket.write(fd, compose_message { cmd, ... }) + self.__batch = self.__batch + 1 else - local err, result = skynet.call( self.__handle, "lua", cmd, ...) - assert(err, result) - return result + socket.lock(fd) + socket.write(fd, compose_message { cmd, ... }) + local ok, ret = read_response(fd) + socket.unlock(fd) + return ok, ret end end - t[k] = f + t[k] = f return f end}) function command:exists(key) - assert(not batch, "exists can't used in batch mode") - local result , exists = skynet.call( self.__handle, "lua" , "EXISTS", key) - assert(result, exists) + assert(not self.__mode, "exists can't used in batch mode") + local fd = self.__handle + 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 return exists end function command:batch(mode) if mode == "end" then - assert(batch, "Open batch mode first") - batch = false - local err, result = skynet.unpack(skynet.rawcall( self.__handle, "text", mode)) - assert(err, result) - return result + local fd = self.__handle + if self.__mode == "read" then + local allok = true + local allret = {} + 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 assert(mode == "read" or mode == "write") - batch = mode - skynet.send( self.__handle, "text", mode) + self.__mode = mode + self.__batch = 0 + socket.lock(self.__handle) 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 diff --git a/lualib/socket.lua b/lualib/socket.lua index 9dda19e5..3951a9d7 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -1,79 +1,189 @@ +local buffer = require "socketbuffer" 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 fd -local object -local data = {} -local function presend() - if next(data) then - local tmp = data - data = {} - for _,v in ipairs(tmp) do - socket.write(fd, v) +function socket.open(addr, port) + local cmd = "open" .. " " .. (port and (addr..":"..port) or addr) + local r = skynet.call(".socket", "text", cmd) + if r == "" then + error(cmd .. " failed") + end + return tonumber(r) +end + +function socket.stdin() + local r = skynet.call(".socket", "text", "bind 1") + if r == "" then + error("stdin bind failed") + end + return tonumber(r) +end + +function socket.close(fd) + socket.lock(fd) + skynet.call(".socket", "text", "close", fd) + READBUF[fd] = true + READLOCK[fd] = nil +end + +function socket.read(fd, sz) + assert(coroutine.running() == READTHREAD[fd], "call socket.lock first") + + local str = buffer.pop(READBUF[fd],sz) + if str then + return str + end + + READREQUEST[fd] = sz + 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 + return str + end + + READREQUEST[fd] = sep + skynet.call(selfaddr, "system",fd,2,sep) -- singal sep 2 + READREQUEST[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 -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 - -function socket.stdin() - skynet.send(".connection", "text", "ADD", 1 , skynet.address(skynet.self())) - object = c.new() -end - -function socket.push(msg,sz) - if msg then - c.push(object, msg, sz) - end -end - -function socket.read(bytes) - return c.read(object, bytes) -end - -function socket.readline(sep) - 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 - socket.close() - table.insert(data, str) - end -end - -function socket.writeblock(...) - local str = c.writeblock(fd, ...) - if str then - socket.close() - table.insert(data, str) - end -end - -function socket.close() - if fd then - skynet.send(".connection","text", "DEL", fd) - fd = nil - end -end - return socket - diff --git a/service-src/event.h b/service-src/event.h new file mode 100644 index 00000000..af639efa --- /dev/null +++ b/service-src/event.h @@ -0,0 +1,136 @@ +#ifndef skynet_event_h +#define skynet_event_h + +#include +#include + +/* 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 + +#ifdef HAVE_EPOLL +#include +#elif HAVE_KQUEUE +#include +#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 +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#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;icap;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;iid; // 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;iev[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;icap;i++) { + if (pool->s[i].status != STATUS_INVALID && pool->s[i].fd >=0) { + close(pool->s[i].fd); + } + } + free(pool->s); + free(pool); +} diff --git a/service/console.lua b/service/console.lua index effaa462..c95abae1 100644 --- a/service/console.lua +++ b/service/console.lua @@ -1,46 +1,15 @@ local skynet = require "skynet" 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.dispatch("text", function (session, address, cmd) - local handle = skynet.newservice(cmd) + local stdin = socket.stdin() + socket.lock(stdin) + while true do + local cmdline = socket.readline(stdin, "\n") + local handle = skynet.newservice(cmdline) if handle == nil then - print("Launch error:",cmd) + print("Launch error:",cmdline) end - end) - socket.stdin() + end + socket.unlock(stdin) end) diff --git a/service/main.lua b/service/main.lua index dbfdc594..bd3c03ce 100644 --- a/service/main.lua +++ b/service/main.lua @@ -2,8 +2,8 @@ local skynet = require "skynet" skynet.start(function() print("Server start") + skynet.launch("socket",128) local service = skynet.launch("snlua","service_mgr") - local connection = skynet.launch("connection","256") local lualog = skynet.launch("snlua","lualog") local console = skynet.launch("snlua","console") local remoteroot = skynet.launch("snlua","remote_root") diff --git a/service/redis-cli.lua b/service/redis-cli.lua deleted file mode 100644 index 52ae14c9..00000000 --- a/service/redis-cli.lua +++ /dev/null @@ -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) - diff --git a/service/redis-mgr.lua b/service/redis-mgr.lua deleted file mode 100644 index f6bb9327..00000000 --- a/service/redis-mgr.lua +++ /dev/null @@ -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) - diff --git a/service/testredis.lua b/service/testredis.lua index 0d4011f8..cbddf815 100644 --- a/service/testredis.lua +++ b/service/testredis.lua @@ -3,10 +3,11 @@ local redis = require "redis" skynet.start(function() local db = redis.connect "main" + print(db:select(0)) db:batch "write" -- ignore results db:set("A", "hello") db:set("B", "world") - db:batch "end" + print(db:batch "end") db:batch "read" db:get("A") @@ -19,6 +20,7 @@ skynet.start(function() print(db:get "A") print(db:set("A","hello world")) print(db:get("A")) + db:disconnect() skynet.exit() end)