bahman/scylla
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

db: view_updates: coroutinize move_to

Browse Source 
And allow yielding in-between freezing each update mutation.

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>
This commit is contained in:
Benny Halevy
2022-04-08 11:29:25 +03:00
parent 0e570d6ffa
commit 6454c8d67f
2 changed files with 9 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
db/view/view.cc
View File

@@ -388,12 +388,14 @@ bool matches_view_filter(const schema& base, const view_info& view, const partit
&& visitor.matches_view_filter();
}
void view_updates::move_to(utils::chunked_vector<frozen_mutation_and_schema>& mutations) {
std::transform(_updates.begin(), _updates.end(), std::back_inserter(mutations), [&, this] (auto&& m) {
future<> view_updates::move_to(utils::chunked_vector<frozen_mutation_and_schema>& mutations) {
mutations.reserve(mutations.size() + _updates.size());
for (auto it = _updates.begin(); it != _updates.end(); it = _updates.erase(it)) {
auto&& m = std::move(*it);
auto mut = mutation(_view, dht::decorate_key(*_view, std::move(m.first)), std::move(m.second));
return frozen_mutation_and_schema{freeze(mut), _view};
});
_updates.clear();
mutations.emplace_back(frozen_mutation_and_schema{freeze(mut), _view});
co_await coroutine::maybe_yield();
}
_op_count = 0;
}
@@ -928,8 +930,7 @@ future<utils::chunked_vector<frozen_mutation_and_schema>> view_update_builder::b
utils::chunked_vector<frozen_mutation_and_schema> mutations;
for (auto& update : _view_updates) {
update.move_to(mutations);
co_await coroutine::maybe_yield();
co_await update.move_to(mutations);
}
co_return mutations;
}

2
db/view/view.hh
View File

@@ -139,7 +139,7 @@ public:
, _updates(8, partition_key::hashing(*_view), partition_key::equality(*_view)) {
}
void move_to(utils::chunked_vector<frozen_mutation_and_schema>& mutations);
future<> move_to(utils::chunked_vector<frozen_mutation_and_schema>& mutations);
void generate_update(const partition_key& base_key, const clustering_row& update, const std::optional<clustering_row>& existing, gc_clock::time_point now);