Export fiber_cond to public C API

A part of #1451

extra/exports

@@ -99,6 +99,12 @@ fiber_is_cancelled
 fiber_time
 fiber_time64
 fiber_reschedule
+fiber_cond_new
+fiber_cond_delete
+fiber_cond_signal
+fiber_cond_broadcast
+fiber_cond_wait_timeout
+fiber_cond_wait
 cord_slab_cache
 coio_wait
 coio_close

src/CMakeLists.txt

@@ -160,6 +160,7 @@ set(api_headers
     ${CMAKE_SOURCE_DIR}/src/trivia/util.h
     ${CMAKE_SOURCE_DIR}/src/say.h
     ${CMAKE_SOURCE_DIR}/src/fiber.h
+    ${CMAKE_SOURCE_DIR}/src/fiber_cond.h
     ${CMAKE_SOURCE_DIR}/src/coio.h
     ${CMAKE_SOURCE_DIR}/src/coio_task.h
     ${CMAKE_SOURCE_DIR}/src/lua/utils.h

src/fiber_cond.c

@@ -31,6 +31,7 @@
 
 #include "fiber_cond.h"
 
+#include <small/mempool.h>
 #include <tarantool_ev.h>
 
 #include "fiber.h"
@@ -48,6 +49,36 @@ fiber_cond_destroy(struct fiber_cond *c)
 	assert(rlist_empty(&c->waiters));
 }
 
+static __thread struct mempool cond_pool;
+
+struct fiber_cond *
+fiber_cond_new()
+{
+	struct fiber_cond *cond;
+	if (! mempool_is_initialized(&cond_pool)) {
+		/*
+		 * We don't need to bother with
+		 * destruction since the entire slab cache
+		 * is freed when the thread ends.
+		 */
+		mempool_create(&cond_pool, &cord()->slabc, sizeof(*cond));
+	}
+	cond = mempool_alloc(&cond_pool);
+	if (cond == NULL) {
+		diag_set(OutOfMemory, sizeof(*cond), "fiber_cond_pool",
+			 "struct fiber_cond");
+		return NULL;
+	}
+	fiber_cond_create(cond);
+	return cond;
+}
+
+void
+fiber_cond_delete(struct fiber_cond *cond)
+{
+	mempool_free(&cond_pool, cond);
+}
+
 void
 fiber_cond_signal(struct fiber_cond *e)
 {

src/fiber_cond.h

@@ -37,40 +37,96 @@
 extern "C" {
 #endif /* defined(__cplusplus) */
 
+/** \cond public */
+
+/**
+ * Conditional variable for cooperative multitasking (fibers).
+ *
+ * A cond (short for "condition variable") is a synchronization primitive
+ * that allow fibers to yield until some predicate is satisfied. Fiber
+ * conditions have two basic operations - wait() and signal(). wait()
+ * suspends execution of fiber (i.e. yields) until signal() is called.
+ * Unlike pthread_cond, fiber_cond doesn't require mutex/latch wrapping.
+ * 
+ */
+struct fiber_cond;
+
+/** \endcond public */
+
 struct fiber_cond {
+	/** Waiting fibers */
 	struct rlist waiters;
 };
 
 /**
- * Initialize a cond - semantics as in POSIX condition variable.
+ * Initialize the fiber condition variable.
+ *
+ * @param cond condition
  */
 void
-fiber_cond_create(struct fiber_cond *c);
+fiber_cond_create(struct fiber_cond *cond);
 
 /**
- * Finalize a cond. UB if there are fibers waiting for a cond.
+ * Finalize the cond.
+ * Behaviour is undefined if there are fiber waiting for the cond.
+ * @param cond condition
  */
 void
-fiber_cond_destroy(struct fiber_cond *c);
+fiber_cond_destroy(struct fiber_cond *cond);
+
+/** \cond public */
+
+/**
+ * Instantiate a new fiber cond object.
+ */
+struct fiber_cond *
+fiber_cond_new(void);
+
+/**
+ * Delete the fiber cond object.
+ * Behaviour is undefined if there are fiber waiting for the cond.
+ */
+void
+fiber_cond_delete(struct fiber_cond *cond);
 
 /**
  * Wake one fiber waiting for the cond.
  * Does nothing if no one is waiting.
+ * @param cond condition
  */
 void
-fiber_cond_signal(struct fiber_cond *c);
+fiber_cond_signal(struct fiber_cond *cond);
 
 /**
- * Wake all fibers waiting for the cond.
+ * Wake up all fibers waiting for the cond.
+ * @param cond condition
  */
 void
-fiber_cond_broadcast(struct fiber_cond *c);
+fiber_cond_broadcast(struct fiber_cond *cond);
 
+/**
+ * Suspend the execution of the current fiber (i.e. yield) until
+ * fiber_cond_signal() is called. Like pthread_cond, fiber_cond can issue
+ * spurious wake ups caused by explicit fiber_wakeup() or fiber_cancel()
+ * calls. It is highly recommended to wrap calls to this function into a loop
+ * and check an actual predicate and fiber_testcancel() on every iteration.
+ *
+ * @param cond condition
+ * @param timeout timeout in seconds
+ * @retval 0 on fiber_cond_signal() call or a spurious wake up
+ * @retval -1 on timeout, diag is set to TimedOut
+ */
 int
-fiber_cond_wait_timeout(struct fiber_cond *c, double timeout);
+fiber_cond_wait_timeout(struct fiber_cond *cond, double timeout);
 
+/**
+ * Shortcut for fiber_cond_wait_timeout().
+ * @see fiber_cond_wait_timeout()
+ */
 int
-fiber_cond_wait(struct fiber_cond *c);
+fiber_cond_wait(struct fiber_cond *cond);
+
+/** \endcond public */
 
 #if defined(__cplusplus)
 } /* extern "C" */

test/unit/CMakeLists.txt

@@ -80,6 +80,9 @@ endif ()
 add_executable(fiber_stress.test fiber_stress.cc)
 target_link_libraries(fiber_stress.test core)
 
+add_executable(fiber_cond.test fiber_cond.c unit.c)
+target_link_libraries(fiber_cond.test core)
+
 add_executable(fiber_channel.test fiber_channel.cc unit.c)
 target_link_libraries(fiber_channel.test core)
 

test/unit/fiber_cond.c

+#include "memory.h"
+#include "fiber.h"
+#include "fiber_cond.h"
+#include "unit.h"
+
+static int
+fiber_cond_basic_f(va_list ap)
+{
+	struct fiber_cond *cond = va_arg(ap, struct fiber_cond *);
+	int *check = va_arg(ap, int *);
+
+	int rc;
+
+	rc = fiber_cond_wait_timeout(cond, 0.0);
+	ok(rc != 0, "timeout");
+
+	rc = fiber_cond_wait(cond);
+	is(rc, 0, "signal");
+
+	(*check)++;
+
+	rc = fiber_cond_wait(cond);
+	is(rc, 0, "broadcast");
+
+	return 0;
+}
+
+static void
+fiber_cond_basic()
+{
+	struct fiber_cond *cond = fiber_cond_new();
+	int check = 0;
+
+	struct fiber *f1 = fiber_new("f1", fiber_cond_basic_f);
+	assert(f1 != NULL);
+	fiber_start(f1, cond, &check);
+	fiber_set_joinable(f1, true);
+
+	struct fiber *f2 = fiber_new("f2", fiber_cond_basic_f);
+	assert(f2 != NULL);
+	fiber_start(f2, cond, &check);
+	fiber_set_joinable(f2, true);
+
+	/* check timeout */
+	fiber_sleep(0.0);
+	fiber_sleep(0.0);
+
+	/* Wake up the first fiber */
+	fiber_cond_signal(cond);
+	fiber_sleep(0.0);
+
+	/* Wake ip the second fiber */
+	fiber_cond_signal(cond);
+	fiber_sleep(0.0);
+
+	/* Check that fiber scheduling is fair */
+	is(check, 2, "order");
+
+	fiber_cond_broadcast(cond);
+	fiber_sleep(0.0);
+
+	fiber_join(f1);
+	fiber_join(f2);
+
+	fiber_cond_delete(cond);
+}
+
+static int
+main_f(va_list ap)
+{
+	(void) ap;
+	fiber_cond_basic();
+	ev_break(loop(), EVBREAK_ALL);
+	return 0;
+}
+
+int
+main()
+{
+	plan(7);
+	memory_init();
+	fiber_init(fiber_c_invoke);
+	struct fiber *f = fiber_new("main", main_f);
+	fiber_wakeup(f);
+	ev_run(loop(), 0);
+	fiber_free();
+	memory_free();
+	return check_plan();
+}

test/unit/fiber_cond.result

+1..7
+ok 1 - timeout
+ok 2 - timeout
+ok 3 - signal
+ok 4 - signal
+ok 5 - order
+ok 6 - broadcast
+ok 7 - broadcast