Stress test insertions into dbengine and bugfixes (#6814)

* Fix memory corruption during deallocation of page cache

* Refactored dataset generator in order to support the upcoming self-validating stress test and multithreading.

* Fix starvation in database engine loop when the command queues are continuously populated

* Fixing disk quota limits for dbengine dataset generator

2 files changed, 25 insertions, 4 deletions

diff --git a/database/engine/pagecache.c b/database/engine/pagecache.c
index 1bd4c94361..457bcb2185 100644
--- a/database/engine/pagecache.c
+++ b/database/engine/pagecache.c
@@ -216,6 +216,9 @@ static void pg_cache_release_pages(struct rrdengine_instance *ctx, unsigned numb
 static void pg_cache_reserve_pages(struct rrdengine_instance *ctx, unsigned number)
 {
     struct page_cache *pg_cache = &ctx->pg_cache;
+    unsigned failures = 0;
+    const unsigned FAILURES_CEILING = 10; /* truncates exponential backoff to (2^FAILURES_CEILING x slot) */
+    unsigned long exp_backoff_slot_usec = USEC_PER_MS * 10;
 
     assert(number < ctx->max_cache_pages);
 
@@ -224,11 +227,13 @@ static void pg_cache_reserve_pages(struct rrdengine_instance *ctx, unsigned numb
         debug(D_RRDENGINE, "==Page cache full. Reserving %u pages.==",
                 number);
     while (pg_cache->populated_pages + number >= ctx->max_cache_pages + 1) {
+
         if (!pg_cache_try_evict_one_page_unsafe(ctx)) {
             /* failed to evict */
             struct completion compl;
             struct rrdeng_cmd cmd;
 
+            ++failures;
             uv_rwlock_wrunlock(&pg_cache->pg_cache_rwlock);
 
             init_completion(&compl);
@@ -240,6 +245,12 @@ static void pg_cache_reserve_pages(struct rrdengine_instance *ctx, unsigned numb
             wait_for_completion(&compl);
             destroy_completion(&compl);
 
+            if (unlikely(failures > 1)) {
+                unsigned long slots;
+                /* exponential backoff */
+                slots = random() % (2LU << MIN(failures, FAILURES_CEILING));
+                (void)sleep_usec(slots * exp_backoff_slot_usec);
+            }
             uv_rwlock_wrlock(&pg_cache->pg_cache_rwlock);
         }
     }
@@ -1042,9 +1053,8 @@ void free_page_cache(struct rrdengine_instance *ctx)
         /* Find first page in range */
         Index = (Word_t) 0;
         PValue = JudyLFirst(page_index->JudyL_array, &Index, PJE0);
-        if (likely(NULL != PValue)) {
-            descr = *PValue;
-        }
+        descr = unlikely(NULL == PValue) ? NULL : *PValue;
+
         while (descr != NULL) {
             /* Iterate all page descriptors of this metric */
 
diff --git a/database/engine/rrdengine.c b/database/engine/rrdengine.c
index 39d1485781..36d917541d 100644
--- a/database/engine/rrdengine.c
+++ b/database/engine/rrdengine.c
@@ -676,7 +676,7 @@ void timer_cb(uv_timer_t* handle)
 /* Flushes dirty pages when timer expires */
 #define TIMER_PERIOD_MS (1000)
 
-#define CMD_BATCH_SIZE (256)
+#define MAX_CMD_BATCH_SIZE (256)
 
 void rrdeng_worker(void* arg)
 {
@@ -687,6 +687,7 @@ void rrdeng_worker(void* arg)
     enum rrdeng_opcode opcode;
     uv_timer_t timer_req;
     struct rrdeng_cmd cmd;
+    unsigned cmd_batch_size;
 
     rrdeng_init_cmd_queue(wc);
 
@@ -723,10 +724,20 @@ void rrdeng_worker(void* arg)
     shutdown = 0;
     while (shutdown == 0 || uv_loop_alive(loop)) {
         uv_run(loop, UV_RUN_DEFAULT);
+
         /* wait for commands */
+        cmd_batch_size = 0;
         do {
+            /*
+             * Avoid starving the loop when there are too many commands coming in.
+             * timer_cb will interrupt the loop again to allow serving more commands.
+             */
+            if (unlikely(cmd_batch_size >= MAX_CMD_BATCH_SIZE))
+                break;
+
             cmd = rrdeng_deq_cmd(wc);
             opcode = cmd.opcode;
+            ++cmd_batch_size;
 
             switch (opcode) {
             case RRDENG_NOOP: