mirrors/scylladb
1
0
Fork 0
mirror of https://github.com/scylladb/scylladb.git synced 2026-05-31 12:06:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

repair: repair_reader: close underlying reader on_end_of_stream

Browse Source 
Need to close the reader before reassigning it with
an empty f_m_r.

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>
This commit is contained in:
Benny Halevy
2021-04-04 15:21:40 +03:00
parent 391f942b2a
commit 0c19f788e5
1 changed files with 9 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
repair/row_level.cc
View File

@@ -492,9 +492,11 @@ public:
});
}
void on_end_of_stream() {
future<> on_end_of_stream() noexcept {
return _reader.close().then([this] {
_reader = make_empty_flat_reader(_schema, _permit);
_reader_handle.reset();
});
}
lw_shared_ptr<const decorated_key_with_hash>& get_current_dk() {
@@ -1192,15 +1194,17 @@ private:
_gate.check();
return _repair_reader.read_mutation_fragment().then([this, &cur_size, &new_rows_size, &cur_rows] (mutation_fragment_opt mfopt) mutable {
if (!mfopt) {
_repair_reader.on_end_of_stream();
return _repair_reader.on_end_of_stream().then([] {
return stop_iteration::yes;
});
}
return handle_mutation_fragment(*mfopt, cur_size, new_rows_size, cur_rows);
return make_ready_future<stop_iteration>(handle_mutation_fragment(*mfopt, cur_size, new_rows_size, cur_rows));
});
}).then_wrapped([this, &cur_rows, &new_rows_size] (future<> fut) mutable {
if (fut.failed()) {
_repair_reader.on_end_of_stream();
return make_exception_future<value_type>(fut.get_exception());
return make_exception_future<value_type>(fut.get_exception()).finally([this] {
return _repair_reader.on_end_of_stream();
});
}
_repair_reader.pause();
return make_ready_future<value_type>(value_type(std::move(cur_rows), new_rows_size));