reader_concurrency_semaphore: add disk_reads and sstables_read stats

And the infrastructure to reader_permit to update them. The
infrastructure is not wired in yet.
These metrics will be used to count the number of reads gone to disk and
the number of sstables read currently respectively.

reader_concurrency_semaphore.cc

@@ -82,6 +82,7 @@ class reader_permit::impl
     bool _marked_as_blocked = false;
     db::timeout_clock::time_point _timeout;
     query::max_result_size _max_result_size{query::result_memory_limiter::unlimited_result_size};
+    uint64_t _sstables_read = 0;
 
 private:
     void on_permit_used() {
@@ -169,6 +170,10 @@ public:
             _semaphore.on_permit_unblocked();
         }
 
+        // Should probably make a scene here, but its not worth it.
+        _semaphore._stats.sstables_read -= _sstables_read;
+        _semaphore._stats.disk_reads -= bool(_sstables_read);
+
         _semaphore.on_permit_destroyed(*this);
     }
 
@@ -326,6 +331,22 @@ public:
     void set_max_result_size(query::max_result_size s) {
         _max_result_size = std::move(s);
     }
+
+    void on_start_sstable_read() noexcept {
+        if (!_sstables_read) {
+            ++_semaphore._stats.disk_reads;
+        }
+        ++_sstables_read;
+        ++_semaphore._stats.sstables_read;
+    }
+
+    void on_finish_sstable_read() noexcept {
+        --_sstables_read;
+        --_semaphore._stats.sstables_read;
+        if (!_sstables_read) {
+            --_semaphore._stats.disk_reads;
+        }
+    }
 };
 
 static_assert(std::is_nothrow_copy_constructible_v<reader_permit>);
@@ -434,6 +455,14 @@ void reader_permit::set_max_result_size(query::max_result_size s) {
     _impl->set_max_result_size(std::move(s));
 }
 
+void reader_permit::on_start_sstable_read() noexcept {
+    _impl->on_start_sstable_read();
+}
+
+void reader_permit::on_finish_sstable_read() noexcept {
+    _impl->on_finish_sstable_read();
+}
+
 std::ostream& operator<<(std::ostream& os, reader_permit::state s) {
     switch (s) {
         case reader_permit::state::waiting:

reader_concurrency_semaphore.hh

@@ -83,6 +83,10 @@ public:
         uint64_t used_permits = 0;
         // Current number of blocked permits.
         uint64_t blocked_permits = 0;
+        // Current number of reads reading from the disk.
+        uint64_t disk_reads = 0;
+        // The number of sstables read currently.
+        uint64_t sstables_read = 0;
     };
 
     using permit_list_type = bi::list<

reader_permit.hh

@@ -157,6 +157,9 @@ public:
 
     query::max_result_size max_result_size() const;
     void set_max_result_size(query::max_result_size);
+
+    void on_start_sstable_read() noexcept;
+    void on_finish_sstable_read() noexcept;
 };
 
 using reader_permit_opt = optimized_optional<reader_permit>;