From b39ca29b3c100cce4692a5385b2720af4d772835 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Botond=20D=C3=A9nes?= <bdenes@scylladb.com>
Date: Fri, 18 Nov 2022 10:03:33 +0200
Subject: [PATCH] reader_concurrency_semaphore: do_wait_admission(): detect
 admission-waiter anomaly

The semaphore should admit readers as soon as it can. So at any point in
time there should be either no waiters, or the semaphore shouldn't be
able to admit new reads. Otherwise something went wrong. Detect this
when queuing up reads and dump the diagnostics if detected.
Even though tests should ensure this should never happen, recently we've
seen a race between eviction and enqueuing producing such situations.
This is very hard to write tests for, so add built-in detection and
protection instead. Detecting this is very cheap anyway.
---
 reader_concurrency_semaphore.cc | 21 ++++++++++++---------
 reader_concurrency_semaphore.hh | 10 +++++-----
 2 files changed, 17 insertions(+), 14 deletions(-)

diff --git a/reader_concurrency_semaphore.cc b/reader_concurrency_semaphore.cc
index aadbe831d0..076cc570c4 100644
--- a/reader_concurrency_semaphore.cc
+++ b/reader_concurrency_semaphore.cc
@@ -884,11 +884,7 @@ void reader_concurrency_semaphore::evict_readers_in_background() {
 }
 
 reader_concurrency_semaphore::can_admit
-reader_concurrency_semaphore::can_admit_read(const reader_permit& permit, require_empty_waitlist wait_list_empty) const noexcept {
-    if (wait_list_empty && !_wait_list.empty()) {
-        return can_admit::no;
-    }
-
+reader_concurrency_semaphore::can_admit_read(const reader_permit& permit) const noexcept {
     if (!_ready_list.empty()) {
         return can_admit::no;
     }
@@ -913,10 +909,17 @@ future<> reader_concurrency_semaphore::do_wait_admission(reader_permit permit, r
         _execution_loop_future.emplace(execution_loop());
     }
 
-    const auto admit = can_admit_read(permit, require_empty_waitlist::yes);
-    if (admit != can_admit::yes) {
+    const auto admit = can_admit_read(permit);
+    if (admit != can_admit::yes || !_wait_list.empty()) {
         auto fut = enqueue_waiter(std::move(permit), std::move(func));
-        if (admit == can_admit::maybe) {
+        if (admit == can_admit::yes && !_wait_list.empty()) {
+            // This is a contradiction: the semaphore could admit waiters yet it has waiters.
+            // Normally, the semaphore should admit waiters as soon as it can.
+            // So at any point in time, there should either be no waiters, or it
+            // shouldn't be able to admit new reads. Otherwise something went wrong.
+            maybe_dump_reader_permit_diagnostics(*this, _permit_list, "semaphore could admit new reads yet there are waiters");
+            maybe_admit_waiters();
+        } else if (admit == can_admit::maybe) {
             evict_readers_in_background();
         }
         return fut;
@@ -932,7 +935,7 @@ future<> reader_concurrency_semaphore::do_wait_admission(reader_permit permit, r
 
 void reader_concurrency_semaphore::maybe_admit_waiters() noexcept {
     auto admit = can_admit::no;
-    while (!_wait_list.empty() && (admit = can_admit_read(_wait_list.front().permit, require_empty_waitlist::no)) == can_admit::yes) {
+    while (!_wait_list.empty() && (admit = can_admit_read(_wait_list.front().permit)) == can_admit::yes) {
         auto& x = _wait_list.front();
         try {
             x.permit.on_admission();
diff --git a/reader_concurrency_semaphore.hh b/reader_concurrency_semaphore.hh
index 5ba134f7dc..e9cbe9371d 100644
--- a/reader_concurrency_semaphore.hh
+++ b/reader_concurrency_semaphore.hh
@@ -205,12 +205,12 @@ private:
     future<> do_wait_admission(reader_permit permit, read_func func = {});
 
     // Check whether permit can be admitted or not.
-    // Caller can specify whether wait list should be empty or not for admission
-    // to be possible. can_admit::maybe means admission might be possible if some
-    // of the inactive readers are evicted.
+    // The wait list is not taken into consideration, this is the caller's
+    // responsibility.
+    // A return value of can_admit::maybe means admission might be possible if
+    // some of the inactive readers are evicted.
     enum class can_admit { no, maybe, yes };
-    using require_empty_waitlist = bool_class<class require_empty_waitlist_tag>;
-    can_admit can_admit_read(const reader_permit& permit, require_empty_waitlist wait_list_empty) const noexcept;
+    can_admit can_admit_read(const reader_permit& permit) const noexcept;
 
     void maybe_admit_waiters() noexcept;