fdsdump: make aggregate mode use threaded aggregator

Author: sedmicha

src/tools/fdsdump/src/aggregator/mode.cpp

@@ -7,60 +7,41 @@
  * SPDX-License-Identifier: BSD-3-Clause
  */
 
-#include <common/flowProvider.hpp>
+#include <vector>
+
 #include <aggregator/aggregator.hpp>
 #include <aggregator/mode.hpp>
+#include <aggregator/print.hpp>
 #include <aggregator/printer.hpp>
-#include <aggregator/view.hpp>
+#include <aggregator/threadedAggregator.hpp>
 #include <aggregator/viewFactory.hpp>
+#include <common/channel.hpp>
+#include <common/flowProvider.hpp>
 
 #include <memory>
 #include <iomanip>
 
 namespace fdsdump {
 namespace aggregator {
 
-void
-mode_aggregate(const Options &opts)
+static void print_results(const Options &opts, std::vector<uint8_t *> items)
 {
     View view = ViewFactory::create_view(
             opts.get_aggregation_keys(),
             opts.get_aggregation_values(),
-            opts.get_order_by());
+            opts.get_order_by(),
+            opts.get_output_limit());
 
     std::unique_ptr<Printer> printer = printer_factory(
             view,
             opts.get_output_specifier());
-    FlowProvider flows;
-    Aggregator aggr(view);
 
     const size_t rec_limit = opts.get_output_limit();
     size_t rec_printed = 0;
 
-    flows.set_biflow_autoignore(opts.get_biflow_autoignore());
-
-    if (!opts.get_input_filter().empty()) {
-        flows.set_filter(opts.get_input_filter());
-    }
-
-    for (const auto &it : glob_files(opts.get_input_file_patterns())) {
-        flows.add_file(it);
-    }
-
-    while (true) {
-        Flow *flow = flows.next_record();
-
-        if (!flow) {
-            break;
-        }
-
-        aggr.process_record(*flow);
-    }
-
-    aggr.sort_items();
     printer->print_prologue();
 
-    for (uint8_t *record : aggr.items()) {
+    for (uint8_t *record : items) {
         if (rec_limit != 0 && rec_printed >= rec_limit) {
             break;
         }
@@ -72,5 +53,48 @@ mode_aggregate(const Options &opts)
     printer->print_epilogue();
 }
 
+void
+mode_aggregate(const Options &opts)
+{
+	Channel<ThreadedAggregator *> notify_channel;
+	ThreadedAggregator aggregator(
+			opts.get_aggregation_keys(),
+			opts.get_aggregation_values(),
+			opts.get_input_filter(),
+			opts.get_input_file_patterns(),
+			opts.get_order_by(),
+			opts.get_num_threads(),
+			opts.get_biflow_autoignore(),
+			true,
+			opts.get_output_limit(),
+			notify_channel);
+	aggregator.start();
+
+	while (true) {
+		notify_channel.get(std::chrono::milliseconds(1000));
+		auto state = aggregator.get_aggregator_state();
+
+        auto state_str = aggregator_state_to_str(state);
+        if (state == AggregatorState::aggregating) {
+            auto percent = (double(aggregator.get_processed_flows()) / double(aggregator.get_total_flows())) * 100.0;
+            LOG_INFO << "Status: " << state_str << " (" << std::fixed << std::setprecision(2) << percent << "%)";
+        } else {
+            LOG_INFO << "Status: " << state_str;
+        }
+
+		if (state == AggregatorState::errored) {
+			std::rethrow_exception(aggregator.get_exception());
+		}
+
+		if (state == AggregatorState::finished) {
+			break;
+		}
+	}
+
+	aggregator.join();
+
+	print_results(opts, aggregator.get_results());
+}
+
 } // aggregator
 } // fdsdump

src/tools/fdsdump/src/aggregator/threadedAggregator.cpp

@@ -168,7 +168,7 @@ void ThreadedAggregator::thread_worker(unsigned int thread_id)
                 continue;
             }
 
-            info.processed_flows.fetch_add(1, std::memory_order_relaxed);
+            info.processed_flows.store(flows.get_processed_flow_count(), std::memory_order_relaxed);
             aggregator.process_record(*flow);
         }