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

redesign connection module

Browse Source
This commit is contained in:
云风
2012-08-14 22:37:30 +08:00
parent 4b3ce3ba5b
commit 4a0265020f
13 changed files with 466 additions and 705 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
Makefile
View File

@@ -7,9 +7,10 @@ service/logger.so \
lualib/skynet.so \
service/gate.so \
service/client.so \
service/connection.so \
service/broker.so \
service/connection.so \
client \
lualib/socket.so \
lualib/lpeg.so \
lualib/protobuf.so \
lualib/int64.so \
@@ -36,7 +37,7 @@ service/snlua.so : service-src/service_lua.c
gcc -Wall -g -fPIC --shared $^ -o $@ -Iskynet-src
service/gate.so : gate/mread.c gate/ringbuffer.c gate/main.c
gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Igate -Iskynet-src
gcc -Wall -g -fPIC --shared -o $@ $^ -Igate -Iskynet-src
lualib/skynet.so : lualib-src/lua-skynet.c lua-serialize/serialize.c
gcc -Wall -g -fPIC --shared $^ -o $@ -Iskynet-src -Ilua-serialize
@@ -45,7 +46,10 @@ service/client.so : service-src/service_client.c
gcc -Wall -g -fPIC --shared $^ -o $@ -Iskynet-src
service/connection.so : connection/connection.c connection/main.c
gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Iconnection -Iskynet-src
gcc -Wall -g -fPIC --shared -o $@ $^ -Iconnection -Iskynet-src
lualib/socket.so : connection/lua-socket.c
gcc -Wall -g -fPIC --shared -o $@ $^ -Iconnection -Iskynet-src
service/broker.so : service-src/service_broker.c
gcc -Wall -g -fPIC --shared $^ -o $@ -Iskynet-src

10
connection/Makefile
View File

@@ -1,10 +0,0 @@
all : test
test : connection.c test.c
gcc -g -Wall -o $@ $^
clean :
rm test

378
connection/connection.c
View File

@@ -2,259 +2,55 @@
#include <stdio.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <unistd.h>
#define EPOLLQUEUE 32
#define SEPLEN 8
#define DEFAULT_LINE 64
struct connection {
int handle;
int fd;
short in_epoll;
short read_complete;
int readbuffer_cap;
int read_request;
int read_sz;
int yield_sz;
char sep[SEPLEN];
char * readbuffer;
};
struct connection_pool {
int size;
struct connection * conn;
int epoll_fd;
int handle_index;
int queue_len;
int queue_head;
struct epoll_event ev[EPOLLQUEUE];
};
struct connection_pool *
struct connection_pool *
connection_newpool(int max) {
int epoll_fd = epoll_create(max);
if (epoll_fd == -1) {
return NULL;
}
struct connection_pool * pool = malloc(sizeof(*pool));
pool->size = max;
pool->conn = malloc(max * sizeof(struct connection));
memset(pool->conn, 0, max * sizeof(struct connection));
pool->epoll_fd = epoll_fd;
pool->handle_index =0;
pool->queue_len = 0;
pool->queue_head = 0;
return pool;
}
void
connection_deletepool(struct connection_pool *pool) {
int i;
for (i=0;i<pool->size;i++) {
struct connection * c = &pool->conn[i];
if (c->handle) {
close(c->fd);
}
if (c->readbuffer) {
free(c->readbuffer);
}
}
void
connection_deletepool(struct connection_pool * pool) {
close(pool->epoll_fd);
free(pool->conn);
free(pool);
}
static struct connection *
_new_connection(struct connection_pool *pool) {
int i;
for (i=1;i<=pool->size;i++) {
int handle = pool->handle_index + i;
struct connection *c = &pool->conn[handle % pool->size];
if (c->handle == 0) {
c->handle = handle;
c->in_epoll = 0;
c->read_complete = 0;
pool->handle_index = handle;
return c;
}
}
return NULL;
}
int
connection_open(struct connection_pool *pool, const char * address) {
char * sep = strchr(address,':');
if (sep == NULL)
return 0;
int len = sep-address;
char tmp[len+1];
memcpy(tmp, address, len);
tmp[len] = '\0';
int port = strtol(sep+1, NULL, 10);
connection_add(struct connection_pool * pool, int fd, void *ud) {
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.ptr = ud;
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(tmp);
int fd = socket(AF_INET,SOCK_STREAM,0);
int r = connect(fd,(struct sockaddr *)&my_addr,sizeof(struct sockaddr_in));
if (r == -1) {
return 0;
if (epoll_ctl(pool->epoll_fd, EPOLL_CTL_ADD, fd, &ev) == -1) {
return 1;
}
struct connection * c = _new_connection(pool);
if (c == NULL) {
close(fd);
return 0;
}
c->fd = fd;
return c->handle;
}
static void
_remove_connection(struct connection_pool * pool, struct connection *c) {
c->handle = 0;
if (c->in_epoll) {
epoll_ctl(pool->epoll_fd, EPOLL_CTL_DEL, c->fd , NULL);
}
close(c->fd);
}
struct connection *
_get_connection(struct connection_pool *pool, int handle) {
assert(handle != 0);
struct connection * c = &pool->conn[handle % pool->size];
if (c->handle != handle)
return NULL;
return c;
return 0;
}
void
connection_close(struct connection_pool * pool, int handle) {
struct connection * c = _get_connection(pool, handle);
if (c) {
_remove_connection(pool, c);
}
}
static void
_add_epoll(struct connection_pool * pool, struct connection *c) {
if (!c->in_epoll) {
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.ptr = c;
if (epoll_ctl(pool->epoll_fd, EPOLL_CTL_ADD, c->fd, &ev) == -1) {
close(c->fd);
c->handle = 0;
return;
}
c->in_epoll = 1;
}
}
static void
_yield(struct connection *c) {
c->read_complete = 0;
int len = c->read_sz - c->yield_sz;
if (len == 0) {
c->read_sz = 0;
return;
}
assert(len > 0);
memmove(c->readbuffer, c->readbuffer + c->yield_sz, len);
c->read_sz = len;
}
void *
connection_read(struct connection_pool * pool, int handle, size_t sz) {
struct connection * c = _get_connection(pool, handle);
if (c == NULL)
return NULL;
if (c->read_complete) {
_yield(c);
}
if (c->readbuffer_cap < sz) {
int cap = DEFAULT_LINE;
while (cap <= sz) {
cap *=2;
}
c->readbuffer = realloc(c->readbuffer, cap);
c->readbuffer_cap = cap;
}
if (c->read_sz >= sz) {
c->yield_sz = sz;
c->read_complete = 1;
c->read_request = 0;
return c->readbuffer;
}
c->read_request = sz;
_add_epoll(pool,c);
return NULL;
}
static const char *
_try_readline(struct connection_pool * pool, struct connection *c) {
int seplen = strlen(c->sep);
int i;
for (i=0;i<= c->read_sz - seplen;i++) {
if (memcmp(c->readbuffer+i,c->sep,seplen) == 0) {
c->readbuffer[i] = '\0';
c->read_request = i;
c->yield_sz = i + seplen;
c->read_complete = 1;
return c->readbuffer;
}
}
return NULL;
}
void *
connection_readline(struct connection_pool * pool, int handle, const char * sep, size_t *sz) {
struct connection * c = _get_connection(pool, handle);
if (c == NULL)
return NULL;
if (c->read_complete) {
_yield(c);
}
int i;
for (i=0;i<SEPLEN;i++) {
c->sep[i] = sep[i];
if (sep[i] == '\0')
break;
}
assert(i < SEPLEN);
const char * line = _try_readline(pool,c);
if (line) {
if (sz) {
*sz = c->read_request;
}
c->read_request = 0;
return c->readbuffer;
}
c->read_request = -1;
_add_epoll(pool,c);
return NULL;
connection_del(struct connection_pool * pool, int fd) {
epoll_ctl(pool->epoll_fd, EPOLL_CTL_DEL, fd , NULL);
close(fd);
}
static int
@@ -269,155 +65,13 @@ _read_queue(struct connection_pool * pool, int timeout) {
return n;
}
inline static struct connection *
_read_one(struct connection_pool * pool) {
if (pool->queue_head >= pool->queue_len) {
return NULL;
}
return pool->ev[pool->queue_head ++].data.ptr;
}
static int
_read_buffer(struct connection_pool * pool, struct connection *c) {
int bytes = recv(c->fd, c->readbuffer + c->read_sz, c->readbuffer_cap - c->read_sz, MSG_DONTWAIT);
if (bytes > 0) {
c->read_sz += bytes;
}
if (bytes == 0) {
// closed
_remove_connection(pool, c);
}
return bytes;
}
static int
_read_block(struct connection_pool * pool, struct connection *c, size_t * id) {
assert(c->read_complete == 0);
int need_read = c->read_request - c->read_sz;
if (need_read <= 0) {
c->yield_sz = c->read_request;
c->read_complete = 1;
return 1;
}
int handle = c->handle;
int bytes = _read_buffer(pool,c);
if (bytes > 0) {
if (c->read_request - c->read_sz <= 0) {
c->yield_sz = c->read_request;
c->read_complete = 1;
return 1;
}
} else if (bytes == 0) {
if (id) {
*id = c - pool->conn;
}
return -handle;
}
return 0;
}
static int
_read_line(struct connection_pool * pool, struct connection *c, size_t * id) {
assert(c->read_complete == 0);
if (c->readbuffer_cap == c->read_sz) {
int cap = DEFAULT_LINE;
while (cap <= c->read_sz) {
cap *=2;
}
c->readbuffer = realloc(c->readbuffer, cap);
c->readbuffer_cap = cap;
}
int handle = c->handle;
int bytes = _read_buffer(pool,c);
if (bytes == 0) {
if (id) {
*id = c - pool->conn;
}
return -handle; // close
}
const char * line = _try_readline(pool,c);
if (line)
return 1;
return 0;
}
static int
_read_request(struct connection_pool * pool, struct connection *c, size_t * id) {
if (c->read_request < 0) {
return _read_line(pool, c, id);
} else {
return _read_block(pool, c, id);
}
}
void *
connection_poll(struct connection_pool * pool, int timeout, int *phandle, size_t * id) {
connection_poll(struct connection_pool * pool, int timeout) {
if (pool->queue_head >= pool->queue_len) {
if (_read_queue(pool, timeout) == -1) {
return NULL;
}
}
for (;;) {
struct connection * c = _read_one(pool);
if (c == NULL) {
*phandle = 0;
return NULL;
}
if (c->read_complete) {
assert(c->in_epoll);
epoll_ctl(pool->epoll_fd, EPOLL_CTL_DEL, c->fd , NULL);
c->in_epoll = 0;
continue;
}
int result = _read_request(pool, c, id);
switch (result) {
case 0:
// block
*phandle = 0;
break;
case 1:
*phandle = c->handle;
if (id) {
*id = c->read_request;
}
c->read_request = 0;
return c->readbuffer;
default:
// nagetive close
*phandle = -result;
return NULL;
}
}
return pool->ev[pool->queue_head ++].data.ptr;
}
void
connection_write(struct connection_pool * pool, int handle, const void * buffer, size_t sz) {
struct connection * c = _get_connection(pool, handle);
if (c) {
for (;;) {
int err = send(c->fd, buffer, sz, 0);
if (err < 0) {
switch (errno) {
case EAGAIN:
case EINTR:
continue;
}
}
assert(err == sz);
return;
}
}
}
int
connection_id(struct connection_pool * pool, int handle) {
struct connection * c = _get_connection(pool, handle);
if (c == NULL) {
return 0;
}
return c - pool->conn + 1;
}

13
connection/connection.h
View File

@@ -8,16 +8,9 @@ struct connection_pool;
struct connection_pool * connection_newpool(int max);
void connection_deletepool(struct connection_pool *);
int connection_open(struct connection_pool *, const char * ipaddr);
void connection_close(struct connection_pool *, int handle);
int connection_add(struct connection_pool *, int fd, void *ud);
void connection_del(struct connection_pool *, int fd);
void * connection_read(struct connection_pool *, int handle, size_t sz);
void * connection_readline(struct connection_pool *, int handle, const char * sep, size_t *sz);
void * connection_poll(struct connection_pool *, int timeout, int *handle, size_t *sz);
int connection_closed(struct connection_pool *, int handle);
void connection_write(struct connection_pool *, int handle, const void * buffer, size_t sz);
int connection_id(struct connection_pool *, int handle);
void * connection_poll(struct connection_pool *, int timeout);
#endif

186
connection/lua-socket.c Normal file
View File

@@ -0,0 +1,186 @@
#include "skynet.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <arpa/inet.h>
#include <errno.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) {
return 0;
}
lua_pushinteger(L,fd);
return 1;
}
static int
_write(lua_State *L) {
int fd = luaL_checkinteger(L,1);
int type = lua_type(L,2);
const char * buffer;
size_t sz;
if (type == LUA_TSTRING) {
buffer = lua_tolstring(L,2,&sz);
} else if (type == LUA_TLIGHTUSERDATA) {
buffer = lua_touserdata(L,2);
sz = luaL_checkinteger(L,3);
}
for (;;) {
int err = send(fd, buffer, sz, 0);
if (err < 0) {
switch (errno) {
case EAGAIN:
case EINTR:
continue;
}
}
assert(err == sz);
return 0;
}
}
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);
}
buffer->read = 0;
if (size + buffer->size > buffer->cap) {
if (buffer->cap == 0) {
lua_rawgetp(L, LUA_REGISTRYINDEX, _delete);
lua_setmetatable(L, 1);
}
buffer->buffer = realloc(buffer->buffer, buffer->size + size);
memcpy(buffer->buffer + buffer->size, buf, size);
buffer->size += size;
buffer->cap = buffer->size;
return 0;
}
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;
}
static int
_yield(lua_State *L) {
luaL_checktype(L,1,LUA_TUSERDATA);
struct buffer * buffer = lua_touserdata(L,1);
if (buffer->read) {
int size = buffer->size - buffer->read;
memmove(buffer->buffer, buffer->buffer + buffer->read, size);
buffer->size -= buffer->read;
buffer->read = 0;
}
return 0;
}
int
luaopen_socket_c(lua_State *L) {
luaL_Reg l[] = {
{ "open", _open },
{ "write", _write },
{ "new", _new },
{ "push", _push },
{ "read", _read },
{ "readline", _readline },
{ "yield", _yield },
{ 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;
}

304
connection/main.c
View File

@@ -1,30 +1,30 @@
#include "connection.h"
#include "skynet.h"
#include <sys/types.h>
#include <sys/socket.h>
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
struct reply {
int session;
uint32_t dest;
#define DEFAULT_BUFFER_SIZE 1024
#define DEFAULT_CONNECTION 16
struct connection {
int fd;
char * addr;
};
struct connection_server {
int poll;
int max_connection;
int current_connection;
struct connection_pool *pool;
struct skynet_context *ctx;
struct reply * reply;
};
typedef void (*command_func)(struct connection_server *server, const char * param, size_t sz, int session, const char * reply);
struct command {
const char * name;
command_func func;
struct connection * conn;
};
struct connection_server *
@@ -39,183 +39,86 @@ connection_release(struct connection_server * server) {
if (server->pool) {
connection_deletepool(server->pool);
}
free(server->reply);
int i;
for (i=0;i<server->max_connection;i++) {
free(server->conn[i].addr);
}
free(server->conn);
free(server);
}
static inline const char *
_command(const char * cmd, const char * msg, size_t sz) {
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<sz;i++) {
if (cmd[i] == '\0') {
if (msg[i] != ' ')
return NULL;
return msg+i+1;
}
if (cmd[i] != msg[i])
return NULL;
for (i=0;i<server->max_connection;i++) {
struct connection * c = &server->conn[i];
connection_add(server->pool, c->fd , c);
}
return NULL;
server->max_connection *= 2;
}
static void
_connect(struct connection_server *server, const char * ipaddr, size_t sz, int session, const char * reply) {
char tmp[sz+1];
memcpy(tmp, ipaddr, sz);
tmp[sz] = '\0';
int id = connection_open(server->pool, ipaddr);
if (id == 0) {
skynet_send(server->ctx, NULL, reply, session, NULL, 0, 0);
return;
_add(struct connection_server * server, int fd , char * addr) {
++server->current_connection;
if (server->current_connection > server->max_connection) {
_expand(server);
}
char idstring[20];
int n = sprintf(idstring, "%d", id);
skynet_send(server->ctx, NULL, reply, session, idstring, n, 0);
}
static void
_close(struct connection_server *server, const char * param, size_t sz, int session, const char * reply) {
int handle = strtol(param, NULL, 10);
if (handle <= 0) {
skynet_error(server->ctx, "[connection] Close invalid handle from %s", reply);
return;
}
connection_close(server->pool, handle);
}
static void
_write(struct connection_server *server, const char * param, size_t sz, int session, const char * reply) {
char * endptr = NULL;
int handle = strtol(param, &endptr, 10);
if (handle <= 0 || endptr == NULL || *endptr !=' ') {
skynet_error(server->ctx, "[connection] Write invalid handle from %s", reply);
return;
}
++endptr;
sz -= endptr - param;
connection_write(server->pool, handle, endptr, sz);
}
static void
_read(struct connection_server *server, const char * param, size_t sz, int session, const char * reply) {
char * endptr = NULL;
int handle = strtol(param, &endptr, 10);
if (handle <= 0 || endptr == NULL || *endptr !=' ') {
skynet_error(server->ctx, "[connection] Read invalid handle from %s", reply);
return;
}
int size = strtol(endptr+1, &endptr, 10);
if (size <= 0 || endptr == NULL) {
skynet_error(server->ctx, "[connection] Read invalid size (%d) from %s", size, reply);
return;
}
void * buffer = connection_read(server->pool, handle, size);
if (buffer == NULL) {
int id = connection_id(server->pool, handle);
if (id == 0) {
skynet_send(server->ctx, NULL, reply, session, NULL, 0, 0);
return;
}
--id;
assert(id < server->max_connection);
assert(reply[0] == ':');
server->reply[id].session = session;
server->reply[id].dest = strtoul(reply+1, NULL, 16);
++server->poll;
if (server->poll == 1) {
skynet_command(server->ctx, "TIMEOUT","0");
return;
}
} else {
skynet_send(server->ctx, NULL, reply, session, buffer, size, 0);
}
}
static void
_readline(struct connection_server *server, const char * param, size_t sz, int session, const char * reply) {
char * endptr = NULL;
int handle = strtol(param, &endptr, 10);
if (handle <= 0 || endptr == NULL || *endptr !=' ') {
skynet_error(server->ctx, "[connection] Readline invalid handle from %s", reply);
return;
}
sz -= endptr - param + 1;
if (sz < 1 || sz > 7) {
skynet_error(server->ctx, "[connection] Readline invalid sep (size = %d) from %s", sz, reply);
return;
}
char sep[sz+1];
memcpy(sep, endptr+1, sz);
sep[sz] = '\0';
void * buffer = connection_readline(server->pool, handle, sep, &sz);
if (buffer == NULL) {
int id = connection_id(server->pool, handle);
if (id == 0) {
skynet_send(server->ctx, NULL, reply, session, NULL, 0, 0);
return;
}
--id;
assert(id < server->max_connection);
assert(reply[0] == ':');
server->reply[id].session = session;
server->reply[id].dest = strtoul(reply+1, NULL, 16);
++server->poll;
if (server->poll == 1) {
skynet_command(server->ctx, "TIMEOUT","0");
return;
}
} else {
skynet_send(server->ctx, NULL, reply, session, buffer, sz, 0);
}
}
static void
_id_to_hex(char * str, uint32_t id) {
int i;
static char hex[16] = { '0','1','2','3','4','5','6','7','8','9','A','B','C','D','E','F' };
for (i=0;i<8;i++) {
str[i] = hex[(id >> ((7-i) * 4))&0xf];
for (i=0;i<server->max_connection;i++) {
struct connection * c = &server->conn[i];
if (c->addr == NULL) {
c->fd = fd;
c->addr = addr;
int err = connection_add(server->pool, fd , c);
assert(err == 0);
return;
}
}
str[8] = '\0';
assert(0);
}
static void
_poll(struct connection_server *server) {
int handle = 0;
size_t sz = 0;
_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) {
free(c->addr);
c->addr = NULL;
c->fd = 0;
connection_del(server->pool, fd);
return;
}
}
skynet_error(server->ctx, "[connection] Delete invalid handle %d", fd);
}
static void
_poll(struct connection_server * server) {
int timeout = 100;
char addr[10];
for (;;) {
void * buffer = connection_poll(server->pool, timeout, &handle, &sz);
struct connection * c = connection_poll(server->pool, timeout);
if (c==NULL) {
skynet_command(server->ctx,"TIMEOUT","1");
return;
}
timeout = 0;
if (buffer == NULL) {
if (handle) {
--server->poll;
int id = sz;
struct reply * r = &server->reply[id];
addr[0] = ':';
_id_to_hex(addr+1, id);
skynet_send(server->ctx, NULL, addr , r->session, NULL, 0, 0);
} else {
assert(server->poll >= 0);
if (server->poll > 0) {
skynet_command(server->ctx, "TIMEOUT","1");
}
return;
}
void * buffer = malloc(DEFAULT_BUFFER_SIZE);
int size = recv(c->fd, buffer, DEFAULT_BUFFER_SIZE, MSG_DONTWAIT);
if (size < 0) {
continue;
}
if (size == 0) {
printf("disconnect\n");
free(buffer);
skynet_send(server->ctx, NULL, c->addr, 0x7fffffff, NULL, 0, DONTCOPY);
} else {
--server->poll;
int id = connection_id(server->pool, handle);
assert(id > 0);
--id;
struct reply * r = &server->reply[id];
addr[0] = ':';
_id_to_hex(addr+1, r->dest);
skynet_send(server->ctx, NULL, addr, r->session, buffer, sz, 0);
skynet_send(server->ctx, NULL, c->addr, 0x7fffffff, buffer, size, DONTCOPY);
}
}
}
@@ -223,52 +126,49 @@ _poll(struct connection_server *server) {
static void
_main(struct skynet_context * ctx, void * ud, int session, const char * uid, const void * msg, size_t sz) {
if (msg == NULL) {
assert(session >= 0);
_poll(ud);
return;
}
if (session > 0) {
skynet_error(ctx, "[connection] Invalid response (session = %d) from %s", session, uid);
return;
}
struct command cmd[] = {
{ "CONNECT", _connect },
{ "CLOSE" , _close },
{ "READ" , _read },
{ "READLINE", _readline },
{ "WRITE", _write },
{ NULL, NULL },
};
struct command * p = cmd;
while (p->name) {
const char * param = _command(p->name, msg, sz);
if (param) {
p->func(ud, param, sz - (param-(const char *)msg), -session, uid);
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 %s (session = %d)", uid, session);
return;
}
++p;
int addr_sz = sz - (endptr - (char *)msg);
char * addr = malloc(addr_sz);
memcpy(addr, endptr+1, addr_sz-1);
addr[addr_sz] = '\0';
_add(ud, fd, addr);
} 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 %s (session = %d)", uid, session);
return;
}
_del(ud, fd);
} else {
skynet_error(ctx, "[connection] Invalid command from %s (session = %d)", uid, session);
}
skynet_error(ctx, "[connection] Invalid command from %s (session = %d)", uid, -session);
}
int
connection_init(struct connection_server * server, struct skynet_context * ctx, char * param) {
int max = strtol(param, NULL, 10);
if (max <=0) {
return 1;
}
server->pool = connection_newpool(max);
server->pool = connection_newpool(DEFAULT_CONNECTION);
if (server->pool == NULL)
return 1;
server->max_connection = max;
server->reply = malloc(sizeof(struct reply) * max);
memset(server->reply, 0, sizeof(struct reply) * max);
server->max_connection = DEFAULT_CONNECTION;
server->current_connection = 0;
server->ctx = ctx;
server->poll = 0;
server->conn = malloc(server->max_connection * sizeof(struct connection));
memset(server->conn, 0, server->max_connection * sizeof(struct connection));
skynet_callback(ctx, server, _main);
skynet_command(ctx,"REG",".connection");
skynet_command(ctx,"TIMEOUT","0");
return 0;
}

63
connection/test.c
View File

@@ -1,63 +0,0 @@
#include "connection.h"
#include <stdio.h>
#include <stdint.h>
static void
test(struct connection_pool *p, int id) {
int i=0;
while (i<5) {
int handle = id;
const char * line = connection_readline(p, handle, "\n", NULL);
if (line == NULL) {
line = connection_poll(p,1000,&handle,NULL);
}
if (line) {
printf("%d %d: %s\n",i,handle, line);
connection_write(p,handle,"readline\n",9);
++i;
} else {
if (handle) {
printf("Close %d\n",handle);
return;
}
}
}
i=0;
while (i<5) {
int handle = id;
uint8_t * buffer = connection_read(p, handle, 8);
if (buffer == NULL) {
buffer = connection_poll(p,1000,&handle,NULL);
}
if (buffer) {
int j;
printf("%d %d: ",i,handle);
for (j=0;j<8;j++) {
printf("%02x ",buffer[j]);
}
printf("\n");
connection_write(p,handle,"readblock\n",10);
++i;
} else {
if (handle) {
printf("Close %d\n", handle);
return;
}
}
}
}
int
main() {
struct connection_pool * p = connection_newpool(16);
int handle = connection_open(p, "127.0.0.1:8888");
test(p,handle);
connection_deletepool(p);
return 0;
};

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

@@ -97,6 +97,14 @@ _command(lua_State *L) {
return 0;
}
static int
_genid(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
int session = skynet_send(context, NULL, NULL, -1, NULL, 0 , 0);
lua_pushinteger(L, session);
return 1;
}
static int
_send(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
@@ -183,6 +191,7 @@ luaopen_skynet_c(lua_State *L) {
luaL_checkversion(L);
luaL_Reg l[] = {
{ "send" , _send },
{ "genid", _genid },
{ "redirect", _redirect },
{ "command" , _command },
{ "callback" , _callback },

9
lualib/skynet.lua
View File

@@ -18,10 +18,11 @@ local function suspend(co, result, command, param, size)
local co_address = session_coroutine_address[co]
c.send(co_address, co_session, param, size)
return suspend(co, coroutine.resume(co))
else
assert(command == nil, command)
elseif command == nil then
session_coroutine_id[co] = nil
session_coroutine_address[co] = nil
else
error("Unknown command : " .. command)
end
end
@@ -99,6 +100,7 @@ function skynet.setenv(key, value)
end
skynet.send = assert(c.send)
skynet.genid = assert(c.genid)
skynet.redirect = assert(c.redirect)
skynet.pack = assert(c.pack)
skynet.tostring = assert(c.tostring)
@@ -126,6 +128,9 @@ end
local function default_dispatch(f)
return function(session, address , msg, sz)
if session == nil then
return
end
if session <= 0 then
session = - session
co = coroutine.create(f)

46
lualib/socket.lua Normal file
View File

@@ -0,0 +1,46 @@
local skynet = require "skynet"
local c = require "socket.c"
local socket = {}
local fd
local object
function socket.connect(addr)
local ip, port = string.match(addr,"([^:]+):(.+)")
port = tonumber(port)
fd = c.open(ip,port)
skynet.send(".connection","ADD "..fd.." "..skynet.self())
object = c.new()
end
function socket.push(msg,sz)
if msg == nil then
socket.close()
else
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.write(...)
c.write(fd, ...)
end
function socket.yield()
c.yield(object)
end
function socket.close()
skynet.send(".connection","DEL "..fd)
fd = nil
end
return socket

138
service/redis-cli.lua
View File

@@ -1,34 +1,31 @@
local skynet = require "skynet"
local socket = require "socket"
local string = string
local table = table
local tonumber = tonumber
local ipairs = ipairs
local unpack = unpack
local redis_server = ...
local fd
local write_fd
local readline_fd
local read_fd
local close_fd
local function init_fd(fdstr)
fd = fdstr
write_fd = "WRITE "..fd.." "
readline_fd = "READLINE ".. fd .." \r\n"
read_fd = "READ " .. fd .. " "
close_fd = "CLOSE "..fd
end
local function init()
fd = skynet.call(".connection", "CONNECT " .. redis_server)
if fd == nil then
print("Connect to redis server error : ", redis_server)
skynet.exit()
return true
local function read(bytes)
while true do
local result = socket.read(bytes)
if result then
return result
end
coroutine.yield()
end
init_fd(fd)
end
local function readline(sep)
while true do
local result = socket.readline(sep)
if result then
return result
end
coroutine.yield()
end
end
local function compose_message(msg)
local lines = { "*" .. #msg }
@@ -42,72 +39,109 @@ local function compose_message(msg)
return cmd
end
local function init()
socket.connect(redis_server)
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 response(...)
local reply = pop_request_queue()
skynet.send(reply[2],reply[1],skynet.pack(...))
end
local redcmd = {}
redcmd[42] = function(data) -- '*'
local n = tonumber(data)
if n < 1 then
skynet.ret(skynet.pack(true, nil))
response(true, nil)
return
end
local bulk = {}
for i = 1,n do
local line = skynet.call(".connection", readline_fd)
local line = socket.readline "\r\n"
if line == nil then
return "BLOCK"
end
local bytes = tonumber(string.sub(line,2) + 2)
local data = skynet.call(".connection", read_fd .. bytes)
local data = socket.read(bytes)
if data == nil then
return "BLOCK"
end
table.insert(bulk, string.sub(data,1,-3))
end
skynet.ret(skynet.pack(true,bulk))
response(true, bulk)
end
redcmd[36] = function(data) -- '$'
local bytes = tonumber(data)
if bytes < 0 then
skynet.ret(skynet.pack(true, nil))
response(true,nil)
return
end
local firstline = skynet.call(".connection", read_fd .. (bytes + 2))
skynet.ret(skynet.pack(true,string.sub(firstline,1,-3)))
local firstline = socket.read(bytes+2)
if firstline == nil then
return "BLOCK"
end
response(true,string.sub(firstline,1,-3))
end
redcmd[43] = function(data) -- '+'
skynet.ret(skynet.pack(true, data))
response(true,data)
end
redcmd[45] = function(data) -- '-'
skynet.ret(skynet.pack(false, data))
response(false,data)
end
redcmd[58] = function(data) -- ':'
skynet.ret(skynet.pack(true, tonumber(data)))
response(true, tonumber(data))
end
skynet.dispatch(function(msg, sz, session, address)
local message = { skynet.unpack(msg,sz) }
local write_cmd = write_fd .. compose_message(message)
local result
while true do
skynet.send(".connection", write_cmd )
result = skynet.call(".connection", readline_fd)
if result then
break
end
-- reconnect
if init() then
skynet.ret(skynet.pack(false , "Disconnected"))
return
end
local function split_package()
local result = socket.readline "\r\n"
if result == nil then
return
end
local firstchar = string.byte(result)
local data = string.sub(result,2)
local f = redcmd[firstchar]
if f == nil then
skynet.ret(skynet.pack(false , "Invalid result"))
skynet.send(".connection", close_fd)
init()
else
f(data)
assert(f)
if f(data) then
return
end
end)
socket.yield()
return true
end
skynet.filter(
function(session, address , msg, sz)
if session == 0x7fffffff then
socket.push(msg,sz)
while split_package() do end
elseif session < 0 then
local message = { skynet.unpack(msg,sz) }
local cmd = compose_message(message)
socket.write(cmd)
push_request_queue { -session , address }
else
return session, address, msg , sz
end
end
)
skynet.start(init)

2
skynet-src/skynet_harbor.c
View File

@@ -275,7 +275,7 @@ _remote_harbor_update(int harbor_id, const char * addr) {
void *socket = zmq_socket( Z->zmq_context, ZMQ_PUSH);
int rc = zmq_connect(socket, addr);
if (rc<0) {
skynet_error(NULL, "Can't connect to %d %s",harbor_id,addr);
skynet_error(NULL, "Can't connect to harbor %d %s",harbor_id,addr);
zmq_close(socket);
socket = NULL;
}

3
skynet-src/skynet_server.c
View File

@@ -399,6 +399,9 @@ skynet_send(struct skynet_context * context, const char * source, const char * a
session = skynet_context_newsession(context);
session_id = - session;
}
if (addr == NULL) {
return session;
}
uint32_t des = 0;
if (addr[0] == ':') {
des = strtoul(addr+1, NULL, 16);