Merge branch 'fdsdump-threading' into devel

Author: sedmicha

src/tools/fdsdump/src/aggregator/CMakeLists.txt

@@ -17,6 +17,8 @@ set(AGGREGATOR_SRC
     stdAllocator.cpp
     stdHashTable.cpp
     tablePrinter.cpp
+    threadedAggregator.cpp
+    thresholdAlgorithm.cpp
     value.cpp
     view.cpp
     viewFactory.cpp

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

@@ -28,6 +28,38 @@
 namespace fdsdump {
 namespace aggregator {
 
+// Merge two hash tables containing records defined by view
+void merge_hash_tables(const View &view, HashTable &dst_table, HashTable &src_table)
+{
+    uint8_t *dst_record = nullptr;
+    for (uint8_t *src_record : src_table.items()) {
+        bool found = dst_table.find_or_create(src_record, dst_record);
+        if (found) {
+            // If found - merge
+            for (auto x : view.iter_values(dst_record, src_record)) {
+                x.field.merge(x.value1, x.value2);
+            }
+        } else {
+            // If not found - copy over
+            // Key is already copied by find_or_create, so only value needs to be copied
+            unsigned int key_size = view.key_size(src_record);
+            unsigned int value_size = view.value_size();
+            //TODO: Possible optimalization: Can we just move pointers instead of memcpy?
+            std::memcpy(dst_record + key_size, src_record + key_size, value_size);
+        }
+    }
+}
+
+// Sort records
+void sort_records(const View &view, std::vector<uint8_t *>& records)
+{
+    if (view.order_fields().empty()) {
+        return;
+    }
+    std::sort(records.begin(), records.end(),
+            [&view](uint8_t *a, uint8_t *b) { return view.ordered_before(a, b); });
+}
+
 Aggregator::Aggregator(const View &view) :
     m_table(view),
     m_key_buffer(65535),
@@ -104,45 +136,14 @@ Aggregator::aggregate(FlowContext &ctx)
 void
 Aggregator::sort_items()
 {
-    if (!m_view.order_fields().empty()) {
-        std::sort(items().begin(), items().end(),
-            [this](uint8_t *a, uint8_t *b) { return m_view.ordered_before(a, b); });
-    }
+	sort_records(m_view, items());
 }
 
-
-//     uint8_t *record;
-
-//     if (!m_table.find_or_create(&m_key_buffer[0], record)) {
-//         init_values(*m_view_def.get(), record + m_view_def->keys_size);
-//     }
-
-//     for (const auto &iter : m_view_def->iter_values(record + m_view_def->keys_size)) {
-//         aggregate_value(iter.field, drec, &iter.value, direction, drec_find_flags);
-//     }
-// }
-
-// void
-// Aggregator::merge(Aggregator &other, unsigned int max_num_items)
-// {
-//     unsigned int n = 0;
-//     for (uint8_t *other_record : other.items()) {
-//         if (max_num_items != 0 && n == max_num_items) {
-//             break;
-//         }
-
-//         uint8_t *record;
-
-//         if (!m_table.find_or_create(other_record, record)) {
-//             //TODO: this copy is unnecessary, we could just take the already allocated record from the other table instead
-//             memcpy(record, other_record, m_view_def->keys_size + m_view_def->values_size);
-//         } else {
-//             merge_records(*m_view_def.get(), record, other_record);
-//         }
-
-//         n++;
-//     }
-// }
+void
+Aggregator::merge(Aggregator &other)
+{
+	merge_hash_tables(m_view, m_table, other.m_table);
+}
 
 } // aggregator
 } // fdsdump

src/tools/fdsdump/src/aggregator/aggregator.hpp

@@ -21,6 +21,10 @@
 namespace fdsdump {
 namespace aggregator {
 
+void merge_hash_tables(const View &view, HashTable &dst_table, HashTable &src_table);
+
+void sort_records(const View &view, std::vector<uint8_t *>& records);
+
 /**
  * @brief A class performing aggregation of fds data records based on a view definition.
  */
@@ -43,8 +47,8 @@ class Aggregator {
      * @brief Merge other aggregator into this one.
      * @param other The other aggregator
      */
-    // void
-    // merge(Aggregator &other, unsigned int max_num_items = 0);
+    void
+    merge(Aggregator &other);
 
     /**
      * @brief The underlying hash table.

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

@@ -7,59 +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;
         }
@@ -71,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