cached_file: Evict unused pages that aren't linked to LRU yet

It was found that cached_file dtor can hit the following assert
after OOM

cached_file_test: utils/cached_file.hh:379: cached_file::~cached_file(): Assertion _cache.empty()' failed.`

cached_file's dtor iterates through all entries and evict those
that are linked to LRU, under the assumption that all unused
entries were linked to LRU.

That's partially correct. get_page_ptr() may fetch more than 1
page due to read ahead, but it will only call cached_page::share()
on the first page, the one that will be consumed now.

share() is responsible for automatically placing the page into
LRU once refcount drops to zero.

If the read is aborted midway, before cached_file has a chance
to hit the 2nd page (read ahead) in cache, it will remain there
with refcount 0 and unlinked to LRU, in hope that a subsequent
read will bring it out of that state.

Our main user of cached_file is per-sstable index caching.
If the scenario above happens, and the sstable and its associated
cached_file is destroyed, before the 2nd page is hit, cached_file
will not be able to clear all the cache because some of the
pages are unused and not linked.

A page read ahead will be linked into LRU so it doesn't sit in
memory indefinitely. Also allowing for cached_file dtor to
clear all cache if some of those pages brought in advance
aren't fetched later.

A reproducer was added.

Fixes #14814.

Signed-off-by: Raphael S. Carvalho <raphaelsc@scylladb.com>

Closes #14818

(cherry picked from commit 050ce9e)

test/boost/cached_file_test.cc

@@ -120,6 +120,25 @@ SEASTAR_THREAD_TEST_CASE(test_file_wrapper) {
     BOOST_CHECK_THROW(read_to_string(f, 0, cf.size() + 1), seastar::file::eof_error);
 }
 
+/* Reproducer for issue https://github.com/scylladb/scylladb/issues/14814 */
+SEASTAR_THREAD_TEST_CASE(test_no_crash_on_dtor_after_oom) {
+    auto page_size = cached_file::page_size;
+    cached_file::metrics metrics;
+    test_file tf = make_test_file(page_size * 32); // 128k.
+    logalloc::region region;
+    cached_file cf(tf.f, metrics, cf_lru, region, page_size * 32);
+    seastar::file f = make_cached_seastar_file(cf);
+
+    utils::get_local_injector().enable("cached_file_get_first_page", true /* oneshot */);
+
+    try {
+        BOOST_REQUIRE_EQUAL(tf.contents.substr(0, page_size * 32),
+                            read_to_string(f, 0, page_size * 32));
+    } catch (...) {
+        testlog.info("exception caught: {}", std::current_exception());
+    }
+}
+
 SEASTAR_THREAD_TEST_CASE(test_concurrent_population) {
     auto page_size = cached_file::page_size;
     cached_file::metrics metrics;

utils/cached_file.hh

@@ -12,6 +12,7 @@
 #include "utils/div_ceil.hh"
 #include "utils/bptree.hh"
 #include "utils/lru.hh"
+#include "utils/error_injection.hh"
 #include "tracing/trace_state.hh"
 
 #include <seastar/core/file.hh>
@@ -193,10 +194,15 @@ private:
                         _metrics.cached_bytes += cp.size_in_allocator();
                         _cached_bytes += cp.size_in_allocator();
                     }
+                    // pages read ahead will be placed into LRU, as there's no guarantee they will be fetched later.
+                    cached_page::ptr_type ptr = cp.share();
                     if (!first_page) {
-                        first_page = cp.share();
+                        first_page = std::move(ptr);
                     }
                 }
+                utils::get_local_injector().inject("cached_file_get_first_page", []() {
+                    throw std::bad_alloc();
+                });
                 return first_page;
             });
     }