[PATCH v2 05/11] vinyl: implement quota wait queue without fiber_cond

[Vladimir Davydov]

Using fiber_cond as a wait queue isn't very convenient, because:
 - It doesn't allow us to put a spuriously woken up fiber back to the
   same position in the queue where it was, thus violating fairness.
 - It doesn't allow us to check whether we actually need to wake up a
   fiber or it will have to go back to sleep anyway as it needs more
   memory than currently available.
 - It doesn't allow us to implement a multi-queue approach where fibers
   that have different priorities are put to different queues.

So let's rewrite the wait queue with plain rlist and fiber_yield.

Needed for #1862
---
 src/box/vy_quota.c | 33 +++++++++++++++++++++++++++------
 src/box/vy_quota.h | 23 +++++++++++++++++------
 2 files changed, 44 insertions(+), 12 deletions(-)

diff --git a/src/box/vy_quota.c b/src/box/vy_quota.c
index 2afed6b7..99811ae9 100644
--- a/src/box/vy_quota.c
+++ b/src/box/vy_quota.c
@@ -39,7 +39,6 @@
 #include "error.h"
 #include "errcode.h"
 #include "fiber.h"
-#include "fiber_cond.h"
 #include "say.h"
 #include "trivia/util.h"
 
@@ -92,7 +91,17 @@ vy_quota_check_limit(struct vy_quota *q)
 static void
 vy_quota_signal(struct vy_quota *q)
 {
-	fiber_cond_signal(&q->cond);
+	if (!rlist_empty(&q->wait_queue)) {
+		struct vy_quota_wait_node *n;
+		n = rlist_first_entry(&q->wait_queue,
+				      struct vy_quota_wait_node, in_wait_queue);
+		/*
+		 * No need in waking up a consumer if it will have
+		 * to go back to sleep immediately.
+		 */
+		if (vy_quota_may_use(q, n->size))
+			fiber_wakeup(n->fiber);
+	}
 }
 
 void
@@ -102,14 +111,13 @@ vy_quota_create(struct vy_quota *q, vy_quota_exceeded_f quota_exceeded_cb)
 	q->used = 0;
 	q->too_long_threshold = TIMEOUT_INFINITY;
 	q->quota_exceeded_cb = quota_exceeded_cb;
-	fiber_cond_create(&q->cond);
+	rlist_create(&q->wait_queue);
 }
 
 void
 vy_quota_destroy(struct vy_quota *q)
 {
-	fiber_cond_broadcast(&q->cond);
-	fiber_cond_destroy(&q->cond);
+	(void)q;
 }
 
 void
@@ -155,6 +163,12 @@ vy_quota_use(struct vy_quota *q, size_t size, double timeout)
 	double wait_start = ev_monotonic_now(loop());
 	double deadline = wait_start + timeout;
 
+	struct vy_quota_wait_node wait_node = {
+		.fiber = fiber(),
+		.size = size,
+	};
+	rlist_add_tail_entry(&q->wait_queue, &wait_node, in_wait_queue);
+
 	do {
 		/*
 		 * If the requested amount of memory cannot be
@@ -164,12 +178,19 @@ vy_quota_use(struct vy_quota *q, size_t size, double timeout)
 		 */
 		if (q->used + size > q->limit)
 			q->quota_exceeded_cb(q);
-		if (fiber_cond_wait_deadline(&q->cond, deadline) != 0) {
+
+		double now = ev_monotonic_now(loop());
+		fiber_yield_timeout(deadline - now);
+
+		if (now >= deadline) {
+			rlist_del_entry(&wait_node, in_wait_queue);
 			diag_set(ClientError, ER_VY_QUOTA_TIMEOUT);
 			return -1;
 		}
 	} while (!vy_quota_may_use(q, size));
 
+	rlist_del_entry(&wait_node, in_wait_queue);
+
 	double wait_time = ev_monotonic_now(loop()) - wait_start;
 	if (wait_time > q->too_long_threshold) {
 		say_warn("waited for %zu bytes of vinyl memory quota "
diff --git a/src/box/vy_quota.h b/src/box/vy_quota.h
index 59fe075f..d46b6748 100644
--- a/src/box/vy_quota.h
+++ b/src/box/vy_quota.h
@@ -32,18 +32,28 @@
  */
 
 #include <stddef.h>
+#include <small/rlist.h>
 #include <tarantool_ev.h>
-#include "fiber_cond.h"
 
 #if defined(__cplusplus)
 extern "C" {
 #endif /* defined(__cplusplus) */
 
+struct fiber;
 struct vy_quota;
 
 typedef void
 (*vy_quota_exceeded_f)(struct vy_quota *quota);
 
+struct vy_quota_wait_node {
+	/** Link in vy_quota::wait_queue. */
+	struct rlist in_wait_queue;
+	/** Fiber waiting for quota. */
+	struct fiber *fiber;
+	/** Amount of requested memory. */
+	size_t size;
+};
+
 /**
  * Quota used for accounting and limiting memory consumption
  * in the vinyl engine. It is NOT multi-threading safe.
@@ -62,15 +72,16 @@ struct vy_quota {
 	 */
 	double too_long_threshold;
 	/**
-	 * Condition variable used for throttling consumers when
-	 * there is no quota left.
-	 */
-	struct fiber_cond cond;
-	/**
 	 * Called if the limit is hit when quota is consumed.
 	 * It is supposed to trigger memory reclaim.
 	 */
 	vy_quota_exceeded_f quota_exceeded_cb;
+	/**
+	 * Queue of consumers waiting for quota, linked by
+	 * vy_quota_wait_node::state. Newcomers are added
+	 * to the tail.
+	 */
+	struct rlist wait_queue;
 };
 
 void
-- 
2.11.0