Merge "Fix deadlocking multishard readers" from Botond

"
Multishard combining readers, running concurrently, with limited
concurrency and no timeout may deadlock, due to inactive shard readers
sitting on permits. To avoid this we have to make sure that all shard
readers belonging to a multishard combining readers, that are not
currently active, can be evicted to free up their permits, ensuring that
all readers can make progress.
Making inactive shard readers evictable is the solution for this
problem, however the original series introducing this solution
(414b14a6bd) did not go all they way and
left some loose ends. These loose ends are tied up by this mini-series.
Namely, two issues remained:
* The last reader to reach EOS was not paused (made evictable).
* Readers created/resumed as part of a read-ahead were not paused
  immediately after finishing the read-ahead.

This series fixes both of these.

Fixes: #3865
Tests: unit(release, debug)
"

* 'fix-multishard-reader-deadlock/v1' of https://github.com/denesb/scylla:
  multishard_combining_reader: pause readers after reading ahead
  multishard_combining_reader: pause *all* EOS'd readers

mutation_reader.cc

@@ -1309,7 +1309,12 @@ void multishard_combining_reader::shard_reader::read_ahead(db::timeout_clock::ti
         if (state->stopped) {
             return make_ready_future<>();
         }
-        return do_fill_buffer(timeout);
+        return do_fill_buffer(timeout).then([this, state = std::move(state)] {
+            // Read ahead is still in the background, so pause the reader.
+            if (!state->stopped && _read_ahead) {
+                pause();
+            }
+        });
     }));
 }
 
@@ -1351,8 +1356,8 @@ future<> multishard_combining_reader::handle_empty_reader_buffer(db::timeout_clo
             _end_of_stream = true;
         } else {
             move_to_next_shard();
-            reader.pause();
         }
+        reader.pause();
         return make_ready_future<>();
     } else if (reader.is_read_ahead_in_progress()) {
         return reader.fill_buffer(timeout);