[PATCH 10/11] vinyl: implement quota wait queue without fiber_cond
Vladimir Davydov
vdavydov.dev at gmail.com
Thu Sep 20 12:34:15 MSK 2018
In the next patch we will need to split the queue of fibers waiting for
quota in two: one for high priority consumers and the other for low
priority ones. To balance wake-ups between the two queues, we'll need to
attach some additional info to fibers waiting on the queue (timestamp).
With fiber_cond it is impossible to do. So let's rewrite the quota wait
queue using plain rlist and fiber_yield.
---
src/box/vy_quota.c | 47 ++++++++++++++++++++++++++++++++++++-----------
src/box/vy_quota.h | 17 +++++++++++++----
2 files changed, 49 insertions(+), 15 deletions(-)
diff --git a/src/box/vy_quota.c b/src/box/vy_quota.c
index 18cf4f35..3374aacc 100644
--- a/src/box/vy_quota.c
+++ b/src/box/vy_quota.c
@@ -34,11 +34,11 @@
#include <stddef.h>
#include <stdint.h>
#include <math.h>
+#include <small/rlist.h>
#include <tarantool_ev.h>
#include "diag.h"
#include "fiber.h"
-#include "fiber_cond.h"
#include "say.h"
#include "histogram.h"
#include "trivia/util.h"
@@ -141,6 +141,21 @@ vy_quota_check_watermark(struct vy_quota *q)
vy_quota_trigger_dump(q);
}
+/**
+ * Wake up the next consumer in the line waiting for quota.
+ */
+static void
+vy_quota_signal(struct vy_quota *q)
+{
+ 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);
+ fiber_wakeup(n->fiber);
+ }
+}
+
static void
vy_quota_timer_cb(ev_loop *loop, ev_timer *timer, int events)
{
@@ -179,7 +194,7 @@ vy_quota_timer_cb(ev_loop *loop, ev_timer *timer, int events)
* Replenish quota and wake up throttled fibers, if any.
*/
vy_quota_rl_refill(&q->rl);
- fiber_cond_signal(&q->cond);
+ vy_quota_signal(q);
}
int
@@ -213,7 +228,7 @@ vy_quota_create(struct vy_quota *q, size_t limit,
q->too_long_threshold = TIMEOUT_INFINITY;
q->dump_bw = VY_DEFAULT_DUMP_BANDWIDTH;
q->trigger_dump_cb = trigger_dump_cb;
- fiber_cond_create(&q->cond);
+ rlist_create(&q->wait_queue);
ev_timer_init(&q->timer, vy_quota_timer_cb, 0,
VY_QUOTA_UPDATE_INTERVAL);
q->timer.data = q;
@@ -237,8 +252,6 @@ vy_quota_destroy(struct vy_quota *q)
{
ev_timer_stop(loop(), &q->timer);
histogram_delete(q->dump_bw_hist);
- fiber_cond_broadcast(&q->cond);
- fiber_cond_destroy(&q->cond);
}
void
@@ -246,7 +259,7 @@ vy_quota_set_limit(struct vy_quota *q, size_t limit)
{
q->limit = q->watermark = limit;
vy_quota_check_watermark(q);
- fiber_cond_signal(&q->cond);
+ vy_quota_signal(q);
}
void
@@ -260,7 +273,7 @@ vy_quota_dump(struct vy_quota *q, size_t size, double duration)
*/
assert(q->used >= size);
q->used -= size;
- fiber_cond_signal(&q->cond);
+ vy_quota_signal(q);
/*
* Update dump bandwidth.
@@ -340,8 +353,20 @@ vy_quota_use(struct vy_quota *q, size_t size, double timeout)
do {
if (q->used + size > q->limit)
vy_quota_trigger_dump(q);
- if (fiber_cond_wait_deadline(&q->cond, deadline) != 0)
- return -1; /* timed out */
+
+ double now = ev_monotonic_now(loop());
+ if (now >= deadline) {
+ diag_set(TimedOut);
+ return -1;
+ }
+
+ struct vy_quota_wait_node node = {
+ .fiber = fiber(),
+ };
+ rlist_add_tail_entry(&q->wait_queue,
+ &node, in_wait_queue);
+ fiber_yield_timeout(deadline - now);
+ rlist_del_entry(&node, in_wait_queue);
} while (!vy_quota_may_use(q, size));
double wait_time = ev_monotonic_now(loop()) - start_time;
@@ -353,7 +378,7 @@ vy_quota_use(struct vy_quota *q, size_t size, double timeout)
* Wake up the next fiber in the line waiting
* for quota.
*/
- fiber_cond_signal(&q->cond);
+ vy_quota_signal(q);
}
vy_quota_force_use(q, size);
return 0;
@@ -367,7 +392,7 @@ vy_quota_unuse(struct vy_quota *q, size_t size)
/* use_curr could have been reset on timeout. */
q->use_curr -= MIN(q->use_curr, size);
vy_quota_rl_unuse(&q->rl, size);
- fiber_cond_signal(&q->cond);
+ vy_quota_signal(q);
}
void
diff --git a/src/box/vy_quota.h b/src/box/vy_quota.h
index 3070ff82..f53bb8a9 100644
--- a/src/box/vy_quota.h
+++ b/src/box/vy_quota.h
@@ -33,8 +33,8 @@
#include <stdbool.h>
#include <stddef.h>
+#include <small/rlist.h>
#include <tarantool_ev.h>
-#include "fiber_cond.h"
#if defined(__cplusplus)
extern "C" {
@@ -42,6 +42,7 @@ extern "C" {
struct vy_quota;
struct histogram;
+struct fiber;
typedef int
(*vy_quota_trigger_dump_f)(struct vy_quota *quota);
@@ -60,6 +61,13 @@ struct vy_quota_rl {
ssize_t value;
};
+struct vy_quota_wait_node {
+ /** Link in vy_quota::wait_queue. */
+ struct rlist in_wait_queue;
+ /** Fiber waiting for quota. */
+ struct fiber *fiber;
+};
+
/**
* Quota used for accounting and limiting memory consumption
* in the vinyl engine. It is NOT multi-threading safe.
@@ -86,10 +94,11 @@ struct vy_quota {
*/
double too_long_threshold;
/**
- * Condition variable used for throttling consumers when
- * there is no quota left.
+ * Queue of consumers waiting for quota, linked by
+ * vy_quota_wait_node::state. Newcomers are added
+ * to the tail.
*/
- struct fiber_cond cond;
+ struct rlist wait_queue;
/**
* Timer used for updating average use rate, calculating
* quota watermark and refilling rate limit value.
--
2.11.0
More information about the Tarantool-patches
mailing list