Merge pull request #221 from cloudwu/dev

release 0.9.3

HISTORY.md

@@ -1,4 +1,17 @@
-v0.9.0 (2014-11-17)
+v0.9.3 (2014-1-5)
+-----------
+* Add : mongo createIndex
+* Update : sproto
+* bugfix : sharedata check dirty flag when len/pairs metamethod
+* bugfix : multicast
+
+v0.9.2 (2014-12-8)
+-----------
+* Simplify the message queue
+* Add create_index in mongo driver
+* Fix a bug in big-endian architecture (sproto)
+
+v0.9.0 / v0.9.1 (2014-11-17)
 -----------
 * Add UDP support
 * Add IPv6 support

lualib-src/lua-multicast.c

@@ -83,7 +83,7 @@ mc_unpacklocal(lua_State *L) {
 	if (sz != sizeof(*pack)) {
 		return luaL_error(L, "Invalid multicast package size %d", sz);
 	}
-	lua_settop(L, 1);
+	lua_pushlightuserdata(L, *pack);
 	lua_pushlightuserdata(L, (*pack)->data);
 	lua_pushunsigned(L, (*pack)->size);
 	return 3;
@@ -92,6 +92,8 @@ mc_unpacklocal(lua_State *L) {
 /*
 	lightuserdata struct mc_package **
 	integer reference
+
+	return mc_package *
  */
 static int
 mc_bindrefer(lua_State *L) {
@@ -102,16 +104,17 @@ mc_bindrefer(lua_State *L) {
 	}
 	(*pack)->reference = ref;
 
-	return 0;
+	lua_pushlightuserdata(L, *pack);
+
+	return 1;
 }
 
 /*
-	lightuserdata struct mc_package **
+	lightuserdata struct mc_package *
  */
 static int
 mc_closelocal(lua_State *L) {
-	struct mc_package **ptr = lua_touserdata(L,1);
-	struct mc_package *pack = *ptr;
+	struct mc_package *pack = lua_touserdata(L,1);
 
 	int ref = __sync_sub_and_fetch(&pack->reference, 1);
 	if (ref <= 0) {

lualib-src/lua-socket.c

@@ -9,6 +9,7 @@
 #include <lua.h>
 #include <lauxlib.h>
 
+#include <sys/socket.h>
 #include <arpa/inet.h>
 
 #include "skynet_socket.h"

lualib-src/sproto/lsproto.c

@@ -87,7 +87,7 @@ struct encode_ud {
 	int deep;
 };
 
-static int 
+static int
 encode(void *ud, const char *tagname, int type, int index, struct sproto_type *st, void *value, int length) {
 	struct encode_ud *self = ud;
 	lua_State *L = self->L;
@@ -189,7 +189,7 @@ expand_buffer(lua_State *L, int osz, int nsz) {
 	lightuserdata sproto_type
 	table source
 
-	return string 
+	return string
  */
 static int
 lencode(lua_State *L) {
@@ -229,7 +229,7 @@ struct decode_ud {
 	int deep;
 };
 
-static int 
+static int
 decode(void *ud, const char *tagname, int type, int index, struct sproto_type *st, void *value, int length) {
 	struct decode_ud * self = ud;
 	lua_State *L = self->L;
@@ -252,12 +252,12 @@ decode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
 	switch (type) {
 	case SPROTO_TINTEGER: {
 		// notice: in lua 5.2, 52bit integer support (not 64)
-		lua_Integer v = *(lua_Integer *)value;
+		lua_Integer v = *(uint64_t*)value;
 		lua_pushinteger(L, v);
 		break;
 	}
 	case SPROTO_TBOOLEAN: {
-		int v = *(lua_Integer*)value;
+		int v = *(uint64_t*)value;
 		lua_pushboolean(L,v);
 		break;
 	}

lualib-src/sproto/sproto.c

@@ -500,7 +500,7 @@ sproto_dump(struct sproto *s) {
 }
 
 // query
-int 
+int
 sproto_prototag(struct sproto *sp, const char * name) {
 	int i;
 	for (i=0;i<sp->protocol_n;i++) {
@@ -529,7 +529,7 @@ query_proto(struct sproto *sp, int tag) {
 	return NULL;
 }
 
-struct sproto_type * 
+struct sproto_type *
 sproto_protoquery(struct sproto *sp, int proto, int what) {
 	struct protocol * p;
 	if (what <0 || what >1) {
@@ -542,7 +542,7 @@ sproto_protoquery(struct sproto *sp, int proto, int what) {
 	return NULL;
 }
 
-const char * 
+const char *
 sproto_protoname(struct sproto *sp, int proto) {
 	struct protocol * p = query_proto(sp, proto);
 	if (p) {
@@ -551,7 +551,7 @@ sproto_protoname(struct sproto *sp, int proto) {
 	return NULL;
 }
 
-struct sproto_type * 
+struct sproto_type *
 sproto_type(struct sproto *sp, const char * type_name) {
 	int i;
 	for (i=0;i<sp->type_n;i++) {
@@ -806,7 +806,7 @@ encode_array(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int s
 	return fill_size(data, sz);
 }
 
-int 
+int
 sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback cb, void *ud) {
 	uint8_t * header = buffer;
 	uint8_t * data;
@@ -830,7 +830,7 @@ sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback c
 			sz = encode_array(cb,ud, f, data, size);
 		} else {
 			switch(type) {
-			case SPROTO_TINTEGER: 
+			case SPROTO_TINTEGER:
 			case SPROTO_TBOOLEAN: {
 				union {
 					uint64_t u64;
@@ -971,7 +971,7 @@ decode_array(sproto_callback cb, void *ud, struct field *f, uint8_t * stream) {
 	}
 	case SPROTO_TBOOLEAN:
 		for (i=0;i<sz;i++) {
-			int value = stream[i];
+			uint64_t value = stream[i];
 			cb(ud, f->name, SPROTO_TBOOLEAN, i+1, NULL, &value, sizeof(value));
 		}
 		break;
@@ -1050,7 +1050,7 @@ sproto_decode(struct sproto_type *st, const void * data, int size, sproto_callba
 					}
 					break;
 				}
-				case SPROTO_TSTRING: 
+				case SPROTO_TSTRING:
 				case SPROTO_TSTRUCT: {
 					uint32_t sz = todword(currentdata);
 					if (cb(ud, f->name, f->type, 0, f->st, currentdata+SIZEOF_LENGTH, sz))
@@ -1124,7 +1124,7 @@ write_ff(const uint8_t * src, uint8_t * des, int n) {
 	}
 }
 
-int 
+int
 sproto_pack(const void * srcv, int srcsz, void * bufferv, int bufsz) {
 	uint8_t tmp[8];
 	int i;
@@ -1181,7 +1181,7 @@ sproto_pack(const void * srcv, int srcsz, void * bufferv, int bufsz) {
 	return size;
 }
 
-int 
+int
 sproto_unpack(const void * srcv, int srcsz, void * bufferv, int bufsz) {
 	const uint8_t * src = srcv;
 	uint8_t * buffer = bufferv;

lualib/mongo.lua

@@ -120,6 +120,8 @@ function mongo.client( conf	)
 	local obj =	{
 		host = first.host,
 		port = first.port or 27017,
+		username = first.username,
+		password = first.password,
 	}
 
 	obj.__id = 0
@@ -227,7 +229,11 @@ function mongo_collection:insert(doc)
 	sock:request(pack)
 end
 
-function mongo_collection:batch_insert(docs)
+function mongo_collection:safe_insert(doc)
+	return self.database:runCommand("insert", self.name, "documents", {bson_encode(doc)})	
+end
+
+function mongo_collection:batch_insert(docs)		
 	for	i=1,#docs do
 		if docs[i]._id == nil then
 			docs[i]._id	= bson.objectid()
@@ -276,6 +282,62 @@ function mongo_collection:find(query, selector)
 	} ,	cursor_meta)
 end
 
+-- collection:createIndex({username = 1}, {unique = true})
+function mongo_collection:createIndex(keys, option)
+	local name = option.name
+	option.name = nil
+
+	if not name then
+		for k, v in pairs(keys) do
+			name = (name == nil) and k or (name .. "_" .. k)
+			name = name  .. "_" .. v
+		end		
+	end
+
+
+	local doc = {};
+	doc.name = name
+	doc.key = keys
+	for k, v in pairs(option) do
+		doc[k] = v
+	end
+	return self.database:runCommand("createIndexes", self.name, "indexes", {doc})
+end
+
+mongo_collection.ensureIndex = mongo_collection.createIndex;
+
+
+function mongo_collection:drop()
+	return self.database:runCommand("drop", self.name)
+end
+
+-- collection:dropIndex("age_1")
+-- collection:dropIndex("*")
+function mongo_collection:dropIndex(indexName)
+	return self.database:runCommand("dropIndexes", self.name, "index", indexName)
+end
+
+-- collection:findAndModify({query = {name = "userid"}, update = {["$inc"] = {nextid = 1}}, })
+-- keys, value type
+-- query, table	
+-- sort, table
+-- remove, bool 
+-- update, table 
+-- new, bool 
+-- fields, bool 
+-- upsert, boolean 
+function mongo_collection:findAndModify(doc)
+	assert(doc.query)
+	assert(doc.update or doc.remove)
+
+	local cmd = {"findAndModify", self.name};
+	for k, v in pairs(doc) do
+		table.insert(cmd, k)
+		table.insert(cmd, v)
+	end
+	return self.database:runCommand(unpack(cmd))
+end
+
 function mongo_cursor:hasNext()
 	if self.__ptr == nil then
 		if self.__document == nil then

lualib/sharedata/corelib.lua

@@ -53,7 +53,7 @@ local function genkey(self)
 	return key
 end
 
-function meta:__index(key)
+local function getcobj(self)
 	local obj = self.__obj
 	if isdirty(obj) then
 		local newobj, newtbl = needupdate(self.__gcobj)
@@ -67,6 +67,11 @@ function meta:__index(key)
 			obj = self.__obj
 		end
 	end
+	return obj
+end
+
+function meta:__index(key)
+	local obj = getcobj(self)
 	local v = index(obj, key)
 	if type(v) == "userdata" then
 		local r = setmetatable({
@@ -83,11 +88,11 @@ function meta:__index(key)
 end
 
 function meta:__len()
-	return len(self.__obj)
+	return len(getcobj(self))
 end
 
 local function conf_ipairs(self, index)
-	local obj = self.__obj
+	local obj = getcobj(self)
 	index = index + 1
 	local value =  rawget(self, index)
 	if value then
@@ -109,7 +114,8 @@ function meta:__pairs()
 end
 
 function conf.next(obj, key)
-	local nextkey = core.nextkey(obj.__obj, key)
+	local cobj = getcobj(obj)
+	local nextkey = core.nextkey(cobj, key)
 	if nextkey then
 		return nextkey, obj[nextkey]
 	end

lualib/socket.lua

@@ -58,7 +58,7 @@ socket_message[1] = function(id, size, data)
 		end
 	else
 		if s.buffer_limit and sz > s.buffer_limit then
-			skynet.error(string.format("socket buffer overlow: fd=%d size=%d", id , sz))
+			skynet.error(string.format("socket buffer overflow: fd=%d size=%d", id , sz))
 			driver.clear(s.buffer,buffer_pool)
 			driver.close(id)
 			return

lualib/sprotoparser.lua

@@ -313,9 +313,12 @@ local function packgroup(t,p)
 	local tt, tp
 	local alltypes = {}
 	for name in pairs(t) do
-		alltypes[name] = #alltypes
 		table.insert(alltypes, name)
 	end
+	table.sort(alltypes)	-- make result stable
+	for idx, name in ipairs(alltypes) do
+		alltypes[name] = idx - 1
+	end
 	tt = {}
 	for _,name in ipairs(alltypes) do
 		table.insert(tt, packtype(name, t[name], alltypes))

service/multicastd.lua

@@ -68,18 +68,20 @@ end
 local function publish(c , source, pack, size)
 	local group = channel[c]
 	if group == nil then
-		-- dead channel, delete the pack
-		mc.bind(pack, 1)
+		-- dead channel, delete the pack. mc.bind returns the pointer in pack
+		local pack = mc.bind(pack, 1)
 		mc.close(pack)
 		return
 	end
 	mc.bind(pack, channel_n[c])
 	local msg = skynet.tostring(pack, size)
 	for k in pairs(group) do
+		-- the msg is a pointer to the real message, publish pointer in local is ok.
 		skynet.redirect(k, source, "multicast", c , msg)
 	end
 	local remote = channel_remote[c]
 	if remote then
+		-- remote publish should unpack the pack, because we should not publish the pointer out.
 		local _, msg, sz = mc.unpack(pack, size)
 		local msg = skynet.tostring(msg,sz)
 		for node in pairs(remote) do

skynet-src/skynet_mq.c

@@ -32,12 +32,9 @@ struct message_queue {
 };
 
 struct global_queue {
-	uint32_t head;
-	uint32_t tail;
-	struct message_queue ** queue;
-	// We use a separated flag array to ensure the mq is pushed.
-	// See the comments below.
-	struct message_queue *list;
+	struct message_queue *head;
+	struct message_queue *tail;
+	int lock;
 };
 
 static struct global_queue *Q = NULL;
@@ -45,63 +42,36 @@ static struct global_queue *Q = NULL;
 #define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
 #define UNLOCK(q) __sync_lock_release(&(q)->lock);
 
-#define GP(p) ((p) % MAX_GLOBAL_MQ)
-
 void 
 skynet_globalmq_push(struct message_queue * queue) {
 	struct global_queue *q= Q;
 
-	uint32_t tail = GP(__sync_fetch_and_add(&q->tail,1));
-
-	// only one thread can set the slot (change q->queue[tail] from NULL to queue)
-	if (!__sync_bool_compare_and_swap(&q->queue[tail], NULL, queue)) {
-		// The queue may full seldom, save queue in list
-		assert(queue->next == NULL);
-		struct message_queue * last;
-		do {
-			last = q->list;
-			queue->next = last;
-		} while(!__sync_bool_compare_and_swap(&q->list, last, queue));
-
-		return;
+	LOCK(q)
+	assert(queue->next == NULL);
+	if(q->tail) {
+		q->tail->next = queue;
+		q->tail = queue;
+	} else {
+		q->head = q->tail = queue;
 	}
+	UNLOCK(q)
 }
 
 struct message_queue * 
 skynet_globalmq_pop() {
 	struct global_queue *q = Q;
-	uint32_t head =  q->head;
 
-	if (head == q->tail) {
-		// The queue is empty.
-		return NULL;
-	}
-
-	uint32_t head_ptr = GP(head);
-
-	struct message_queue * list = q->list;
-	if (list) {
-		// If q->list is not empty, try to load it back to the queue
-		struct message_queue *newhead = list->next;
-		if (__sync_bool_compare_and_swap(&q->list, list, newhead)) {
-			// try load list only once, if success , push it back to the queue.
-			list->next = NULL;
-			skynet_globalmq_push(list);
+	LOCK(q)
+	struct message_queue *mq = q->head;
+	if(mq) {
+		q->head = mq->next;
+		if(q->head == NULL) {
+			assert(mq == q->tail);
+			q->tail = NULL;
 		}
+		mq->next = NULL;
 	}
-
-	struct message_queue * mq = q->queue[head_ptr];
-	if (mq == NULL) {
-		// globalmq push not complete
-		return NULL;
-	}
-	if (!__sync_bool_compare_and_swap(&q->head, head, head+1)) {
-		return NULL;
-	}
-	// only one thread can get the slot (change q->queue[head_ptr] to NULL)
-	if (!__sync_bool_compare_and_swap(&q->queue[head_ptr], mq, NULL)) {
-		return NULL;
-	}
+	UNLOCK(q)
 
 	return mq;
 }
@@ -243,8 +213,6 @@ void
 skynet_mq_init() {
 	struct global_queue *q = skynet_malloc(sizeof(*q));
 	memset(q,0,sizeof(*q));
-	q->queue = skynet_malloc(MAX_GLOBAL_MQ * sizeof(struct message_queue *));
-	memset(q->queue, 0, sizeof(struct message_queue *) * MAX_GLOBAL_MQ);
 	Q=q;
 }
 

skynet-src/socket_server.c

@@ -628,7 +628,7 @@ append_sendbuffer_(struct socket_server *ss, struct wb_list *s, struct request_s
 	struct write_buffer * buf = MALLOC(size);
 	struct send_object so;
 	buf->userobject = send_object_init(ss, &so, request->buffer, request->sz);
-	buf->ptr = so.buffer+n;
+	buf->ptr = (char*)so.buffer+n;
 	buf->sz = so.sz - n;
 	buf->buffer = request->buffer;
 	buf->next = NULL;

test/testdeadloop.lua

@@ -0,0 +1,10 @@
+local skynet = require "skynet"
+local function dead_loop()
+    while true do
+        skynet.sleep(0)
+    end
+end
+
+skynet.start(function()
+    skynet.fork(dead_loop)
+end)

test/testmongodb.lua

@@ -0,0 +1,90 @@
+local skynet = require "skynet"
+local mongo = require "mongo"
+local bson = require "bson"
+
+local host, db_name = ...
+
+function test_insert_without_index()
+	local db = mongo.client({host = host})
+
+	db[db_name].testdb:dropIndex("*")
+	db[db_name].testdb:drop()
+
+	local ret = db[db_name].testdb:safe_insert({test_key = 1});
+	assert(ret and ret.n == 1)
+
+	local ret = db[db_name].testdb:safe_insert({test_key = 1});
+	assert(ret and ret.n == 1)
+end
+
+function test_insert_with_index()
+	local db = mongo.client({host = host})
+
+	db[db_name].testdb:dropIndex("*")
+	db[db_name].testdb:drop()
+
+	db[db_name].testdb:ensureIndex({test_key = 1}, {unique = true, name = "test_key_index"})
+
+	local ret = db[db_name].testdb:safe_insert({test_key = 1})
+	assert(ret and ret.n == 1)
+
+	local ret = db[db_name].testdb:safe_insert({test_key = 1})
+	assert(ret and ret.n == 0)
+end
+
+function test_find_and_remove()
+	local db = mongo.client({host = host})
+
+	db[db_name].testdb:dropIndex("*")
+	db[db_name].testdb:drop()
+
+	db[db_name].testdb:ensureIndex({test_key = 1}, {unique = true, name = "test_key_index"})
+
+	local ret = db[db_name].testdb:safe_insert({test_key = 1})
+	assert(ret and ret.n == 1)
+
+	local ret = db[db_name].testdb:findOne({test_key = 1})
+	assert(ret and ret.test_key == 1)
+
+	db[db_name].testdb:delete({test_key = 1})
+
+	local ret = db[db_name].testdb:findOne({test_key = 1})
+	assert(ret == nil)
+end
+
+
+function test_expire_index()
+	local db = mongo.client({host = host})
+
+	db[db_name].testdb:dropIndex("*")
+	db[db_name].testdb:drop()
+
+	db[db_name].testdb:ensureIndex({test_key = 1}, {unique = true, name = "test_key_index", expireAfterSeconds = 1, })
+	db[db_name].testdb:ensureIndex({test_date = 1}, {expireAfterSeconds = 1, })
+
+	local ret = db[db_name].testdb:safe_insert({test_key = 1, test_date = bson.date(os.time())})
+	assert(ret and ret.n == 1)
+
+	local ret = db[db_name].testdb:findOne({test_key = 1})
+	assert(ret and ret.test_key == 1)
+
+	for i = 1, 1000 do
+		skynet.sleep(11);
+		
+		local ret = db[db_name].testdb:findOne({test_key = 1})
+		if ret == nil then
+			return
+		end
+	end
+
+	assert(false, "test expire index failed");
+end
+
+skynet.start(function()
+	test_insert_without_index()
+	test_insert_with_index()
+	test_find_and_remove()
+	test_expire_index()
+
+	print("mongodb test finish.");
+end)

test/testmysql.lua

@@ -63,14 +63,14 @@ local function test3( db)
         local    res = db:query("select * from cats order by id asc")
         print ( "test3 loop times=" ,i,"\n","query result=",dump( res ) )
         res = db:query("select * from cats order by id asc")
-        print ( "test3 loop times=" ,i,"\n","query result=",dump( res ) )        
+        print ( "test3 loop times=" ,i,"\n","query result=",dump( res ) )
         skynet.sleep(1000)
         i=i+1
     end
 end
 skynet.start(function()
 
-	local db=mysql.connect{	
+	local db=mysql.connect{
 		host="127.0.0.1",
 		port=3306,
 		database="skynet",
@@ -83,8 +83,10 @@ skynet.start(function()
 	end
 	print("testmysql success to connect to mysql server")
 
+	db:query("set names utf8")
+
 	local res = db:query("drop table if exists cats")
-	res = db:query("create table cats " 
+	res = db:query("create table cats "
 		               .."(id serial primary key, ".. "name varchar(5))")
 	print( dump( res ) )
 
@@ -112,7 +114,7 @@ skynet.start(function()
     while true do
         local    res = db:query("select * from cats order by id asc")
         print ( "test1 loop times=" ,i,"\n","query result=",dump( res ) )
-      
+
         res = db:query("select * from cats order by id asc")
         print ( "test1 loop times=" ,i,"\n","query result=",dump( res ) )