reader_concurrency_semaphore: un-bless permits when they become inactive

When the memory consumption of the semaphore reaches the configured
serialize threshold, all but the blessed permit is blocked from
consuming any more memory. This ensures that past this limit, only one
permit at a time can consume memory.
Such a blessed permit can be registered inactive. Before this patch, it
would still retain its belssed status when doing so. This could result
in this permit being req-eueud on the admission queue after a possible
eviction, potentially resulting in a complete deadlock of the semaphore:
* admission queue permits cannot be admitted because there is no memory
* admitter permits are all queued on memory, as none of them are blessed

This patch strips the blessed status from the permit when it is
registered as inactive. It also adds a unit test to verify this happens.

Fixes: scylladb#12603

reader_concurrency_semaphore.cc

@@ -796,6 +796,9 @@ reader_concurrency_semaphore::~reader_concurrency_semaphore() {
 reader_concurrency_semaphore::inactive_read_handle reader_concurrency_semaphore::register_inactive_read(flat_mutation_reader_v2 reader) noexcept {
     auto& permit_impl = *reader.permit()._impl;
     permit_impl.on_register_as_inactive();
+    if (_blessed_permit == &permit_impl) {
+        _blessed_permit = nullptr;
+    }
     // Implies _inactive_reads.empty(), we don't queue new readers before
     // evicting all inactive reads.
     // Checking the _wait_list covers the count resources only, so check memory

reader_concurrency_semaphore.hh

@@ -528,4 +528,6 @@ public:
     }
 
     void foreach_permit(noncopyable_function<void(const reader_permit&)> func);
+
+    uintptr_t get_blessed_permit() const noexcept { return reinterpret_cast<uintptr_t>(_blessed_permit); }
 };

test/boost/reader_concurrency_semaphore_test.cc

@@ -1572,3 +1572,36 @@ SEASTAR_THREAD_TEST_CASE(test_reader_concurrency_semaphore_request_memory_preser
         do_check(permit, 1, 1, std::source_location::current());
     }
 }
+
+SEASTAR_THREAD_TEST_CASE(test_reader_concurrency_semaphore_blessed_read_goes_inactive) {
+    const auto initial_resources = reader_concurrency_semaphore::resources{2, 2 * 1024};
+    const auto serialize_multiplier = 2;
+    const auto kill_multiplier = std::numeric_limits<uint32_t>::max(); // we don't want this to interfere with our test
+    reader_concurrency_semaphore semaphore(initial_resources.count, initial_resources.memory, get_name(), 100,
+            utils::updateable_value<uint32_t>(serialize_multiplier), utils::updateable_value<uint32_t>(kill_multiplier));
+    auto stop_sem = deferred_stop(semaphore);
+
+    simple_schema ss;
+    auto s = ss.schema();
+
+    auto permit = semaphore.obtain_permit(s.get(), get_name(), 1024, db::no_timeout).get();
+
+    std::vector<reader_permit::resource_units> permit_res;
+
+    permit_res.emplace_back(permit.request_memory(1024).get());
+    permit_res.emplace_back(permit.request_memory(1024).get());
+    BOOST_REQUIRE_EQUAL(semaphore.consumed_resources(), reader_resources(1, 3 * 1024));
+    BOOST_REQUIRE_EQUAL(semaphore.get_blessed_permit(), 0);
+
+    // permit is the blessed one
+    permit_res.emplace_back(permit.request_memory(1024).get());
+    BOOST_REQUIRE_EQUAL(semaphore.get_stats().reads_enqueued_for_memory, 0);
+    BOOST_REQUIRE_EQUAL(semaphore.get_blessed_permit(), permit.id());
+
+    // register the blessed permit (permit) as inactive
+    permit_res.clear();
+    auto handle = semaphore.register_inactive_read(make_empty_flat_reader_v2(s, permit));
+
+    // Upon being registered as inactive, the permit should loose the blessed status
+    BOOST_REQUIRE_EQUAL(semaphore.get_blessed_permit(), 0);
+}