connection server

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
 

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
+
+
+	

connection/connection.c

@@ -0,0 +1,420 @@
+#include "connection.h"
+
+#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>
+
+#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;i<pool->size;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;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;
+}
+
+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;
+}

connection/connection.h

@@ -0,0 +1,23 @@
+#ifndef SKYNET_CONNECTION_H
+#define SKYNET_CONNECTION_H
+
+#include <stddef.h>
+
+struct connection_pool;
+
+struct connection_pool * connection_newpool(int max);
+void connection_deletepool(struct connection_pool *);
+
+int connection_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

connection/main.c

@@ -0,0 +1,276 @@
+#include "connection.h"
+#include "skynet.h"
+
+#include <assert.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <string.h>
+
+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;i<sz;i++) {
+		if (cmd[i] == '\0') {
+			if (msg[i] != ' ')
+				return NULL;
+			return msg+i+1;
+		}
+		if (cmd[i] != msg[i])
+			return NULL;
+	}
+	return NULL;
+}
+
+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, 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;
+}
+
+
+

connection/test.c

@@ -0,0 +1,63 @@
+#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;
+};
+

connection/testmap.c

@@ -0,0 +1,52 @@
+#include "map.h"
+
+#include <stdlib.h>
+#include <assert.h>
+#include <stdio.h>
+
+#define MAX 2000
+
+static void
+test(struct map *m) {
+	int a[MAX * 2];
+	int i;
+	int s = 0;
+	for (i=0;i<MAX*2;i++) {
+		int inc = rand() % 3 + 1;
+		s += inc;
+		a[i] = s;
+	}
+	for (i=0;i<MAX * 2;i++) {
+		int x = rand()%(MAX*2);
+		int y = rand()%(MAX*2);
+		int temp = a[x];
+		a[x] = a[y];
+		a[y] = temp;
+	}
+	for (i=0;i<MAX;i++) {
+		map_insert(m, a[i], i);
+	}
+	for (i=0;i<MAX;i++) {
+		int id = map_search(m,a[i]);
+		assert(id == i);
+	}
+	for (i=0;i<MAX/2;i++) {
+		map_erase(m, a[i]);
+	}
+	for (i=0;i<MAX/2;i++) {
+		a[i] = a[i+MAX];
+		map_insert(m,a[i],i);
+	}
+	for (i=0;i<MAX;i++) {
+		int id = map_search(m,a[i]);
+		assert(id == i);
+	}
+}
+
+int
+main() {
+	struct map * m = map_new(MAX);
+	test(m);
+	map_delete(m);
+	return 0;
+}

gate/main.c

@@ -107,26 +107,20 @@ _report(struct skynet_context * ctx, const char * data, ...) {
 	char tmp[1024];
 	int n = vsnprintf(tmp, sizeof(tmp), data, ap);
 	va_end(ap);
-	if (n>=sizeof(tmp)) {
-		n = sizeof(tmp) - 1;
-		tmp[n] = '\0';
-	}
 
-	skynet_send(ctx, WATCHDOG, 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);
 	}
 }
 

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);
 

skynet.h

@@ -4,11 +4,13 @@
 #include <stddef.h>
 #include <stdint.h>
 
+#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);

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);

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)