diff --git a/Makefile b/Makefile index 772c8e48..50407846 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -all : skynet snlua.so logger.so skynet.so gate.so client.so client skynet-master +all : skynet snlua.so logger.so skynet.so gate.so client.so connection.so client skynet-master skynet : skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c skynet_server.c skynet_start.c skynet_timer.c skynet_error.c skynet_harbor.c gcc -Wall -g -Wl,-E -o $@ $^ -lpthread -ldl -lrt -Wl,-E -llua -lm -lzmq @@ -18,6 +18,9 @@ skynet.so : lua-skynet.c client.so : service_client.c gcc -Wall -g -fPIC --shared $^ -o $@ +connection.so : connection/connection.c connection/main.c + gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Iconnection + client : client.c gcc -Wall -g $^ -o $@ -lpthread diff --git a/connection/Makefile b/connection/Makefile new file mode 100644 index 00000000..00f1e55a --- /dev/null +++ b/connection/Makefile @@ -0,0 +1,13 @@ +all : test testmap + +test : connection.c test.c + gcc -g -Wall -o $@ $^ + +testmap : testmap.c map.c + gcc -g -Wall -o $@ $^ + +clean : + rm test testmap + + + \ No newline at end of file diff --git a/connection/connection.c b/connection/connection.c new file mode 100644 index 00000000..447de049 --- /dev/null +++ b/connection/connection.c @@ -0,0 +1,420 @@ +#include "connection.h" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#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 * +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;isize;i++) { + struct connection * c = &pool->conn[i]; + if (c->handle) { + close(c->fd); + } + if (c->readbuffer) { + free(c->readbuffer); + } + } + 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); + + 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; + } + + 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; +} + +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;isep[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; +} + +static int +_read_queue(struct connection_pool * pool, int timeout) { + pool->queue_head = 0; + int n = epoll_wait(pool->epoll_fd , pool->ev, EPOLLQUEUE, timeout); + if (n == -1) { + pool->queue_len = 0; + return -1; + } + pool->queue_len = n; + 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->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->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) { + 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_request == 0) { + assert(c->in_epoll); + epoll_ctl(pool->epoll_fd, EPOLL_CTL_DEL, c->fd , NULL); + c->in_epoll = 0; + } + 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; + } + } +} + +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; +} diff --git a/connection/connection.h b/connection/connection.h new file mode 100644 index 00000000..df13e1b3 --- /dev/null +++ b/connection/connection.h @@ -0,0 +1,23 @@ +#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_open(struct connection_pool *, const char * ipaddr); +void connection_close(struct connection_pool *, int handle); + +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); + +#endif diff --git a/connection/main.c b/connection/main.c new file mode 100644 index 00000000..2190f700 --- /dev/null +++ b/connection/main.c @@ -0,0 +1,276 @@ +#include "connection.h" +#include "skynet.h" + +#include +#include +#include +#include +#include + +struct reply { + int session; + uint32_t dest; +}; + +struct connection_server { + int poll; + int max_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_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->reply); + free(server); +} + +static inline const char * +_command(const char * cmd, const char * msg, size_t sz) { + int i; + for (i=0;ipool, ipaddr); + if (id == 0) { + skynet_send(server->ctx, reply, session, NULL, 0, 0); + return; + } + char idstring[20]; + int n = sprintf(idstring, "%d", id); + skynet_send(server->ctx, 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, 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, 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, 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, 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]; + } + str[8] = '\0'; +} + +static void +_poll(struct connection_server *server) { + int handle = 0; + size_t sz = 0; + int timeout = 100; + char addr[10]; + for (;;) { + void * buffer = connection_poll(server->pool, timeout, &handle, &sz); + 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, addr , r->session, NULL, 0, 0); + } else { + assert(server->poll >= 0); + if (server->poll > 0) { + skynet_command(server->ctx, "TIMEOUT","1"); + } + return; + } + } 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, addr, r->session, buffer, sz, 0); + } + } +} + +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); + return; + } + ++p; + } + + 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); + 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->ctx = ctx; + server->poll = 0; + skynet_callback(ctx, server, _main); + skynet_command(ctx,"REG",".connection"); + return 0; +} + + + diff --git a/connection/test.c b/connection/test.c new file mode 100644 index 00000000..8d325efc --- /dev/null +++ b/connection/test.c @@ -0,0 +1,63 @@ +#include "connection.h" + +#include +#include + +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; +}; + diff --git a/connection/testmap.c b/connection/testmap.c new file mode 100644 index 00000000..cd4f4b9e --- /dev/null +++ b/connection/testmap.c @@ -0,0 +1,52 @@ +#include "map.h" + +#include +#include +#include + +#define MAX 2000 + +static void +test(struct map *m) { + int a[MAX * 2]; + int i; + int s = 0; + for (i=0;i=sizeof(tmp)) { - n = sizeof(tmp) - 1; - tmp[n] = '\0'; - } - skynet_send(ctx, WATCHDOG, 0, strdup(tmp), n); + skynet_send(ctx, WATCHDOG, 0, tmp, n, 0); } static void _forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_t len) { struct connection * agent = _id_to_agent(g,uid); if (agent->agent) { - char * tmp = malloc(len); - memcpy(tmp,data,len); - skynet_send(ctx, agent->agent, 0, tmp, len); + skynet_send(ctx, agent->agent, 0, data, len, 0); } else { char * tmp = malloc(len + 32); int n = snprintf(tmp,len+32,"%d data ",uid); memcpy(tmp+n,data,len); - skynet_send(ctx, WATCHDOG, 0, tmp, len + n); + skynet_send(ctx, WATCHDOG, 0, tmp, len + n, DONTCOPY); } } diff --git a/lua-skynet.c b/lua-skynet.c index 968f49d5..c3e8838a 100644 --- a/lua-skynet.c +++ b/lua-skynet.c @@ -20,17 +20,18 @@ traceback (lua_State *L) { static void _cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { lua_State *L = ud; - lua_pushcfunction(L, traceback); + lua_rawgetp(L, LUA_REGISTRYINDEX, traceback); + int trace = lua_gettop(L); lua_rawgetp(L, LUA_REGISTRYINDEX, _cb); int r; if (msg == NULL) { lua_pushinteger(L, session); - r = lua_pcall(L, 1, 0 , -3); + r = lua_pcall(L, 1, 0 , trace); } else { lua_pushinteger(L, session); lua_pushstring(L, addr); lua_pushlstring(L, msg, sz); - r = lua_pcall(L, 3, 0 , -5); + r = lua_pcall(L, 3, 0 , trace); } if (r == LUA_OK) return; @@ -85,24 +86,20 @@ _send(lua_State *L) { ++index; } if (lua_gettop(L) == index + 1) { - session = skynet_send(context, dest, session , NULL, 0); + session = skynet_send(context, dest, session , NULL, 0, 0); } else { int type = lua_type(L,index+2); if (type == LUA_TSTRING) { size_t len = 0; void * msg = (void *)lua_tolstring(L,index+2,&len); - void * message = malloc(len); - memcpy(message, msg, len); - session = skynet_send(context, dest, session , message, len); + session = skynet_send(context, dest, session , msg, len, 0); } else if (type == LUA_TNIL) { - session = skynet_send(context, dest, session , NULL, 0); + session = skynet_send(context, dest, session , NULL, 0, 0); } else { + luaL_checktype(L,index+2, LUA_TLIGHTUSERDATA); void * msg = lua_touserdata(L,index+2); - if (msg == NULL) { - return luaL_error(L, "skynet.send need userdata or string (%s)", lua_typename(L,type)); - } int size = luaL_checkinteger(L,index+3); - session = skynet_send(context, dest, session, msg, size); + session = skynet_send(context, dest, session, msg, size, DONTCOPY); } } if (session < 0) { @@ -129,6 +126,8 @@ luaopen_skynet_c(lua_State *L) { { "error", _error }, { NULL, NULL }, }; + lua_pushcfunction(L, traceback); + lua_rawsetp(L, LUA_REGISTRYINDEX, traceback); luaL_newlibtable(L,l); diff --git a/skynet.h b/skynet.h index 9dd4dcbb..8bf39d0b 100644 --- a/skynet.h +++ b/skynet.h @@ -4,11 +4,13 @@ #include #include +#define DONTCOPY 1 + struct skynet_context; void skynet_error(struct skynet_context * context, const char *msg, ...); const char * skynet_command(struct skynet_context * context, const char * cmd , const char * parm); -int skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz); +int skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz, int flags); typedef void (*skynet_cb)(struct skynet_context * context, void *ud, int session, const char * addr , const void * msg, size_t sz); void skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb); diff --git a/skynet_server.c b/skynet_server.c index 92e814ee..f1ba3899 100644 --- a/skynet_server.c +++ b/skynet_server.c @@ -29,7 +29,7 @@ struct skynet_context { }; static void -_id_to_hex(char * str, int id) { +_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++) { @@ -268,7 +268,15 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * } int -skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz) { +skynet_send(struct skynet_context * context, const char * addr , int session, void * data, size_t sz, int flags) { + char * msg; + if ((flags & DONTCOPY) || data == NULL) { + msg = data; + } else { + msg = malloc(sz+1); + memcpy(msg, data, sz); + msg[sz] = '\0'; + } int session_id = session; if (session < 0) { session = skynet_context_newsession(context); diff --git a/testconn.lua b/testconn.lua new file mode 100644 index 00000000..116e3cbb --- /dev/null +++ b/testconn.lua @@ -0,0 +1,18 @@ +local skynet = require "skynet" + +-- register a dummy callback function +skynet.dispatch() + +local connection = skynet.launch("connection","16") +print("connection",connection) + +skynet.start(function() + local fd = skynet.call(".connection","CONNECT 127.0.0.1:8000") + print("connect to ",fd) + skynet.send(".connection","WRITE "..fd.." Welcome\n") + local line = skynet.call(".connection","READLINE "..fd .." \n") + skynet.send(".connection","WRITE "..fd.." "..line.."\n") + skynet.send(".connection","CLOSE "..fd) + + skynet.exit() +end)