Merge pull request #325 from cloudwu/pthreadlock

add pthread lock

lualib-src/lua-bson.c

@@ -8,6 +8,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <stdbool.h>
+#include "atomic.h"
 
 #define DEFAULT_CAP 64
 #define MAX_NUMBER 1024
@@ -1140,7 +1141,7 @@ lobjectid(lua_State *L) {
 	} else {
 		time_t ti = time(NULL);
 		// old_counter is a static var, use atom inc.
-		uint32_t id = __sync_fetch_and_add(&oid_counter,1);
+		uint32_t id = ATOM_FINC(&oid_counter);
 
 		oid[2] = (ti>>24) & 0xff;
 		oid[3] = (ti>>16) & 0xff;

lualib-src/lua-clientsocket.c

@@ -128,11 +128,8 @@ lusleep(lua_State *L) {
 
 #define QUEUE_SIZE 1024
 
-#define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
-#define UNLOCK(q) __sync_lock_release(&(q)->lock);
-
 struct queue {
-	int lock;
+	pthread_mutex_t lock;
 	int head;
 	int tail;
 	char * queue[QUEUE_SIZE];
@@ -153,7 +150,7 @@ readline_stdin(void * arg) {
 		memcpy(str, tmp, n);
 		str[n] = 0;
 
-		LOCK(q);
+		pthread_mutex_lock(&q->lock);
 		q->queue[q->tail] = str;
 
 		if (++q->tail >= QUEUE_SIZE) {
@@ -163,7 +160,7 @@ readline_stdin(void * arg) {
 			// queue overflow
 			exit(1);
 		}
-		UNLOCK(q);
+		pthread_mutex_unlock(&q->lock);
 	}
 	return NULL;
 }
@@ -171,16 +168,16 @@ readline_stdin(void * arg) {
 static int
 lreadstdin(lua_State *L) {
 	struct queue *q = lua_touserdata(L, lua_upvalueindex(1));
-	LOCK(q);
+	pthread_mutex_lock(&q->lock);
 	if (q->head == q->tail) {
-		UNLOCK(q);
+		pthread_mutex_unlock(&q->lock);
 		return 0;
 	}
 	char * str = q->queue[q->head];
 	if (++q->head >= QUEUE_SIZE) {
 		q->head = 0;
 	}
-	UNLOCK(q);
+	pthread_mutex_unlock(&q->lock);
 	lua_pushstring(L, str);
 	free(str);
 	return 1;
@@ -201,6 +198,7 @@ luaopen_clientsocket(lua_State *L) {
 
 	struct queue * q = lua_newuserdata(L, sizeof(*q));
 	memset(q, 0, sizeof(*q));
+	pthread_mutex_init(&q->lock, NULL);
 	lua_pushcclosure(L, lreadstdin, 1);
 	lua_setfield(L, -2, "readstdin");
 

lualib-src/lua-debugchannel.c

@@ -4,10 +4,9 @@
 #include <unistd.h>
 #include <stdlib.h>
 #include <string.h>
+#include "spinlock.h"
 
 #define METANAME "debugchannel"
-#define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
-#define UNLOCK(q) __sync_lock_release(&(q)->lock);
 
 struct command {
 	struct command * next;
@@ -15,7 +14,7 @@ struct command {
 };
 
 struct channel {
-	int lock;
+	struct spinlock lock;
 	int ref;
 	struct command * head;
 	struct command * tail;
@@ -26,6 +25,7 @@ channel_new() {
 	struct channel * c = malloc(sizeof(*c));
 	memset(c, 0 , sizeof(*c));
 	c->ref = 1;
+	SPIN_INIT(c)
 
 	return c;
 }
@@ -33,21 +33,21 @@ channel_new() {
 static struct channel *
 channel_connect(struct channel *c) {
 	struct channel * ret = NULL;
-	LOCK(c)
+	SPIN_LOCK(c)
 	if (c->ref == 1) {
 		++c->ref;
 		ret = c;
 	}
-	UNLOCK(c)
+	SPIN_UNLOCK(c)
 	return ret;
 }
 
 static struct channel *
 channel_release(struct channel *c) {
-	LOCK(c)
+	SPIN_LOCK(c)
 	--c->ref;
 	if (c->ref > 0) {
-		UNLOCK(c)
+		SPIN_UNLOCK(c)
 		return c;
 	}
 	// never unlock while reference is 0
@@ -59,6 +59,8 @@ channel_release(struct channel *c) {
 		free(p);
 		p = next;
 	}
+	SPIN_UNLOCK(c)
+	SPIN_DESTROY(c)
 	free(c);
 	return NULL;
 }
@@ -67,9 +69,9 @@ channel_release(struct channel *c) {
 static struct command *
 channel_read(struct channel *c, double timeout) {
 	struct command * ret = NULL;
-	LOCK(c)
+	SPIN_LOCK(c)
 	if (c->head == NULL) {
-		UNLOCK(c)
+		SPIN_UNLOCK(c)
 		int ti = (int)(timeout * 100000);
 		usleep(ti);
 		return NULL;
@@ -79,7 +81,7 @@ channel_read(struct channel *c, double timeout) {
 	if (c->head == NULL) {
 		c->tail = NULL;
 	}
-	UNLOCK(c)
+	SPIN_UNLOCK(c)
 	
 	return ret;
 }
@@ -90,14 +92,14 @@ channel_write(struct channel *c, const char * s, size_t sz) {
 	cmd->sz = sz;
 	cmd->next = NULL;
 	memcpy(cmd+1, s, sz);
-	LOCK(c)
+	SPIN_LOCK(c)
 	if (c->tail == NULL) {
 		c->head = c->tail = cmd;
 	} else {
 		c->tail->next = cmd;
 		c->tail = cmd;
 	}
-	UNLOCK(c)
+	SPIN_UNLOCK(c)
 }
 
 struct channel_box {

lualib-src/lua-multicast.c

@@ -5,6 +5,8 @@
 #include <stdint.h>
 #include <string.h>
 
+#include "atomic.h"
+
 struct mc_package {
 	int reference;
 	uint32_t size;
@@ -116,7 +118,7 @@ static int
 mc_closelocal(lua_State *L) {
 	struct mc_package *pack = lua_touserdata(L,1);
 
-	int ref = __sync_sub_and_fetch(&pack->reference, 1);
+	int ref = ATOM_DEC(&pack->reference);
 	if (ref <= 0) {
 		skynet_free(pack->data);
 		skynet_free(pack);

lualib-src/lua-sharedata.c

@@ -4,6 +4,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <assert.h>
+#include "atomic.h"
 
 #define KEYTYPE_INTEGER 0
 #define KEYTYPE_STRING 1
@@ -663,7 +664,7 @@ releaseobj(lua_State *L) {
 	struct ctrl *c = lua_touserdata(L, 1);
 	struct table *tbl = c->root;
 	struct state *s = lua_touserdata(tbl->L, 1);
-	__sync_fetch_and_sub(&s->ref, 1);
+	ATOM_DEC(&s->ref);
 	c->root = NULL;
 	c->update = NULL;
 
@@ -674,7 +675,7 @@ static int
 lboxconf(lua_State *L) {
 	struct table * tbl = get_table(L,1);	
 	struct state * s = lua_touserdata(tbl->L, 1);
-	__sync_fetch_and_add(&s->ref, 1);
+	ATOM_INC(&s->ref);
 
 	struct ctrl * c = lua_newuserdata(L, sizeof(*c));
 	c->root = tbl;
@@ -719,7 +720,7 @@ static int
 lincref(lua_State *L) {
 	struct table *tbl = get_table(L,1);
 	struct state * s = lua_touserdata(tbl->L, 1);
-	int ref = __sync_add_and_fetch(&s->ref, 1);
+	int ref = ATOM_INC(&s->ref);
 	lua_pushinteger(L , ref);
 
 	return 1;
@@ -729,7 +730,7 @@ static int
 ldecref(lua_State *L) {
 	struct table *tbl = get_table(L,1);
 	struct state * s = lua_touserdata(tbl->L, 1);
-	int ref = __sync_sub_and_fetch(&s->ref, 1);
+	int ref = ATOM_DEC(&s->ref);
 	lua_pushinteger(L , ref);
 
 	return 1;

lualib-src/lua-stm.c

@@ -7,6 +7,7 @@
 
 #include "rwlock.h"
 #include "skynet_malloc.h"
+#include "atomic.h"
 
 struct stm_object {
 	struct rwlock lock;
@@ -45,7 +46,7 @@ static void
 stm_releasecopy(struct stm_copy *copy) {
 	if (copy == NULL)
 		return;
-	if (__sync_sub_and_fetch(&copy->reference, 1) == 0) {
+	if (ATOM_DEC(&copy->reference) == 0) {
 		skynet_free(copy->msg);
 		skynet_free(copy);
 	}
@@ -70,7 +71,7 @@ stm_release(struct stm_object *obj) {
 static void
 stm_releasereader(struct stm_object *obj) {
 	rwlock_rlock(&obj->lock);
-	if (__sync_sub_and_fetch(&obj->reference,1) == 0) {
+	if (ATOM_DEC(&obj->reference) == 0) {
 		// last reader, no writer. so no need to unlock
 		assert(obj->copy == NULL);
 		skynet_free(obj);
@@ -82,7 +83,7 @@ stm_releasereader(struct stm_object *obj) {
 static void
 stm_grab(struct stm_object *obj) {
 	rwlock_rlock(&obj->lock);
-	int ref = __sync_fetch_and_add(&obj->reference,1);
+	int ref = ATOM_FINC(&obj->reference);
 	rwlock_runlock(&obj->lock);
 	assert(ref > 0);
 }
@@ -92,7 +93,7 @@ stm_copy(struct stm_object *obj) {
 	rwlock_rlock(&obj->lock);
 	struct stm_copy * ret = obj->copy;
 	if (ret) {
-		int ref = __sync_fetch_and_add(&ret->reference,1);
+		int ref = ATOM_FINC(&ret->reference);
 		assert(ref > 0);
 	}
 	rwlock_runlock(&obj->lock);