Adding upstream version 1.38.0.upstream/1.38.0

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1465 insertions, 1169 deletions

diff --git a/database/engine/rrdengine.c b/database/engine/rrdengine.c
index a6840f38c..d64868f03 100644
--- a/database/engine/rrdengine.c
+++ b/database/engine/rrdengine.c
@@ -2,6 +2,7 @@
 #define NETDATA_RRD_INTERNALS
 
 #include "rrdengine.h"
+#include "pdc.h"
 
 rrdeng_stats_t global_io_errors = 0;
 rrdeng_stats_t global_fs_errors = 0;
@@ -11,31 +12,74 @@ rrdeng_stats_t global_flushing_pressure_page_deletions = 0;
 
 unsigned rrdeng_pages_per_extent = MAX_PAGES_PER_EXTENT;
 
-#if WORKER_UTILIZATION_MAX_JOB_TYPES < (RRDENG_MAX_OPCODE + 2)
+#if WORKER_UTILIZATION_MAX_JOB_TYPES < (RRDENG_OPCODE_MAX + 2)
 #error Please increase WORKER_UTILIZATION_MAX_JOB_TYPES to at least (RRDENG_MAX_OPCODE + 2)
 #endif
 
-void *dbengine_page_alloc() {
-    void *page = NULL;
-    if (unlikely(db_engine_use_malloc))
-        page = mallocz(RRDENG_BLOCK_SIZE);
-    else {
-        page = netdata_mmap(NULL, RRDENG_BLOCK_SIZE, MAP_PRIVATE, enable_ksm);
-        if(!page) fatal("Cannot allocate dbengine page cache page, with mmap()");
-    }
-    return page;
-}
-
-void dbengine_page_free(void *page) {
-    if (unlikely(db_engine_use_malloc))
-        freez(page);
-    else
-        netdata_munmap(page, RRDENG_BLOCK_SIZE);
-}
+struct rrdeng_main {
+    uv_thread_t thread;
+    uv_loop_t loop;
+    uv_async_t async;
+    uv_timer_t timer;
+    pid_t tid;
+
+    size_t flushes_running;
+    size_t evictions_running;
+    size_t cleanup_running;
+
+    struct {
+        ARAL *ar;
+
+        struct {
+            SPINLOCK spinlock;
+
+            size_t waiting;
+            struct rrdeng_cmd *waiting_items_by_priority[STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE];
+            size_t executed_by_priority[STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE];
+        } unsafe;
+    } cmd_queue;
+
+    struct {
+        ARAL *ar;
+
+        struct {
+            size_t dispatched;
+            size_t executing;
+            size_t pending_cb;
+        } atomics;
+    } work_cmd;
+
+    struct {
+        ARAL *ar;
+    } handles;
+
+    struct {
+        ARAL *ar;
+    } descriptors;
+
+    struct {
+        ARAL *ar;
+    } xt_io_descr;
+
+} rrdeng_main = {
+        .thread = 0,
+        .loop = {},
+        .async = {},
+        .timer = {},
+        .flushes_running = 0,
+        .evictions_running = 0,
+        .cleanup_running = 0,
+
+        .cmd_queue = {
+                .unsafe = {
+                        .spinlock = NETDATA_SPINLOCK_INITIALIZER,
+                },
+        }
+};
 
 static void sanity_check(void)
 {
-    BUILD_BUG_ON(WORKER_UTILIZATION_MAX_JOB_TYPES < (RRDENG_MAX_OPCODE + 2));
+    BUILD_BUG_ON(WORKER_UTILIZATION_MAX_JOB_TYPES < (RRDENG_OPCODE_MAX + 2));
 
     /* Magic numbers must fit in the super-blocks */
     BUILD_BUG_ON(strlen(RRDENG_DF_MAGIC) > RRDENG_MAGIC_SZ);
@@ -54,519 +98,489 @@ static void sanity_check(void)
     BUILD_BUG_ON(MAX_PAGES_PER_EXTENT > 255);
 
     /* extent cache count must fit in 32 bits */
-    BUILD_BUG_ON(MAX_CACHED_EXTENTS > 32);
+//    BUILD_BUG_ON(MAX_CACHED_EXTENTS > 32);
 
     /* page info scratch space must be able to hold 2 32-bit integers */
     BUILD_BUG_ON(sizeof(((struct rrdeng_page_info *)0)->scratch) < 2 * sizeof(uint32_t));
 }
 
-/* always inserts into tail */
-static inline void xt_cache_replaceQ_insert(struct rrdengine_worker_config* wc,
-                                            struct extent_cache_element *xt_cache_elem)
-{
-    struct extent_cache *xt_cache = &wc->xt_cache;
+// ----------------------------------------------------------------------------
+// work request cache
 
-    xt_cache_elem->prev = NULL;
-    xt_cache_elem->next = NULL;
+typedef void *(*work_cb)(struct rrdengine_instance *ctx, void *data, struct completion *completion, uv_work_t* req);
+typedef void (*after_work_cb)(struct rrdengine_instance *ctx, void *data, struct completion *completion, uv_work_t* req, int status);
 
-    if (likely(NULL != xt_cache->replaceQ_tail)) {
-        xt_cache_elem->prev = xt_cache->replaceQ_tail;
-        xt_cache->replaceQ_tail->next = xt_cache_elem;
-    }
-    if (unlikely(NULL == xt_cache->replaceQ_head)) {
-        xt_cache->replaceQ_head = xt_cache_elem;
-    }
-    xt_cache->replaceQ_tail = xt_cache_elem;
+struct rrdeng_work {
+    uv_work_t req;
+
+    struct rrdengine_instance *ctx;
+    void *data;
+    struct completion *completion;
+
+    work_cb work_cb;
+    after_work_cb after_work_cb;
+    enum rrdeng_opcode opcode;
+};
+
+static void work_request_init(void) {
+    rrdeng_main.work_cmd.ar = aral_create(
+            "dbengine-work-cmd",
+            sizeof(struct rrdeng_work),
+            0,
+            65536, NULL,
+            NULL, NULL, false, false
+    );
 }
 
-static inline void xt_cache_replaceQ_delete(struct rrdengine_worker_config* wc,
-                                            struct extent_cache_element *xt_cache_elem)
-{
-    struct extent_cache *xt_cache = &wc->xt_cache;
-    struct extent_cache_element *prev, *next;
+static inline bool work_request_full(void) {
+    return __atomic_load_n(&rrdeng_main.work_cmd.atomics.dispatched, __ATOMIC_RELAXED) >= (size_t)(libuv_worker_threads - RESERVED_LIBUV_WORKER_THREADS);
+}
 
-    prev = xt_cache_elem->prev;
-    next = xt_cache_elem->next;
+static inline void work_done(struct rrdeng_work *work_request) {
+    aral_freez(rrdeng_main.work_cmd.ar, work_request);
+}
 
-    if (likely(NULL != prev)) {
-        prev->next = next;
-    }
-    if (likely(NULL != next)) {
-        next->prev = prev;
-    }
-    if (unlikely(xt_cache_elem == xt_cache->replaceQ_head)) {
-        xt_cache->replaceQ_head = next;
-    }
-    if (unlikely(xt_cache_elem == xt_cache->replaceQ_tail)) {
-        xt_cache->replaceQ_tail = prev;
-    }
-    xt_cache_elem->prev = xt_cache_elem->next = NULL;
+static void work_standard_worker(uv_work_t *req) {
+    __atomic_add_fetch(&rrdeng_main.work_cmd.atomics.executing, 1, __ATOMIC_RELAXED);
+
+    register_libuv_worker_jobs();
+    worker_is_busy(UV_EVENT_WORKER_INIT);
+
+    struct rrdeng_work *work_request = req->data;
+    work_request->data = work_request->work_cb(work_request->ctx, work_request->data, work_request->completion, req);
+    worker_is_idle();
+
+    __atomic_sub_fetch(&rrdeng_main.work_cmd.atomics.dispatched, 1, __ATOMIC_RELAXED);
+    __atomic_sub_fetch(&rrdeng_main.work_cmd.atomics.executing, 1, __ATOMIC_RELAXED);
+    __atomic_add_fetch(&rrdeng_main.work_cmd.atomics.pending_cb, 1, __ATOMIC_RELAXED);
+
+    // signal the event loop a worker is available
+    fatal_assert(0 == uv_async_send(&rrdeng_main.async));
 }
 
-static inline void xt_cache_replaceQ_set_hot(struct rrdengine_worker_config* wc,
-                                             struct extent_cache_element *xt_cache_elem)
-{
-    xt_cache_replaceQ_delete(wc, xt_cache_elem);
-    xt_cache_replaceQ_insert(wc, xt_cache_elem);
+static void after_work_standard_callback(uv_work_t* req, int status) {
+    struct rrdeng_work *work_request = req->data;
+
+    worker_is_busy(RRDENG_OPCODE_MAX + work_request->opcode);
+
+    if(work_request->after_work_cb)
+        work_request->after_work_cb(work_request->ctx, work_request->data, work_request->completion, req, status);
+
+    work_done(work_request);
+    __atomic_sub_fetch(&rrdeng_main.work_cmd.atomics.pending_cb, 1, __ATOMIC_RELAXED);
+
+    worker_is_idle();
 }
 
-/* Returns the index of the cached extent if it was successfully inserted in the extent cache, otherwise -1 */
-static int try_insert_into_xt_cache(struct rrdengine_worker_config* wc, struct extent_info *extent)
-{
-    struct extent_cache *xt_cache = &wc->xt_cache;
-    struct extent_cache_element *xt_cache_elem;
-    unsigned idx;
-    int ret;
+static bool work_dispatch(struct rrdengine_instance *ctx, void *data, struct completion *completion, enum rrdeng_opcode opcode, work_cb work_cb, after_work_cb after_work_cb) {
+    struct rrdeng_work *work_request = NULL;
 
-    ret = find_first_zero(xt_cache->allocation_bitmap);
-    if (-1 == ret || ret >= MAX_CACHED_EXTENTS) {
-        for (xt_cache_elem = xt_cache->replaceQ_head ; NULL != xt_cache_elem ; xt_cache_elem = xt_cache_elem->next) {
-            idx = xt_cache_elem - xt_cache->extent_array;
-            if (!check_bit(xt_cache->inflight_bitmap, idx)) {
-                xt_cache_replaceQ_delete(wc, xt_cache_elem);
-                break;
-            }
-        }
-        if (NULL == xt_cache_elem)
-            return -1;
-    } else {
-        idx = (unsigned)ret;
-        xt_cache_elem = &xt_cache->extent_array[idx];
+    internal_fatal(rrdeng_main.tid != gettid(), "work_dispatch() can only be run from the event loop thread");
+
+    work_request = aral_mallocz(rrdeng_main.work_cmd.ar);
+    memset(work_request, 0, sizeof(struct rrdeng_work));
+    work_request->req.data = work_request;
+    work_request->ctx = ctx;
+    work_request->data = data;
+    work_request->completion = completion;
+    work_request->work_cb = work_cb;
+    work_request->after_work_cb = after_work_cb;
+    work_request->opcode = opcode;
+
+    if(uv_queue_work(&rrdeng_main.loop, &work_request->req, work_standard_worker, after_work_standard_callback)) {
+        internal_fatal(true, "DBENGINE: cannot queue work");
+        work_done(work_request);
+        return false;
     }
-    xt_cache_elem->extent = extent;
-    xt_cache_elem->fileno = extent->datafile->fileno;
-    xt_cache_elem->inflight_io_descr = NULL;
-    xt_cache_replaceQ_insert(wc, xt_cache_elem);
-    modify_bit(&xt_cache->allocation_bitmap, idx, 1);
 
-    return (int)idx;
+    __atomic_add_fetch(&rrdeng_main.work_cmd.atomics.dispatched, 1, __ATOMIC_RELAXED);
+
+    return true;
 }
 
-/**
- * Returns 0 if the cached extent was found in the extent cache, 1 otherwise.
- * Sets *idx to point to the position of the extent inside the cache.
- **/
-static uint8_t lookup_in_xt_cache(struct rrdengine_worker_config* wc, struct extent_info *extent, unsigned *idx)
-{
-    struct extent_cache *xt_cache = &wc->xt_cache;
-    struct extent_cache_element *xt_cache_elem;
-    unsigned i;
+// ----------------------------------------------------------------------------
+// page descriptor cache
+
+void page_descriptors_init(void) {
+    rrdeng_main.descriptors.ar = aral_create(
+            "dbengine-descriptors",
+            sizeof(struct page_descr_with_data),
+            0,
+            65536 * 4,
+            NULL,
+            NULL, NULL, false, false);
+}
 
-    for (i = 0 ; i < MAX_CACHED_EXTENTS ; ++i) {
-        xt_cache_elem = &xt_cache->extent_array[i];
-        if (check_bit(xt_cache->allocation_bitmap, i) && xt_cache_elem->extent == extent &&
-            xt_cache_elem->fileno == extent->datafile->fileno) {
-            *idx = i;
-            return 0;
-        }
-    }
-    return 1;
+struct page_descr_with_data *page_descriptor_get(void) {
+    struct page_descr_with_data *descr = aral_mallocz(rrdeng_main.descriptors.ar);
+    memset(descr, 0, sizeof(struct page_descr_with_data));
+    return descr;
 }
 
-#if 0 /* disabled code */
-static void delete_from_xt_cache(struct rrdengine_worker_config* wc, unsigned idx)
-{
-    struct extent_cache *xt_cache = &wc->xt_cache;
-    struct extent_cache_element *xt_cache_elem;
+static inline void page_descriptor_release(struct page_descr_with_data *descr) {
+    aral_freez(rrdeng_main.descriptors.ar, descr);
+}
 
-    xt_cache_elem = &xt_cache->extent_array[idx];
-    xt_cache_replaceQ_delete(wc, xt_cache_elem);
-    xt_cache_elem->extent = NULL;
-    modify_bit(&wc->xt_cache.allocation_bitmap, idx, 0); /* invalidate it */
-    modify_bit(&wc->xt_cache.inflight_bitmap, idx, 0); /* not in-flight anymore */
+// ----------------------------------------------------------------------------
+// extent io descriptor cache
+
+static void extent_io_descriptor_init(void) {
+    rrdeng_main.xt_io_descr.ar = aral_create(
+            "dbengine-extent-io",
+            sizeof(struct extent_io_descriptor),
+            0,
+            65536,
+            NULL,
+            NULL, NULL, false, false
+            );
 }
-#endif
 
-void enqueue_inflight_read_to_xt_cache(struct rrdengine_worker_config* wc, unsigned idx,
-                                       struct extent_io_descriptor *xt_io_descr)
-{
-    struct extent_cache *xt_cache = &wc->xt_cache;
-    struct extent_cache_element *xt_cache_elem;
-    struct extent_io_descriptor *old_next;
+static struct extent_io_descriptor *extent_io_descriptor_get(void) {
+    struct extent_io_descriptor *xt_io_descr = aral_mallocz(rrdeng_main.xt_io_descr.ar);
+    memset(xt_io_descr, 0, sizeof(struct extent_io_descriptor));
+    return xt_io_descr;
+}
 
-    xt_cache_elem = &xt_cache->extent_array[idx];
-    old_next = xt_cache_elem->inflight_io_descr->next;
-    xt_cache_elem->inflight_io_descr->next = xt_io_descr;
-    xt_io_descr->next = old_next;
+static inline void extent_io_descriptor_release(struct extent_io_descriptor *xt_io_descr) {
+    aral_freez(rrdeng_main.xt_io_descr.ar, xt_io_descr);
 }
 
-void read_cached_extent_cb(struct rrdengine_worker_config* wc, unsigned idx, struct extent_io_descriptor *xt_io_descr)
-{
-    unsigned i, j, page_offset;
-    struct rrdengine_instance *ctx = wc->ctx;
-    struct rrdeng_page_descr *descr;
-    struct page_cache_descr *pg_cache_descr;
-    void *page;
-    struct extent_info *extent = xt_io_descr->descr_array[0]->extent;
-
-    for (i = 0 ; i < xt_io_descr->descr_count; ++i) {
-        page = dbengine_page_alloc();
-        descr = xt_io_descr->descr_array[i];
-        for (j = 0, page_offset = 0 ; j < extent->number_of_pages ; ++j) {
-            /* care, we don't hold the descriptor mutex */
-            if (!uuid_compare(*extent->pages[j]->id, *descr->id) &&
-                extent->pages[j]->page_length == descr->page_length &&
-                extent->pages[j]->start_time_ut == descr->start_time_ut &&
-                extent->pages[j]->end_time_ut == descr->end_time_ut) {
-                break;
-            }
-            page_offset += extent->pages[j]->page_length;
+// ----------------------------------------------------------------------------
+// query handle cache
+
+void rrdeng_query_handle_init(void) {
+    rrdeng_main.handles.ar = aral_create(
+            "dbengine-query-handles",
+            sizeof(struct rrdeng_query_handle),
+            0,
+            65536,
+            NULL,
+            NULL, NULL, false, false);
+}
 
-        }
-        /* care, we don't hold the descriptor mutex */
-       (void) memcpy(page, wc->xt_cache.extent_array[idx].pages + page_offset, descr->page_length);
-
-        rrdeng_page_descr_mutex_lock(ctx, descr);
-        pg_cache_descr = descr->pg_cache_descr;
-        pg_cache_descr->page = page;
-        pg_cache_descr->flags |= RRD_PAGE_POPULATED;
-        pg_cache_descr->flags &= ~RRD_PAGE_READ_PENDING;
-        rrdeng_page_descr_mutex_unlock(ctx, descr);
-        pg_cache_replaceQ_insert(ctx, descr);
-        if (xt_io_descr->release_descr) {
-            pg_cache_put(ctx, descr);
-        } else {
-            debug(D_RRDENGINE, "%s: Waking up waiters.", __func__);
-            pg_cache_wake_up_waiters(ctx, descr);
-        }
+struct rrdeng_query_handle *rrdeng_query_handle_get(void) {
+    struct rrdeng_query_handle *handle = aral_mallocz(rrdeng_main.handles.ar);
+    memset(handle, 0, sizeof(struct rrdeng_query_handle));
+    return handle;
+}
+
+void rrdeng_query_handle_release(struct rrdeng_query_handle *handle) {
+    aral_freez(rrdeng_main.handles.ar, handle);
+}
+
+// ----------------------------------------------------------------------------
+// WAL cache
+
+static struct {
+    struct {
+        SPINLOCK spinlock;
+        WAL *available_items;
+        size_t available;
+    } protected;
+
+    struct {
+        size_t allocated;
+    } atomics;
+} wal_globals = {
+        .protected = {
+                .spinlock = NETDATA_SPINLOCK_INITIALIZER,
+                .available_items = NULL,
+                .available = 0,
+        },
+        .atomics = {
+                .allocated = 0,
+        },
+};
+
+static void wal_cleanup1(void) {
+    WAL *wal = NULL;
+
+    if(!netdata_spinlock_trylock(&wal_globals.protected.spinlock))
+        return;
+
+    if(wal_globals.protected.available_items && wal_globals.protected.available > storage_tiers) {
+        wal = wal_globals.protected.available_items;
+        DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wal_globals.protected.available_items, wal, cache.prev, cache.next);
+        wal_globals.protected.available--;
     }
-    if (xt_io_descr->completion)
-        completion_mark_complete(xt_io_descr->completion);
-    freez(xt_io_descr);
-}
-
-static void fill_page_with_nulls(void *page, uint32_t page_length, uint8_t type) {
-    switch(type) {
-        case PAGE_METRICS: {
-            storage_number n = pack_storage_number(NAN, SN_FLAG_NONE);
-            storage_number *array = (storage_number *)page;
-            size_t slots = page_length / sizeof(n);
-            for(size_t i = 0; i < slots ; i++)
-                array[i] = n;
-        }
-        break;
-
-        case PAGE_TIER: {
-            storage_number_tier1_t n = {
-                .min_value = NAN,
-                .max_value = NAN,
-                .sum_value = NAN,
-                .count = 1,
-                .anomaly_count = 0,
-            };
-            storage_number_tier1_t *array = (storage_number_tier1_t *)page;
-            size_t slots = page_length / sizeof(n);
-            for(size_t i = 0; i < slots ; i++)
-                array[i] = n;
-        }
-        break;
 
-        default: {
-            static bool logged = false;
-            if(!logged) {
-                error("DBENGINE: cannot fill page with nulls on unknown page type id %d", type);
-                logged = true;
-            }
-            memset(page, 0, page_length);
-        }
+    netdata_spinlock_unlock(&wal_globals.protected.spinlock);
+
+    if(wal) {
+        posix_memfree(wal->buf);
+        freez(wal);
+        __atomic_sub_fetch(&wal_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
     }
 }
 
-struct rrdeng_page_descr *get_descriptor(struct pg_cache_page_index *page_index, time_t start_time_s)
-{
-    uv_rwlock_rdlock(&page_index->lock);
-    Pvoid_t *PValue = JudyLGet(page_index->JudyL_array, start_time_s, PJE0);
-    struct rrdeng_page_descr *descr = unlikely(NULL == PValue) ? NULL : *PValue;
-    uv_rwlock_rdunlock(&page_index->lock);
-    return descr;
-};
+WAL *wal_get(struct rrdengine_instance *ctx, unsigned size) {
+    if(!size || size > RRDENG_BLOCK_SIZE)
+        fatal("DBENGINE: invalid WAL size requested");
 
-static void do_extent_processing (struct rrdengine_worker_config *wc, struct extent_io_descriptor *xt_io_descr, bool read_failed)
-{
-    struct rrdengine_instance *ctx = wc->ctx;
-    struct page_cache *pg_cache = &ctx->pg_cache;
-    struct rrdeng_page_descr *descr;
-    struct page_cache_descr *pg_cache_descr;
-    int ret;
-    unsigned i, j, count;
-    void *page, *uncompressed_buf = NULL;
-    uint32_t payload_length, payload_offset, page_offset, uncompressed_payload_length = 0;
-    uint8_t have_read_error = 0;
-    /* persistent structures */
-    struct rrdeng_df_extent_header *header;
-    struct rrdeng_df_extent_trailer *trailer;
-    uLong crc;
+    WAL *wal = NULL;
 
-    header = xt_io_descr->buf;
-    payload_length = header->payload_length;
-    count = header->number_of_pages;
-    payload_offset = sizeof(*header) + sizeof(header->descr[0]) * count;
-    trailer = xt_io_descr->buf + xt_io_descr->bytes - sizeof(*trailer);
-
-    if (unlikely(read_failed)) {
-        struct rrdengine_datafile *datafile = xt_io_descr->descr_array[0]->extent->datafile;
-
-        ++ctx->stats.io_errors;
-        rrd_stat_atomic_add(&global_io_errors, 1);
-        have_read_error = 1;
-        error("%s: uv_fs_read - extent at offset %"PRIu64"(%u) in datafile %u-%u.", __func__, xt_io_descr->pos,
-              xt_io_descr->bytes, datafile->tier, datafile->fileno);
-        goto after_crc_check;
+    netdata_spinlock_lock(&wal_globals.protected.spinlock);
+
+    if(likely(wal_globals.protected.available_items)) {
+        wal = wal_globals.protected.available_items;
+        DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wal_globals.protected.available_items, wal, cache.prev, cache.next);
+        wal_globals.protected.available--;
     }
-    crc = crc32(0L, Z_NULL, 0);
-    crc = crc32(crc, xt_io_descr->buf, xt_io_descr->bytes - sizeof(*trailer));
-    ret = crc32cmp(trailer->checksum, crc);
-#ifdef NETDATA_INTERNAL_CHECKS
-    {
-        struct rrdengine_datafile *datafile = xt_io_descr->descr_array[0]->extent->datafile;
-        debug(D_RRDENGINE, "%s: Extent at offset %"PRIu64"(%u) was read from datafile %u-%u. CRC32 check: %s", __func__,
-              xt_io_descr->pos, xt_io_descr->bytes, datafile->tier, datafile->fileno, ret ? "FAILED" : "SUCCEEDED");
+
+    uint64_t transaction_id = __atomic_fetch_add(&ctx->atomic.transaction_id, 1, __ATOMIC_RELAXED);
+    netdata_spinlock_unlock(&wal_globals.protected.spinlock);
+
+    if(unlikely(!wal)) {
+        wal = mallocz(sizeof(WAL));
+        wal->buf_size = RRDENG_BLOCK_SIZE;
+        int ret = posix_memalign((void *)&wal->buf, RRDFILE_ALIGNMENT, wal->buf_size);
+        if (unlikely(ret))
+            fatal("DBENGINE: posix_memalign:%s", strerror(ret));
+        __atomic_add_fetch(&wal_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
     }
-#endif
-    if (unlikely(ret)) {
-        struct rrdengine_datafile *datafile = xt_io_descr->descr_array[0]->extent->datafile;
 
-        ++ctx->stats.io_errors;
-        rrd_stat_atomic_add(&global_io_errors, 1);
-        have_read_error = 1;
-        error("%s: Extent at offset %"PRIu64"(%u) was read from datafile %u-%u. CRC32 check: FAILED", __func__,
-              xt_io_descr->pos, xt_io_descr->bytes, datafile->tier, datafile->fileno);
+    // these need to survive
+    unsigned buf_size = wal->buf_size;
+    void *buf = wal->buf;
+
+    memset(wal, 0, sizeof(WAL));
+
+    // put them back
+    wal->buf_size = buf_size;
+    wal->buf = buf;
+
+    memset(wal->buf, 0, wal->buf_size);
+
+    wal->transaction_id = transaction_id;
+    wal->size = size;
+
+    return wal;
+}
+
+void wal_release(WAL *wal) {
+    if(unlikely(!wal)) return;
+
+    netdata_spinlock_lock(&wal_globals.protected.spinlock);
+    DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(wal_globals.protected.available_items, wal, cache.prev, cache.next);
+    wal_globals.protected.available++;
+    netdata_spinlock_unlock(&wal_globals.protected.spinlock);
+}
+
+// ----------------------------------------------------------------------------
+// command queue cache
+
+struct rrdeng_cmd {
+    struct rrdengine_instance *ctx;
+    enum rrdeng_opcode opcode;
+    void *data;
+    struct completion *completion;
+    enum storage_priority priority;
+    dequeue_callback_t dequeue_cb;
+
+    struct {
+        struct rrdeng_cmd *prev;
+        struct rrdeng_cmd *next;
+    } queue;
+};
+
+static void rrdeng_cmd_queue_init(void) {
+    rrdeng_main.cmd_queue.ar = aral_create("dbengine-opcodes",
+                                           sizeof(struct rrdeng_cmd),
+                                           0,
+                                           65536,
+                                           NULL,
+                                           NULL, NULL, false, false);
+}
+
+static inline STORAGE_PRIORITY rrdeng_enq_cmd_map_opcode_to_priority(enum rrdeng_opcode opcode, STORAGE_PRIORITY priority) {
+    if(unlikely(priority >= STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE))
+        priority = STORAGE_PRIORITY_BEST_EFFORT;
+
+    switch(opcode) {
+        case RRDENG_OPCODE_QUERY:
+            priority = STORAGE_PRIORITY_INTERNAL_QUERY_PREP;
+            break;
+
+        default:
+            break;
     }
 
-after_crc_check:
-    if (!have_read_error && RRD_NO_COMPRESSION != header->compression_algorithm) {
-        uncompressed_payload_length = 0;
-        for (i = 0 ; i < count ; ++i) {
-            uncompressed_payload_length += header->descr[i].page_length;
+    return priority;
+}
+
+void rrdeng_enqueue_epdl_cmd(struct rrdeng_cmd *cmd) {
+    epdl_cmd_queued(cmd->data, cmd);
+}
+
+void rrdeng_dequeue_epdl_cmd(struct rrdeng_cmd *cmd) {
+    epdl_cmd_dequeued(cmd->data);
+}
+
+void rrdeng_req_cmd(requeue_callback_t get_cmd_cb, void *data, STORAGE_PRIORITY priority) {
+    netdata_spinlock_lock(&rrdeng_main.cmd_queue.unsafe.spinlock);
+
+    struct rrdeng_cmd *cmd = get_cmd_cb(data);
+    if(cmd) {
+        priority = rrdeng_enq_cmd_map_opcode_to_priority(cmd->opcode, priority);
+
+        if (cmd->priority > priority) {
+            DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[cmd->priority], cmd, queue.prev, queue.next);
+            DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority], cmd, queue.prev, queue.next);
+            cmd->priority = priority;
         }
-        uncompressed_buf = mallocz(uncompressed_payload_length);
-        ret = LZ4_decompress_safe(xt_io_descr->buf + payload_offset, uncompressed_buf,
-                                  payload_length, uncompressed_payload_length);
-        ctx->stats.before_decompress_bytes += payload_length;
-        ctx->stats.after_decompress_bytes += ret;
-        debug(D_RRDENGINE, "LZ4 decompressed %u bytes to %d bytes.", payload_length, ret);
-        /* care, we don't hold the descriptor mutex */
     }
-    {
-        uint8_t xt_is_cached = 0;
-        unsigned xt_idx;
-        struct extent_info *extent = xt_io_descr->descr_array[0]->extent;
-
-        xt_is_cached = !lookup_in_xt_cache(wc, extent, &xt_idx);
-        if (xt_is_cached && check_bit(wc->xt_cache.inflight_bitmap, xt_idx)) {
-            struct extent_cache *xt_cache = &wc->xt_cache;
-            struct extent_cache_element *xt_cache_elem = &xt_cache->extent_array[xt_idx];
-            struct extent_io_descriptor *curr, *next;
-
-            if (have_read_error) {
-                memset(xt_cache_elem->pages, 0, sizeof(xt_cache_elem->pages));
-            } else if (RRD_NO_COMPRESSION == header->compression_algorithm) {
-                (void)memcpy(xt_cache_elem->pages, xt_io_descr->buf + payload_offset, payload_length);
-            } else {
-                (void)memcpy(xt_cache_elem->pages, uncompressed_buf, uncompressed_payload_length);
-            }
-            /* complete all connected in-flight read requests */
-            for (curr = xt_cache_elem->inflight_io_descr->next ; curr ; curr = next) {
-                next = curr->next;
-                read_cached_extent_cb(wc, xt_idx, curr);
+
+    netdata_spinlock_unlock(&rrdeng_main.cmd_queue.unsafe.spinlock);
+}
+
+void rrdeng_enq_cmd(struct rrdengine_instance *ctx, enum rrdeng_opcode opcode, void *data, struct completion *completion,
+               enum storage_priority priority, enqueue_callback_t enqueue_cb, dequeue_callback_t dequeue_cb) {
+
+    priority = rrdeng_enq_cmd_map_opcode_to_priority(opcode, priority);
+
+    struct rrdeng_cmd *cmd = aral_mallocz(rrdeng_main.cmd_queue.ar);
+    memset(cmd, 0, sizeof(struct rrdeng_cmd));
+    cmd->ctx = ctx;
+    cmd->opcode = opcode;
+    cmd->data = data;
+    cmd->completion = completion;
+    cmd->priority = priority;
+    cmd->dequeue_cb = dequeue_cb;
+
+    netdata_spinlock_lock(&rrdeng_main.cmd_queue.unsafe.spinlock);
+    DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority], cmd, queue.prev, queue.next);
+    rrdeng_main.cmd_queue.unsafe.waiting++;
+    if(enqueue_cb)
+        enqueue_cb(cmd);
+    netdata_spinlock_unlock(&rrdeng_main.cmd_queue.unsafe.spinlock);
+
+    fatal_assert(0 == uv_async_send(&rrdeng_main.async));
+}
+
+static inline bool rrdeng_cmd_has_waiting_opcodes_in_lower_priorities(STORAGE_PRIORITY priority, STORAGE_PRIORITY max_priority) {
+    for(; priority <= max_priority ; priority++)
+        if(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority])
+            return true;
+
+    return false;
+}
+
+static inline struct rrdeng_cmd rrdeng_deq_cmd(void) {
+    struct rrdeng_cmd *cmd = NULL;
+
+    STORAGE_PRIORITY max_priority = work_request_full() ? STORAGE_PRIORITY_INTERNAL_DBENGINE : STORAGE_PRIORITY_BEST_EFFORT;
+
+    // find an opcode to execute from the queue
+    netdata_spinlock_lock(&rrdeng_main.cmd_queue.unsafe.spinlock);
+    for(STORAGE_PRIORITY priority = STORAGE_PRIORITY_INTERNAL_DBENGINE; priority <= max_priority ; priority++) {
+        cmd = rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority];
+        if(cmd) {
+
+            // avoid starvation of lower priorities
+            if(unlikely(priority >= STORAGE_PRIORITY_HIGH &&
+                        priority < STORAGE_PRIORITY_BEST_EFFORT &&
+                        ++rrdeng_main.cmd_queue.unsafe.executed_by_priority[priority] % 50 == 0 &&
+                        rrdeng_cmd_has_waiting_opcodes_in_lower_priorities(priority + 1, max_priority))) {
+                // let the others run 2% of the requests
+                cmd = NULL;
+                continue;
             }
-            xt_cache_elem->inflight_io_descr = NULL;
-            modify_bit(&xt_cache->inflight_bitmap, xt_idx, 0); /* not in-flight anymore */
+
+            // remove it from the queue
+            DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority], cmd, queue.prev, queue.next);
+            rrdeng_main.cmd_queue.unsafe.waiting--;
+            break;
         }
     }
 
-    uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
-    Pvoid_t *PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, xt_io_descr->descr_array[0]->id, sizeof(uuid_t));
-    struct pg_cache_page_index *page_index = likely( NULL != PValue) ? *PValue : NULL;
-    uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
-
-
-    for (i = 0, page_offset = 0; i < count; page_offset += header->descr[i++].page_length) {
-        uint8_t is_prefetched_page;
-        descr = NULL;
-        for (j = 0 ; j < xt_io_descr->descr_count; ++j) {
-            struct rrdeng_page_descr descrj;
-
-            descrj = xt_io_descr->descr_read_array[j];
-            /* care, we don't hold the descriptor mutex */
-            if (!uuid_compare(*(uuid_t *) header->descr[i].uuid, *descrj.id) &&
-                header->descr[i].page_length == descrj.page_length &&
-                header->descr[i].start_time_ut == descrj.start_time_ut &&
-                header->descr[i].end_time_ut == descrj.end_time_ut) {
-                //descr = descrj;
-                descr = get_descriptor(page_index, (time_t) (descrj.start_time_ut / USEC_PER_SEC));
-                if (unlikely(!descr)) {
-                    error_limit_static_thread_var(erl, 1, 0);
-                    error_limit(&erl, "%s: Required descriptor is not in the page index anymore", __FUNCTION__);
-                }
-                break;
-            }
-        }
-        is_prefetched_page = 0;
-        if (!descr) { /* This extent page has not been requested. Try populating it for locality (best effort). */
-            descr = pg_cache_lookup_unpopulated_and_lock(ctx, (uuid_t *)header->descr[i].uuid,
-                                                         header->descr[i].start_time_ut);
-            if (!descr)
-                continue; /* Failed to reserve a suitable page */
-            is_prefetched_page = 1;
-        }
-        page = dbengine_page_alloc();
-
-        /* care, we don't hold the descriptor mutex */
-        if (have_read_error) {
-            fill_page_with_nulls(page, descr->page_length, descr->type);
-        } else if (RRD_NO_COMPRESSION == header->compression_algorithm) {
-            (void) memcpy(page, xt_io_descr->buf + payload_offset + page_offset, descr->page_length);
-        } else {
-            (void) memcpy(page, uncompressed_buf + page_offset, descr->page_length);
-        }
-        rrdeng_page_descr_mutex_lock(ctx, descr);
-        pg_cache_descr = descr->pg_cache_descr;
-        pg_cache_descr->page = page;
-        pg_cache_descr->flags |= RRD_PAGE_POPULATED;
-        pg_cache_descr->flags &= ~RRD_PAGE_READ_PENDING;
-        rrdeng_page_descr_mutex_unlock(ctx, descr);
-        pg_cache_replaceQ_insert(ctx, descr);
-        if (xt_io_descr->release_descr || is_prefetched_page) {
-            pg_cache_put(ctx, descr);
-        } else {
-            debug(D_RRDENGINE, "%s: Waking up waiters.", __func__);
-            pg_cache_wake_up_waiters(ctx, descr);
-        }
+    if(cmd && cmd->dequeue_cb) {
+        cmd->dequeue_cb(cmd);
+        cmd->dequeue_cb = NULL;
     }
-    if (!have_read_error && RRD_NO_COMPRESSION != header->compression_algorithm) {
-        freez(uncompressed_buf);
+
+    netdata_spinlock_unlock(&rrdeng_main.cmd_queue.unsafe.spinlock);
+
+    struct rrdeng_cmd ret;
+    if(cmd) {
+        // copy it, to return it
+        ret = *cmd;
+
+        aral_freez(rrdeng_main.cmd_queue.ar, cmd);
     }
-    if (xt_io_descr->completion)
-        completion_mark_complete(xt_io_descr->completion);
+    else
+        ret = (struct rrdeng_cmd) {
+                .ctx = NULL,
+                .opcode = RRDENG_OPCODE_NOOP,
+                .priority = STORAGE_PRIORITY_BEST_EFFORT,
+                .completion = NULL,
+                .data = NULL,
+        };
+
+    return ret;
 }
 
-static void read_extent_cb(uv_fs_t *req)
-{
-    struct rrdengine_worker_config *wc = req->loop->data;
-    struct extent_io_descriptor *xt_io_descr;
 
-    xt_io_descr = req->data;
-    do_extent_processing(wc, xt_io_descr, req->result < 0);
-    uv_fs_req_cleanup(req);
-    posix_memfree(xt_io_descr->buf);
-    freez(xt_io_descr);
+// ----------------------------------------------------------------------------
+
+struct {
+    ARAL *aral[RRD_STORAGE_TIERS];
+} dbengine_page_alloc_globals = {};
+
+static inline ARAL *page_size_lookup(size_t size) {
+    for(size_t tier = 0; tier < storage_tiers ;tier++)
+        if(size == tier_page_size[tier])
+            return dbengine_page_alloc_globals.aral[tier];
+
+    return NULL;
 }
 
-static void read_mmap_extent_cb(uv_work_t *req, int status __maybe_unused)
-{
-    struct rrdengine_worker_config *wc = req->loop->data;
-    struct rrdengine_instance *ctx = wc->ctx;
-    struct extent_io_descriptor *xt_io_descr;
-    xt_io_descr = req->data;
+static void dbengine_page_alloc_init(void) {
+    for(size_t i = storage_tiers; i > 0 ;i--) {
+        size_t tier = storage_tiers - i;
 
-    if (likely(xt_io_descr->map_base)) {
-        do_extent_processing(wc, xt_io_descr, false);
-        munmap(xt_io_descr->map_base, xt_io_descr->map_length);
-        freez(xt_io_descr);
-        return;
-    }
+        char buf[20 + 1];
+        snprintfz(buf, 20, "tier%zu-pages", tier);
 
-    // MMAP failed, so do uv_fs_read
-    int ret = posix_memalign((void *)&xt_io_descr->buf, RRDFILE_ALIGNMENT, ALIGN_BYTES_CEILING(xt_io_descr->bytes));
-    if (unlikely(ret)) {
-        fatal("posix_memalign:%s", strerror(ret));
+        dbengine_page_alloc_globals.aral[tier] = aral_create(
+                buf,
+                tier_page_size[tier],
+                64,
+                512 * tier_page_size[tier],
+                pgc_aral_statistics(),
+                NULL, NULL, false, false);
     }
-    unsigned real_io_size = ALIGN_BYTES_CEILING( xt_io_descr->bytes);
-    xt_io_descr->iov = uv_buf_init((void *)xt_io_descr->buf, real_io_size);
-    xt_io_descr->req.data = xt_io_descr;
-    ret = uv_fs_read(req->loop, &xt_io_descr->req, xt_io_descr->file, &xt_io_descr->iov, 1, (unsigned) xt_io_descr->pos, read_extent_cb);
-    fatal_assert(-1 != ret);
-    ctx->stats.io_read_bytes += real_io_size;
-    ctx->stats.io_read_extent_bytes += real_io_size;
 }
 
-static void do_mmap_read_extent(uv_work_t *req)
-{
-    struct extent_io_descriptor *xt_io_descr = (struct extent_io_descriptor * )req->data;
-    struct rrdengine_worker_config *wc = req->loop->data;
-    struct rrdengine_instance *ctx = wc->ctx;
-
-    off_t map_start =  ALIGN_BYTES_FLOOR(xt_io_descr->pos);
-    size_t length = ALIGN_BYTES_CEILING(xt_io_descr->pos + xt_io_descr->bytes) - map_start;
-    unsigned real_io_size = xt_io_descr->bytes;
-
-    void *data = mmap(NULL, length, PROT_READ, MAP_SHARED, xt_io_descr->file, map_start);
-    if (likely(data != MAP_FAILED)) {
-        xt_io_descr->map_base = data;
-        xt_io_descr->map_length = length;
-        xt_io_descr->buf = data + (xt_io_descr->pos - map_start);
-        ctx->stats.io_read_bytes += real_io_size;
-        ctx->stats.io_read_extent_bytes += real_io_size;
-    }
+void *dbengine_page_alloc(size_t size) {
+    ARAL *ar = page_size_lookup(size);
+    if(ar) return aral_mallocz(ar);
+
+    return mallocz(size);
 }
 
-static void do_read_extent(struct rrdengine_worker_config* wc,
-                           struct rrdeng_page_descr **descr,
-                           unsigned count,
-                           uint8_t release_descr)
-{
-    struct rrdengine_instance *ctx = wc->ctx;
-    struct page_cache_descr *pg_cache_descr;
-    int ret;
-    unsigned i, size_bytes, pos;
-    struct extent_io_descriptor *xt_io_descr;
-    struct rrdengine_datafile *datafile;
-    struct extent_info *extent = descr[0]->extent;
-    uint8_t xt_is_cached = 0, xt_is_inflight = 0;
-    unsigned xt_idx;
-
-    datafile = extent->datafile;
-    pos = extent->offset;
-    size_bytes = extent->size;
-
-    xt_io_descr = callocz(1, sizeof(*xt_io_descr));
-    for (i = 0 ; i < count; ++i) {
-        rrdeng_page_descr_mutex_lock(ctx, descr[i]);
-        pg_cache_descr = descr[i]->pg_cache_descr;
-        pg_cache_descr->flags |= RRD_PAGE_READ_PENDING;
-        rrdeng_page_descr_mutex_unlock(ctx, descr[i]);
-        xt_io_descr->descr_array[i] = descr[i];
-        xt_io_descr->descr_read_array[i] = *(descr[i]);
-    }
-    xt_io_descr->descr_count = count;
-    xt_io_descr->file = datafile->file;
-    xt_io_descr->bytes = size_bytes;
-    xt_io_descr->pos = pos;
-    xt_io_descr->req_worker.data = xt_io_descr;
-    xt_io_descr->completion = NULL;
-    xt_io_descr->release_descr = release_descr;
-    xt_io_descr->buf = NULL;
-
-    xt_is_cached = !lookup_in_xt_cache(wc, extent, &xt_idx);
-    if (xt_is_cached) {
-        xt_cache_replaceQ_set_hot(wc, &wc->xt_cache.extent_array[xt_idx]);
-        xt_is_inflight = check_bit(wc->xt_cache.inflight_bitmap, xt_idx);
-        if (xt_is_inflight) {
-            enqueue_inflight_read_to_xt_cache(wc, xt_idx, xt_io_descr);
-            return;
-        }
-        return read_cached_extent_cb(wc, xt_idx, xt_io_descr);
-    } else {
-        ret = try_insert_into_xt_cache(wc, extent);
-        if (-1 != ret) {
-            xt_idx = (unsigned)ret;
-            modify_bit(&wc->xt_cache.inflight_bitmap, xt_idx, 1);
-            wc->xt_cache.extent_array[xt_idx].inflight_io_descr = xt_io_descr;
-        }
-    }
+void dbengine_page_free(void *page, size_t size __maybe_unused) {
+    if(unlikely(!page || page == DBENGINE_EMPTY_PAGE))
+        return;
 
-    ret = uv_queue_work(wc->loop, &xt_io_descr->req_worker, do_mmap_read_extent, read_mmap_extent_cb);
-    fatal_assert(-1 != ret);
+    ARAL *ar = page_size_lookup(size);
+    if(ar)
+        aral_freez(ar, page);
+    else
+        freez(page);
+}
 
-    ++ctx->stats.io_read_requests;
-    ++ctx->stats.io_read_extents;
-    ctx->stats.pg_cache_backfills += count;
+// ----------------------------------------------------------------------------
+
+void *dbengine_extent_alloc(size_t size) {
+    void *extent = mallocz(size);
+    return extent;
 }
 
-static void commit_data_extent(struct rrdengine_worker_config* wc, struct extent_io_descriptor *xt_io_descr)
-{
-    struct rrdengine_instance *ctx = wc->ctx;
+void dbengine_extent_free(void *extent, size_t size __maybe_unused) {
+    freez(extent);
+}
+
+static void journalfile_extent_build(struct rrdengine_instance *ctx, struct extent_io_descriptor *xt_io_descr) {
     unsigned count, payload_length, descr_size, size_bytes;
     void *buf;
     /* persistent structures */
@@ -582,12 +596,13 @@ static void commit_data_extent(struct rrdengine_worker_config* wc, struct extent
     payload_length = sizeof(*jf_metric_data) + descr_size;
     size_bytes = sizeof(*jf_header) + payload_length + sizeof(*jf_trailer);
 
-    buf = wal_get_transaction_buffer(wc, size_bytes);
+    xt_io_descr->wal = wal_get(ctx, size_bytes);
+    buf = xt_io_descr->wal->buf;
 
     jf_header = buf;
     jf_header->type = STORE_DATA;
     jf_header->reserved = 0;
-    jf_header->id = ctx->commit_log.transaction_id++;
+    jf_header->id = xt_io_descr->wal->transaction_id;
     jf_header->payload_length = payload_length;
 
     jf_metric_data = buf + sizeof(*jf_header);
@@ -602,265 +617,210 @@ static void commit_data_extent(struct rrdengine_worker_config* wc, struct extent
     crc32set(jf_trailer->checksum, crc);
 }
 
-static void do_commit_transaction(struct rrdengine_worker_config* wc, uint8_t type, void *data)
-{
-    switch (type) {
-    case STORE_DATA:
-        commit_data_extent(wc, (struct extent_io_descriptor *)data);
-        break;
-    default:
-        fatal_assert(type == STORE_DATA);
-        break;
-    }
-}
+static void after_extent_flushed_to_open(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    if(completion)
+        completion_mark_complete(completion);
 
-static void after_invalidate_oldest_committed(struct rrdengine_worker_config* wc)
-{
-    int error;
-
-    error = uv_thread_join(wc->now_invalidating_dirty_pages);
-    if (error) {
-        error("uv_thread_join(): %s", uv_strerror(error));
-    }
-    freez(wc->now_invalidating_dirty_pages);
-    wc->now_invalidating_dirty_pages = NULL;
-    wc->cleanup_thread_invalidating_dirty_pages = 0;
+    if(ctx_is_available_for_queries(ctx))
+        rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
 }
 
-static void invalidate_oldest_committed(void *arg)
-{
-    struct rrdengine_instance *ctx = arg;
-    struct rrdengine_worker_config *wc = &ctx->worker_config;
-    struct page_cache *pg_cache = &ctx->pg_cache;
-    int ret;
-    struct rrdeng_page_descr *descr;
-    struct page_cache_descr *pg_cache_descr;
-    Pvoid_t *PValue;
-    Word_t Index;
-    unsigned nr_committed_pages;
+static void *extent_flushed_to_open_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    worker_is_busy(UV_EVENT_DBENGINE_FLUSHED_TO_OPEN);
 
-    do {
-        uv_rwlock_wrlock(&pg_cache->committed_page_index.lock);
-        for (Index = 0,
-             PValue = JudyLFirst(pg_cache->committed_page_index.JudyL_array, &Index, PJE0),
-             descr = unlikely(NULL == PValue) ? NULL : *PValue;
+    uv_fs_t *uv_fs_request = data;
+    struct extent_io_descriptor *xt_io_descr = uv_fs_request->data;
+    struct page_descr_with_data *descr;
+    struct rrdengine_datafile *datafile;
+    unsigned i;
 
-             descr != NULL;
+    datafile = xt_io_descr->datafile;
 
-             PValue = JudyLNext(pg_cache->committed_page_index.JudyL_array, &Index, PJE0),
-                     descr = unlikely(NULL == PValue) ? NULL : *PValue) {
-            fatal_assert(0 != descr->page_length);
+    bool still_running = ctx_is_available_for_queries(ctx);
 
-            rrdeng_page_descr_mutex_lock(ctx, descr);
-            pg_cache_descr = descr->pg_cache_descr;
-            if (!(pg_cache_descr->flags & RRD_PAGE_WRITE_PENDING) && pg_cache_try_get_unsafe(descr, 1)) {
-                rrdeng_page_descr_mutex_unlock(ctx, descr);
+    for (i = 0 ; i < xt_io_descr->descr_count ; ++i) {
+        descr = xt_io_descr->descr_array[i];
 
-                ret = JudyLDel(&pg_cache->committed_page_index.JudyL_array, Index, PJE0);
-                fatal_assert(1 == ret);
-                break;
-            }
-            rrdeng_page_descr_mutex_unlock(ctx, descr);
-        }
-        uv_rwlock_wrunlock(&pg_cache->committed_page_index.lock);
+        if (likely(still_running))
+            pgc_open_add_hot_page(
+                    (Word_t)ctx, descr->metric_id,
+                    (time_t) (descr->start_time_ut / USEC_PER_SEC),
+                    (time_t) (descr->end_time_ut / USEC_PER_SEC),
+                    descr->update_every_s,
+                    datafile,
+                    xt_io_descr->pos, xt_io_descr->bytes, descr->page_length);
 
-        if (!descr) {
-            info("Failed to invalidate any dirty pages to relieve page cache pressure.");
+        page_descriptor_release(descr);
+    }
 
-            goto out;
-        }
-        pg_cache_punch_hole(ctx, descr, 1, 1, NULL);
+    uv_fs_req_cleanup(uv_fs_request);
+    posix_memfree(xt_io_descr->buf);
+    extent_io_descriptor_release(xt_io_descr);
 
-        uv_rwlock_wrlock(&pg_cache->committed_page_index.lock);
-        nr_committed_pages = --pg_cache->committed_page_index.nr_committed_pages;
-        uv_rwlock_wrunlock(&pg_cache->committed_page_index.lock);
-        rrd_stat_atomic_add(&ctx->stats.flushing_pressure_page_deletions, 1);
-        rrd_stat_atomic_add(&global_flushing_pressure_page_deletions, 1);
+    netdata_spinlock_lock(&datafile->writers.spinlock);
+    datafile->writers.flushed_to_open_running--;
+    netdata_spinlock_unlock(&datafile->writers.spinlock);
 
-    } while (nr_committed_pages >= pg_cache_committed_hard_limit(ctx));
-out:
-    wc->cleanup_thread_invalidating_dirty_pages = 1;
-    /* wake up event loop */
-    fatal_assert(0 == uv_async_send(&wc->async));
-}
+    if(datafile->fileno != ctx_last_fileno_get(ctx) && still_running)
+        // we just finished a flushing on a datafile that is not the active one
+        rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_INDEX, datafile, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
 
-void rrdeng_invalidate_oldest_committed(struct rrdengine_worker_config* wc)
-{
-    struct rrdengine_instance *ctx = wc->ctx;
-    struct page_cache *pg_cache = &ctx->pg_cache;
-    unsigned nr_committed_pages;
-    int error;
+    return data;
+}
 
-    if (unlikely(ctx->quiesce != NO_QUIESCE)) /* Shutting down */
-        return;
+// Main event loop callback
+static void after_extent_write_datafile_io(uv_fs_t *uv_fs_request) {
+    worker_is_busy(RRDENG_OPCODE_MAX + RRDENG_OPCODE_EXTENT_WRITE);
 
-    uv_rwlock_rdlock(&pg_cache->committed_page_index.lock);
-    nr_committed_pages = pg_cache->committed_page_index.nr_committed_pages;
-    uv_rwlock_rdunlock(&pg_cache->committed_page_index.lock);
+    struct extent_io_descriptor *xt_io_descr = uv_fs_request->data;
+    struct rrdengine_datafile *datafile = xt_io_descr->datafile;
+    struct rrdengine_instance *ctx = datafile->ctx;
 
-    if (nr_committed_pages >= pg_cache_committed_hard_limit(ctx)) {
-        /* delete the oldest page in memory */
-        if (wc->now_invalidating_dirty_pages) {
-            /* already deleting a page */
-            return;
-        }
-        errno = 0;
-        error("Failed to flush dirty buffers quickly enough in dbengine instance \"%s\". "
-              "Metric data are being deleted, please reduce disk load or use a faster disk.", ctx->dbfiles_path);
-
-        wc->now_invalidating_dirty_pages = mallocz(sizeof(*wc->now_invalidating_dirty_pages));
-        wc->cleanup_thread_invalidating_dirty_pages = 0;
-
-        error = uv_thread_create(wc->now_invalidating_dirty_pages, invalidate_oldest_committed, ctx);
-        if (error) {
-            error("uv_thread_create(): %s", uv_strerror(error));
-            freez(wc->now_invalidating_dirty_pages);
-            wc->now_invalidating_dirty_pages = NULL;
-        }
+    if (uv_fs_request->result < 0) {
+        ctx_io_error(ctx);
+        error("DBENGINE: %s: uv_fs_write(): %s", __func__, uv_strerror((int)uv_fs_request->result));
     }
+
+    journalfile_v1_extent_write(ctx, xt_io_descr->datafile, xt_io_descr->wal, &rrdeng_main.loop);
+
+    netdata_spinlock_lock(&datafile->writers.spinlock);
+    datafile->writers.running--;
+    datafile->writers.flushed_to_open_running++;
+    netdata_spinlock_unlock(&datafile->writers.spinlock);
+
+    rrdeng_enq_cmd(xt_io_descr->ctx,
+                   RRDENG_OPCODE_FLUSHED_TO_OPEN,
+                   uv_fs_request,
+                   xt_io_descr->completion,
+                   STORAGE_PRIORITY_INTERNAL_DBENGINE,
+                   NULL,
+                   NULL);
+
+    worker_is_idle();
 }
 
-void flush_pages_cb(uv_fs_t* req)
-{
-    struct rrdengine_worker_config* wc = req->loop->data;
-    struct rrdengine_instance *ctx = wc->ctx;
-    struct page_cache *pg_cache = &ctx->pg_cache;
-    struct extent_io_descriptor *xt_io_descr;
-    struct rrdeng_page_descr *descr;
-    struct page_cache_descr *pg_cache_descr;
-    unsigned i, count;
-
-    xt_io_descr = req->data;
-    if (req->result < 0) {
-        ++ctx->stats.io_errors;
-        rrd_stat_atomic_add(&global_io_errors, 1);
-        error("%s: uv_fs_write: %s", __func__, uv_strerror((int)req->result));
-    }
-#ifdef NETDATA_INTERNAL_CHECKS
-    {
-        struct rrdengine_datafile *datafile = xt_io_descr->descr_array[0]->extent->datafile;
-        debug(D_RRDENGINE, "%s: Extent at offset %"PRIu64"(%u) was written to datafile %u-%u. Waking up waiters.",
-              __func__, xt_io_descr->pos, xt_io_descr->bytes, datafile->tier, datafile->fileno);
-    }
-#endif
-    count = xt_io_descr->descr_count;
-    for (i = 0 ; i < count ; ++i) {
-        /* care, we don't hold the descriptor mutex */
-        descr = xt_io_descr->descr_array[i];
+static bool datafile_is_full(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile) {
+    bool ret = false;
+    netdata_spinlock_lock(&datafile->writers.spinlock);
 
-        pg_cache_replaceQ_insert(ctx, descr);
+    if(ctx_is_available_for_queries(ctx) && datafile->pos > rrdeng_target_data_file_size(ctx))
+        ret = true;
 
-        rrdeng_page_descr_mutex_lock(ctx, descr);
-        pg_cache_descr = descr->pg_cache_descr;
-        pg_cache_descr->flags &= ~(RRD_PAGE_DIRTY | RRD_PAGE_WRITE_PENDING);
-        /* wake up waiters, care no reference being held */
-        pg_cache_wake_up_waiters_unsafe(descr);
-        rrdeng_page_descr_mutex_unlock(ctx, descr);
-    }
-    if (xt_io_descr->completion)
-        completion_mark_complete(xt_io_descr->completion);
-    uv_fs_req_cleanup(req);
-    posix_memfree(xt_io_descr->buf);
-    freez(xt_io_descr);
+    netdata_spinlock_unlock(&datafile->writers.spinlock);
+
+    return ret;
+}
+
+static struct rrdengine_datafile *get_datafile_to_write_extent(struct rrdengine_instance *ctx) {
+    struct rrdengine_datafile *datafile;
 
-    uv_rwlock_wrlock(&pg_cache->committed_page_index.lock);
-    pg_cache->committed_page_index.nr_committed_pages -= count;
-    uv_rwlock_wrunlock(&pg_cache->committed_page_index.lock);
-    wc->inflight_dirty_pages -= count;
+    // get the latest datafile
+    uv_rwlock_rdlock(&ctx->datafiles.rwlock);
+    datafile = ctx->datafiles.first->prev;
+    // become a writer on this datafile, to prevent it from vanishing
+    netdata_spinlock_lock(&datafile->writers.spinlock);
+    datafile->writers.running++;
+    netdata_spinlock_unlock(&datafile->writers.spinlock);
+    uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
+
+    if(datafile_is_full(ctx, datafile)) {
+        // remember the datafile we have become writers to
+        struct rrdengine_datafile *old_datafile = datafile;
+
+        // only 1 datafile creation at a time
+        static netdata_mutex_t mutex = NETDATA_MUTEX_INITIALIZER;
+        netdata_mutex_lock(&mutex);
+
+        // take the latest datafile again - without this, multiple threads may create multiple files
+        uv_rwlock_rdlock(&ctx->datafiles.rwlock);
+        datafile = ctx->datafiles.first->prev;
+        uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
+
+        if(datafile_is_full(ctx, datafile) && create_new_datafile_pair(ctx) == 0)
+            rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_INDEX, datafile, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL,
+                           NULL);
+
+        netdata_mutex_unlock(&mutex);
+
+        // get the new latest datafile again, like above
+        uv_rwlock_rdlock(&ctx->datafiles.rwlock);
+        datafile = ctx->datafiles.first->prev;
+        // become a writer on this datafile, to prevent it from vanishing
+        netdata_spinlock_lock(&datafile->writers.spinlock);
+        datafile->writers.running++;
+        netdata_spinlock_unlock(&datafile->writers.spinlock);
+        uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
+
+        // release the writers on the old datafile
+        netdata_spinlock_lock(&old_datafile->writers.spinlock);
+        old_datafile->writers.running--;
+        netdata_spinlock_unlock(&old_datafile->writers.spinlock);
+    }
+
+    return datafile;
 }
 
 /*
- * completion must be NULL or valid.
- * Returns 0 when no flushing can take place.
- * Returns datafile bytes to be written on successful flushing initiation.
+ * Take a page list in a judy array and write them
  */
-static int do_flush_pages(struct rrdengine_worker_config* wc, int force, struct completion *completion)
-{
-    struct rrdengine_instance *ctx = wc->ctx;
-    struct page_cache *pg_cache = &ctx->pg_cache;
+static struct extent_io_descriptor *datafile_extent_build(struct rrdengine_instance *ctx, struct page_descr_with_data *base, struct completion *completion) {
     int ret;
     int compressed_size, max_compressed_size = 0;
     unsigned i, count, size_bytes, pos, real_io_size;
     uint32_t uncompressed_payload_length, payload_offset;
-    struct rrdeng_page_descr *descr, *eligible_pages[MAX_PAGES_PER_EXTENT];
-    struct page_cache_descr *pg_cache_descr;
+    struct page_descr_with_data *descr, *eligible_pages[MAX_PAGES_PER_EXTENT];
     struct extent_io_descriptor *xt_io_descr;
+    struct extent_buffer *eb = NULL;
     void *compressed_buf = NULL;
-    Word_t descr_commit_idx_array[MAX_PAGES_PER_EXTENT];
-    Pvoid_t *PValue;
     Word_t Index;
-    uint8_t compression_algorithm = ctx->global_compress_alg;
-    struct extent_info *extent;
+    uint8_t compression_algorithm = ctx->config.global_compress_alg;
     struct rrdengine_datafile *datafile;
     /* persistent structures */
     struct rrdeng_df_extent_header *header;
     struct rrdeng_df_extent_trailer *trailer;
     uLong crc;
 
-    if (force) {
-        debug(D_RRDENGINE, "Asynchronous flushing of extent has been forced by page pressure.");
-    }
-    uv_rwlock_wrlock(&pg_cache->committed_page_index.lock);
-    for (Index = 0, count = 0, uncompressed_payload_length = 0,
-         PValue = JudyLFirst(pg_cache->committed_page_index.JudyL_array, &Index, PJE0),
-         descr = unlikely(NULL == PValue) ? NULL : *PValue ;
-
-         descr != NULL && count != rrdeng_pages_per_extent;
-
-         PValue = JudyLNext(pg_cache->committed_page_index.JudyL_array, &Index, PJE0),
-         descr = unlikely(NULL == PValue) ? NULL : *PValue) {
-        uint8_t page_write_pending;
-
-        fatal_assert(0 != descr->page_length);
-        page_write_pending = 0;
-
-        rrdeng_page_descr_mutex_lock(ctx, descr);
-        pg_cache_descr = descr->pg_cache_descr;
-        if (!(pg_cache_descr->flags & RRD_PAGE_WRITE_PENDING)) {
-            page_write_pending = 1;
-            /* care, no reference being held */
-            pg_cache_descr->flags |= RRD_PAGE_WRITE_PENDING;
-            uncompressed_payload_length += descr->page_length;
-            descr_commit_idx_array[count] = Index;
-            eligible_pages[count++] = descr;
-        }
-        rrdeng_page_descr_mutex_unlock(ctx, descr);
+    for(descr = base, Index = 0, count = 0, uncompressed_payload_length = 0;
+        descr && count != rrdeng_pages_per_extent;
+        descr = descr->link.next, Index++) {
+
+        uncompressed_payload_length += descr->page_length;
+        eligible_pages[count++] = descr;
 
-        if (page_write_pending) {
-            ret = JudyLDel(&pg_cache->committed_page_index.JudyL_array, Index, PJE0);
-            fatal_assert(1 == ret);
-        }
     }
-    uv_rwlock_wrunlock(&pg_cache->committed_page_index.lock);
 
     if (!count) {
-        debug(D_RRDENGINE, "%s: no pages eligible for flushing.", __func__);
         if (completion)
             completion_mark_complete(completion);
-        return 0;
+
+        __atomic_sub_fetch(&ctx->atomic.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
+        return NULL;
     }
-    wc->inflight_dirty_pages += count;
 
-    xt_io_descr = mallocz(sizeof(*xt_io_descr));
+    xt_io_descr = extent_io_descriptor_get();
+    xt_io_descr->ctx = ctx;
     payload_offset = sizeof(*header) + count * sizeof(header->descr[0]);
     switch (compression_algorithm) {
-    case RRD_NO_COMPRESSION:
-        size_bytes = payload_offset + uncompressed_payload_length + sizeof(*trailer);
-        break;
-    default: /* Compress */
-        fatal_assert(uncompressed_payload_length < LZ4_MAX_INPUT_SIZE);
-        max_compressed_size = LZ4_compressBound(uncompressed_payload_length);
-        compressed_buf = mallocz(max_compressed_size);
-        size_bytes = payload_offset + MAX(uncompressed_payload_length, (unsigned)max_compressed_size) + sizeof(*trailer);
-        break;
+        case RRD_NO_COMPRESSION:
+            size_bytes = payload_offset + uncompressed_payload_length + sizeof(*trailer);
+            break;
+
+        default: /* Compress */
+            fatal_assert(uncompressed_payload_length < LZ4_MAX_INPUT_SIZE);
+            max_compressed_size = LZ4_compressBound(uncompressed_payload_length);
+            eb = extent_buffer_get(max_compressed_size);
+            compressed_buf = eb->data;
+            size_bytes = payload_offset + MAX(uncompressed_payload_length, (unsigned)max_compressed_size) + sizeof(*trailer);
+            break;
     }
+
     ret = posix_memalign((void *)&xt_io_descr->buf, RRDFILE_ALIGNMENT, ALIGN_BYTES_CEILING(size_bytes));
     if (unlikely(ret)) {
-        fatal("posix_memalign:%s", strerror(ret));
+        fatal("DBENGINE: posix_memalign:%s", strerror(ret));
         /* freez(xt_io_descr);*/
     }
     memset(xt_io_descr->buf, 0, ALIGN_BYTES_CEILING(size_bytes));
-    (void) memcpy(xt_io_descr->descr_array, eligible_pages, sizeof(struct rrdeng_page_descr *) * count);
+    (void) memcpy(xt_io_descr->descr_array, eligible_pages, sizeof(struct page_descr_with_data *) * count);
     xt_io_descr->descr_count = count;
 
     pos = 0;
@@ -869,17 +829,7 @@ static int do_flush_pages(struct rrdengine_worker_config* wc, int force, struct
     header->number_of_pages = count;
     pos += sizeof(*header);
 
-    extent = mallocz(sizeof(*extent) + count * sizeof(extent->pages[0]));
-    datafile = ctx->datafiles.last;  /* TODO: check for exceeded size quota */
-    extent->offset = datafile->pos;
-    extent->number_of_pages = count;
-    extent->datafile = datafile;
-    extent->next = NULL;
-
     for (i = 0 ; i < count ; ++i) {
-        /* This is here for performance reasons */
-        xt_io_descr->descr_commit_idx_array[i] = descr_commit_idx_array[i];
-
         descr = xt_io_descr->descr_array[i];
         header->descr[i].type = descr->type;
         uuid_copy(*(uuid_t *)header->descr[i].uuid, *descr->id);
@@ -890,35 +840,40 @@ static int do_flush_pages(struct rrdengine_worker_config* wc, int force, struct
     }
     for (i = 0 ; i < count ; ++i) {
         descr = xt_io_descr->descr_array[i];
-        /* care, we don't hold the descriptor mutex */
-        (void) memcpy(xt_io_descr->buf + pos, descr->pg_cache_descr->page, descr->page_length);
-        descr->extent = extent;
-        extent->pages[i] = descr;
-
+        (void) memcpy(xt_io_descr->buf + pos, descr->page, descr->page_length);
         pos += descr->page_length;
     }
-    df_extent_insert(extent);
 
-    switch (compression_algorithm) {
-    case RRD_NO_COMPRESSION:
-        header->payload_length = uncompressed_payload_length;
-        break;
-    default: /* Compress */
-        compressed_size = LZ4_compress_default(xt_io_descr->buf + payload_offset, compressed_buf,
-                                               uncompressed_payload_length, max_compressed_size);
-        ctx->stats.before_compress_bytes += uncompressed_payload_length;
-        ctx->stats.after_compress_bytes += compressed_size;
-        debug(D_RRDENGINE, "LZ4 compressed %"PRIu32" bytes to %d bytes.", uncompressed_payload_length, compressed_size);
+    if(likely(compression_algorithm == RRD_LZ4)) {
+        compressed_size = LZ4_compress_default(
+                xt_io_descr->buf + payload_offset,
+                compressed_buf,
+                (int)uncompressed_payload_length,
+                max_compressed_size);
+
+        __atomic_add_fetch(&ctx->stats.before_compress_bytes, uncompressed_payload_length, __ATOMIC_RELAXED);
+        __atomic_add_fetch(&ctx->stats.after_compress_bytes, compressed_size, __ATOMIC_RELAXED);
+
         (void) memcpy(xt_io_descr->buf + payload_offset, compressed_buf, compressed_size);
-        freez(compressed_buf);
+        extent_buffer_release(eb);
         size_bytes = payload_offset + compressed_size + sizeof(*trailer);
         header->payload_length = compressed_size;
-        break;
     }
-    extent->size = size_bytes;
-    xt_io_descr->bytes = size_bytes;
+    else { // RRD_NO_COMPRESSION
+        header->payload_length = uncompressed_payload_length;
+    }
+
+    real_io_size = ALIGN_BYTES_CEILING(size_bytes);
+
+    datafile = get_datafile_to_write_extent(ctx);
+    netdata_spinlock_lock(&datafile->writers.spinlock);
+    xt_io_descr->datafile = datafile;
     xt_io_descr->pos = datafile->pos;
-    xt_io_descr->req.data = xt_io_descr;
+    datafile->pos += real_io_size;
+    netdata_spinlock_unlock(&datafile->writers.spinlock);
+
+    xt_io_descr->bytes = size_bytes;
+    xt_io_descr->uv_fs_request.data = xt_io_descr;
     xt_io_descr->completion = completion;
 
     trailer = xt_io_descr->buf + size_bytes - sizeof(*trailer);
@@ -926,324 +881,508 @@ static int do_flush_pages(struct rrdengine_worker_config* wc, int force, struct
     crc = crc32(crc, xt_io_descr->buf, size_bytes - sizeof(*trailer));
     crc32set(trailer->checksum, crc);
 
-    real_io_size = ALIGN_BYTES_CEILING(size_bytes);
     xt_io_descr->iov = uv_buf_init((void *)xt_io_descr->buf, real_io_size);
-    ret = uv_fs_write(wc->loop, &xt_io_descr->req, datafile->file, &xt_io_descr->iov, 1, datafile->pos, flush_pages_cb);
-    fatal_assert(-1 != ret);
-    ctx->stats.io_write_bytes += real_io_size;
-    ++ctx->stats.io_write_requests;
-    ctx->stats.io_write_extent_bytes += real_io_size;
-    ++ctx->stats.io_write_extents;
-    do_commit_transaction(wc, STORE_DATA, xt_io_descr);
-    datafile->pos += ALIGN_BYTES_CEILING(size_bytes);
-    ctx->disk_space += ALIGN_BYTES_CEILING(size_bytes);
-    rrdeng_test_quota(wc);
+    journalfile_extent_build(ctx, xt_io_descr);
+
+    ctx_last_flush_fileno_set(ctx, datafile->fileno);
+    ctx_current_disk_space_increase(ctx, real_io_size);
+    ctx_io_write_op_bytes(ctx, real_io_size);
 
-    return ALIGN_BYTES_CEILING(size_bytes);
+    return xt_io_descr;
 }
 
-static void after_delete_old_data(struct rrdengine_worker_config* wc)
-{
-    struct rrdengine_instance *ctx = wc->ctx;
-    struct rrdengine_datafile *datafile;
-    struct rrdengine_journalfile *journalfile;
-    unsigned deleted_bytes, journalfile_bytes, datafile_bytes;
-    int ret, error;
-    char path[RRDENG_PATH_MAX];
+static void after_extent_write(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* uv_work_req __maybe_unused, int status __maybe_unused) {
+    struct extent_io_descriptor *xt_io_descr = data;
 
-    datafile = ctx->datafiles.first;
-    journalfile = datafile->journalfile;
-    datafile_bytes = datafile->pos;
-    journalfile_bytes = journalfile->pos;
-    deleted_bytes = 0;
+    if(xt_io_descr) {
+        int ret = uv_fs_write(&rrdeng_main.loop,
+                              &xt_io_descr->uv_fs_request,
+                              xt_io_descr->datafile->file,
+                              &xt_io_descr->iov,
+                              1,
+                              (int64_t) xt_io_descr->pos,
+                              after_extent_write_datafile_io);
 
-    info("Deleting data and journal file pair.");
-    datafile_list_delete(ctx, datafile);
-    ret = destroy_journal_file(journalfile, datafile);
-    if (!ret) {
-        generate_journalfilepath(datafile, path, sizeof(path));
-        info("Deleted journal file \"%s\".", path);
-        deleted_bytes += journalfile_bytes;
-    }
-    ret = destroy_data_file(datafile);
-    if (!ret) {
-        generate_datafilepath(datafile, path, sizeof(path));
-        info("Deleted data file \"%s\".", path);
-        deleted_bytes += datafile_bytes;
+        fatal_assert(-1 != ret);
     }
-    freez(journalfile);
-    freez(datafile);
+}
 
-    ctx->disk_space -= deleted_bytes;
-    info("Reclaimed %u bytes of disk space.", deleted_bytes);
+static void *extent_write_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    worker_is_busy(UV_EVENT_DBENGINE_EXTENT_WRITE);
+    struct page_descr_with_data *base = data;
+    struct extent_io_descriptor *xt_io_descr = datafile_extent_build(ctx, base, completion);
+    return xt_io_descr;
+}
 
-    error = uv_thread_join(wc->now_deleting_files);
-    if (error) {
-        error("uv_thread_join(): %s", uv_strerror(error));
-    }
-    freez(wc->now_deleting_files);
-    /* unfreeze command processing */
-    wc->now_deleting_files = NULL;
+static void after_database_rotate(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    __atomic_store_n(&ctx->atomic.now_deleting_files, false, __ATOMIC_RELAXED);
+}
 
-    wc->cleanup_thread_deleting_files = 0;
-    rrdcontext_db_rotation();
+struct uuid_first_time_s {
+    uuid_t *uuid;
+    time_t first_time_s;
+    METRIC *metric;
+    size_t pages_found;
+    size_t df_matched;
+    size_t df_index_oldest;
+};
 
-    /* interrupt event loop */
-    uv_stop(wc->loop);
+static int journal_metric_compare(const void *key, const void *metric)
+{
+    return uuid_compare(*(uuid_t *) key, ((struct journal_metric_list *) metric)->uuid);
 }
 
-static void delete_old_data(void *arg)
-{
-    struct rrdengine_instance *ctx = arg;
-    struct rrdengine_worker_config* wc = &ctx->worker_config;
-    struct rrdengine_datafile *datafile;
-    struct extent_info *extent, *next;
-    struct rrdeng_page_descr *descr;
-    unsigned count, i;
-    uint8_t can_delete_metric;
-    uuid_t metric_id;
-
-    /* Safe to use since it will be deleted after we are done */
-    datafile = ctx->datafiles.first;
-
-    for (extent = datafile->extents.first ; extent != NULL ; extent = next) {
-        count = extent->number_of_pages;
-        for (i = 0 ; i < count ; ++i) {
-            descr = extent->pages[i];
-            can_delete_metric = pg_cache_punch_hole(ctx, descr, 0, 0, &metric_id);
-            if (unlikely(can_delete_metric)) {
-                /*
-                 * If the metric is empty, has no active writers and if the metadata log has been initialized then
-                 * attempt to delete the corresponding netdata dimension.
-                 */
-                metaqueue_delete_dimension_uuid(&metric_id);
-            }
-        }
-        next = extent->next;
-        freez(extent);
-    }
-    wc->cleanup_thread_deleting_files = 1;
-    /* wake up event loop */
-    fatal_assert(0 == uv_async_send(&wc->async));
+struct rrdengine_datafile *datafile_release_and_acquire_next_for_retention(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile) {
+
+    uv_rwlock_rdlock(&ctx->datafiles.rwlock);
+
+    struct rrdengine_datafile *next_datafile = datafile->next;
+
+    while(next_datafile && !datafile_acquire(next_datafile, DATAFILE_ACQUIRE_RETENTION))
+        next_datafile = next_datafile->next;
+
+    uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
+
+    datafile_release(datafile, DATAFILE_ACQUIRE_RETENTION);
+
+    return next_datafile;
 }
 
-void rrdeng_test_quota(struct rrdengine_worker_config* wc)
+void find_uuid_first_time(
+    struct rrdengine_instance *ctx,
+    struct rrdengine_datafile *datafile,
+    struct uuid_first_time_s *uuid_first_entry_list,
+    size_t count)
 {
-    struct rrdengine_instance *ctx = wc->ctx;
-    struct rrdengine_datafile *datafile;
-    unsigned current_size, target_size;
-    uint8_t out_of_space, only_one_datafile;
-    int ret, error;
-
-    out_of_space = 0;
-    /* Do not allow the pinned pages to exceed the disk space quota to avoid deadlocks */
-    if (unlikely(ctx->disk_space > MAX(ctx->max_disk_space, 2 * ctx->metric_API_max_producers * RRDENG_BLOCK_SIZE))) {
-        out_of_space = 1;
-    }
-    datafile = ctx->datafiles.last;
-    current_size = datafile->pos;
-    target_size = ctx->max_disk_space / TARGET_DATAFILES;
-    target_size = MIN(target_size, MAX_DATAFILE_SIZE);
-    target_size = MAX(target_size, MIN_DATAFILE_SIZE);
-    only_one_datafile = (datafile == ctx->datafiles.first) ? 1 : 0;
-    if (unlikely(current_size >= target_size || (out_of_space && only_one_datafile))) {
-        /* Finalize data and journal file and create a new pair */
-        wal_flush_transaction_buffer(wc);
-        ret = create_new_datafile_pair(ctx, 1, ctx->last_fileno + 1);
-        if (likely(!ret)) {
-            ++ctx->last_fileno;
+    // acquire the datafile to work with it
+    uv_rwlock_rdlock(&ctx->datafiles.rwlock);
+    while(datafile && !datafile_acquire(datafile, DATAFILE_ACQUIRE_RETENTION))
+        datafile = datafile->next;
+    uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
+
+    if (unlikely(!datafile))
+        return;
+
+    unsigned journalfile_count = 0;
+    size_t binary_match = 0;
+    size_t not_matching_bsearches = 0;
+
+    while (datafile) {
+        struct journal_v2_header *j2_header = journalfile_v2_data_acquire(datafile->journalfile, NULL, 0, 0);
+        if (!j2_header) {
+            datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
+            continue;
+        }
+
+        time_t journal_start_time_s = (time_t) (j2_header->start_time_ut / USEC_PER_SEC);
+        struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) j2_header + j2_header->metric_offset);
+        struct uuid_first_time_s *uuid_original_entry;
+
+        size_t journal_metric_count = j2_header->metric_count;
+
+        for (size_t index = 0; index < count; ++index) {
+            uuid_original_entry = &uuid_first_entry_list[index];
+
+            // Check here if we should skip this
+            if (uuid_original_entry->df_matched > 3 || uuid_original_entry->pages_found > 5)
+                continue;
+
+            struct journal_metric_list *live_entry = bsearch(uuid_original_entry->uuid,uuid_list,journal_metric_count,sizeof(*uuid_list), journal_metric_compare);
+            if (!live_entry) {
+                // Not found in this journal
+                not_matching_bsearches++;
+                continue;
+            }
+
+            uuid_original_entry->pages_found += live_entry->entries;
+            uuid_original_entry->df_matched++;
+
+            time_t old_first_time_s = uuid_original_entry->first_time_s;
+
+            // Calculate first / last for this match
+            time_t first_time_s = live_entry->delta_start_s + journal_start_time_s;
+            uuid_original_entry->first_time_s = MIN(uuid_original_entry->first_time_s, first_time_s);
+
+            if (uuid_original_entry->first_time_s != old_first_time_s)
+                uuid_original_entry->df_index_oldest = uuid_original_entry->df_matched;
+
+            binary_match++;
         }
+
+        journalfile_count++;
+        journalfile_v2_data_release(datafile->journalfile);
+        datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
     }
-    if (unlikely(out_of_space && NO_QUIESCE == ctx->quiesce)) {
-        /* delete old data */
-        if (wc->now_deleting_files) {
-            /* already deleting data */
-            return;
+
+    // Let's scan the open cache for almost exact match
+    size_t open_cache_count = 0;
+
+    size_t df_index[10] = { 0 };
+    size_t without_metric = 0;
+    size_t open_cache_gave_first_time_s = 0;
+    size_t metric_count = 0;
+    size_t without_retention = 0;
+    size_t not_needed_bsearches = 0;
+
+    for (size_t index = 0; index < count; ++index) {
+        struct uuid_first_time_s *uuid_first_t_entry = &uuid_first_entry_list[index];
+
+        metric_count++;
+
+        size_t idx = uuid_first_t_entry->df_index_oldest;
+        if(idx >= 10)
+            idx = 9;
+
+        df_index[idx]++;
+
+        not_needed_bsearches += uuid_first_t_entry->df_matched - uuid_first_t_entry->df_index_oldest;
+
+        if (unlikely(!uuid_first_t_entry->metric)) {
+            without_metric++;
+            continue;
         }
-        if (NULL == ctx->datafiles.first->next) {
-            error("Cannot delete data file \"%s/"DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION"\""
-                 " to reclaim space, there are no other file pairs left.",
-                 ctx->dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);
-            return;
+
+        PGC_PAGE *page = pgc_page_get_and_acquire(
+                open_cache, (Word_t)ctx,
+                (Word_t)uuid_first_t_entry->metric, 0,
+                PGC_SEARCH_FIRST);
+
+        if (page) {
+            time_t old_first_time_s = uuid_first_t_entry->first_time_s;
+
+            time_t first_time_s = pgc_page_start_time_s(page);
+            uuid_first_t_entry->first_time_s = MIN(uuid_first_t_entry->first_time_s, first_time_s);
+            pgc_page_release(open_cache, page);
+            open_cache_count++;
+
+            if(uuid_first_t_entry->first_time_s != old_first_time_s) {
+                open_cache_gave_first_time_s++;
+            }
         }
-        info("Deleting data file \"%s/"DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION"\".",
-             ctx->dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);
-        wc->now_deleting_files = mallocz(sizeof(*wc->now_deleting_files));
-        wc->cleanup_thread_deleting_files = 0;
-
-        error = uv_thread_create(wc->now_deleting_files, delete_old_data, ctx);
-        if (error) {
-            error("uv_thread_create(): %s", uv_strerror(error));
-            freez(wc->now_deleting_files);
-            wc->now_deleting_files = NULL;
+        else {
+            if(!uuid_first_t_entry->df_index_oldest)
+                without_retention++;
         }
     }
+    internal_error(true,
+         "DBENGINE: analyzed the retention of %zu rotated metrics of tier %d, "
+         "did %zu jv2 matching binary searches (%zu not matching, %zu overflown) in %u journal files, "
+         "%zu metrics with entries in open cache, "
+         "metrics first time found per datafile index ([not in jv2]:%zu, [1]:%zu, [2]:%zu, [3]:%zu, [4]:%zu, [5]:%zu, [6]:%zu, [7]:%zu, [8]:%zu, [bigger]: %zu), "
+         "open cache found first time %zu, "
+         "metrics without any remaining retention %zu, "
+         "metrics not in MRG %zu",
+         metric_count,
+         ctx->config.tier,
+         binary_match,
+         not_matching_bsearches,
+         not_needed_bsearches,
+         journalfile_count,
+         open_cache_count,
+         df_index[0], df_index[1], df_index[2], df_index[3], df_index[4], df_index[5], df_index[6], df_index[7], df_index[8], df_index[9],
+         open_cache_gave_first_time_s,
+         without_retention,
+         without_metric
+    );
 }
 
-static inline int rrdeng_threads_alive(struct rrdengine_worker_config* wc)
-{
-    if (wc->now_invalidating_dirty_pages || wc->now_deleting_files) {
-        return 1;
+static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile_to_delete, struct rrdengine_datafile *first_datafile_remaining, bool worker) {
+    __atomic_add_fetch(&rrdeng_cache_efficiency_stats.metrics_retention_started, 1, __ATOMIC_RELAXED);
+
+    if(worker)
+        worker_is_busy(UV_EVENT_DBENGINE_FIND_ROTATED_METRICS);
+
+    struct rrdengine_journalfile *journalfile = datafile_to_delete->journalfile;
+    struct journal_v2_header *j2_header = journalfile_v2_data_acquire(journalfile, NULL, 0, 0);
+    struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) j2_header + j2_header->metric_offset);
+
+    size_t count = j2_header->metric_count;
+    struct uuid_first_time_s *uuid_first_t_entry;
+    struct uuid_first_time_s *uuid_first_entry_list = callocz(count, sizeof(struct uuid_first_time_s));
+
+    size_t added = 0;
+    for (size_t index = 0; index < count; ++index) {
+        METRIC *metric = mrg_metric_get_and_acquire(main_mrg, &uuid_list[index].uuid, (Word_t) ctx);
+        if (!metric)
+            continue;
+
+        uuid_first_entry_list[added].metric = metric;
+        uuid_first_entry_list[added].first_time_s = LONG_MAX;
+        uuid_first_entry_list[added].df_matched = 0;
+        uuid_first_entry_list[added].df_index_oldest = 0;
+        uuid_first_entry_list[added].uuid = mrg_metric_uuid(main_mrg, metric);
+        added++;
+    }
+
+    info("DBENGINE: recalculating tier %d retention for %zu metrics starting with datafile %u",
+         ctx->config.tier, count, first_datafile_remaining->fileno);
+
+    journalfile_v2_data_release(journalfile);
+
+    // Update the first time / last time for all metrics we plan to delete
+
+    if(worker)
+        worker_is_busy(UV_EVENT_DBENGINE_FIND_REMAINING_RETENTION);
+
+    find_uuid_first_time(ctx, first_datafile_remaining, uuid_first_entry_list, added);
+
+    if(worker)
+        worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);
+
+    info("DBENGINE: updating tier %d metrics registry retention for %zu metrics",
+         ctx->config.tier, added);
+
+    size_t deleted_metrics = 0, zero_retention_referenced = 0, zero_disk_retention = 0, zero_disk_but_live = 0;
+    for (size_t index = 0; index < added; ++index) {
+        uuid_first_t_entry = &uuid_first_entry_list[index];
+        if (likely(uuid_first_t_entry->first_time_s != LONG_MAX)) {
+            mrg_metric_set_first_time_s_if_bigger(main_mrg, uuid_first_t_entry->metric, uuid_first_t_entry->first_time_s);
+            mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
+        }
+        else {
+            zero_disk_retention++;
+
+            // there is no retention for this metric
+            bool has_retention = mrg_metric_zero_disk_retention(main_mrg, uuid_first_t_entry->metric);
+            if (!has_retention) {
+                bool deleted = mrg_metric_release_and_delete(main_mrg, uuid_first_t_entry->metric);
+                if(deleted)
+                    deleted_metrics++;
+                else
+                    zero_retention_referenced++;
+            }
+            else {
+                zero_disk_but_live++;
+                mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
+            }
+        }
     }
-    return 0;
+    freez(uuid_first_entry_list);
+
+    internal_error(zero_disk_retention,
+                   "DBENGINE: deleted %zu metrics, zero retention but referenced %zu (out of %zu total, of which %zu have main cache retention) zero on-disk retention tier %d metrics from metrics registry",
+                   deleted_metrics, zero_retention_referenced, zero_disk_retention, zero_disk_but_live, ctx->config.tier);
+
+    if(worker)
+        worker_is_idle();
 }
 
-static void rrdeng_cleanup_finished_threads(struct rrdengine_worker_config* wc)
-{
-    struct rrdengine_instance *ctx = wc->ctx;
+void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, bool update_retention, bool worker) {
+    if(worker)
+        worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT);
+
+    bool datafile_got_for_deletion = datafile_acquire_for_deletion(datafile);
+
+    if (update_retention)
+        update_metrics_first_time_s(ctx, datafile, datafile->next, worker);
 
-    if (unlikely(wc->cleanup_thread_invalidating_dirty_pages)) {
-        after_invalidate_oldest_committed(wc);
+    while (!datafile_got_for_deletion) {
+        if(worker)
+            worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT);
+
+        datafile_got_for_deletion = datafile_acquire_for_deletion(datafile);
+
+        if (!datafile_got_for_deletion) {
+            info("DBENGINE: waiting for data file '%s/"
+                         DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION
+                         "' to be available for deletion, "
+                         "it is in use currently by %u users.",
+                 ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno, datafile->users.lockers);
+
+            __atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_spin, 1, __ATOMIC_RELAXED);
+            sleep_usec(1 * USEC_PER_SEC);
+        }
     }
-    if (unlikely(wc->cleanup_thread_deleting_files)) {
-        after_delete_old_data(wc);
+
+    __atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_started, 1, __ATOMIC_RELAXED);
+    info("DBENGINE: deleting data file '%s/"
+         DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION
+         "'.",
+         ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);
+
+    if(worker)
+        worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE);
+
+    struct rrdengine_journalfile *journal_file;
+    unsigned deleted_bytes, journal_file_bytes, datafile_bytes;
+    int ret;
+    char path[RRDENG_PATH_MAX];
+
+    uv_rwlock_wrlock(&ctx->datafiles.rwlock);
+    datafile_list_delete_unsafe(ctx, datafile);
+    uv_rwlock_wrunlock(&ctx->datafiles.rwlock);
+
+    journal_file = datafile->journalfile;
+    datafile_bytes = datafile->pos;
+    journal_file_bytes = journalfile_current_size(journal_file);
+    deleted_bytes = journalfile_v2_data_size_get(journal_file);
+
+    info("DBENGINE: deleting data and journal files to maintain disk quota");
+    ret = journalfile_destroy_unsafe(journal_file, datafile);
+    if (!ret) {
+        journalfile_v1_generate_path(datafile, path, sizeof(path));
+        info("DBENGINE: deleted journal file \"%s\".", path);
+        journalfile_v2_generate_path(datafile, path, sizeof(path));
+        info("DBENGINE: deleted journal file \"%s\".", path);
+        deleted_bytes += journal_file_bytes;
     }
-    if (unlikely(SET_QUIESCE == ctx->quiesce && !rrdeng_threads_alive(wc))) {
-        ctx->quiesce = QUIESCED;
-        completion_mark_complete(&ctx->rrdengine_completion);
+    ret = destroy_data_file_unsafe(datafile);
+    if (!ret) {
+        generate_datafilepath(datafile, path, sizeof(path));
+        info("DBENGINE: deleted data file \"%s\".", path);
+        deleted_bytes += datafile_bytes;
     }
+    freez(journal_file);
+    freez(datafile);
+
+    ctx_current_disk_space_decrease(ctx, deleted_bytes);
+    info("DBENGINE: reclaimed %u bytes of disk space.", deleted_bytes);
 }
 
-/* return 0 on success */
-int init_rrd_files(struct rrdengine_instance *ctx)
-{
-    int ret = init_data_files(ctx);
-
-    BUFFER *wb = buffer_create(1000);
-    size_t all_errors = 0;
-    usec_t now = now_realtime_usec();
-
-    if(ctx->load_errors[LOAD_ERRORS_PAGE_FLIPPED_TIME].counter) {
-        buffer_sprintf(wb, "%s%zu pages had start time > end time (latest: %llu secs ago)"
-                       , (all_errors)?", ":""
-                       , ctx->load_errors[LOAD_ERRORS_PAGE_FLIPPED_TIME].counter
-                       , (now - ctx->load_errors[LOAD_ERRORS_PAGE_FLIPPED_TIME].latest_end_time_ut) / USEC_PER_SEC
-                       );
-        all_errors += ctx->load_errors[LOAD_ERRORS_PAGE_FLIPPED_TIME].counter;
-    }
+static void *database_rotate_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    datafile_delete(ctx, ctx->datafiles.first, ctx_is_available_for_queries(ctx), true);
 
-    if(ctx->load_errors[LOAD_ERRORS_PAGE_EQUAL_TIME].counter) {
-        buffer_sprintf(wb, "%s%zu pages had start time = end time with more than 1 entries (latest: %llu secs ago)"
-                       , (all_errors)?", ":""
-                       , ctx->load_errors[LOAD_ERRORS_PAGE_EQUAL_TIME].counter
-                       , (now - ctx->load_errors[LOAD_ERRORS_PAGE_EQUAL_TIME].latest_end_time_ut) / USEC_PER_SEC
-        );
-        all_errors += ctx->load_errors[LOAD_ERRORS_PAGE_EQUAL_TIME].counter;
-    }
+    if (rrdeng_ctx_exceeded_disk_quota(ctx))
+        rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
 
-    if(ctx->load_errors[LOAD_ERRORS_PAGE_ZERO_ENTRIES].counter) {
-        buffer_sprintf(wb, "%s%zu pages had zero points (latest: %llu secs ago)"
-                       , (all_errors)?", ":""
-                       , ctx->load_errors[LOAD_ERRORS_PAGE_ZERO_ENTRIES].counter
-                       , (now - ctx->load_errors[LOAD_ERRORS_PAGE_ZERO_ENTRIES].latest_end_time_ut) / USEC_PER_SEC
-        );
-        all_errors += ctx->load_errors[LOAD_ERRORS_PAGE_ZERO_ENTRIES].counter;
-    }
+    rrdcontext_db_rotation();
 
-    if(ctx->load_errors[LOAD_ERRORS_PAGE_UPDATE_ZERO].counter) {
-        buffer_sprintf(wb, "%s%zu pages had update every == 0 with entries > 1 (latest: %llu secs ago)"
-                       , (all_errors)?", ":""
-                       , ctx->load_errors[LOAD_ERRORS_PAGE_UPDATE_ZERO].counter
-                       , (now - ctx->load_errors[LOAD_ERRORS_PAGE_UPDATE_ZERO].latest_end_time_ut) / USEC_PER_SEC
-        );
-        all_errors += ctx->load_errors[LOAD_ERRORS_PAGE_UPDATE_ZERO].counter;
-    }
+    return data;
+}
 
-    if(ctx->load_errors[LOAD_ERRORS_PAGE_FLEXY_TIME].counter) {
-        buffer_sprintf(wb, "%s%zu pages had a different number of points compared to their timestamps (latest: %llu secs ago; these page have been loaded)"
-                       , (all_errors)?", ":""
-                       , ctx->load_errors[LOAD_ERRORS_PAGE_FLEXY_TIME].counter
-                       , (now - ctx->load_errors[LOAD_ERRORS_PAGE_FLEXY_TIME].latest_end_time_ut) / USEC_PER_SEC
-        );
-        all_errors += ctx->load_errors[LOAD_ERRORS_PAGE_FLEXY_TIME].counter;
-    }
+static void after_flush_all_hot_and_dirty_pages_of_section(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    ;
+}
+
+static void *flush_all_hot_and_dirty_pages_of_section_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    worker_is_busy(UV_EVENT_DBENGINE_QUIESCE);
+    pgc_flush_all_hot_and_dirty_pages(main_cache, (Word_t)ctx);
+    completion_mark_complete(&ctx->quiesce.completion);
+    return data;
+}
+
+static void after_populate_mrg(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    ;
+}
+
+static void *populate_mrg_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);
+
+    do {
+        struct rrdengine_datafile *datafile = NULL;
+
+        // find a datafile to work
+        uv_rwlock_rdlock(&ctx->datafiles.rwlock);
+        for(datafile = ctx->datafiles.first; datafile ; datafile = datafile->next) {
+            if(!netdata_spinlock_trylock(&datafile->populate_mrg.spinlock))
+                continue;
+
+            if(datafile->populate_mrg.populated) {
+                netdata_spinlock_unlock(&datafile->populate_mrg.spinlock);
+                continue;
+            }
+
+            // we have the spinlock and it is not populated
+            break;
+        }
+        uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
+
+        if(!datafile)
+            break;
+
+        journalfile_v2_populate_retention_to_mrg(ctx, datafile->journalfile);
+        datafile->populate_mrg.populated = true;
+        netdata_spinlock_unlock(&datafile->populate_mrg.spinlock);
 
-    if(ctx->load_errors[LOAD_ERRORS_DROPPED_EXTENT].counter) {
-        buffer_sprintf(wb, "%s%zu extents have been dropped because they didn't have any valid pages"
-                       , (all_errors)?", ":""
-                       , ctx->load_errors[LOAD_ERRORS_DROPPED_EXTENT].counter
-        );
-        all_errors += ctx->load_errors[LOAD_ERRORS_DROPPED_EXTENT].counter;
+    } while(1);
+
+    completion_mark_complete(completion);
+
+    return data;
+}
+
+static void after_ctx_shutdown(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    ;
+}
+
+static void *ctx_shutdown_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    worker_is_busy(UV_EVENT_DBENGINE_SHUTDOWN);
+
+    completion_wait_for(&ctx->quiesce.completion);
+    completion_destroy(&ctx->quiesce.completion);
+
+    bool logged = false;
+    while(__atomic_load_n(&ctx->atomic.extents_currently_being_flushed, __ATOMIC_RELAXED) ||
+            __atomic_load_n(&ctx->atomic.inflight_queries, __ATOMIC_RELAXED)) {
+        if(!logged) {
+            logged = true;
+            info("DBENGINE: waiting for %zu inflight queries to finish to shutdown tier %d...",
+                 __atomic_load_n(&ctx->atomic.inflight_queries, __ATOMIC_RELAXED),
+                 (ctx->config.legacy) ? -1 : ctx->config.tier);
+        }
+        sleep_usec(1 * USEC_PER_MS);
     }
 
-    if(all_errors)
-        info("DBENGINE: tier %d: %s", ctx->tier, buffer_tostring(wb));
+    completion_mark_complete(completion);
 
-    buffer_free(wb);
-    return ret;
+    return data;
 }
 
-void finalize_rrd_files(struct rrdengine_instance *ctx)
-{
-    return finalize_data_files(ctx);
+static void *cache_flush_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    if (!main_cache)
+        return data;
+
+    worker_is_busy(UV_EVENT_DBENGINE_FLUSH_MAIN_CACHE);
+    pgc_flush_pages(main_cache, 0);
+
+    return data;
 }
 
-void rrdeng_init_cmd_queue(struct rrdengine_worker_config* wc)
-{
-    wc->cmd_queue.head = wc->cmd_queue.tail = 0;
-    wc->queue_size = 0;
-    fatal_assert(0 == uv_cond_init(&wc->cmd_cond));
-    fatal_assert(0 == uv_mutex_init(&wc->cmd_mutex));
+static void *cache_evict_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *req __maybe_unused) {
+    if (!main_cache)
+        return data;
+
+    worker_is_busy(UV_EVENT_DBENGINE_EVICT_MAIN_CACHE);
+    pgc_evict_pages(main_cache, 0, 0);
+
+    return data;
 }
 
-void rrdeng_enq_cmd(struct rrdengine_worker_config* wc, struct rrdeng_cmd *cmd)
-{
-    unsigned queue_size;
+static void after_prep_query(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    ;
+}
 
-    /* wait for free space in queue */
-    uv_mutex_lock(&wc->cmd_mutex);
-    while ((queue_size = wc->queue_size) == RRDENG_CMD_Q_MAX_SIZE) {
-        uv_cond_wait(&wc->cmd_cond, &wc->cmd_mutex);
-    }
-    fatal_assert(queue_size < RRDENG_CMD_Q_MAX_SIZE);
-    /* enqueue command */
-    wc->cmd_queue.cmd_array[wc->cmd_queue.tail] = *cmd;
-    wc->cmd_queue.tail = wc->cmd_queue.tail != RRDENG_CMD_Q_MAX_SIZE - 1 ?
-                         wc->cmd_queue.tail + 1 : 0;
-    wc->queue_size = queue_size + 1;
-    uv_mutex_unlock(&wc->cmd_mutex);
+static void *query_prep_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *req __maybe_unused) {
+    worker_is_busy(UV_EVENT_DBENGINE_QUERY);
+    PDC *pdc = data;
+    rrdeng_prep_query(pdc);
+    return data;
+}
 
-    /* wake up event loop */
-    fatal_assert(0 == uv_async_send(&wc->async));
+unsigned rrdeng_target_data_file_size(struct rrdengine_instance *ctx) {
+    unsigned target_size = ctx->config.max_disk_space / TARGET_DATAFILES;
+    target_size = MIN(target_size, MAX_DATAFILE_SIZE);
+    target_size = MAX(target_size, MIN_DATAFILE_SIZE);
+    return target_size;
 }
 
-struct rrdeng_cmd rrdeng_deq_cmd(struct rrdengine_worker_config* wc)
+bool rrdeng_ctx_exceeded_disk_quota(struct rrdengine_instance *ctx)
 {
-    struct rrdeng_cmd ret;
-    unsigned queue_size;
-
-    uv_mutex_lock(&wc->cmd_mutex);
-    queue_size = wc->queue_size;
-    if (queue_size == 0) {
-        ret.opcode = RRDENG_NOOP;
-    } else {
-        /* dequeue command */
-        ret = wc->cmd_queue.cmd_array[wc->cmd_queue.head];
-        if (queue_size == 1) {
-            wc->cmd_queue.head = wc->cmd_queue.tail = 0;
-        } else {
-            wc->cmd_queue.head = wc->cmd_queue.head != RRDENG_CMD_Q_MAX_SIZE - 1 ?
-                                 wc->cmd_queue.head + 1 : 0;
-        }
-        wc->queue_size = queue_size - 1;
+    uint64_t estimated_disk_space = ctx_current_disk_space_get(ctx) + rrdeng_target_data_file_size(ctx) -
+                                    (ctx->datafiles.first->prev ? ctx->datafiles.first->prev->pos : 0);
 
-        /* wake up producers */
-        uv_cond_signal(&wc->cmd_cond);
-    }
-    uv_mutex_unlock(&wc->cmd_mutex);
+    return estimated_disk_space > ctx->config.max_disk_space;
+}
 
-    return ret;
+/* return 0 on success */
+int init_rrd_files(struct rrdengine_instance *ctx)
+{
+    return init_data_files(ctx);
 }
 
-static void load_configuration_dynamic(void)
+void finalize_rrd_files(struct rrdengine_instance *ctx)
 {
-    unsigned read_num = (unsigned)config_get_number(CONFIG_SECTION_DB, "dbengine pages per extent", MAX_PAGES_PER_EXTENT);
-    if (read_num > 0 && read_num <= MAX_PAGES_PER_EXTENT)
-        rrdeng_pages_per_extent = read_num;
-    else {
-        error("Invalid dbengine pages per extent %u given. Using %u.", read_num, rrdeng_pages_per_extent);
-        config_set_number(CONFIG_SECTION_DB, "dbengine pages per extent", rrdeng_pages_per_extent);
-    }
+    return finalize_data_files(ctx);
 }
 
 void async_cb(uv_async_t *handle)
@@ -1253,256 +1392,413 @@ void async_cb(uv_async_t *handle)
     debug(D_RRDENGINE, "%s called, active=%d.", __func__, uv_is_active((uv_handle_t *)handle));
 }
 
-/* Flushes dirty pages when timer expires */
 #define TIMER_PERIOD_MS (1000)
 
-void timer_cb(uv_timer_t* handle)
-{
-    worker_is_busy(RRDENG_MAX_OPCODE + 1);
 
-    struct rrdengine_worker_config* wc = handle->data;
-    struct rrdengine_instance *ctx = wc->ctx;
+static void *extent_read_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    EPDL *epdl = data;
+    epdl_find_extent_and_populate_pages(ctx, epdl, true);
+    return data;
+}
 
-    uv_stop(handle->loop);
-    uv_update_time(handle->loop);
-    rrdeng_test_quota(wc);
-    debug(D_RRDENGINE, "%s: timeout reached.", __func__);
-    if (likely(!wc->now_deleting_files && !wc->now_invalidating_dirty_pages)) {
-        /* There is free space so we can write to disk and we are not actively deleting dirty buffers */
-        struct page_cache *pg_cache = &ctx->pg_cache;
-        unsigned long total_bytes, bytes_written, nr_committed_pages, bytes_to_write = 0, producers, low_watermark,
-                      high_watermark;
-
-        uv_rwlock_rdlock(&pg_cache->committed_page_index.lock);
-        nr_committed_pages = pg_cache->committed_page_index.nr_committed_pages;
-        uv_rwlock_rdunlock(&pg_cache->committed_page_index.lock);
-        producers = ctx->metric_API_max_producers;
-        /* are flushable pages more than 25% of the maximum page cache size */
-        high_watermark = (ctx->max_cache_pages * 25LLU) / 100;
-        low_watermark = (ctx->max_cache_pages * 5LLU) / 100; /* 5%, must be smaller than high_watermark */
-
-        /* Flush more pages only if disk can keep up */
-        if (wc->inflight_dirty_pages < high_watermark + producers) {
-            if (nr_committed_pages > producers &&
-                /* committed to be written pages are more than the produced number */
-                nr_committed_pages - producers > high_watermark) {
-                /* Flushing speed must increase to stop page cache from filling with dirty pages */
-                bytes_to_write = (nr_committed_pages - producers - low_watermark) * RRDENG_BLOCK_SIZE;
-            }
-            bytes_to_write = MAX(DATAFILE_IDEAL_IO_SIZE, bytes_to_write);
+static void epdl_populate_pages_asynchronously(struct rrdengine_instance *ctx, EPDL *epdl, STORAGE_PRIORITY priority) {
+    rrdeng_enq_cmd(ctx, RRDENG_OPCODE_EXTENT_READ, epdl, NULL, priority,
+                   rrdeng_enqueue_epdl_cmd, rrdeng_dequeue_epdl_cmd);
+}
 
-            debug(D_RRDENGINE, "Flushing pages to disk.");
-            for (total_bytes = bytes_written = do_flush_pages(wc, 0, NULL);
-                 bytes_written && (total_bytes < bytes_to_write);
-                 total_bytes += bytes_written) {
-                bytes_written = do_flush_pages(wc, 0, NULL);
-            }
+void pdc_route_asynchronously(struct rrdengine_instance *ctx, struct page_details_control *pdc) {
+    pdc_to_epdl_router(ctx, pdc, epdl_populate_pages_asynchronously, epdl_populate_pages_asynchronously);
+}
+
+void epdl_populate_pages_synchronously(struct rrdengine_instance *ctx, EPDL *epdl, enum storage_priority priority __maybe_unused) {
+    epdl_find_extent_and_populate_pages(ctx, epdl, false);
+}
+
+void pdc_route_synchronously(struct rrdengine_instance *ctx, struct page_details_control *pdc) {
+    pdc_to_epdl_router(ctx, pdc, epdl_populate_pages_synchronously, epdl_populate_pages_synchronously);
+}
+
+#define MAX_RETRIES_TO_START_INDEX (100)
+static void *journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    unsigned count = 0;
+    worker_is_busy(UV_EVENT_DBENGINE_JOURNAL_INDEX_WAIT);
+
+    while (__atomic_load_n(&ctx->atomic.now_deleting_files, __ATOMIC_RELAXED) && count++ < MAX_RETRIES_TO_START_INDEX)
+        sleep_usec(100 * USEC_PER_MS);
+
+    if (count == MAX_RETRIES_TO_START_INDEX) {
+        worker_is_idle();
+        return data;
+    }
+
+    struct rrdengine_datafile *datafile = ctx->datafiles.first;
+    worker_is_busy(UV_EVENT_DBENGINE_JOURNAL_INDEX);
+    count = 0;
+    while (datafile && datafile->fileno != ctx_last_fileno_get(ctx) && datafile->fileno != ctx_last_flush_fileno_get(ctx)) {
+
+        netdata_spinlock_lock(&datafile->writers.spinlock);
+        bool available = (datafile->writers.running || datafile->writers.flushed_to_open_running) ? false : true;
+        netdata_spinlock_unlock(&datafile->writers.spinlock);
+
+        if(!available)
+            continue;
+
+        if (unlikely(!journalfile_v2_data_available(datafile->journalfile))) {
+            info("DBENGINE: journal file %u is ready to be indexed", datafile->fileno);
+            pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->config.page_type,
+                                         journalfile_migrate_to_v2_callback, (void *) datafile->journalfile);
+            count++;
         }
+
+        datafile = datafile->next;
+
+        if (unlikely(!ctx_is_available_for_queries(ctx)))
+            break;
     }
-    load_configuration_dynamic();
-#ifdef NETDATA_INTERNAL_CHECKS
+
+    errno = 0;
+    internal_error(count, "DBENGINE: journal indexing done; %u files processed", count);
+
+    worker_is_idle();
+
+    return data;
+}
+
+static void after_do_cache_flush(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    rrdeng_main.flushes_running--;
+}
+
+static void after_do_cache_evict(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    rrdeng_main.evictions_running--;
+}
+
+static void after_extent_read(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    ;
+}
+
+static void after_journal_v2_indexing(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    __atomic_store_n(&ctx->atomic.migration_to_v2_running, false, __ATOMIC_RELAXED);
+    rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
+}
+
+struct rrdeng_buffer_sizes rrdeng_get_buffer_sizes(void) {
+    return (struct rrdeng_buffer_sizes) {
+            .pgc         = pgc_aral_overhead() + pgc_aral_structures(),
+            .mrg         = mrg_aral_overhead() + mrg_aral_structures(),
+            .opcodes     = aral_overhead(rrdeng_main.cmd_queue.ar) + aral_structures(rrdeng_main.cmd_queue.ar),
+            .handles     = aral_overhead(rrdeng_main.handles.ar) + aral_structures(rrdeng_main.handles.ar),
+            .descriptors = aral_overhead(rrdeng_main.descriptors.ar) + aral_structures(rrdeng_main.descriptors.ar),
+            .wal         = __atomic_load_n(&wal_globals.atomics.allocated, __ATOMIC_RELAXED) * (sizeof(WAL) + RRDENG_BLOCK_SIZE),
+            .workers     = aral_overhead(rrdeng_main.work_cmd.ar),
+            .pdc         = pdc_cache_size(),
+            .xt_io       = aral_overhead(rrdeng_main.xt_io_descr.ar) + aral_structures(rrdeng_main.xt_io_descr.ar),
+            .xt_buf      = extent_buffer_cache_size(),
+            .epdl        = epdl_cache_size(),
+            .deol        = deol_cache_size(),
+            .pd          = pd_cache_size(),
+
+#ifdef PDC_USE_JULYL
+            .julyl       = julyl_cache_size(),
+#endif
+    };
+}
+
+static void after_cleanup(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
+    rrdeng_main.cleanup_running--;
+}
+
+static void *cleanup_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
+    worker_is_busy(UV_EVENT_DBENGINE_BUFFERS_CLEANUP);
+
+    wal_cleanup1();
+    extent_buffer_cleanup1();
+
     {
-        char buf[4096];
-        debug(D_RRDENGINE, "%s", get_rrdeng_statistics(wc->ctx, buf, sizeof(buf)));
+        static time_t last_run_s = 0;
+        time_t now_s = now_monotonic_sec();
+        if(now_s - last_run_s >= 10) {
+            last_run_s = now_s;
+            journalfile_v2_data_unmount_cleanup(now_s);
+        }
     }
+
+#ifdef PDC_USE_JULYL
+    julyl_cleanup1();
 #endif
 
+    return data;
+}
+
+void timer_cb(uv_timer_t* handle) {
+    worker_is_busy(RRDENG_TIMER_CB);
+    uv_stop(handle->loop);
+    uv_update_time(handle->loop);
+
+    worker_set_metric(RRDENG_OPCODES_WAITING, (NETDATA_DOUBLE)rrdeng_main.cmd_queue.unsafe.waiting);
+    worker_set_metric(RRDENG_WORKS_DISPATCHED, (NETDATA_DOUBLE)__atomic_load_n(&rrdeng_main.work_cmd.atomics.dispatched, __ATOMIC_RELAXED));
+    worker_set_metric(RRDENG_WORKS_EXECUTING, (NETDATA_DOUBLE)__atomic_load_n(&rrdeng_main.work_cmd.atomics.executing, __ATOMIC_RELAXED));
+
+    rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_INIT, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
+    rrdeng_enq_cmd(NULL, RRDENG_OPCODE_EVICT_INIT, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
+    rrdeng_enq_cmd(NULL, RRDENG_OPCODE_CLEANUP, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
+
     worker_is_idle();
 }
 
-#define MAX_CMD_BATCH_SIZE (256)
+static void dbengine_initialize_structures(void) {
+    pgc_and_mrg_initialize();
+
+    pdc_init();
+    page_details_init();
+    epdl_init();
+    deol_init();
+    rrdeng_cmd_queue_init();
+    work_request_init();
+    rrdeng_query_handle_init();
+    page_descriptors_init();
+    extent_buffer_init();
+    dbengine_page_alloc_init();
+    extent_io_descriptor_init();
+}
 
-void rrdeng_worker(void* arg)
-{
-    worker_register("DBENGINE");
-    worker_register_job_name(RRDENG_NOOP,                          "noop");
-    worker_register_job_name(RRDENG_READ_PAGE,                     "page read");
-    worker_register_job_name(RRDENG_READ_EXTENT,                   "extent read");
-    worker_register_job_name(RRDENG_COMMIT_PAGE,                   "commit");
-    worker_register_job_name(RRDENG_FLUSH_PAGES,                   "flush");
-    worker_register_job_name(RRDENG_SHUTDOWN,                      "shutdown");
-    worker_register_job_name(RRDENG_INVALIDATE_OLDEST_MEMORY_PAGE, "page lru");
-    worker_register_job_name(RRDENG_QUIESCE,                       "quiesce");
-    worker_register_job_name(RRDENG_MAX_OPCODE,                    "cleanup");
-    worker_register_job_name(RRDENG_MAX_OPCODE + 1,                "timer");
-
-    struct rrdengine_worker_config* wc = arg;
-    struct rrdengine_instance *ctx = wc->ctx;
-    uv_loop_t* loop;
-    int shutdown, ret;
-    enum rrdeng_opcode opcode;
-    uv_timer_t timer_req;
-    struct rrdeng_cmd cmd;
-    unsigned cmd_batch_size;
+bool rrdeng_dbengine_spawn(struct rrdengine_instance *ctx __maybe_unused) {
+    static bool spawned = false;
+    static SPINLOCK spinlock = NETDATA_SPINLOCK_INITIALIZER;
 
-    rrdeng_init_cmd_queue(wc);
+    netdata_spinlock_lock(&spinlock);
 
-    loop = wc->loop = mallocz(sizeof(uv_loop_t));
-    ret = uv_loop_init(loop);
-    if (ret) {
-        error("uv_loop_init(): %s", uv_strerror(ret));
-        goto error_after_loop_init;
-    }
-    loop->data = wc;
+    if(!spawned) {
+        int ret;
 
-    ret = uv_async_init(wc->loop, &wc->async, async_cb);
-    if (ret) {
-        error("uv_async_init(): %s", uv_strerror(ret));
-        goto error_after_async_init;
-    }
-    wc->async.data = wc;
+        ret = uv_loop_init(&rrdeng_main.loop);
+        if (ret) {
+            error("DBENGINE: uv_loop_init(): %s", uv_strerror(ret));
+            return false;
+        }
+        rrdeng_main.loop.data = &rrdeng_main;
 
-    wc->now_deleting_files = NULL;
-    wc->cleanup_thread_deleting_files = 0;
+        ret = uv_async_init(&rrdeng_main.loop, &rrdeng_main.async, async_cb);
+        if (ret) {
+            error("DBENGINE: uv_async_init(): %s", uv_strerror(ret));
+            fatal_assert(0 == uv_loop_close(&rrdeng_main.loop));
+            return false;
+        }
+        rrdeng_main.async.data = &rrdeng_main;
+
+        ret = uv_timer_init(&rrdeng_main.loop, &rrdeng_main.timer);
+        if (ret) {
+            error("DBENGINE: uv_timer_init(): %s", uv_strerror(ret));
+            uv_close((uv_handle_t *)&rrdeng_main.async, NULL);
+            fatal_assert(0 == uv_loop_close(&rrdeng_main.loop));
+            return false;
+        }
+        rrdeng_main.timer.data = &rrdeng_main;
 
-    wc->now_invalidating_dirty_pages = NULL;
-    wc->cleanup_thread_invalidating_dirty_pages = 0;
-    wc->inflight_dirty_pages = 0;
+        dbengine_initialize_structures();
 
-    /* dirty page flushing timer */
-    ret = uv_timer_init(loop, &timer_req);
-    if (ret) {
-        error("uv_timer_init(): %s", uv_strerror(ret));
-        goto error_after_timer_init;
+        fatal_assert(0 == uv_thread_create(&rrdeng_main.thread, dbengine_event_loop, &rrdeng_main));
+        spawned = true;
     }
-    timer_req.data = wc;
 
-    wc->error = 0;
-    /* wake up initialization thread */
-    completion_mark_complete(&ctx->rrdengine_completion);
+    netdata_spinlock_unlock(&spinlock);
+    return true;
+}
 
-    fatal_assert(0 == uv_timer_start(&timer_req, timer_cb, TIMER_PERIOD_MS, TIMER_PERIOD_MS));
-    shutdown = 0;
-    int set_name = 0;
-    while (likely(shutdown == 0 || rrdeng_threads_alive(wc))) {
+void dbengine_event_loop(void* arg) {
+    sanity_check();
+    uv_thread_set_name_np(pthread_self(), "DBENGINE");
+    service_register(SERVICE_THREAD_TYPE_EVENT_LOOP, NULL, NULL, NULL, true);
+
+    worker_register("DBENGINE");
+
+    // opcode jobs
+    worker_register_job_name(RRDENG_OPCODE_NOOP,                                     "noop");
+
+    worker_register_job_name(RRDENG_OPCODE_QUERY,                                    "query");
+    worker_register_job_name(RRDENG_OPCODE_EXTENT_WRITE,                             "extent write");
+    worker_register_job_name(RRDENG_OPCODE_EXTENT_READ,                              "extent read");
+    worker_register_job_name(RRDENG_OPCODE_FLUSHED_TO_OPEN,                          "flushed to open");
+    worker_register_job_name(RRDENG_OPCODE_DATABASE_ROTATE,                          "db rotate");
+    worker_register_job_name(RRDENG_OPCODE_JOURNAL_INDEX,                            "journal index");
+    worker_register_job_name(RRDENG_OPCODE_FLUSH_INIT,                               "flush init");
+    worker_register_job_name(RRDENG_OPCODE_EVICT_INIT,                               "evict init");
+    worker_register_job_name(RRDENG_OPCODE_CTX_SHUTDOWN,                             "ctx shutdown");
+    worker_register_job_name(RRDENG_OPCODE_CTX_QUIESCE,                              "ctx quiesce");
+
+    worker_register_job_name(RRDENG_OPCODE_MAX,                                      "get opcode");
+
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_QUERY,                "query cb");
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_EXTENT_WRITE,         "extent write cb");
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_EXTENT_READ,          "extent read cb");
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_FLUSHED_TO_OPEN,      "flushed to open cb");
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_DATABASE_ROTATE,      "db rotate cb");
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_JOURNAL_INDEX,        "journal index cb");
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_FLUSH_INIT,           "flush init cb");
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_EVICT_INIT,           "evict init cb");
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_SHUTDOWN,         "ctx shutdown cb");
+    worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_QUIESCE,          "ctx quiesce cb");
+
+    // special jobs
+    worker_register_job_name(RRDENG_TIMER_CB,                                        "timer");
+    worker_register_job_name(RRDENG_FLUSH_TRANSACTION_BUFFER_CB,                     "transaction buffer flush cb");
+
+    worker_register_job_custom_metric(RRDENG_OPCODES_WAITING,  "opcodes waiting",  "opcodes", WORKER_METRIC_ABSOLUTE);
+    worker_register_job_custom_metric(RRDENG_WORKS_DISPATCHED, "works dispatched", "works",   WORKER_METRIC_ABSOLUTE);
+    worker_register_job_custom_metric(RRDENG_WORKS_EXECUTING,  "works executing",  "works",   WORKER_METRIC_ABSOLUTE);
+
+    struct rrdeng_main *main = arg;
+    enum rrdeng_opcode opcode;
+    struct rrdeng_cmd cmd;
+    main->tid = gettid();
+
+    fatal_assert(0 == uv_timer_start(&main->timer, timer_cb, TIMER_PERIOD_MS, TIMER_PERIOD_MS));
+
+    bool shutdown = false;
+    while (likely(!shutdown)) {
         worker_is_idle();
-        uv_run(loop, UV_RUN_DEFAULT);
-        worker_is_busy(RRDENG_MAX_OPCODE);
-        rrdeng_cleanup_finished_threads(wc);
+        uv_run(&main->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);
+            worker_is_busy(RRDENG_OPCODE_MAX);
+            cmd = rrdeng_deq_cmd();
             opcode = cmd.opcode;
-            ++cmd_batch_size;
 
-            if(likely(opcode != RRDENG_NOOP))
-                worker_is_busy(opcode);
+            worker_is_busy(opcode);
 
             switch (opcode) {
-            case RRDENG_NOOP:
-                /* the command queue was empty, do nothing */
-                break;
-            case RRDENG_SHUTDOWN:
-                shutdown = 1;
-                break;
-            case RRDENG_QUIESCE:
-                ctx->drop_metrics_under_page_cache_pressure = 0;
-                ctx->quiesce = SET_QUIESCE;
-                fatal_assert(0 == uv_timer_stop(&timer_req));
-                uv_close((uv_handle_t *)&timer_req, NULL);
-                while (do_flush_pages(wc, 1, NULL)) {
-                    ; /* Force flushing of all committed pages. */
+                case RRDENG_OPCODE_EXTENT_READ: {
+                    struct rrdengine_instance *ctx = cmd.ctx;
+                    EPDL *epdl = cmd.data;
+                    work_dispatch(ctx, epdl, NULL, opcode, extent_read_tp_worker, after_extent_read);
+                    break;
                 }
-                wal_flush_transaction_buffer(wc);
-                if (!rrdeng_threads_alive(wc)) {
-                    ctx->quiesce = QUIESCED;
-                    completion_mark_complete(&ctx->rrdengine_completion);
+
+                case RRDENG_OPCODE_QUERY: {
+                    struct rrdengine_instance *ctx = cmd.ctx;
+                    PDC *pdc = cmd.data;
+                    work_dispatch(ctx, pdc, NULL, opcode, query_prep_tp_worker, after_prep_query);
+                    break;
                 }
-                break;
-            case RRDENG_READ_PAGE:
-                do_read_extent(wc, &cmd.read_page.page_cache_descr, 1, 0);
-                break;
-            case RRDENG_READ_EXTENT:
-                do_read_extent(wc, cmd.read_extent.page_cache_descr, cmd.read_extent.page_count, 1);
-                if (unlikely(!set_name)) {
-                    set_name = 1;
-                    uv_thread_set_name_np(ctx->worker_config.thread, "DBENGINE");
+
+                case RRDENG_OPCODE_EXTENT_WRITE: {
+                    struct rrdengine_instance *ctx = cmd.ctx;
+                    struct page_descr_with_data *base = cmd.data;
+                    struct completion *completion = cmd.completion; // optional
+                    work_dispatch(ctx, base, completion, opcode, extent_write_tp_worker, after_extent_write);
+                    break;
                 }
-                break;
-            case RRDENG_COMMIT_PAGE:
-                do_commit_transaction(wc, STORE_DATA, NULL);
-                break;
-            case RRDENG_FLUSH_PAGES: {
-                if (wc->now_invalidating_dirty_pages) {
-                    /* Do not flush if the disk cannot keep up */
-                    completion_mark_complete(cmd.completion);
-                } else {
-                    (void)do_flush_pages(wc, 1, cmd.completion);
+
+                case RRDENG_OPCODE_FLUSHED_TO_OPEN: {
+                    struct rrdengine_instance *ctx = cmd.ctx;
+                    uv_fs_t *uv_fs_request = cmd.data;
+                    struct extent_io_descriptor *xt_io_descr = uv_fs_request->data;
+                    struct completion *completion = xt_io_descr->completion;
+                    work_dispatch(ctx, uv_fs_request, completion, opcode, extent_flushed_to_open_tp_worker, after_extent_flushed_to_open);
+                    break;
+                }
+
+                case RRDENG_OPCODE_FLUSH_INIT: {
+                    if(rrdeng_main.flushes_running < (size_t)(libuv_worker_threads / 4)) {
+                        rrdeng_main.flushes_running++;
+                        work_dispatch(NULL, NULL, NULL, opcode, cache_flush_tp_worker, after_do_cache_flush);
+                    }
+                    break;
+                }
+
+                case RRDENG_OPCODE_EVICT_INIT: {
+                    if(!rrdeng_main.evictions_running) {
+                        rrdeng_main.evictions_running++;
+                        work_dispatch(NULL, NULL, NULL, opcode, cache_evict_tp_worker, after_do_cache_evict);
+                    }
+                    break;
+                }
+
+                case RRDENG_OPCODE_CLEANUP: {
+                    if(!rrdeng_main.cleanup_running) {
+                        rrdeng_main.cleanup_running++;
+                        work_dispatch(NULL, NULL, NULL, opcode, cleanup_tp_worker, after_cleanup);
+                    }
+                    break;
+                }
+
+                case RRDENG_OPCODE_JOURNAL_INDEX: {
+                    struct rrdengine_instance *ctx = cmd.ctx;
+                    struct rrdengine_datafile *datafile = cmd.data;
+                    if(!__atomic_load_n(&ctx->atomic.migration_to_v2_running, __ATOMIC_RELAXED)) {
+
+                        __atomic_store_n(&ctx->atomic.migration_to_v2_running, true, __ATOMIC_RELAXED);
+                        work_dispatch(ctx, datafile, NULL, opcode, journal_v2_indexing_tp_worker, after_journal_v2_indexing);
+                    }
+                    break;
+                }
+
+                case RRDENG_OPCODE_DATABASE_ROTATE: {
+                    struct rrdengine_instance *ctx = cmd.ctx;
+                    if (!__atomic_load_n(&ctx->atomic.now_deleting_files, __ATOMIC_RELAXED) &&
+                         ctx->datafiles.first->next != NULL &&
+                         ctx->datafiles.first->next->next != NULL &&
+                         rrdeng_ctx_exceeded_disk_quota(ctx)) {
+
+                        __atomic_store_n(&ctx->atomic.now_deleting_files, true, __ATOMIC_RELAXED);
+                        work_dispatch(ctx, NULL, NULL, opcode, database_rotate_tp_worker, after_database_rotate);
+                    }
+                    break;
+                }
+
+                case RRDENG_OPCODE_CTX_POPULATE_MRG: {
+                    struct rrdengine_instance *ctx = cmd.ctx;
+                    struct completion *completion = cmd.completion;
+                    work_dispatch(ctx, NULL, completion, opcode, populate_mrg_tp_worker, after_populate_mrg);
+                    break;
+                }
+
+                case RRDENG_OPCODE_CTX_QUIESCE: {
+                    // a ctx will shutdown shortly
+                    struct rrdengine_instance *ctx = cmd.ctx;
+                    __atomic_store_n(&ctx->quiesce.enabled, true, __ATOMIC_RELEASE);
+                    work_dispatch(ctx, NULL, NULL, opcode,
+                                      flush_all_hot_and_dirty_pages_of_section_tp_worker,
+                                      after_flush_all_hot_and_dirty_pages_of_section);
+                    break;
+                }
+
+                case RRDENG_OPCODE_CTX_SHUTDOWN: {
+                    // a ctx is shutting down
+                    struct rrdengine_instance *ctx = cmd.ctx;
+                    struct completion *completion = cmd.completion;
+                    work_dispatch(ctx, NULL, completion, opcode, ctx_shutdown_tp_worker, after_ctx_shutdown);
+                    break;
+                }
+
+                case RRDENG_OPCODE_NOOP: {
+                    /* the command queue was empty, do nothing */
+                    break;
+                }
+
+                // not opcodes
+                case RRDENG_OPCODE_MAX:
+                default: {
+                    internal_fatal(true, "DBENGINE: unknown opcode");
+                    break;
                 }
-                break;
-            case RRDENG_INVALIDATE_OLDEST_MEMORY_PAGE:
-                rrdeng_invalidate_oldest_committed(wc);
-                break;
-            }
-            default:
-                debug(D_RRDENGINE, "%s: default.", __func__);
-                break;
             }
-        } while (opcode != RRDENG_NOOP);
+
+        } while (opcode != RRDENG_OPCODE_NOOP);
     }
 
     /* cleanup operations of the event loop */
-    info("Shutting down RRD engine event loop for tier %d", ctx->tier);
+    info("DBENGINE: shutting down dbengine thread");
 
     /*
      * uv_async_send after uv_close does not seem to crash in linux at the moment,
      * it is however undocumented behaviour and we need to be aware if this becomes
      * an issue in the future.
      */
-    uv_close((uv_handle_t *)&wc->async, NULL);
-
-    while (do_flush_pages(wc, 1, NULL)) {
-        ; /* Force flushing of all committed pages. */
-    }
-    wal_flush_transaction_buffer(wc);
-    uv_run(loop, UV_RUN_DEFAULT);
-
-    info("Shutting down RRD engine event loop for tier %d complete", ctx->tier);
-    /* TODO: don't let the API block by waiting to enqueue commands */
-    uv_cond_destroy(&wc->cmd_cond);
-/*  uv_mutex_destroy(&wc->cmd_mutex); */
-    fatal_assert(0 == uv_loop_close(loop));
-    freez(loop);
-
+    uv_close((uv_handle_t *)&main->async, NULL);
+    uv_timer_stop(&main->timer);
+    uv_close((uv_handle_t *)&main->timer, NULL);
+    uv_run(&main->loop, UV_RUN_DEFAULT);
+    uv_loop_close(&main->loop);
     worker_unregister();
-    return;
-
-error_after_timer_init:
-    uv_close((uv_handle_t *)&wc->async, NULL);
-error_after_async_init:
-    fatal_assert(0 == uv_loop_close(loop));
-error_after_loop_init:
-    freez(loop);
-
-    wc->error = UV_EAGAIN;
-    /* wake up initialization thread */
-    completion_mark_complete(&ctx->rrdengine_completion);
-    worker_unregister();
-}
-
-/* C entry point for development purposes
- * make "LDFLAGS=-errdengine_main"
- */
-void rrdengine_main(void)
-{
-    int ret;
-    struct rrdengine_instance *ctx;
-
-    sanity_check();
-    ret = rrdeng_init(NULL, &ctx, "/tmp", RRDENG_MIN_PAGE_CACHE_SIZE_MB, RRDENG_MIN_DISK_SPACE_MB, 0);
-    if (ret) {
-        exit(ret);
-    }
-    rrdeng_exit(ctx);
-    fprintf(stderr, "Hello world!");
-    exit(0);
 }