Merge from acl/libfiber.

Author: zhengshuxin

c/include/fiber/fiber_sem.h

@@ -21,6 +21,14 @@ typedef struct ACL_FIBER_SEM ACL_FIBER_SEM;
 FIBER_API ACL_FIBER_SEM* acl_fiber_sem_create2(int num, unsigned flags);
 #define ACL_FIBER_SEM_F_ASYNC	(1 << 0)	/* If notifying in async mode */
 
+/**
+ * Create one fiber semaphore with the specified buff count and flags.
+ * @param num {int} the initial value of the semaphore, must >= 0
+ * @param buf {int} the buffed count before signal the waiters.
+ * @param flags {unsigned} the flags defined as ACL_FIBER_SEM_F_XXX
+ */
+FIBER_API ACL_FIBER_SEM* acl_fiber_sem_create3(int num, int buf, unsigned flags);
+
 FIBER_API ACL_FIBER_SEM* acl_fiber_sem_create(int num);
 
 /**
@@ -65,9 +73,17 @@ FIBER_API int acl_fiber_sem_wait(ACL_FIBER_SEM* sem);
  */
 FIBER_API int acl_fiber_sem_trywait(ACL_FIBER_SEM* sem);
 
+/**
+ * Wait for semaphore until > 0 or the timer arriving.
+ * @param sem {ACL_FIBER_SEM *} created by acl_fiber_sem_create
+ * @param milliseconds {int} specify the timeout to wait
+ * @return {int} return >= 0 if waiting successfully, or -1 if waiting timed out.
+ */
+FIBER_API int acl_fiber_sem_timed_wait(ACL_FIBER_SEM *sem, int milliseconds);
+
 /**
  * Add 1 to the semaphore, if there are other fibers waiting for semaphore,
- * one waiter will be wakeuped
+ * one waiter will be wakeup
  * @param sem {ACL_FIBER_SEM *} created by acl_fiber_sem_create
  * @return {int} the current semaphore value returned, -1 returned if the
  *  current thread ID is not same as the semaphore's owner ID

c/src/fiber.h

@@ -35,6 +35,7 @@ struct SYNC_WAITER;
 struct ACL_FIBER {
 	FIBER_BASE    *base;
 	RING           me;
+	RING           me2;
 	long           tid;
 	unsigned       fid;
 	unsigned       slot;

c/src/sync/fiber_cond.c

@@ -159,6 +159,8 @@ static int fiber_cond_timedwait(ACL_FIBER_COND *cond, ACL_FIBER_MUTEX *mutex,
 	UNLOCK_COND(cond);
 
 	if (fiber->flag & FIBER_F_TIMER) {
+		acl_fiber_set_errno(fiber, FIBER_EAGAIN);
+		acl_fiber_set_error(FIBER_EAGAIN);
 		// The obj has been deleted in sync_timer.c when timeout.
 		fiber->flag &= ~FIBER_F_TIMER;
 		sync_obj_unrefer(obj);

c/src/sync/fiber_sem.c

@@ -6,6 +6,7 @@
 
 struct ACL_FIBER_SEM {
 	int num;
+	int buf;
 	unsigned flags;
 	RING waiting;
 	unsigned long tid;
@@ -17,11 +18,18 @@ ACL_FIBER_SEM *acl_fiber_sem_create(int num)
 }
 
 ACL_FIBER_SEM *acl_fiber_sem_create2(int num, unsigned flags)
+{
+	int buf = (flags & ACL_FIBER_SEM_F_ASYNC) ? 50000000 : 0;
+	return acl_fiber_sem_create3(num, buf, flags);
+}
+
+ACL_FIBER_SEM *acl_fiber_sem_create3(int num, int buf, unsigned flags)
 {
 	ACL_FIBER_SEM *sem = (ACL_FIBER_SEM *) mem_malloc(sizeof(ACL_FIBER_SEM));
 
 	sem->tid   = 0;
 	sem->num   = num;
+	sem->buf   = buf;
 	sem->flags = flags;
 	ring_init(&sem->waiting);
 	return sem;
@@ -58,21 +66,14 @@ int acl_fiber_sem_waiters_num(ACL_FIBER_SEM *sem)
 	return ring_size(&sem->waiting);
 }
 
-int acl_fiber_sem_wait(ACL_FIBER_SEM *sem)
+int acl_fiber_sem_timed_wait(ACL_FIBER_SEM *sem, int milliseconds)
 {
 	ACL_FIBER *curr;
 	EVENT *ev;
 
 	if (sem->tid == 0) {
 		sem->tid = thread_self();
 	}
-#if 0
-	else if (sem->tid != (unsigned long) thread_self()) {
-		msg_error("%s(%d): current tid=%lu, sem tid=%lu",
-			__FUNCTION__, __LINE__, thread_self(), sem->tid);
-		return -1;
-	}
-#endif
 
 	if (sem->num > 0) {
 		sem->num--;
@@ -84,35 +85,55 @@ int acl_fiber_sem_wait(ACL_FIBER_SEM *sem)
 		return -1;
 	}
 
-	// Sanity check befor suspending.
+	if (milliseconds == 0) {
+		acl_fiber_set_errno(curr, FIBER_EAGAIN);
+		acl_fiber_set_error(FIBER_EAGAIN);
+		return -1;
+	}
+
+	// Sanity check before suspending.
 	if (acl_fiber_canceled(curr)) {
 		acl_fiber_set_error(curr->errnum);
-		//msg_info("%s(%d): fiber-%d be killed",
-		//	__FUNCTION__, __LINE__, acl_fiber_id(curr));
 		return -1;
 	}
 
-	ring_prepend(&sem->waiting, &curr->me);
+	ring_prepend(&sem->waiting, &curr->me2);
 
 	curr->wstatus |= FIBER_WAIT_SEM;
 
+	if (milliseconds > 0) {
+		fiber_timer_add(curr, (size_t) milliseconds);
+	}
+
+	// Make sure to start wakeup_timers in fiber_io.c by the following:
+	// fiber_io_event -> fiber_io_check -> fiber_io_loop -> wakeup_timers.
 	ev = fiber_io_event();
 	WAITER_INC(ev);  // Just for avoiding fiber_io_loop to exit
 	acl_fiber_switch();
 	WAITER_DEC(ev);
 
+	if (milliseconds > 0) {
+		fiber_timer_del(curr);
+	}
+
 	curr->wstatus &= ~FIBER_WAIT_SEM;
 
 	/* If switch to me because other killed me, I should detach myself;
-	 * else if because other unlock, I'll be detached twice which is
-	 * hamless because RIGN can deal with it.
+ 	 * else if because other unlock, I'll be detached twice which is
+	 * harmless because RING can deal with it.
 	 */
 	ring_detach(&curr->me);
+	ring_detach(&curr->me2);
 
 	if (acl_fiber_canceled(curr)) {
 		acl_fiber_set_error(curr->errnum);
-		//msg_info("%s(%d): fiber-%d be killed",
-		//	__FUNCTION__, __LINE__, acl_fiber_id(curr));
+		return -1;
+	} else if (curr->flag & FIBER_F_TIMER) {
+		// Clear FIBER_F_TIMER flag been set in wakeup_timers.
+		curr->flag &= ~FIBER_F_TIMER;
+
+		acl_fiber_set_errno(curr, FIBER_EAGAIN);
+		acl_fiber_set_error(FIBER_EAGAIN);
 		return -1;
 	}
 
@@ -122,18 +143,16 @@ int acl_fiber_sem_wait(ACL_FIBER_SEM *sem)
 	return sem->num;
 }
 
+int acl_fiber_sem_wait(ACL_FIBER_SEM *sem)
+{
+	return acl_fiber_sem_timed_wait(sem, -1);
+}
+
 int acl_fiber_sem_trywait(ACL_FIBER_SEM *sem)
 {
 	if (sem->tid == 0) {
 		sem->tid = thread_self();
 	}
-#if 0
-	else if (sem->tid != thread_self()) {
-		msg_error("%s(%d): current tid=%lu, sem tid=%lu",
-			__FUNCTION__, __LINE__, thread_self(), sem->tid);
-		return -1;
-	}
-#endif
 
 	if (sem->num > 0) {
 		sem->num--;
@@ -144,7 +163,7 @@ int acl_fiber_sem_trywait(ACL_FIBER_SEM *sem)
 }
 
 #define RING_TO_FIBER(r) \
-    ((ACL_FIBER *) ((char *) (r) - offsetof(ACL_FIBER, me)))
+    ((ACL_FIBER *) ((char *) (r) - offsetof(ACL_FIBER, me2)))
 
 #define FIRST_FIBER(head) \
     (ring_succ(head) != (head) ? RING_TO_FIBER(ring_succ(head)) : 0)
@@ -168,16 +187,19 @@ int acl_fiber_sem_post(ACL_FIBER_SEM *sem)
 	sem->num++;
 
 	if ((ready = FIRST_FIBER(&sem->waiting)) == NULL) {
+		if (sem->num >= sem->buf) {
+			acl_fiber_yield();
+		}
 		return sem->num;
 	}
 
-	ring_detach(&ready->me);
+	ring_detach(&ready->me2);
 	FIBER_READY(ready);
 
 	/* Help the fiber to be wakeup to decrease the sem number. */
 	num = sem->num--;
 
-	if (!(sem->flags & ACL_FIBER_SEM_F_ASYNC)) {
+	if (num >= sem->buf) {
 		acl_fiber_yield();
 	}
 

cpp/include/fiber/channel.hpp

@@ -1,4 +1,4 @@
-#pragma once
+#pragma once
 
 #include "fiber_cpp_define.hpp"
 
@@ -14,11 +14,11 @@ int channel_recv(ACL_CHANNEL *c, void *v);
 template <typename T>
 class channel {
 public:
-	channel(void) {
+	channel() {
 		chan_ = channel_create(sizeof(T), 100);
 	}
 
-	~channel(void) {
+	~channel() {
 		channel_free(chan_);
 	}