Fixed bugs in sync_timer.c and sync_waiter.c when multiple fibers in the same thread to wake up others.

Author: zhengshuxin

lib_fiber/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;
 	}
 

lib_fiber/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);
 
 /**
  * 从消息队列中读取消息

lib_fiber/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)

lib_fiber/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
@@ -866,7 +867,10 @@ void fiber_file_cache_put(FILE_EVENT *fe)
 	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);
 	}

lib_fiber/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();
 

lib_fiber/c/src/sync/sync_timer.c

@@ -11,7 +11,7 @@ struct SYNC_TIMER {
 	ACL_FIBER *fb;
 	MBOX *box;
 	int stop;
-	TIMER_CACHE *waiters;
+//	TIMER_CACHE *waiters;
 	long tid;
 };
 
@@ -23,7 +23,7 @@ static SYNC_TIMER *sync_timer_new(void)
 
 	pthread_mutex_init(&timer->lock, NULL);
 	timer->box = mbox_create(MBOX_T_MPSC);
-	timer->waiters = timer_cache_create();
+//	timer->waiters = timer_cache_create();
 	timer->tid = thread_self();
 
 	out = mbox_out(timer->box);
@@ -47,7 +47,7 @@ static void sync_timer_free(SYNC_TIMER *timer)
 {
 	pthread_mutex_destroy(&timer->lock);
 	mbox_free(timer->box, NULL);
-	timer_cache_free(timer->waiters);
+//	timer_cache_free(timer->waiters);
 	mem_free(timer);
 }
 
@@ -69,7 +69,7 @@ static void thread_init(void)
 
 static void wakeup_waiter(SYNC_TIMER *timer UNUSED, SYNC_OBJ *obj)
 {
-	// The fiber must has been awakened by the other fiber or thread.
+	// The fiber must have been awakened by the other fiber or thread.
 
 	if (obj->delay < 0) {
 		// No timer has been set if delay < 0,
@@ -117,9 +117,11 @@ static void fiber_waiting(ACL_FIBER *fiber fiber_unused, void *ctx)
 		sync_obj_unrefer(obj);
 		mem_free(msg);
 
+/*
 		if (timer_cache_size(timer->waiters) == 0) {
 			delay = -1;
 		}
+*/
 	}
 }
 
@@ -160,16 +162,35 @@ void sync_timer_wakeup(SYNC_TIMER *timer, SYNC_OBJ *obj)
 		// message with the temporary FILE_EVENT.
 
 		SYNC_MSG *msg = (SYNC_MSG*) mem_malloc(sizeof(SYNC_MSG));
+		socket_t out, out2 = INVALID_SOCKET;
+		FILE_EVENT *fe;
+
 		msg->obj = obj;
 		msg->action = SYNC_ACTION_WAKEUP;
-		socket_t out = mbox_out(timer->box);
-		FILE_EVENT *fe = fiber_file_cache_get(out);
+		sync_obj_refer(obj);
 
-		fe->mask |= EVENT_SYSIO;
+		out = mbox_out(timer->box);
+
+		// Check if the out fd has been bound by the other fiber that
+		// it was yield when calling acl_fiber_write in mbox_send2,
+		// if so, we should duplicate another out fd to bind the new one.
+		fe = fiber_file_get(out);
+		if (fe) {
+			out2 = dup(out);
+			msg_warn("%s(%d): fe exists for out=%d, dup's out2=%d",
+				__FUNCTION__, __LINE__, (int) out, (int) out2);
+			fe = fiber_file_cache_get(out2);
+		} else {
+			fe = fiber_file_cache_get(out);
+		}
 
-		sync_obj_refer(obj);
-		mbox_send(timer->box, msg);
+		fe->mask |= EVENT_SYSIO;
+		mbox_send2(timer->box, out2 != INVALID_SOCKET ? out2 : out, msg);
 		fiber_file_cache_put(fe);
+
+		if (out2 != INVALID_SOCKET) {
+			acl_fiber_close(out2);
+		}
 	} else {
 		// If the current notifier is a fiber in the same thread with
 		// the one to be awakened, just wakeup it directly.

lib_fiber/c/src/sync/sync_waiter.c

@@ -97,7 +97,9 @@ SYNC_WAITER *sync_waiter_get(void)
 
 void sync_waiter_wakeup(SYNC_WAITER *waiter, ACL_FIBER *fb)
 {
-	if (var_hook_sys_api) {
+	if (!var_hook_sys_api) {
+		mbox_send(waiter->box, fb);
+	} else {
 		// When using io_uring, we should call the system API of write
 		// to send data, because the fd is shared by multiple threads
 		// and which can't use io_uring directly, so we set the mask
@@ -123,6 +125,8 @@ void sync_waiter_wakeup(SYNC_WAITER *waiter, ACL_FIBER *fb)
 			assert(fe->mbox_wsem);
 		}
 
+		file_event_refer(fe);
+
 		// Reduce the sem number and maybe be suspended if sem is 0.
 		acl_fiber_sem_wait(fe->mbox_wsem);
 
@@ -136,7 +140,6 @@ void sync_waiter_wakeup(SYNC_WAITER *waiter, ACL_FIBER *fb)
 		if (acl_fiber_sem_post(fe->mbox_wsem) == 1) {
 			fiber_file_cache_put(fe);
 		}
-	} else {
-		mbox_send(waiter->box, fb);
+		file_event_unrefer(fe);
 	}
 }