rpc: sink_impl: extend backpressure until snd_buf destroy

Gleb wrote:
> backpresure mechanism in sink_impl<Serializer, Out...>::operator()
> does not work as expected. It uses semaphore _sem to limit the
> amount of work on the other shard but units are released before
> foreign_ptr is freed, so another shard may accumulate a lot of them.
> The solution as I see it is to make semaphore_units lifetime to be
> the same as foreign_ptr (by holding it in the snd_buf for instance).

Fixes #2979
Refs scylladb/scylladb#24818

Signed-off-by: Benny Halevy <[email protected]>

Author: bhalevy

include/seastar/rpc/rpc_impl.hh

@@ -840,6 +840,7 @@ future<> sink_impl<Serializer, Out...>::operator()(const Out&... args) {
         if (this->_ex) {
             return make_exception_future(this->_ex);
         }
+        data->su = std::move(su);
         // It is OK to discard this future. The user is required to
         // wait for it when closing.
         (void)smp::submit_to(this->_con->get_owner_shard(), [this, data = std::move(data), seq_num] () mutable {
@@ -871,7 +872,7 @@ future<> sink_impl<Serializer, Out...>::operator()(const Out&... args) {
                 out_of_order_bufs.erase(it);
             }
             return ret_fut;
-        }).then_wrapped([su = std::move(su), this] (future<> f) {
+        }).then_wrapped([this] (future<> f) {
             if (f.failed() && !this->_ex) { // first error is the interesting one
                 this->_ex = f.get_exception();
             } else {

include/seastar/rpc/rpc_types.hh

@@ -41,6 +41,7 @@
 #include <seastar/core/lowres_clock.hh>
 #include <boost/functional/hash.hpp>
 #include <seastar/core/sharded.hh>
+#include <seastar/core/semaphore.hh>
 
 namespace seastar {
 
@@ -254,6 +255,7 @@ struct snd_buf {
     static constexpr size_t chunk_size = 128*1024;
     uint32_t size = 0;
     std::variant<std::vector<temporary_buffer<char>>, temporary_buffer<char>> bufs;
+    semaphore_units<> su;
     using iterator = std::vector<temporary_buffer<char>>::iterator;
     snd_buf() {}
     snd_buf(snd_buf&&) noexcept;