From 2d181da656d335de38e0825ccecdf03cd6440fbb Mon Sep 17 00:00:00 2001
From: Tomasz Grabiec <tgrabiec@scylladb.com>
Date: Tue, 15 Jan 2019 11:21:40 +0100
Subject: [PATCH] row_cache: Fix crash on memtable flush with LCS

Presence checker is constructed and destroyed in the standard
allocator context, but the presence check was invoked in the LSA
context. If the presence checker allocates and caches some managed
objects, there will be alloc-dealloc mismatch.

That is the case with LeveledCompactionStrategy, which uses
incremental_selector.

Fix by invoking the presence check in the standard allocator context.

Fixes #4063.

Message-Id: <1547547700-16599-1-git-send-email-tgrabiec@scylladb.com>
(cherry picked from commit 32f711ce5669b604e41f3fd70e2e678027392509)
---
 row_cache.cc            |  4 +++-
 tests/row_cache_test.cc | 40 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 43 insertions(+), 1 deletion(-)

diff --git a/row_cache.cc b/row_cache.cc
index ac2ba54875..1a1f6fe19e 100644
--- a/row_cache.cc
+++ b/row_cache.cc
@@ -1029,7 +1029,9 @@ future<> row_cache::update(external_updater eu, memtable& m) {
             _tracker.on_partition_merge();
             return entry.partition().apply_to_incomplete(*_schema, std::move(mem_e.partition()), *mem_e.schema(), _tracker.memtable_cleaner(),
                 alloc, _tracker.region(), _tracker, _underlying_phase, acc);
-        } else if (cache_i->continuous() || is_present(mem_e.key()) == partition_presence_checker_result::definitely_doesnt_exist) {
+        } else if (cache_i->continuous()
+                   || with_allocator(standard_allocator(), [&] { return is_present(mem_e.key()); })
+                      == partition_presence_checker_result::definitely_doesnt_exist) {
             // Partition is absent in underlying. First, insert a neutral partition entry.
             cache_entry* entry = current_allocator().construct<cache_entry>(cache_entry::evictable_tag(),
                 _schema, dht::decorated_key(mem_e.key()),
diff --git a/tests/row_cache_test.cc b/tests/row_cache_test.cc
index a765463ee5..e1c2bdafcd 100644
--- a/tests/row_cache_test.cc
+++ b/tests/row_cache_test.cc
@@ -688,6 +688,46 @@ SEASTAR_TEST_CASE(test_reading_from_random_partial_partition) {
     });
 }
 
+SEASTAR_TEST_CASE(test_presence_checker_runs_under_right_allocator) {
+    return seastar::async([] {
+        cache_tracker tracker;
+        random_mutation_generator gen(random_mutation_generator::generate_counters::no);
+
+        memtable_snapshot_source underlying(gen.schema());
+
+        // Create a snapshot source whose presence checker allocates and stores a managed object.
+        // The presence checker may assume that it runs and is destroyed in the context
+        // of the standard allocator. If that isn't the case, there will be alloc-dealloc mismatch.
+        auto src = snapshot_source([&] {
+            auto ms = underlying();
+            return mutation_source([ms = std::move(ms)] (schema_ptr s,
+                const dht::partition_range& pr,
+                const query::partition_slice& slice,
+                const io_priority_class& pc,
+                tracing::trace_state_ptr tr,
+                streamed_mutation::forwarding fwd,
+                mutation_reader::forwarding mr_fwd,
+                reader_resource_tracker rrt) mutable {
+                return ms.make_reader(s, pr, slice, pc, std::move(tr), fwd, mr_fwd, std::move(rrt));
+            }, [] {
+                return [saved = managed_bytes()] (const dht::decorated_key& key) mutable {
+                    // size large enough to defeat the small blob optimization
+                    saved = managed_bytes(managed_bytes::initialized_later(), 1024);
+                    return partition_presence_checker_result::maybe_exists;
+                };
+            });
+        });
+
+        row_cache cache(gen.schema(), std::move(src), tracker);
+
+        auto m1 = make_fully_continuous(gen());
+
+        auto mt = make_lw_shared<memtable>(gen.schema());
+        mt->apply(m1);
+        cache.update([&] { underlying.apply(m1); }, *mt).get();
+    });
+}
+
 SEASTAR_TEST_CASE(test_random_partition_population) {
     return seastar::async([] {
         cache_tracker tracker;