Fixed bugs in fiber mutex and fiber cond.

Author: zhengshuxin

c/include/fiber/fiber_mutex.h

@@ -10,7 +10,8 @@ extern "C" {
 typedef struct ACL_FIBER_MUTEX ACL_FIBER_MUTEX;
 
 #define	FIBER_MUTEX_F_LOCK_TRY		(1 << 0)
-#define	FIBER_MUTEX_F_SWITCH_FIRST	(1 << 1)
+#define	FIBER_MUTEX_F_LOCK_ONCE		(1 << 1)
+#define	FIBER_MUTEX_F_CHECK_DEADLOCK	(1 << 2)
 
 FIBER_API ACL_FIBER_MUTEX *acl_fiber_mutex_create(unsigned flag);
 FIBER_API void acl_fiber_mutex_free(ACL_FIBER_MUTEX *mutex);

c/src/common/mbox.c

@@ -105,6 +105,11 @@ void mbox_free(MBOX *mbox, void (*free_fn)(void*))
 }
 
 int mbox_send(MBOX *mbox, void *msg)
+{
+	return mbox_send2(mbox, mbox->out, msg);
+}
+
+int mbox_send2(MBOX *mbox, socket_t out, void *msg)
 {
 	int ret;
 	long long n = 1;
@@ -134,9 +139,9 @@ int mbox_send(MBOX *mbox, void *msg)
 	mbox->nsend++;
 
 #if defined(_WIN32) || defined(_WIN64)
-	ret = (int) acl_fiber_send(mbox->out, (const char*) &n, (int) sizeof(n), 0);
+	ret = (int) acl_fiber_send(out, (const char*) &n, (int) sizeof(n), 0);
 #else
-	ret = (int) acl_fiber_write(mbox->out, &n, sizeof(n));
+	ret = (int) acl_fiber_write(out, &n, sizeof(n));
 #endif
 
 #if !defined(HAS_EVENTFD)
@@ -147,7 +152,7 @@ int mbox_send(MBOX *mbox, void *msg)
 
 	if (ret == -1) {
 		msg_error("%s(%d), %s: mbox write %d error %s", __FILE__,
-			__LINE__, __FUNCTION__, mbox->out, last_serror());
+			__LINE__, __FUNCTION__, (int) out, last_serror());
 		return -1;
 	}
 

c/src/common/mbox.h

@@ -27,6 +27,7 @@ void mbox_free(MBOX *mbox, void (*free_fn)(void*));
  * @return {int} 发送成功返回 0，否则返回 -1
  */
 int mbox_send(MBOX *mbox, void *msg);
+int mbox_send2(MBOX *mbox, socket_t out, void *msg);
 
 /**
  * 从消息队列中读取消息

c/src/event.h

@@ -130,6 +130,7 @@ struct FILE_EVENT {
 #define	STATUS_WRITEWAIT	(unsigned) (1 << 6)	// Wait for Writable
 #define	STATUS_CLOSING		(unsigned) (1 << 7)	// In closing status
 #define	STATUS_CLOSED		(unsigned) (1 << 8)	// In closed status
+#define STATUS_BUFFED		(unsigned) (1 << 9)
 
 #define	SET_CONNECTING(x)	((x)->status |= STATUS_CONNECTING)
 #define	SET_READABLE(x)		((x)->status |= STATUS_READABLE)

c/src/fiber_io.c

@@ -178,7 +178,7 @@ void fiber_timer_add(ACL_FIBER *fb, size_t milliseconds)
 	long long now = event_get_stamp(ev);
 	TIMER_CACHE_NODE *timer;
 
-	fb->when = now + (ssize_t) milliseconds;
+	fb->when = now + (long long) milliseconds;
 	ring_detach(&fb->me);  // Detach the previous binding.
 	timer_cache_add(__thread_fiber->ev_timer, fb->when, &fb->me);
 
@@ -849,6 +849,7 @@ FILE_EVENT *fiber_file_cache_get(socket_t fd)
 		fe = file_event_alloc(fd);
 	} else {
 		file_event_init(fe, fd);
+		fe->status &= ~STATUS_BUFFED;
 	}
 
 #ifdef	HAS_IO_URING
@@ -862,11 +863,22 @@ FILE_EVENT *fiber_file_cache_get(socket_t fd)
 
 void fiber_file_cache_put(FILE_EVENT *fe)
 {
-	fiber_file_del(fe, fe->fd);
-	fe->fd = INVALID_SOCKET;
+	// If the fe is being refered by more than one fibers, don't put it
+	// back to cache or free it, because the fe is still aliving now.
+	if (fe->refer > 1) {
+		return;
+	}
+
+	if (fe->fd != INVALID_SOCKET) {
+		fiber_file_del(fe, fe->fd);
+		fe->fd = INVALID_SOCKET;
+	}
 
 	if (array_size(__thread_fiber->cache) < __thread_fiber->cache_max) {
-		array_push_back(__thread_fiber->cache, fe);
+		if (!(fe->status & STATUS_BUFFED)) {
+			array_push_back(__thread_fiber->cache, fe);
+			fe->status |= STATUS_BUFFED;
+		}
 	} else {
 		file_event_unrefer(fe);
 	}

c/src/sync/fiber_cond.c

@@ -144,7 +144,7 @@ static int fiber_cond_timedwait(ACL_FIBER_COND *cond, ACL_FIBER_MUTEX *mutex,
 
 	WAITER_INC(ev);
 
-	// Hang the current fiber and will wakeup if the timer arrives or
+	// Hang the current fiber and will wake up if the timer arrives or
 	// be awakened by the other fiber or thread.
 	acl_fiber_switch();
 

c/src/sync/fiber_mutex.c

@@ -373,20 +373,34 @@ void acl_fiber_mutex_stats_show(const ACL_FIBER_MUTEX_STATS *stats)
 
 /****************************************************************************/
 
+#if 1
+#define	LOCK(m) while(pthread_mutex_trylock(&(m)->lock) != 0) {}
+#else
+#define	LOCK(m) do { \
+	pthread_mutex_lock(&(m)->lock); \
+} while (0)
+#endif
+
+#define	UNLOCK(m) do { \
+	pthread_mutex_unlock(&(m)->lock); \
+} while (0)
+
+// Just for checking deadlock.
 static void thread_waiter_add(ACL_FIBER_MUTEX *mutex, unsigned long tid)
 {
 	THREAD_WAITER *waiter = (THREAD_WAITER*) mem_malloc(sizeof(THREAD_WAITER));
 	waiter->tid = tid;
-	pthread_mutex_lock(&mutex->lock);
+	LOCK(mutex);
 	array_append(mutex->waiting_threads, waiter);
-	pthread_mutex_unlock(&mutex->lock);
+	UNLOCK(mutex);
 }
 
+// Just for checking deadlock.
 static void thread_waiter_remove(ACL_FIBER_MUTEX *mutex, unsigned long tid)
 {
 	ITER iter;
 
-	pthread_mutex_lock(&mutex->lock);
+	LOCK(mutex);
 	foreach(iter, mutex->waiting_threads) {
 		THREAD_WAITER *waiter = (THREAD_WAITER*) iter.data;
 		if (waiter->tid == tid) {
@@ -395,7 +409,7 @@ static void thread_waiter_remove(ACL_FIBER_MUTEX *mutex, unsigned long tid)
 			break;
 		}
 	}
-	pthread_mutex_unlock(&mutex->lock);
+	UNLOCK(mutex);
 }
 
 static void free_locks_onexit(void)
@@ -427,6 +441,12 @@ ACL_FIBER_MUTEX *acl_fiber_mutex_create(unsigned flags)
 	mutex = (ACL_FIBER_MUTEX*) mem_calloc(1, sizeof(ACL_FIBER_MUTEX));
 	ring_init(&mutex->me);
 
+	if (flags & FIBER_MUTEX_F_LOCK_ONCE) {
+		flags &= ~FIBER_MUTEX_F_LOCK_TRY;
+	} else {
+		flags |= FIBER_MUTEX_F_LOCK_TRY;
+	}
+
 	mutex->flags  = flags;
 
 	mutex->waiters = array_create(5, ARRAY_F_UNORDER);
@@ -456,21 +476,23 @@ void acl_fiber_mutex_free(ACL_FIBER_MUTEX *mutex)
 
 static int fiber_mutex_lock_once(ACL_FIBER_MUTEX *mutex)
 {
-	int wakeup = 0, pos;
+	int pos;
 	EVENT *ev;
 	ACL_FIBER *fiber;
 
 	while (1) {
-		pthread_mutex_lock(&mutex->lock);
+		LOCK(mutex);
 		if (pthread_mutex_trylock(&mutex->thread_lock) == 0) {
-			pthread_mutex_unlock(&mutex->lock);
+			UNLOCK(mutex);
 			return 0;
 		}
 
 		// For the independent thread, only lock the thread mutex.
 		if (!var_hook_sys_api) {
-			pthread_mutex_unlock(&mutex->lock);
-			thread_waiter_add(mutex, thread_self());
+			UNLOCK(mutex);
+			if (mutex->flags & FIBER_MUTEX_F_CHECK_DEADLOCK) {
+				thread_waiter_add(mutex, thread_self());
+			}
 			return pthread_mutex_lock(&mutex->thread_lock);
 		}
 
@@ -481,11 +503,11 @@ static int fiber_mutex_lock_once(ACL_FIBER_MUTEX *mutex)
 
 		if (pthread_mutex_trylock(&mutex->thread_lock) == 0) {
 			array_delete(mutex->waiters, pos, NULL);
-			pthread_mutex_unlock(&mutex->lock);
+			UNLOCK(mutex);
 			return 0;
 		}
 
-		pthread_mutex_unlock(&mutex->lock);
+		UNLOCK(mutex);
 
 		fiber->wstatus |= FIBER_WAIT_MUTEX;
 
@@ -495,19 +517,14 @@ static int fiber_mutex_lock_once(ACL_FIBER_MUTEX *mutex)
 		WAITER_DEC(ev);
 
 		fiber->wstatus &= ~FIBER_WAIT_MUTEX;
-
-		if (++wakeup > 5) {
-			wakeup = 0;
-			acl_fiber_delay(100);
-		}
 	}
 }
 
 static int fiber_mutex_lock_try(ACL_FIBER_MUTEX *mutex)
 {
-	int wakeup = 0, pos;
+	int pos;
 	EVENT *ev;
-	ACL_FIBER *fiber = acl_fiber_running();
+	ACL_FIBER *fiber;
 
 	while (1) {
 		if (pthread_mutex_trylock(&mutex->thread_lock) == 0) {
@@ -516,22 +533,25 @@ static int fiber_mutex_lock_try(ACL_FIBER_MUTEX *mutex)
 
 		// For the independent thread, only lock the thread mutex.
 		if (!var_hook_sys_api) {
-			thread_waiter_add(mutex, thread_self());
+			if (mutex->flags & FIBER_MUTEX_F_CHECK_DEADLOCK) {
+				thread_waiter_add(mutex, thread_self());
+			}
 			return pthread_mutex_lock(&mutex->thread_lock);
 		}
 
+		fiber = acl_fiber_running();
 		fiber->sync = sync_waiter_get();
 
-		pthread_mutex_lock(&mutex->lock);
+		LOCK(mutex);
 		pos = array_append(mutex->waiters, fiber);
 
 		if (pthread_mutex_trylock(&mutex->thread_lock) == 0) {
 			array_delete(mutex->waiters, pos, NULL);
-			pthread_mutex_unlock(&mutex->lock);
+			UNLOCK(mutex);
 			return 0;
 		}
 
-		pthread_mutex_unlock(&mutex->lock);
+		UNLOCK(mutex);
 
 		fiber->wstatus |= FIBER_WAIT_MUTEX;
 
@@ -541,28 +561,24 @@ static int fiber_mutex_lock_try(ACL_FIBER_MUTEX *mutex)
 		WAITER_DEC(ev);
 
 		fiber->wstatus &= ~FIBER_WAIT_MUTEX;
-
-		if (++wakeup > 5) {
-			wakeup = 0;
-			acl_fiber_delay(100);
-		}
 	}
 }
 
 int acl_fiber_mutex_lock(ACL_FIBER_MUTEX *mutex)
 {
 	int ret;
 
-	if (mutex->flags & FIBER_MUTEX_F_LOCK_TRY) {
-		ret = fiber_mutex_lock_try(mutex);
-	} else {
+	if (mutex->flags & FIBER_MUTEX_F_LOCK_ONCE) {
 		ret = fiber_mutex_lock_once(mutex);
+	} else {
+		ret = fiber_mutex_lock_try(mutex);
 	}
+
 	if (ret == 0) {
 		unsigned id = acl_fiber_self();  // 0 will return in no fiber mode.
 		long me = id == 0 ? -thread_self() : (long) id;
 		mutex->owner = me;
-		if (me < 0) {
+		if (me < 0 && (mutex->flags & FIBER_MUTEX_F_CHECK_DEADLOCK)) {
 			thread_waiter_remove(mutex, thread_self());
 		}
 
@@ -588,7 +604,7 @@ int acl_fiber_mutex_unlock(ACL_FIBER_MUTEX *mutex)
 	ACL_FIBER *fiber;
 	int ret;
 
-	pthread_mutex_lock(&mutex->lock);
+	LOCK(mutex);
 	fiber = (ACL_FIBER*) array_pop_front(mutex->waiters);
 
 	// Just a sanity check!
@@ -605,7 +621,7 @@ int acl_fiber_mutex_unlock(ACL_FIBER_MUTEX *mutex)
 
 	// Unlock the internal prive lock must behind the public thread_lock,
 	// or else the waiter maybe be skipped added in lock waiting process.
-	pthread_mutex_unlock(&mutex->lock);
+	UNLOCK(mutex);
 
 	if (ret != 0) {
 		return ret;