Warmup time

8 files changed, 59 insertions, 29 deletions

diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_full.cpp b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_full.cpp
index 7bc53a89609..5f1fee58b42 100644
--- a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_full.cpp
+++ b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_full.cpp
@@ -45,7 +45,13 @@ void TCommandWorkloadTopicRunFull::Config(TConfig& config) {
         .StoreTrue(&Quiet);
     config.Opts->AddLongOption("print-timestamp", "Print timestamp each second with statistics.")
         .StoreTrue(&PrintTimestamp);
-
+    config.Opts->AddLongOption("percentile", "Percentile for output statistics.")
+        .DefaultValue(50)
+        .StoreResult(&Percentile);
+    config.Opts->AddLongOption("warmup", "Warm-up time in seconds.")
+        .DefaultValue(1)
+        .StoreResult(&Warmup);
+        
     // Specific params
     config.Opts->AddLongOption('p', "producer-threads", "Number of producer threads.")
         .DefaultValue(1)
@@ -69,9 +75,6 @@ void TCommandWorkloadTopicRunFull::Config(TConfig& config) {
         .Optional()
         .DefaultValue((TStringBuilder() << NTopic::ECodec::RAW))
         .StoreMappedResultT<TString>(&Codec, &TCommandWorkloadTopicParams::StrToCodec);
-    config.Opts->AddLongOption("percentile", "Percentile for output statistics.")
-        .DefaultValue(50)
-        .StoreResult(&Percentile);
 
     config.Opts->MutuallyExclusive("message-rate", "byte-rate");
 
@@ -85,6 +88,9 @@ void TCommandWorkloadTopicRunFull::Parse(TConfig& config)
     if (Percentile >= 100) {
         throw TMisuseException() << "--percentile should be less than 100.";
     }
+    if (Warmup >= Seconds) {
+        throw TMisuseException() << "--warmup should be less than --seconds.";
+    }
 }
 
 int TCommandWorkloadTopicRunFull::Run(TConfig& config) {
@@ -92,7 +98,7 @@ int TCommandWorkloadTopicRunFull::Run(TConfig& config) {
     Log->SetFormatter(GetPrefixLogFormatter(""));
     Driver = std::make_unique<NYdb::TDriver>(CreateDriver(config, CreateLogBackend("cerr", TClientCommand::TConfig::VerbosityLevelToELogPriority(config.VerbosityLevel))));
 
-    StatsCollector = std::make_shared<TTopicWorkloadStatsCollector>(ProducerThreadCount, ConsumerCount * ConsumerThreadCount, Quiet, PrintTimestamp, WindowDurationSec, Seconds, Percentile, ErrorFlag);
+    StatsCollector = std::make_shared<TTopicWorkloadStatsCollector>(ProducerThreadCount, ConsumerCount * ConsumerThreadCount, Quiet, PrintTimestamp, WindowDurationSec, Seconds, Warmup, Percentile, ErrorFlag);
     StatsCollector->PrintHeader();
 
     auto describeTopicResult = TCommandWorkloadTopicDescribe::DescribeTopic(config.Database, *Driver);
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_full.h b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_full.h
index 38ccfec8b06..af1053d7a1c 100644
--- a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_full.h
+++ b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_full.h
@@ -16,7 +16,8 @@ namespace NYdb {
             virtual int Run(TConfig& config) override;
 
         private:
-            size_t Seconds;
+            ui32 Seconds;
+            ui32 Warmup;
             ui8 Percentile;
 
             size_t MessageRate;
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.cpp b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.cpp
index 14c7fa5a545..089419d703a 100644
--- a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.cpp
+++ b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.cpp
@@ -42,6 +42,12 @@ void TCommandWorkloadTopicRunRead::Config(TConfig& config) {
         .StoreTrue(&Quiet);
     config.Opts->AddLongOption("print-timestamp", "Print timestamp each second with statistics.")
         .StoreTrue(&PrintTimestamp);
+    config.Opts->AddLongOption("percentile", "Percentile for output statistics.")
+        .DefaultValue(50)
+        .StoreResult(&Percentile);
+    config.Opts->AddLongOption("warmup", "Warm-up time in seconds.")
+        .DefaultValue(1)
+        .StoreResult(&Warmup);
 
     // Specific params
     config.Opts->AddLongOption('c', "consumers", "Number of consumers in a topic.")
@@ -50,9 +56,6 @@ void TCommandWorkloadTopicRunRead::Config(TConfig& config) {
     config.Opts->AddLongOption('t', "threads", "Number of consumer threads.")
         .DefaultValue(1)
         .StoreResult(&ConsumerThreadCount);
-    config.Opts->AddLongOption("percentile", "Percentile for output statistics.")
-        .DefaultValue(50)
-        .StoreResult(&Percentile);
 
     config.IsNetworkIntensive = true;
 }
@@ -64,6 +67,9 @@ void TCommandWorkloadTopicRunRead::Parse(TConfig& config)
     if (Percentile >= 100) {
         throw TMisuseException() << "--percentile should be less than 100.";
     }
+    if (Warmup >= Seconds) {
+        throw TMisuseException() << "--warmup should be less than --seconds.";
+    }
 }
 
 int TCommandWorkloadTopicRunRead::Run(TConfig& config) {
@@ -71,7 +77,7 @@ int TCommandWorkloadTopicRunRead::Run(TConfig& config) {
     Log->SetFormatter(GetPrefixLogFormatter(""));
     Driver = std::make_unique<NYdb::TDriver>(CreateDriver(config, CreateLogBackend("cerr", TClientCommand::TConfig::VerbosityLevelToELogPriority(config.VerbosityLevel))));
 
-    StatsCollector = std::make_shared<TTopicWorkloadStatsCollector>(0, ConsumerCount * ConsumerThreadCount, Quiet, PrintTimestamp, WindowDurationSec, Seconds, Percentile, ErrorFlag);
+    StatsCollector = std::make_shared<TTopicWorkloadStatsCollector>(0, ConsumerCount * ConsumerThreadCount, Quiet, PrintTimestamp, WindowDurationSec, Seconds, Warmup, Percentile, ErrorFlag);
     StatsCollector->PrintHeader();
 
     std::vector<std::future<void>> threads;
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.h b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.h
index 8c68c5154bd..d1adf07ddca 100644
--- a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.h
+++ b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_read.h
@@ -16,7 +16,8 @@ namespace NYdb {
             virtual int Run(TConfig& config) override;
 
         private:
-            size_t Seconds;
+            ui32 Seconds;
+            ui32 Warmup;
             ui8 Percentile;
             
             ui32 ConsumerThreadCount;
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_write.cpp b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_write.cpp
index d3d142bbb9f..dea5fb6971c 100644
--- a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_write.cpp
+++ b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_write.cpp
@@ -44,7 +44,13 @@ void TCommandWorkloadTopicRunWrite::Config(TConfig& config) {
         .StoreTrue(&Quiet);
     config.Opts->AddLongOption("print-timestamp", "Print timestamp each second with statistics.")
         .StoreTrue(&PrintTimestamp);
-
+    config.Opts->AddLongOption("percentile", "Percentile for output statistics.")
+        .DefaultValue(50)
+        .StoreResult(&Percentile);
+    config.Opts->AddLongOption("warmup", "Warm-up time in seconds.")
+        .DefaultValue(1)
+        .StoreResult(&Warmup);
+        
     // Specific params
     config.Opts->AddLongOption('t', "threads", "Number of producer threads.")
         .DefaultValue(1)
@@ -62,9 +68,7 @@ void TCommandWorkloadTopicRunWrite::Config(TConfig& config) {
         .Optional()
         .DefaultValue((TStringBuilder() << NTopic::ECodec::RAW))
         .StoreMappedResultT<TString>(&Codec, &TCommandWorkloadTopicParams::StrToCodec);
-    config.Opts->AddLongOption("percentile", "Percentile for output statistics.")
-        .DefaultValue(50)
-        .StoreResult(&Percentile);
+
 
     config.Opts->MutuallyExclusive("message-rate", "byte-rate");
 
@@ -78,13 +82,16 @@ void TCommandWorkloadTopicRunWrite::Parse(TConfig& config)
     if (Percentile >= 100) {
         throw TMisuseException() << "--percentile should be less than 100.";
     }
+    if (Warmup >= Seconds) {
+        throw TMisuseException() << "--warmup should be less than --seconds.";
+    }
 }
 
 int TCommandWorkloadTopicRunWrite::Run(TConfig& config) {
     Log = std::make_shared<TLog>(CreateLogBackend("cerr", TClientCommand::TConfig::VerbosityLevelToELogPriority(config.VerbosityLevel)));
     Log->SetFormatter(GetPrefixLogFormatter(""));
     Driver = std::make_unique<NYdb::TDriver>(CreateDriver(config, CreateLogBackend("cerr", TClientCommand::TConfig::VerbosityLevelToELogPriority(config.VerbosityLevel))));
-    StatsCollector = std::make_shared<TTopicWorkloadStatsCollector>(ProducerThreadCount, 0, Quiet, PrintTimestamp, WindowDurationSec, Seconds, Percentile, ErrorFlag);
+    StatsCollector = std::make_shared<TTopicWorkloadStatsCollector>(ProducerThreadCount, 0, Quiet, PrintTimestamp, WindowDurationSec, Seconds, Warmup, Percentile, ErrorFlag);
     StatsCollector->PrintHeader();
 
     auto describeTopicResult = TCommandWorkloadTopicDescribe::DescribeTopic(config.Database, *Driver);
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_write.h b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_write.h
index baa0c0586f5..d3f9e8d5bc3 100644
--- a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_write.h
+++ b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_run_write.h
@@ -15,7 +15,8 @@ namespace NYdb {
             virtual void Parse(TConfig& config) override;
             virtual int Run(TConfig& config) override;
         private:
-            size_t Seconds;
+            ui32 Seconds;
+            ui32 Warmup;
             ui8 Percentile;
 
             size_t MessageRate;
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_stats_collector.cpp b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_stats_collector.cpp
index 1b04d53bb1b..9698bb035da 100644
--- a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_stats_collector.cpp
+++ b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_stats_collector.cpp
@@ -7,7 +7,7 @@ using namespace NYdb::NConsoleClient;
 TTopicWorkloadStatsCollector::TTopicWorkloadStatsCollector(
     size_t writerCount, size_t readerCount,
     bool quiet, bool printTimestamp,
-    ui32 windowDurationSec, ui32 totalDurationSec,
+    ui32 windowDurationSec, ui32 totalDurationSec, ui32 warmupSec,
     ui8 percentile,
     std::shared_ptr<std::atomic_bool> errorFlag)
     : WriterCount(writerCount)
@@ -16,6 +16,7 @@ TTopicWorkloadStatsCollector::TTopicWorkloadStatsCollector(
     , PrintTimestamp(printTimestamp)
     , WindowDurationSec(windowDurationSec)
     , TotalDurationSec(totalDurationSec)
+    , WarmupSec(warmupSec)
     , Percentile(percentile)
     , ErrorFlag(errorFlag)
     , WindowStats(MakeHolder<TTopicWorkloadStats>())
@@ -42,20 +43,20 @@ void TTopicWorkloadStatsCollector::PrintHeader(bool total) const {
     header << "Window\t";
     if (WriterCount > 0)
         header << "Write speed\tWrite time\tInflight\t";
-    if(ReaderCount > 0)
+    if (ReaderCount > 0)
         header << "Lag\t\tLag time\tRead speed\tFull time\t";
-    if(PrintTimestamp)
-        header << "Timestamp"; 
+    if (PrintTimestamp)
+        header << "Timestamp";
     header << "\n";
 
     header << "#\t";
     auto percentile = TStringBuilder() << "P" << (ui32)Percentile;
-    if(WriterCount > 0)
+    if (WriterCount > 0)
         header << "msg/s\tMB/s\t" << percentile << "(ms)\t\t" << percentile << "(msg)\t";
-    if(ReaderCount > 0)
+    if (ReaderCount > 0)
         header << percentile << "(msg)\t" << percentile << "(ms)\t\tmsg/s\tMB/s\t" << percentile << "(ms)";
     header << "\n";
-    
+
     Cout << header << Flush;
 }
 
@@ -66,12 +67,12 @@ void TTopicWorkloadStatsCollector::PrintWindowStatsLoop() {
     auto windowDuration = TDuration::Seconds(WindowDurationSec);
     while (Now() < StopTime && !*ErrorFlag) {
         if (Now() > StartTime + windowIt * windowDuration && !*ErrorFlag) {
-            CollectThreadEvents();
+            CollectThreadEvents(windowIt);
             PrintWindowStats(windowIt++);
         }
         Sleep(std::max(TDuration::Zero(), Now() - StartTime - windowIt * windowDuration));
     }
-    CollectThreadEvents();
+    CollectThreadEvents(windowIt);
 }
 
 void TTopicWorkloadStatsCollector::PrintWindowStats(ui32 windowIt) {
@@ -112,11 +113,13 @@ void TTopicWorkloadStatsCollector::PrintStats(TMaybe<ui32> windowIt) const {
     Cout << Endl;
 }
 
-void TTopicWorkloadStatsCollector::CollectThreadEvents()
+void TTopicWorkloadStatsCollector::CollectThreadEvents(ui32 windowIt)
 {
     for (auto& queue : WriterEventQueues) {
         TTopicWorkloadStats::WriterEventRef event;
         while (queue->Dequeue(&event)) {
+            if (windowIt <= WarmupSec)
+                continue;
             WindowStats->AddWriterEvent(*event);
             TotalStats.AddWriterEvent(*event);
         }
@@ -125,6 +128,8 @@ void TTopicWorkloadStatsCollector::CollectThreadEvents()
     {
         TTopicWorkloadStats::ReaderEventRef event;
         while (queue->Dequeue(&event)) {
+            if (windowIt <= WarmupSec)
+                continue;
             WindowStats->AddReaderEvent(*event);
             TotalStats.AddReaderEvent(*event);
         }
@@ -133,6 +138,8 @@ void TTopicWorkloadStatsCollector::CollectThreadEvents()
     {
         TTopicWorkloadStats::LagEventRef event;
         while (queue->Dequeue(&event)) {
+            if (windowIt <= WarmupSec)
+                continue;
             WindowStats->AddLagEvent(*event);
             TotalStats.AddLagEvent(*event);
         }
diff --git a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_stats_collector.h b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_stats_collector.h
index 28ef596d212..1759e2354f8 100644
--- a/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_stats_collector.h
+++ b/ydb/public/lib/ydb_cli/commands/topic_workload/topic_workload_stats_collector.h
@@ -13,7 +13,7 @@ namespace NYdb {
             TTopicWorkloadStatsCollector(
                 size_t producerCount, size_t consumerCount,
                 bool quiet, bool printTimestamp,
-                ui32 windowDurationSec, ui32 totalDurationSec,
+                ui32 windowDurationSec, ui32 totalDurationSec, ui32 warmupSec,
                 ui8 Percentile,
                 std::shared_ptr<std::atomic_bool> errorFlag);
 
@@ -30,7 +30,7 @@ namespace NYdb {
             ui64 GetTotalWriteMessages() const;
 
         private:
-            void CollectThreadEvents();
+            void CollectThreadEvents(ui32 windowIt);
 
             void PrintWindowStats(ui32 windowIt);
             void PrintStats(TMaybe<ui32> windowIt) const;
@@ -47,6 +47,7 @@ namespace NYdb {
 
             double WindowDurationSec;
             double TotalDurationSec;
+            double WarmupSec;
 
             ui8 Percentile;