reader_concurrency_semaphore: fix a deadlock between stop() and execu…

…tion_loop()

Permits added to `_ready_list` remain there until
executed by `execution_loop()`.
But `execution_loop()` exits when `_stopped == true`,
even though nothing prevents new permits from being added
to `_ready_list` after `stop()` sets `_stopped = true`.

Thus, if there are reads concurrent with `stop()`,
it's possible for a permit to be added to `_ready_list`
after `execution_loop()` has already quit. Such a permit will
never be destroyed, and `stop()` will forever block on
`_permit_gate.close()`.

A natural solution is to dismiss `execution_loop()` only after
it's certain that `_ready_list` won't receive any new permits.
This is guaranteed by `_permit_gate.close()`. After this call completes,
it is certain that no permits *exist*.

After this patch, `execution_loop()` no longer looks at `_stopped`.
It only exits when `_ready_list_cv` breaks, and this is triggered
by `stop()` right after `_permit_gate.close()`.

Fixes #15198

Closes #15199

reader_concurrency_semaphore.cc

@@ -903,7 +903,7 @@ struct stop_execution_loop {
 }
 
 future<> reader_concurrency_semaphore::execution_loop() noexcept {
-    while (!_stopped) {
+    while (true) {
         try {
             co_await _ready_list_cv.when();
         } catch (stop_execution_loop) {
@@ -1123,10 +1123,8 @@ future<> reader_concurrency_semaphore::stop() noexcept {
     clear_inactive_reads();
     co_await _close_readers_gate.close();
     co_await _permit_gate.close();
+    _ready_list_cv.broken(std::make_exception_ptr(stop_execution_loop{}));
     if (_execution_loop_future) {
-        if (_ready_list_cv.has_waiters()) {
-            _ready_list_cv.broken(std::make_exception_ptr(stop_execution_loop{}));
-        }
         co_await std::move(*_execution_loop_future);
     }
     broken(std::make_exception_ptr(stopped_exception()));