reader_concurrency_semaphore: inactive_read: keep a flat_mutation_reader

There's no need to hold a unique_ptr<flat_mutation_reader> as
flat_mutation_reader itself holds a unique_ptr<flat_mutation_reader::impl>
and functions as a unique ptr via flat_mutation_reader_opt.

With that, unregister_inactive_read was modified to return a
flat_mutation_reader_opt rather than a std::unique_ptr<flat_mutation_reader>,
keeping exactly the same semantics.

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>

mutation_reader.cc

@@ -1094,11 +1094,7 @@ void evictable_reader::maybe_pause(flat_mutation_reader reader) {
 }
 
 flat_mutation_reader_opt evictable_reader::try_resume() {
-    auto ir_ptr = _permit.semaphore().unregister_inactive_read(std::move(_irh));
-    if (!ir_ptr) {
-        return {};
-    }
-    return std::move(*ir_ptr);
+    return _permit.semaphore().unregister_inactive_read(std::move(_irh));
 }
 
 void evictable_reader::update_next_position(flat_mutation_reader& reader) {
@@ -1932,11 +1928,7 @@ reader_lifecycle_policy::pause(reader_concurrency_semaphore& sem, flat_mutation_
 
 flat_mutation_reader_opt
 reader_lifecycle_policy::try_resume(reader_concurrency_semaphore& sem, reader_concurrency_semaphore::inactive_read_handle irh) {
-    auto ir_ptr = sem.unregister_inactive_read(std::move(irh));
-    if (!ir_ptr) {
-        return {};
-    }
-    return std::move(*ir_ptr);
+    return sem.unregister_inactive_read(std::move(irh));
 }
 
 reader_concurrency_semaphore::inactive_read_handle

querier.cc

@@ -304,11 +304,11 @@ static std::optional<Querier> lookup_querier(
         throw std::runtime_error("lookup_querier(): found querier is not of the expected type");
     }
     auto& q = *q_ptr;
-    auto read_ptr = q.permit().semaphore().unregister_inactive_read(querier_utils::get_inactive_read_handle(q));
-    if (!read_ptr) {
+    auto reader_opt = q.permit().semaphore().unregister_inactive_read(querier_utils::get_inactive_read_handle(q));
+    if (!reader_opt) {
         throw std::runtime_error("lookup_querier(): found querier that is evicted");
     }
-    querier_utils::set_reader(q, std::move(*read_ptr.get()));
+    querier_utils::set_reader(q, std::move(*reader_opt));
     --stats.population;
 
     const auto can_be_used = can_be_used_for_page(q, s, ranges.front(), slice);

reader_concurrency_semaphore.cc

@@ -339,10 +339,6 @@ void reader_concurrency_semaphore::expiry_handler::operator()(entry& e) noexcept
     maybe_dump_reader_permit_diagnostics(_semaphore, *_semaphore._permit_list, "timed out");
 }
 
-reader_concurrency_semaphore::inactive_read::inactive_read(flat_mutation_reader reader)
-    : reader(std::make_unique<flat_mutation_reader>(std::move(reader))) {
-}
-
 reader_concurrency_semaphore::inactive_read::~inactive_read() {
 }
 
@@ -413,7 +409,7 @@ reader_concurrency_semaphore::inactive_read_handle reader_concurrency_semaphore:
     return inactive_read_handle();
 }
 
-std::unique_ptr<flat_mutation_reader> reader_concurrency_semaphore::unregister_inactive_read(inactive_read_handle irh) {
+flat_mutation_reader_opt reader_concurrency_semaphore::unregister_inactive_read(inactive_read_handle irh) {
     if (irh && irh._sem != this) {
         throw std::runtime_error(fmt::format(
                     "reader_concurrency_semaphore::unregister_inactive_read(): "

reader_concurrency_semaphore.hh

@@ -24,11 +24,10 @@
 #include <map>
 #include <seastar/core/future.hh>
 #include "reader_permit.hh"
+#include "flat_mutation_reader.hh"
 
 using namespace seastar;
 
-class flat_mutation_reader;
-
 /// Specific semaphore for controlling reader concurrency
 ///
 /// Use `make_permit()` to create a permit to track the resource consumption
@@ -122,11 +121,13 @@ private:
     };
 
     struct inactive_read {
-        std::unique_ptr<flat_mutation_reader> reader;
+        flat_mutation_reader reader;
         eviction_notify_handler notify_handler;
         std::optional<timer<lowres_clock>> ttl_timer;
 
-        explicit inactive_read(flat_mutation_reader);
+        explicit inactive_read(flat_mutation_reader reader_) noexcept
+            : reader(std::move(reader_))
+        { }
         inactive_read(inactive_read&&) = default;
         ~inactive_read();
     };
@@ -203,7 +204,7 @@ public:
     ///
     /// If the read was not evicted, the inactive read object, passed in to the
     /// register call, will be returned. Otherwise a nullptr is returned.
-    std::unique_ptr<flat_mutation_reader> unregister_inactive_read(inactive_read_handle irh);
+    flat_mutation_reader_opt unregister_inactive_read(inactive_read_handle irh);
 
     /// Try to evict an inactive read.
     ///