PageStorage: make stress test support both v2 and v3

dbms/src/Storages/Page/CMakeLists.txt

@@ -1 +1,11 @@
 add_subdirectory(V2)
+
+# PageStorage Stress test
+if (ENABLE_V3_PAGESTORAGE)
+    add_headers_and_sources(page_stress_testing stress)
+    add_headers_and_sources(page_stress_testing stress/workload)
+    add_executable(page_stress_testing EXCLUDE_FROM_ALL ${page_stress_testing_sources})
+    target_link_libraries(page_stress_testing dbms page_storage_v3)
+    target_include_directories(page_stress_testing PRIVATE stress)
+    target_compile_options(page_stress_testing PRIVATE -Wno-format -lc++) # turn off printf format check
+endif()

dbms/src/Storages/Page/V2/tests/CMakeLists.txt

@@ -15,14 +15,6 @@ target_compile_options(gtests_page_storage PRIVATE -Wno-unknown-pragmas)
 target_compile_definitions(gtests_page_storage PRIVATE DBMS_PUBLIC_GTEST)
 add_check(gtests_page_storage)
 
-# non googletest
-add_headers_and_sources(page_stress_testing stress)
-add_headers_and_sources(page_stress_testing stress/workload)
-add_executable(page_stress_testing EXCLUDE_FROM_ALL ${page_stress_testing_sources})
-target_link_libraries(page_stress_testing page_storage_v2)
-target_include_directories(page_stress_testing PRIVATE stress)
-target_compile_options(page_stress_testing PRIVATE -Wno-format -lc++) # turn off printf format check
-
 # FIXME: mock disk full by FailPoint, and move this test into a unittest case.
 add_executable(test_page_storage_write_disk_full test_page_storage_write_disk_full.cpp)
 target_link_libraries(test_page_storage_write_disk_full dbms)

dbms/src/Storages/Page/V2/tests/stress/PSRunnable.cpp → dbms/src/Storages/Page/stress/PSRunnable.cpp

@@ -66,13 +66,13 @@ DB::ReadBufferPtr PSWriter::genRandomData(const DB::PageId pageId, DB::MemHolder
 
     holder = DB::createMemHolder(buff, [&](char * p) { free(p); });
 
-    return std::make_shared<DB::ReadBufferFromMemory>(buff, buff_sz);
+    return std::make_shared<DB::ReadBufferFromMemory>(const_cast<char *>(buff), buff_sz);
 }
 
 void PSWriter::updatedRandomData()
 {
     size_t memory_size = approx_page_mb * DB::MB * 2;
-    if (unlikely(memory != nullptr))
+    if (memory == nullptr)
     {
         memory = static_cast<char *>(malloc(memory_size));
         if (memory == nullptr)
@@ -131,7 +131,7 @@ void PSCommonWriter::updatedRandomData()
                                                                                            : batch_buffer_size);
     size_t memory_size = single_buff_size * batch_buffer_nums;
 
-    if (likely(memory == nullptr))
+    if (memory == nullptr)
     {
         memory = static_cast<char *>(malloc(memory_size));
         if (memory == nullptr)
@@ -170,7 +170,6 @@ bool PSCommonWriter::runImpl()
         bytes_used += buffptr->buffer().size();
     }
 
-    writing_page[index] = page_id;
     ps->write(std::move(wb));
     return (batch_buffer_limit == 0 || bytes_used < batch_buffer_limit);
 }
@@ -302,7 +301,10 @@ DB::PageId PSWindowWriter::genRandomPageId()
     auto random = std::round(distribution(gen));
     // Move this "random" near the pageid_boundary, If "random" is still negative, then make it positive
     random = std::abs(random + pageid_boundary);
-    return static_cast<DB::PageId>(random > pageid_boundary ? pageid_boundary++ : random);
+
+    auto page_id = static_cast<DB::PageId>(random > pageid_boundary ? pageid_boundary++ : random);
+    writing_page[index] = page_id;
+    return page_id;
 }
 
 void PSWindowReader::setWindowSize(size_t window_size_)

dbms/src/Storages/Page/V2/tests/stress/PSRunnable.h → dbms/src/Storages/Page/stress/PSRunnable.h

@@ -254,5 +254,5 @@ class PSSnapshotReader : public PSReader
 protected:
     size_t snapshots_hold_num;
     size_t snapshot_get_interval_ms = 0;
-    std::list<PageStorage::SnapshotPtr> snapshots;
+    std::list<DB::PageStorage::SnapshotPtr> snapshots;
 };

dbms/src/Storages/Page/V2/tests/stress/PSStressEnv.cpp → dbms/src/Storages/Page/stress/PSStressEnv.cpp

@@ -43,7 +43,8 @@ StressEnv StressEnv::parse(int argc, char ** argv)
         ("failpoints,F", value<std::vector<std::string>>(), "failpoint(s) to enable") //
         ("status_interval,S", value<UInt32>()->default_value(1), "Status statistics interval. 0 means no statistics") //
         ("situation_mask,M", value<UInt64>()->default_value(0), "Run special tests sequentially, example -M 2") //
-        ("verify", value<bool>()->default_value(true), "Run special tests sequentially with verify."); //
+        ("verify", value<bool>()->default_value(true), "Run special tests sequentially with verify.") //
+        ("running_ps_version,V", value<UInt16>()->default_value(3), "Select a version of PageStorage. 2 or 3 can used");
 
     po::variables_map options;
     po::store(po::parse_command_line(argc, argv, desc), options);
@@ -68,6 +69,14 @@ StressEnv StressEnv::parse(int argc, char ** argv)
     opt.status_interval = options["status_interval"].as<UInt32>();
     opt.situation_mask = options["situation_mask"].as<UInt64>();
     opt.verify = options["verify"].as<bool>();
+    opt.running_ps_version = options["running_ps_version"].as<UInt16>();
+
+    if (opt.running_ps_version != 2 && opt.running_ps_version != 3)
+    {
+        std::cerr << "Invalid running_ps_version, this arg should be 2 or 3." << std::endl;
+        std::cerr << desc << std::endl;
+        exit(0);
+    }
 
     if (options.count("paths"))
         opt.paths = options["paths"].as<std::vector<std::string>>();
@@ -93,6 +102,8 @@ void StressEnv::setup()
 #ifdef FIU_ENABLE
     fiu_init(0);
 #endif
+
+
     for (const auto & fp : failpoints)
     {
         DB::FailPointHelper::enableFailPoint(fp);

dbms/src/Storages/Page/V2/tests/stress/PSStressEnv.h → dbms/src/Storages/Page/stress/PSStressEnv.h

@@ -1,18 +1,17 @@
 #pragma once
 
 #include <Storages/Page/PageDefines.h>
-#include <Storages/Page/V2/PageStorage.h>
+#include <Storages/Page/PageStorage.h>
 #include <fmt/format.h>
 
 #include <atomic>
 
-using namespace DB::PS::V2;
 namespace Poco
 {
 class Logger;
 }
 
-using PSPtr = std::shared_ptr<DB::PS::V2::PageStorage>;
+using PSPtr = std::shared_ptr<DB::PageStorage>;
 
 enum StressEnvStat
 {
@@ -70,6 +69,7 @@ struct StressEnv
     size_t status_interval = 1;
     size_t situation_mask = 0;
     bool verify = true;
+    size_t running_ps_version = 3;
 
     std::vector<std::string> paths;
     std::vector<std::string> failpoints;
@@ -81,7 +81,8 @@ struct StressEnv
             "num_writers: {}, num_readers: {}, init_pages: {}, clean_before_run: {}"
             ", timeout_s: {}, read_delay_ms: {}, num_writer_slots: {}"
             ", avg_page_size_mb: {}, paths: [{}], failpoints: [{}]"
-            ", status_interval: {}, situation_mask: {}, verify: {}."
+            ", status_interval: {}, situation_mask: {}, verify: {}"
+            ", running_pagestorage_version : {}."
             "}}",
             num_writers,
             num_readers,
@@ -95,7 +96,8 @@ struct StressEnv
             fmt::join(failpoints.begin(), failpoints.end(), ","),
             status_interval,
             situation_mask,
-            verify
+            verify,
+            running_ps_version
             //
         );
     }

dbms/src/Storages/Page/V2/tests/stress/PSWorkload.cpp → dbms/src/Storages/Page/stress/PSWorkload.cpp

@@ -2,6 +2,8 @@
 #include <Encryption/MockKeyManager.h>
 #include <PSWorkload.h>
 #include <Poco/Logger.h>
+#include <Storages/Page/V2/PageStorage.h>
+#include <Storages/Page/V3/PageStorageImpl.h>
 #include <TestUtils/MockDiskDelegator.h>
 
 void StressWorkload::onDumpResult()
@@ -47,7 +49,7 @@ void StressWorkload::onDumpResult()
     }
 }
 
-void StressWorkload::initPageStorage(PageStorage::Config & config, String path_prefix)
+void StressWorkload::initPageStorage(DB::PageStorage::Config & config, String path_prefix)
 {
     DB::FileProviderPtr file_provider = std::make_shared<DB::FileProvider>(std::make_shared<DB::MockKeyManager>(false), false);
 
@@ -67,8 +69,22 @@ void StressWorkload::initPageStorage(PageStorage::Config & config, String path_p
         delegator = std::make_shared<DB::tests::MockDiskDelegatorSingle>(options.paths[0] + "/" + path_prefix);
     }
 
-    ps = std::make_shared<DB::PS::V2::PageStorage>("stress_test", delegator, config, file_provider);
-    ps->restore();
+    if (options.running_ps_version == 2)
+    {
+        ps = std::make_shared<DB::PS::V2::PageStorage>("stress_test", delegator, config, file_provider);
+        ps->restore();
+    }
+    else if (options.running_ps_version == 3)
+    {
+        ps = std::make_shared<DB::PS::V3::PageStorageImpl>("stress_test", delegator, config, file_provider);
+        // FIXME : we need restore V3
+    }
+    else
+    {
+        throw DB::Exception(fmt::format("Invalid PageStorage version {}",
+                                        options.running_ps_version));
+    }
+
     {
         size_t num_of_pages = 0;
         ps->traverse([&num_of_pages](const DB::Page & page) {

dbms/src/Storages/Page/V2/tests/stress/PSWorkload.h → dbms/src/Storages/Page/stress/PSWorkload.h

@@ -6,11 +6,10 @@
 #include <PSStressEnv.h>
 #include <Poco/ThreadPool.h>
 #include <Storages/Page/PageDefines.h>
-#include <Storages/Page/V2/PageStorage.h>
+#include <Storages/Page/PageStorage.h>
 #include <fmt/format.h>
 
 #define NORMAL_WORKLOAD 0
-using namespace DB::PS::V2;
 template <typename Child>
 class StressWorkloadFunc
 {
@@ -47,7 +46,7 @@ class StressWorkload
     virtual void onDumpResult();
 
 protected:
-    void initPageStorage(PageStorage::Config & config, String path_prefix = "");
+    void initPageStorage(DB::PageStorage::Config & config, String path_prefix = "");
 
     void startBackgroundTimer();
 

dbms/src/Storages/Page/V2/tests/stress/workload/HeavyMemoryCostInGC.cpp → dbms/src/Storages/Page/stress/workload/HeavyMemoryCostInGC.cpp

@@ -33,7 +33,7 @@ class HeavyMemoryCostInGC
     {
         stop_watch.start();
 
-        PageStorage::Config config;
+        DB::PageStorage::Config config;
         initPageStorage(config, name());
 
         metrics_dumper = std::make_shared<PSMetricsDumper>(1);

dbms/src/Storages/Page/V2/tests/stress/workload/HeavyRead.cpp → dbms/src/Storages/Page/stress/workload/HeavyRead.cpp

@@ -32,7 +32,7 @@ class HeavyRead : public StressWorkload
 
     void run() override
     {
-        PageStorage::Config config;
+        DB::PageStorage::Config config;
         initPageStorage(config, name());
         PSWriter::fillAllPages(ps);
 

dbms/src/Storages/Page/V2/tests/stress/workload/HeavySkewWriteRead.cpp → dbms/src/Storages/Page/stress/workload/HeavySkewWriteRead.cpp

@@ -32,7 +32,7 @@ class HeavySkewWriteRead : public StressWorkload
     void run() override
     {
         pool.addCapacity(1 + options.num_writers + options.num_readers);
-        PageStorage::Config config;
+        DB::PageStorage::Config config;
         initPageStorage(config, name());
 
         metrics_dumper = std::make_shared<PSMetricsDumper>(1);

dbms/src/Storages/Page/V2/tests/stress/workload/HeavyWrite.cpp → dbms/src/Storages/Page/stress/workload/HeavyWrite.cpp

@@ -31,7 +31,7 @@ class HeavyWrite : public StressWorkload
 
     void run() override
     {
-        PageStorage::Config config;
+        DB::PageStorage::Config config;
         initPageStorage(config, name());
 
         metrics_dumper = std::make_shared<PSMetricsDumper>(1);

dbms/src/Storages/Page/V2/tests/stress/workload/HighValidBigFileGC.cpp → dbms/src/Storages/Page/stress/workload/HighValidBigFileGC.cpp

@@ -42,7 +42,7 @@ class HighValidBigFileGCWorkload
         {
             stop_watch.start();
 
-            PageStorage::Config config;
+            DB::PageStorage::Config config;
             config.file_max_size = 8ULL * DB::GB;
             config.file_roll_size = 8ULL * DB::GB;
             initPageStorage(config, name());
@@ -63,7 +63,7 @@ class HighValidBigFileGCWorkload
 
         // Generate normal data in the same Pagefile
         {
-            PageStorage::Config config;
+            DB::PageStorage::Config config;
             config.file_max_size = DB::PAGE_FILE_MAX_SIZE;
             config.file_roll_size = DB::PAGE_FILE_ROLL_SIZE;
             initPageStorage(config, name());

dbms/src/Storages/Page/V2/tests/stress/workload/HoldSnapshotsLongTime.cpp → dbms/src/Storages/Page/stress/workload/HoldSnapshotsLongTime.cpp

@@ -33,7 +33,7 @@ class HoldSnapshotsLongTime : public StressWorkload
     void run() override
     {
         pool.addCapacity(1 + options.num_writers + options.num_readers);
-        PageStorage::Config config;
+        DB::PageStorage::Config config;
         initPageStorage(config, name());
 
         metrics_dumper = std::make_shared<PSMetricsDumper>(1);

dbms/src/Storages/Page/V2/tests/stress/workload/Normal.cpp → dbms/src/Storages/Page/stress/workload/Normal.cpp

@@ -28,7 +28,7 @@ class NormalWorkload
     {
         pool.addCapacity(1 + options.num_writers + options.num_readers);
 
-        PageStorage::Config config;
+        DB::PageStorage::Config config;
         config.num_write_slots = options.num_writer_slots;
         initPageStorage(config);
 

dbms/src/Storages/Page/V2/tests/stress/workload/PageStorageInMemoryCapacity.cpp → dbms/src/Storages/Page/stress/workload/PageStorageInMemoryCapacity.cpp

@@ -119,7 +119,7 @@ class PageStorageInMemoryCapacity : public StressWorkload
 
         std::tie(total_mem, begin_virtual_size, begin_resident_size) = getCurrentMemory();
         pool.addCapacity(1 + options.num_writers + options.num_readers);
-        PageStorage::Config config;
+        DB::PageStorage::Config config;
         initPageStorage(config, name());
 
         metrics_dumper = std::make_shared<PSMetricsDumper>(1);

dbms/src/Storages/Page/V2/tests/stress/workload/ThousandsOfOffset.cpp → dbms/src/Storages/Page/stress/workload/ThousandsOfOffset.cpp

@@ -54,7 +54,7 @@ class ThousandsOfOffset : public StressWorkload
     void run() override
     {
         pool.addCapacity(1 + options.num_writers + options.num_readers);
-        PageStorage::Config config;
+        DB::PageStorage::Config config;
         initPageStorage(config, name());
 
         metrics_dumper = std::make_shared<PSMetricsDumper>(1);