[PATCH 04/11] vinyl: don't start quota timer until local recovery is complete

[Vladimir Davydov <vdavydov.dev@gmail.com>]

Although we don't impose the limit during local recovery, the quota
timer, which keeps track of the write rate and recalculates the
watermark, is running. This is incorrect, because the write rate can
rocket sky high during recovery so that we can wind up with the
watermark set too low once recovery is complete. When we introduce
transaction throttling, the timer will also set a throttle limit based
on the write rate, which is pointless during local recovery too.

That said, we shouldn't start the quota timer until local recovery is
complete. To do that, let's introduce new function vy_quota_enable()
that enables the limit and starts the timer. The function is called
instead of vy_quota_set_limit() upon bootstrap and recovery completion.
---
 src/box/vinyl.c    |  8 ++++----
 src/box/vy_quota.c | 25 ++++++++++++++++++++-----
 src/box/vy_quota.h | 25 ++++++++++++++++++++++++-
 3 files changed, 48 insertions(+), 10 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index d0262a03..e1b35f3a 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -2520,7 +2520,7 @@ vy_env_new(const char *path, size_t memory,
 			      vy_squash_schedule, e) != 0)
 		goto error_lsm_env;
 
-	if (vy_quota_create(&e->quota, vy_env_trigger_dump_cb) != 0)
+	if (vy_quota_create(&e->quota, e->memory, vy_env_trigger_dump_cb) != 0)
 		goto error_quota;
 
 	struct slab_cache *slab_cache = cord_slab_cache();
@@ -2723,7 +2723,7 @@ vinyl_engine_bootstrap(struct engine *engine)
 	if (vy_log_bootstrap() != 0)
 		return -1;
 	vy_scheduler_start(&e->scheduler);
-	vy_quota_set_limit(&e->quota, e->memory);
+	vy_quota_enable(&e->quota);
 	e->status = VINYL_ONLINE;
 	return 0;
 }
@@ -2758,7 +2758,7 @@ vinyl_engine_begin_initial_recovery(struct engine *engine,
 		if (vy_log_bootstrap() != 0)
 			return -1;
 		vy_scheduler_start(&e->scheduler);
-		vy_quota_set_limit(&e->quota, e->memory);
+		vy_quota_enable(&e->quota);
 		e->status = VINYL_INITIAL_RECOVERY_REMOTE;
 	}
 	return 0;
@@ -2802,7 +2802,7 @@ vinyl_engine_end_recovery(struct engine *engine)
 		e->recovery = NULL;
 		e->recovery_vclock = NULL;
 		vy_scheduler_start(&e->scheduler);
-		vy_quota_set_limit(&e->quota, e->memory);
+		vy_quota_enable(&e->quota);
 		break;
 	case VINYL_FINAL_RECOVERY_REMOTE:
 		break;
diff --git a/src/box/vy_quota.c b/src/box/vy_quota.c
index 852f381b..7cd64474 100644
--- a/src/box/vy_quota.c
+++ b/src/box/vy_quota.c
@@ -74,6 +74,7 @@ enum { VY_DUMP_BANDWIDTH_PCT = 10 };
 static void
 vy_quota_trigger_dump(struct vy_quota *q)
 {
+	assert(q->is_enabled);
 	if (q->dump_in_progress)
 		return;
 	if (q->trigger_dump_cb(q) != 0)
@@ -87,7 +88,7 @@ vy_quota_trigger_dump(struct vy_quota *q)
 static void
 vy_quota_check_watermark(struct vy_quota *q)
 {
-	if (q->used >= q->watermark)
+	if (q->is_enabled && q->used >= q->watermark)
 		vy_quota_trigger_dump(q);
 }
 
@@ -99,6 +100,8 @@ vy_quota_timer_cb(ev_loop *loop, ev_timer *timer, int events)
 
 	struct vy_quota *q = timer->data;
 
+	assert(q->is_enabled);
+
 	/*
 	 * Update the quota use rate with the new measurement.
 	 */
@@ -125,7 +128,8 @@ vy_quota_timer_cb(ev_loop *loop, ev_timer *timer, int events)
 }
 
 int
-vy_quota_create(struct vy_quota *q, vy_quota_trigger_dump_f trigger_dump_cb)
+vy_quota_create(struct vy_quota *q, size_t limit,
+		vy_quota_trigger_dump_f trigger_dump_cb)
 {
 	memset(q, 0, sizeof(*q));
 
@@ -149,8 +153,8 @@ vy_quota_create(struct vy_quota *q, vy_quota_trigger_dump_f trigger_dump_cb)
 		return -1;
 	}
 
-	q->limit = SIZE_MAX;
-	q->watermark = SIZE_MAX;
+	q->limit = limit;
+	q->watermark = limit;
 	q->too_long_threshold = TIMEOUT_INFINITY;
 	q->dump_bw = VY_DEFAULT_DUMP_BANDWIDTH;
 	q->trigger_dump_cb = trigger_dump_cb;
@@ -158,11 +162,20 @@ vy_quota_create(struct vy_quota *q, vy_quota_trigger_dump_f trigger_dump_cb)
 	ev_timer_init(&q->timer, vy_quota_timer_cb, 0,
 		      VY_QUOTA_UPDATE_INTERVAL);
 	q->timer.data = q;
-	ev_timer_start(loop(), &q->timer);
 	return 0;
 }
 
 void
+vy_quota_enable(struct vy_quota *q)
+{
+	assert(!q->is_enabled);
+	q->is_enabled = true;
+	q->use_curr = 0;
+	ev_timer_start(loop(), &q->timer);
+	vy_quota_check_watermark(q);
+}
+
+void
 vy_quota_destroy(struct vy_quota *q)
 {
 	ev_timer_stop(loop(), &q->timer);
@@ -227,6 +240,8 @@ vy_quota_force_use(struct vy_quota *q, size_t size)
 static bool
 vy_quota_may_use(struct vy_quota *q, size_t size)
 {
+	if (!q->is_enabled)
+		return true;
 	return q->used + size <= q->limit;
 }
 
diff --git a/src/box/vy_quota.h b/src/box/vy_quota.h
index 3b020829..793a4430 100644
--- a/src/box/vy_quota.h
+++ b/src/box/vy_quota.h
@@ -51,6 +51,8 @@ typedef int
  * in the vinyl engine. It is NOT multi-threading safe.
  */
 struct vy_quota {
+	/** Set when the quota is enabled. */
+	bool is_enabled;
 	/**
 	 * Memory limit. Once hit, new transactions are
 	 * throttled until memory is reclaimed.
@@ -115,9 +117,30 @@ struct vy_quota {
 	struct histogram *dump_bw_hist;
 };
 
+/**
+ * Initialize a quota object.
+ *
+ * @limit: max allowed memory usage.
+ * @trigger_dump_cb: callback invoked to trigger memory dump.
+ *
+ * Returns 0 on success, -1 on memory allocation error.
+ *
+ * Note, the limit won't be imposed until vy_quota_enable()
+ * is called.
+ */
 int
-vy_quota_create(struct vy_quota *q, vy_quota_trigger_dump_f trigger_dump_cb);
+vy_quota_create(struct vy_quota *q, size_t limit,
+		vy_quota_trigger_dump_f trigger_dump_cb);
 
+/**
+ * Enable the configured limit for a quota object.
+ */
+void
+vy_quota_enable(struct vy_quota *q);
+
+/**
+ * Destroy a quota object.
+ */
 void
 vy_quota_destroy(struct vy_quota *q);
 
-- 
2.11.0