Merge "Improve parallelizm of mutation source tests" from Pavel E

"
There's a run_mutation_source_tests lib helper that runs a bunch of
tests sequentially. The problem is that it does 4 different flavors
of it each being a certain decoration over provided reader. This
amplification makes some test cases run enormous amount of time
without any chance for parallelizm.

The simplest way to help running those cases in parallel is to teach
the slowest cases to run different flavors of mutation source tests
in dedicated cases. This patch makes it so.

The resulting timings are
                                  dev   debug
                 sequential run:  2m1s  53m50s
--parallel-cases (+ this patch):  1m3s  31m15s

tests: unit(dev, debug)
"

* 'br-parallel-mutation-source-tests' of https://github.com/xemul/scylla:
  test: Split multishard combining reader case
  test: Split database test case
  test: Split run_mutation_source_tests

test/boost/database_test.cc

@@ -146,9 +146,9 @@ SEASTAR_TEST_CASE(test_querying_with_limits) {
     });
 }
 
-SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source) {
-    do_with_cql_env_thread([] (cql_test_env& e) {
-        run_mutation_source_tests([&] (schema_ptr s, const std::vector<mutation>& partitions) -> mutation_source {
+static void test_database(void (*run_tests)(populate_fn_ex, bool)) {
+    do_with_cql_env_thread([run_tests] (cql_test_env& e) {
+        run_tests([&] (schema_ptr s, const std::vector<mutation>& partitions, gc_clock::time_point) -> mutation_source {
             try {
                 e.local_db().find_column_family(s->ks_name(), s->cf_name());
                 e.migration_manager().local().announce_column_family_drop(s->ks_name(), s->cf_name()).get();
@@ -172,10 +172,26 @@ SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_sourc
                     mutation_reader::forwarding fwd_mr) {
                 return cf.make_reader(s, std::move(permit), range, slice, pc, std::move(trace_state), fwd, fwd_mr);
             });
-        });
+        }, true);
     }).get();
 }
 
+SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_plain) {
+    test_database(run_mutation_source_tests_plain);
+}
+
+SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_downgrade) {
+    test_database(run_mutation_source_tests_downgrade);
+}
+
+SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_upgrade) {
+    test_database(run_mutation_source_tests_upgrade);
+}
+
+SEASTAR_THREAD_TEST_CASE(test_database_with_data_in_sstables_is_a_mutation_source_reverse) {
+    test_database(run_mutation_source_tests_reverse);
+}
+
 SEASTAR_THREAD_TEST_CASE(test_distributed_loader_with_incomplete_sstables) {
     using sst = sstables::sstable;
 

test/boost/multishard_combining_reader_as_mutation_source_test.cc

@@ -45,7 +45,7 @@
 static std::list<dummy_sharder> keep_alive_sharder;
 
 static auto make_populate(bool evict_paused_readers, bool single_fragment_buffer) {
-    return [evict_paused_readers, single_fragment_buffer] (schema_ptr s, const std::vector<mutation>& mutations) mutable {
+    return [evict_paused_readers, single_fragment_buffer] (schema_ptr s, const std::vector<mutation>& mutations, gc_clock::time_point) mutable {
         // We need to group mutations that have the same token so they land on the same shard.
         std::map<dht::token, std::vector<frozen_mutation>> mutations_by_token;
 
@@ -135,6 +135,9 @@ SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_evict_paused) {
     }).get();
 }
 
+// Single fragment buffer tests are extremely slow, so the
+// run_mutation_source_tests execution is split
+
 SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer) {
     if (smp::count < 2) {
         std::cerr << "Cannot run test " << get_name() << " with smp::count < 2" << std::endl;
@@ -142,7 +145,43 @@ SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer) {
     }
 
     do_with_cql_env_thread([&] (cql_test_env& env) -> future<> {
-        run_mutation_source_tests(make_populate(true, true));
+        run_mutation_source_tests_plain(make_populate(true, true), true);
+        return make_ready_future<>();
+    }).get();
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer_dowgrade) {
+    if (smp::count < 2) {
+        std::cerr << "Cannot run test " << get_name() << " with smp::count < 2" << std::endl;
+        return;
+    }
+
+    do_with_cql_env_thread([&] (cql_test_env& env) -> future<> {
+        run_mutation_source_tests_downgrade(make_populate(true, true), true);
+        return make_ready_future<>();
+    }).get();
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer_upgrade) {
+    if (smp::count < 2) {
+        std::cerr << "Cannot run test " << get_name() << " with smp::count < 2" << std::endl;
+        return;
+    }
+
+    do_with_cql_env_thread([&] (cql_test_env& env) -> future<> {
+        run_mutation_source_tests_upgrade(make_populate(true, true), true);
+        return make_ready_future<>();
+    }).get();
+}
+
+SEASTAR_THREAD_TEST_CASE(test_multishard_combining_reader_with_tiny_buffer_reverse) {
+    if (smp::count < 2) {
+        std::cerr << "Cannot run test " << get_name() << " with smp::count < 2" << std::endl;
+        return;
+    }
+
+    do_with_cql_env_thread([&] (cql_test_env& env) -> future<> {
+        run_mutation_source_tests_reverse(make_populate(true, true), true);
         return make_ready_future<>();
     }).get();
 }

test/lib/mutation_source_test.cc

@@ -1742,8 +1742,19 @@ void run_mutation_source_tests(populate_fn populate, bool with_partition_range_f
 }
 
 void run_mutation_source_tests(populate_fn_ex populate, bool with_partition_range_forwarding) {
-    run_mutation_reader_tests(populate, with_partition_range_forwarding);
+    run_mutation_source_tests_plain(populate, with_partition_range_forwarding);
+    run_mutation_source_tests_downgrade(populate, with_partition_range_forwarding);
+    run_mutation_source_tests_upgrade(populate, with_partition_range_forwarding);
+    run_mutation_source_tests_reverse(populate, with_partition_range_forwarding);
+    // Some tests call the sub-types individually, mind checking them
+    // if adding new stuff here
+}
 
+void run_mutation_source_tests_plain(populate_fn_ex populate, bool with_partition_range_forwarding) {
+    run_mutation_reader_tests(populate, with_partition_range_forwarding);
+}
+
+void run_mutation_source_tests_downgrade(populate_fn_ex populate, bool with_partition_range_forwarding) {
     // ? -> v2 -> v1 -> *
     run_mutation_reader_tests([populate] (schema_ptr s, const std::vector<mutation>& m, gc_clock::time_point t) -> mutation_source {
         return mutation_source([ms = populate(s, m, t)] (schema_ptr s,
@@ -1758,7 +1769,9 @@ void run_mutation_source_tests(populate_fn_ex populate, bool with_partition_rang
                     ms.make_reader_v2(s, std::move(permit), pr, slice, pc, std::move(tr), fwd, mr_fwd));
         });
     }, with_partition_range_forwarding);
+}
 
+void run_mutation_source_tests_upgrade(populate_fn_ex populate, bool with_partition_range_forwarding) {
     // ? -> v1 -> v2 -> *
     run_mutation_reader_tests([populate] (schema_ptr s, const std::vector<mutation>& m, gc_clock::time_point t) -> mutation_source {
         return mutation_source([ms = populate(s, m, t)] (schema_ptr s,
@@ -1773,7 +1786,9 @@ void run_mutation_source_tests(populate_fn_ex populate, bool with_partition_rang
                     ms.make_reader(s, std::move(permit), pr, slice, pc, std::move(tr), fwd, mr_fwd));
         });
     }, with_partition_range_forwarding);
+}
 
+void run_mutation_source_tests_reverse(populate_fn_ex populate, bool with_partition_range_forwarding) {
     // read in reverse
     run_mutation_reader_tests([populate] (schema_ptr s, const std::vector<mutation>& m, gc_clock::time_point t) -> mutation_source {
         auto table_schema = s->make_reversed();

test/lib/mutation_source_test.hh

@@ -30,6 +30,10 @@ using populate_fn_ex = std::function<mutation_source(schema_ptr s, const std::ve
 // Must be run in a seastar thread
 void run_mutation_source_tests(populate_fn populate, bool with_partition_range_forwarding = true);
 void run_mutation_source_tests(populate_fn_ex populate, bool with_partition_range_forwarding = true);
+void run_mutation_source_tests_plain(populate_fn_ex populate, bool with_partition_range_forwarding = true);
+void run_mutation_source_tests_downgrade(populate_fn_ex populate, bool with_partition_range_forwarding = true);
+void run_mutation_source_tests_upgrade(populate_fn_ex populate, bool with_partition_range_forwarding = true);
+void run_mutation_source_tests_reverse(populate_fn_ex populate, bool with_partition_range_forwarding = true);
 
 enum are_equal { no, yes };