Try to keep all pages from extents read from disk in the cache. (#10558)

3 files changed, 93 insertions, 10 deletions

diff --git a/database/engine/pagecache.c b/database/engine/pagecache.c
index a18207100b..d7698de012 100644
--- a/database/engine/pagecache.c
+++ b/database/engine/pagecache.c
@@ -699,6 +699,74 @@ void pg_cache_get_filtered_info_prev(struct rrdengine_instance *ctx, struct pg_c
     }
     uv_rwlock_rdunlock(&page_index->lock);
 }
+
+/**
+ * Searches for an unallocated page without triggering disk I/O. Attempts to reserve the page and get a reference.
+ * @param ctx DB context
+ * @param id lookup by UUID
+ * @param start_time exact starting time in usec
+ * @param ret_page_indexp Sets the page index pointer (*ret_page_indexp) for the given UUID.
+ * @return the page descriptor or NULL on failure. It can fail if:
+ *         1. The page is already allocated to the page cache.
+ *         2. It did not succeed to get a reference.
+ *         3. It did not succeed to reserve a spot in the page cache.
+ */
+struct rrdeng_page_descr *pg_cache_lookup_unpopulated_and_lock(struct rrdengine_instance *ctx, uuid_t *id,
+                                                               usec_t start_time)
+{
+    struct page_cache *pg_cache = &ctx->pg_cache;
+    struct rrdeng_page_descr *descr = NULL;
+    struct page_cache_descr *pg_cache_descr = NULL;
+    unsigned long flags;
+    Pvoid_t *PValue;
+    struct pg_cache_page_index *page_index = NULL;
+    Word_t Index;
+
+    uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
+    PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, id, sizeof(uuid_t));
+    if (likely(NULL != PValue)) {
+        page_index = *PValue;
+    }
+    uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
+
+    if ((NULL == PValue) || !pg_cache_try_reserve_pages(ctx, 1)) {
+        /* Failed to find page or failed to reserve a spot in the cache */
+        return NULL;
+    }
+
+    uv_rwlock_rdlock(&page_index->lock);
+    Index = (Word_t)(start_time / USEC_PER_SEC);
+    PValue = JudyLGet(page_index->JudyL_array, Index, PJE0);
+    if (likely(NULL != PValue)) {
+        descr = *PValue;
+    }
+    if (NULL == PValue || 0 == descr->page_length) {
+        /* Failed to find non-empty page */
+        uv_rwlock_rdunlock(&page_index->lock);
+
+        pg_cache_release_pages(ctx, 1);
+        return NULL;
+    }
+
+    rrdeng_page_descr_mutex_lock(ctx, descr);
+    pg_cache_descr = descr->pg_cache_descr;
+    flags = pg_cache_descr->flags;
+    uv_rwlock_rdunlock(&page_index->lock);
+
+    if ((flags & RRD_PAGE_POPULATED) || !pg_cache_try_get_unsafe(descr, 1)) {
+        /* Failed to get reference or page is already populated */
+        rrdeng_page_descr_mutex_unlock(ctx, descr);
+
+        pg_cache_release_pages(ctx, 1);
+        return NULL;
+    }
+    /* success */
+    rrdeng_page_descr_mutex_unlock(ctx, descr);
+    rrd_stat_atomic_add(&ctx->stats.pg_cache_misses, 1);
+
+    return descr;
+}
+
 /**
  * Searches for pages in a time range and triggers disk I/O if necessary and possible.
  * Does not get a reference.
diff --git a/database/engine/pagecache.h b/database/engine/pagecache.h
index 31e9739da6..d5350ef562 100644
--- a/database/engine/pagecache.h
+++ b/database/engine/pagecache.h
@@ -172,6 +172,8 @@ extern usec_t pg_cache_oldest_time_in_range(struct rrdengine_instance *ctx, uuid
 extern void pg_cache_get_filtered_info_prev(struct rrdengine_instance *ctx, struct pg_cache_page_index *page_index,
                                             usec_t point_in_time, pg_cache_page_info_filter_t *filter,
                                             struct rrdeng_page_info *page_info);
+extern struct rrdeng_page_descr *pg_cache_lookup_unpopulated_and_lock(struct rrdengine_instance *ctx, uuid_t *id,
+                                                                      usec_t start_time);
 extern unsigned
         pg_cache_preload(struct rrdengine_instance *ctx, uuid_t *id, usec_t start_time, usec_t end_time,
                          struct rrdeng_page_info **page_info_arrayp, struct pg_cache_page_index **ret_page_indexp);
diff --git a/database/engine/rrdengine.c b/database/engine/rrdengine.c
index 43135ff018..c19fe6c793 100644
--- a/database/engine/rrdengine.c
+++ b/database/engine/rrdengine.c
@@ -305,19 +305,32 @@ after_crc_check:
         }
     }
 
-    for (i = 0 ; i < xt_io_descr->descr_count; ++i) {
-        page = mallocz(RRDENG_BLOCK_SIZE);
-        descr = xt_io_descr->descr_array[i];
-        for (j = 0, page_offset = 0; j < count; ++j) {
+    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_array[j];
             /* care, we don't hold the descriptor mutex */
-            if (!uuid_compare(*(uuid_t *) header->descr[j].uuid, *descr->id) &&
-                header->descr[j].page_length == descr->page_length &&
-                header->descr[j].start_time == descr->start_time &&
-                header->descr[j].end_time == descr->end_time) {
+            if (!uuid_compare(*(uuid_t *) header->descr[i].uuid, *descrj->id) &&
+                header->descr[i].page_length == descrj->page_length &&
+                header->descr[i].start_time == descrj->start_time &&
+                header->descr[i].end_time == descrj->end_time) {
+                descr = descrj;
                 break;
             }
-            page_offset += header->descr[j].page_length;
         }
+        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);
+            if (!descr)
+                continue; /* Failed to reserve a suitable page */
+            is_prefetched_page = 1;
+        }
+        page = mallocz(RRDENG_BLOCK_SIZE);
+
         /* care, we don't hold the descriptor mutex */
         if (have_read_error) {
             /* Applications should make sure NULL values match 0 as does SN_EMPTY_SLOT */
@@ -334,7 +347,7 @@ after_crc_check:
         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) {
+        if (xt_io_descr->release_descr || is_prefetched_page) {
             pg_cache_put(ctx, descr);
         } else {
             debug(D_RRDENGINE, "%s: Waking up waiters.", __func__);