feat: support the writer to use a separate queue (#20)

* feat: support the writer to use a separate queue

* feat: compensation for uneven write queue completion

* feat: by default no thread is bound when writing data

* feat: return 0 when showing version

Author: YamingPei

src/actions/components/connector/src/TDengineConnector.cpp

@@ -165,7 +165,9 @@ bool TDengineConnector::execute(const StmtV2InsertData& data) {
     );
     if (code != 0) {
         std::cerr << display_name_ << " failed to bind parameters: "
-                  << taos_stmt2_error(stmt_) << std::endl;
+                  << taos_stmt2_error(stmt_)
+                  << " [code: 0x" << std::hex << code << std::dec << "]"
+                  << std::endl;
         return false;
     }
 
@@ -174,7 +176,9 @@ bool TDengineConnector::execute(const StmtV2InsertData& data) {
     code = taos_stmt2_exec(stmt_, &affected_rows);
     if (code != 0) {
         std::cerr << display_name_ << " execute failed: "
-                  << taos_stmt2_error(stmt_) << std::endl;
+                  << taos_stmt2_error(stmt_)
+                  << " [code: 0x" << std::hex << code << std::dec << "]"
+                  << std::endl;
         return false;
     }
 

src/actions/components/expression/CMakeLists.txt

@@ -17,6 +17,7 @@ target_include_directories(components_expression
 target_link_libraries(components_expression
   PUBLIC
     luajit::luajit
+    Threads::Threads
     m
     dl
 )

src/actions/components/formatter/test/TestTopicGenerator.cpp

@@ -12,8 +12,8 @@ void test_basic_pattern() {
 
     MultiBatch batch;
     std::vector<RowData> rows;
-    rows.push_back({1500000000000, {"f01", 101}});
-    rows.push_back({1500000000001, {"f02", 102}});
+    rows.push_back({1500000000000, {std::string("f01"), 101}});
+    rows.push_back({1500000000001, {std::string("f02"), 102}});
     batch.table_batches.emplace_back("t1", std::move(rows));
     batch.update_metadata();
 
@@ -63,7 +63,7 @@ void test_col_not_found() {
 
     MultiBatch batch;
     std::vector<RowData> rows;
-    rows.push_back({1500000000000, {"x"}});
+    rows.push_back({1500000000000, {std::string("x")}});
     batch.table_batches.emplace_back("my_table", std::move(rows));
     batch.update_metadata();
 
@@ -86,7 +86,7 @@ void test_pattern_edge_cases() {
 
     MultiBatch batch;
     std::vector<RowData> rows;
-    rows.push_back({1500000000000, {"A"}});
+    rows.push_back({1500000000000, {std::string("A")}});
     batch.table_batches.emplace_back("my_table", std::move(rows));
     batch.update_metadata();
 

src/actions/components/generator/test/TestRandomColumnGenerator.cpp

@@ -97,7 +97,6 @@ void test_generate_multiple_values() {
     std::cout << "test_generate_multiple_values passed.\n";
 }
 
-
 void test_generate_int_column_with_values() {
     ColumnConfig config;
     config.type = "int";

src/actions/config/inc/InsertDataConfig.hpp

@@ -27,7 +27,10 @@ struct InsertDataConfig {
     size_t insert_threads = 8;
     size_t queue_capacity = 10;
     double queue_warmup_ratio = 0.0;
+    bool shared_queue = false;
     std::string thread_allocation = "index_range";
+    bool thread_affinity = false;
+    bool thread_realtime = false;
 
     struct FailureHandling {
         size_t max_retries = 0;

src/actions/core/insert/src/InsertDataAction.cpp

@@ -114,6 +114,7 @@ void InsertDataAction::execute() {
         const size_t consumer_thread_count = config_.insert_threads;
         const size_t queue_capacity = config_.queue_capacity;
         const double queue_warmup_ratio = config_.queue_warmup_ratio;
+        const bool shared_queue = config_.shared_queue;
         const size_t per_request_rows = config_.schema.generation.per_batch_rows;
         const size_t interlace_rows = config_.schema.generation.interlace_mode.rows;
         const int64_t per_table_rows = config_.schema.generation.per_table_rows;
@@ -133,7 +134,7 @@ void InsertDataAction::execute() {
         MemoryPool pool(block_count, max_tables_per_block, max_rows_per_table, col_instances_);
 
         // Create data pipeline
-        DataPipeline<FormatResult> pipeline(block_count);
+        DataPipeline<FormatResult> pipeline(shared_queue, producer_thread_count, consumer_thread_count, queue_capacity);
         Barrier sync_barrier(consumer_thread_count + 1);
 
         // Start consumer threads
@@ -189,8 +190,12 @@ void InsertDataAction::execute() {
 
             producer_threads.emplace_back([this, i, &split_names, &pipeline, data_manager, &active_producers, &producer_finished] {
                 try {
-                    set_thread_affinity(i, false, "Producer");
-                    set_realtime_priority();
+                    if (config_.thread_affinity) {
+                        set_thread_affinity(i, false, "Producer");
+                    }
+                    if (config_.thread_realtime) {
+                        set_realtime_priority();
+                    }
                     producer_thread_function(i, split_names[i], pipeline, data_manager);
                     producer_finished[i].store(true);
                 } catch (const std::exception& e) {
@@ -201,7 +206,8 @@ void InsertDataAction::execute() {
         }
 
         (void)ProcessUtils::get_cpu_usage_percent();
-        std::this_thread::sleep_for(std::chrono::seconds(2));
+        int64_t wait_seconds = std::min(static_cast<int64_t>(5), static_cast<int64_t>(producer_thread_count));
+        std::this_thread::sleep_for(std::chrono::seconds(wait_seconds));
 
         while (true) {
             size_t total_queued = pipeline.total_queued();
@@ -270,7 +276,7 @@ void InsertDataAction::execute() {
                     << "Queue: " << std::setw(3) << std::setfill(' ') << pipeline.total_queued() << " items | "
                     << "CPU Usage: " << std::setw(7) << std::fixed << std::setprecision(2) << ProcessUtils::get_cpu_usage_percent() << "% | "
                     << "Memory Usage: " << ProcessUtils::get_memory_usage_human_readable() << " | "
-                    << "Thread Count: " << std::setw(3) << std::setfill(' ') << ProcessUtils::get_thread_count() << "\n";
+                    << "Thread Count: " << std::setw(3) << std::setfill(' ') << ProcessUtils::get_thread_count() << std::endl;
         }
 
         std::cout << "All producer threads have finished." << std::endl;
@@ -294,7 +300,7 @@ void InsertDataAction::execute() {
                           << "Processing Rate: " << std::setw(6) << std::fixed << std::setprecision(2) << process_rate << " items/s | "
                           << "CPU Usage: " << std::setw(7) << std::fixed << std::setprecision(2) << ProcessUtils::get_cpu_usage_percent() << "% | "
                           << "Memory Usage: " << ProcessUtils::get_memory_usage_human_readable() << " | "
-                          << "Thread Count: " << std::setw(3) << std::setfill(' ') << ProcessUtils::get_thread_count() << "\n";
+                          << "Thread Count: " << std::setw(3) << std::setfill(' ') << ProcessUtils::get_thread_count() << std::endl;
 
                 last_queue_size = current_queue_size;
                 last_check_time = current_time;
@@ -360,7 +366,7 @@ void InsertDataAction::execute() {
         double avg_wait_time = total_wait_time / consumer_thread_count; // average per thread
 
         // Calculate total duration (seconds)
-        const auto total_duration = std::chrono::duration<double>(max_end_write_time - min_start_write_time).count();
+        const auto total_duration = std::chrono::duration<double>(max_end_write_time - min_start_write_time).count() - avg_wait_time;
 
         // Calculate average insert rate
         const double avg_rows_per_sec = total_duration > 0 ?
@@ -391,7 +397,7 @@ void InsertDataAction::execute() {
         // Print performance statistics
         double thread_latency = global_write_metrics.get_sum() / consumer_thread_count / 1000;
         double effective_ratio = thread_latency / total_duration * 100.0;
-        double framework_ratio = (1 - (thread_latency + avg_wait_time) / total_duration) * 100.0;
+        double framework_ratio = (1 - thread_latency / total_duration) * 100.0;
         TimeIntervalStrategy time_strategy(config_.time_interval, config_.timestamp_precision);
         std::cout << "\n=============================================== Insert Latency & Efficiency Metrics ==========================================\n"
                 << "Total Operations: " << global_write_metrics.get_samples().size() << "\n"
@@ -485,7 +491,7 @@ void InsertDataAction::producer_thread_function(
         // }, formatted_result);
 
         // Push data to pipeline
-        pipeline.push_data(std::move(formatted_result));
+        pipeline.push_data(producer_id, std::move(formatted_result));
 
         // std::cout << "Producer " << producer_id << ": Pushed batch for table(s): "
         //           << batch_size << ", total rows: " << total_rows
@@ -531,7 +537,7 @@ void InsertDataAction::consumer_thread_function(
     // Data processing loop
     (void)running;
     while (true) {
-        auto result = pipeline.fetch_data();
+        auto result = pipeline.fetch_data(consumer_id);
 
         switch (result.status) {
         case DataPipeline<FormatResult>::Status::Success: