vinyl: enable quota upon recovery completion explicitly

Currently, we create a quota object with the limit maximized, and only
set the configured limit when local recovery is complete, so as to make
sure that no dump is triggered during recovery. As a result, we have to
store the configured limit in vy_env::memory, which looks ugly, because
this member is never used afterwards. Let's introduce a new method
vy_quota_enable to enable quota so that we can set the limit right on
quota object construction. This implies that we add a boolean flag to
vy_quota and only check the limit if it is set.

There's another reason to add such a method. Soon we will implement
quota consumption rate limiting. Rate limiting requires a periodic timer
that would replenish quota. It only makes sense to start such a timer
upon recovery completion, which again leads us to an explicit method for
enabling quota.

vy_env::memory will be removed by the following patch along with a few
other pointless members of vy_env.

Needed for #1862

src/box/vinyl.c

@@ -2518,7 +2518,7 @@ vy_env_new(const char *path, size_t memory,
 			      vy_squash_schedule, e) != 0)
 		goto error_lsm_env;
 
-	vy_quota_create(&e->quota, vy_env_quota_exceeded_cb);
+	vy_quota_create(&e->quota, memory, vy_env_quota_exceeded_cb);
 	vy_regulator_create(&e->regulator, &e->quota,
 			    vy_env_trigger_dump_cb);
 
@@ -2572,7 +2572,7 @@ static void
 vy_env_complete_recovery(struct vy_env *env)
 {
 	vy_scheduler_start(&env->scheduler);
-	vy_quota_set_limit(&env->quota, env->memory);
+	vy_quota_enable(&env->quota);
 	vy_regulator_start(&env->regulator);
 }
 

src/box/vy_quota.c

@@ -49,6 +49,8 @@
 static inline bool
 vy_quota_may_use(struct vy_quota *q, size_t size)
 {
+	if (!q->is_enabled)
+		return true;
 	if (q->used + size > q->limit)
 		return false;
 	return true;
@@ -81,7 +83,7 @@ vy_quota_do_unuse(struct vy_quota *q, size_t size)
 static inline void
 vy_quota_check_limit(struct vy_quota *q)
 {
-	if (q->used > q->limit)
+	if (q->is_enabled && q->used > q->limit)
 		q->quota_exceeded_cb(q);
 }
 
@@ -105,15 +107,25 @@ vy_quota_signal(struct vy_quota *q)
 }
 
 void
-vy_quota_create(struct vy_quota *q, vy_quota_exceeded_f quota_exceeded_cb)
+vy_quota_create(struct vy_quota *q, size_t limit,
+		vy_quota_exceeded_f quota_exceeded_cb)
 {
-	q->limit = SIZE_MAX;
+	q->is_enabled = false;
+	q->limit = limit;
 	q->used = 0;
 	q->too_long_threshold = TIMEOUT_INFINITY;
 	q->quota_exceeded_cb = quota_exceeded_cb;
 	rlist_create(&q->wait_queue);
 }
 
+void
+vy_quota_enable(struct vy_quota *q)
+{
+	assert(!q->is_enabled);
+	q->is_enabled = true;
+	vy_quota_check_limit(q);
+}
+
 void
 vy_quota_destroy(struct vy_quota *q)
 {

src/box/vy_quota.h

@@ -31,6 +31,7 @@
  * SUCH DAMAGE.
  */
 
+#include <stdbool.h>
 #include <stddef.h>
 #include <small/rlist.h>
 #include <tarantool_ev.h>
@@ -59,6 +60,8 @@ struct vy_quota_wait_node {
  * in the vinyl engine. It is NOT multi-threading safe.
  */
 struct vy_quota {
+	/** Set if the quota was enabled. */
+	bool is_enabled;
 	/**
 	 * Memory limit. Once hit, new transactions are
 	 * throttled until memory is reclaimed.
@@ -84,9 +87,25 @@ struct vy_quota {
 	struct rlist wait_queue;
 };
 
+/**
+ * Initialize a quota object.
+ *
+ * Note, the limit won't be imposed until vy_quota_enable()
+ * is called.
+ */
 void
-vy_quota_create(struct vy_quota *q, vy_quota_exceeded_f quota_exceeded_cb);
+vy_quota_create(struct vy_quota *q, size_t limit,
+		vy_quota_exceeded_f quota_exceeded_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);