From 8eeac10ded098ba738c268453c8e82636a5f930f Mon Sep 17 00:00:00 2001 From: Piotr Sarna Date: Mon, 7 Jun 2021 12:58:11 +0200 Subject: [PATCH] cql3: limit the concurrency of indexed statements Indexed select statements fetch primary key information from their internal materialized views and then use it to query the base table. Unfortunately, the current mechanism for retrieving base table rows makes it easy to overwhelm the replicas with unbounded concurrency - the number of concurrent ops is increased exponentially until a short read is encountered, but it's not enough to cap the concurrency - if data is fetched row-by-row, then short reads usually don't occur and as a result it's easy to see concurrency of 1M or higher. In order to avoid overloading the replicas, the concurrency of indexed queries is now capped at 4096. The number can be subject to debate, its reasoning is as follows: for 2KiB rows, so moderately large but not huge, they result in fetching 10MB of data, which is the granularity used by replicas. For 200B rows, which is rather small, the result would still be around 1MB. At the same time, 4096 separate tasks also means 4096 allocations, so increasing the number also strains the allocator. Fixes #8799 Tests: unit(release), manual: observing metrics of modified index_paging_test --- cql3/statements/select_statement.cc | 6 ++++-- cql3/statements/select_statement.hh | 2 ++ 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/cql3/statements/select_statement.cc b/cql3/statements/select_statement.cc index b83309f83b..38d2e82a2b 100644 --- a/cql3/statements/select_statement.cc +++ b/cql3/statements/select_statement.cc @@ -574,7 +574,9 @@ indexed_table_select_statement::do_execute_base_query( command->slice.set_range(*_schema, base_pk, row_ranges); } } - concurrency *= 2; + if (concurrency < max_base_table_query_concurrency) { + concurrency *= 2; + } return proxy.query(_schema, command, std::move(prange), options.get_consistency(), {timeout, state.get_permit(), state.get_client_state(), state.get_trace_state()}) .then([&ranges_to_vnodes, &merger] (service::storage_proxy::coordinator_query_result qr) { auto is_short_read = qr.query_result->is_short_read(); @@ -638,7 +640,7 @@ indexed_table_select_statement::do_execute_base_query( // we continue exponentially, asking for 2x more key than before auto already_done = std::distance(keys.begin(), key_it); auto next_iteration = already_done + 1; - next_iteration = std::min(next_iteration, keys.size() - already_done); + next_iteration = std::min({next_iteration, keys.size() - already_done, max_base_table_query_concurrency}); auto key_it_end = key_it + next_iteration; auto command = ::make_lw_shared(*cmd); diff --git a/cql3/statements/select_statement.hh b/cql3/statements/select_statement.hh index d254fb6cf3..87279ced18 100644 --- a/cql3/statements/select_statement.hh +++ b/cql3/statements/select_statement.hh @@ -188,6 +188,8 @@ class indexed_table_select_statement : public select_statement { noncopyable_function _get_partition_ranges_for_posting_list; noncopyable_function _get_partition_slice_for_posting_list; public: + static constexpr size_t max_base_table_query_concurrency = 4096; + static ::shared_ptr prepare(database& db, schema_ptr schema, uint32_t bound_terms,