repair: row_level: clear_gently: clear_gently each repair_row

Rows might be large so free them gently by:
- add bytes_ostream.clear_gently that may yield in the chunk
  freeing loop.
- use that in frozen_mutation_fragment, contained in repair_row.

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

bytes_ostream.hh

@@ -26,6 +26,7 @@
 #include "bytes.hh"
 #include "hashing.hh"
 #include <seastar/core/simple-stream.hh>
+#include <seastar/core/loop.hh>
 #include <concepts>
 
 /**
@@ -460,4 +461,25 @@ public:
             _begin->next.reset();
         }
     }
+
+    // Makes this instance empty using async continuations, while allowing yielding.
+    //
+    // The first buffer is not deallocated, so callers may rely on the
+    // fact that if they write less than the initial chunk size between
+    // the clear() calls then writes will not involve any memory allocations,
+    // except for the first write made on this instance.
+    future<> clear_gently() noexcept {
+        if (!_begin) {
+            return make_ready_future<>();
+        }
+        _begin->offset = 0;
+        _size = 0;
+        return do_until([this] { return !_begin->next; }, [this] {
+            // move next->next first to avoid it being recursively destroyed
+            // in ~chunk when _begin->next is move-assigned.
+            auto next = std::move(_begin->next->next);
+            _begin->next = std::move(next);
+            return make_ready_future<>();
+        });
+    }
 };

frozen_mutation.hh

@@ -128,6 +128,10 @@ public:
     const bytes_ostream& representation() const { return _bytes; }
 
     mutation_fragment unfreeze(const schema& s, reader_permit permit);
+
+    future<> clear_gently() noexcept {
+        return _bytes.clear_gently();
+    }
 };
 
 frozen_mutation_fragment freeze(const schema& s, const mutation_fragment& mf);

repair/row_level.cc

@@ -400,6 +400,16 @@ class repair_row {
     is_dirty_on_master dirty_on_master() const {
         return _dirty_on_master;
     }
+    future<> clear_gently() noexcept {
+        auto f = _fm ? _fm->clear_gently() : make_ready_future<>();
+        return f.finally([this] {
+            _fm.reset();
+            _dk_with_hash = {};
+            _boundary.reset();
+            _hash.reset();
+            _mf = {};
+        });
+    }
 };
 
 class repair_reader {