Merge branch 'fdsdump-fixes' into devel

Author: sedmicha

src/tools/fdsdump/src/CMakeLists.txt

@@ -10,7 +10,7 @@ add_executable(fdsdump
     main.cpp
     options.cpp
     $<TARGET_OBJECTS:common_obj>
-	$<TARGET_OBJECTS:lister_obj>
+    $<TARGET_OBJECTS:lister_obj>
     $<TARGET_OBJECTS:aggregator_obj>
     $<TARGET_OBJECTS:statistics_obj>
 )
@@ -21,7 +21,8 @@ target_include_directories(fdsdump
 
 target_link_libraries(fdsdump
     PUBLIC
-        ${FDS_LIBRARIES})
+        ${FDS_LIBRARIES}
+        Threads::Threads)
 
 # Installation targets
 install(

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

@@ -136,13 +136,13 @@ Aggregator::aggregate(FlowContext &ctx)
 void
 Aggregator::sort_items()
 {
-	sort_records(m_view, items());
+    sort_records(m_view, items());
 }
 
 void
 Aggregator::merge(Aggregator &other)
 {
-	merge_hash_tables(m_view, m_table, other.m_table);
+    merge_hash_tables(m_view, m_table, other.m_table);
 }
 
 } // aggregator

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

@@ -90,17 +90,17 @@ class InOutValueField : public Field {
     void
     init(Value &value) const override;
 
-	/**
-	 * @brief Aggregate the value retrieved from the provided flow record into the provided
-	 *        aggregated value
-	 *
-	 * @param ctx  The flow record to retrieve the value that will be aggregated from
-	 * @param aggregated_value  The currently accumulated value that the retrieved value will be
-	 *                          aggregated towards
-	 *
-	 * @return true if the field was found in the flow record and the aggregated value was updated,
-	 * false otherwise
-	 */
+    /**
+     * @brief Aggregate the value retrieved from the provided flow record into the provided
+     *        aggregated value
+     *
+     * @param ctx  The flow record to retrieve the value that will be aggregated from
+     * @param aggregated_value  The currently accumulated value that the retrieved value will be
+     *                          aggregated towards
+     *
+     * @return true if the field was found in the flow record and the aggregated value was updated,
+     * false otherwise
+     */
     bool
     aggregate(FlowContext &ctx, Value &aggregated_value) const override;
 

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

@@ -56,23 +56,23 @@ static void print_results(const Options &opts, std::vector<uint8_t *> items)
 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();
+    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) {
@@ -82,18 +82,18 @@ mode_aggregate(const Options &opts)
             LOG_INFO << "Status: " << state_str;
         }
 
-		if (state == AggregatorState::errored) {
-			std::rethrow_exception(aggregator.get_exception());
-		}
+        if (state == AggregatorState::errored) {
+            std::rethrow_exception(aggregator.get_exception());
+        }
 
-		if (state == AggregatorState::finished) {
-			break;
-		}
-	}
+        if (state == AggregatorState::finished) {
+            break;
+        }
+    }
 
-	aggregator.join();
+    aggregator.join();
 
-	print_results(opts, aggregator.get_results());
+    print_results(opts, aggregator.get_results());
 }
 
 } // aggregator

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

@@ -43,31 +43,31 @@ const char *aggregator_state_to_str(AggregatorState aggregator_state);
 class ThreadedAggregator {
     DISABLE_COPY_AND_MOVE(ThreadedAggregator)
 
-	struct ThreadInfo {
-		std::atomic<uint64_t> processed_files{0};
-		std::atomic<uint64_t> processed_flows{0};
-		std::atomic<AggregatorState> state{AggregatorState::none};
-		std::exception_ptr exception = nullptr;
-		std::atomic<bool> cancelled{false};
-	};
+    struct ThreadInfo {
+        std::atomic<uint64_t> processed_files{0};
+        std::atomic<uint64_t> processed_flows{0};
+        std::atomic<AggregatorState> state{AggregatorState::none};
+        std::exception_ptr exception = nullptr;
+        std::atomic<bool> cancelled{false};
+    };
 
 public:
-	/**
-	 * @brief Create a new threaded aggregator instance
-	 *
-	 * @param aggregation_keys The aggregation keys
-	 * @param aggregation_values The aggregation values
-	 * @param input_filter The input record filter
-	 * @param input_file_patterns The input file patterns (paths w/ glob support)
-	 * @param order_by The fields to order by
-	 * @param num_threads The number of threads
-	 * @param biflow_autoignore Whether the biflow autoignore option should be applied
-	 * @param merge_results Merge output results or not
-	 * @param merge_topk If merge results is enabled, how many top K values to obtain (0 = all
-	 *                   values, i.e. no top K is performed and everything is merged)
-	 * @param notify_channel The channel to send state notifications through
-	 */
-	ThreadedAggregator(
+    /**
+     * @brief Create a new threaded aggregator instance
+     *
+     * @param aggregation_keys The aggregation keys
+     * @param aggregation_values The aggregation values
+     * @param input_filter The input record filter
+     * @param input_file_patterns The input file patterns (paths w/ glob support)
+     * @param order_by The fields to order by
+     * @param num_threads The number of threads
+     * @param biflow_autoignore Whether the biflow autoignore option should be applied
+     * @param merge_results Merge output results or not
+     * @param merge_topk If merge results is enabled, how many top K values to obtain (0 = all
+     *                   values, i.e. no top K is performed and everything is merged)
+     * @param notify_channel The channel to send state notifications through
+     */
+    ThreadedAggregator(
         const std::string& aggregation_keys,
         const std::string& aggregation_values,
         const std::string& input_filter,
@@ -76,64 +76,64 @@ class ThreadedAggregator {
         unsigned int num_threads,
         bool biflow_autoignore,
         bool merge_results,
-		unsigned int merge_topk,
+        unsigned int merge_topk,
         Channel<ThreadedAggregator *> &notify_channel
     );
 
-	/**
-	 * @brief Start the aggregation. This is a non-blocking operation that starts the threads that
-	 *        will perform the aggregation.
-	 */
-	void start();
+    /**
+     * @brief Start the aggregation. This is a non-blocking operation that starts the threads that
+     *        will perform the aggregation.
+     */
+    void start();
 
-	/**
-	 * @brief Get the exception that occured if the aggregation state is _errored_
-	 */
+    /**
+     * @brief Get the exception that occured if the aggregation state is _errored_
+     */
     std::exception_ptr get_exception() { return m_exception; }
 
-	/**
-	 * @brief Get the final results if merging was enabled
-	 */
+    /**
+     * @brief Get the final results if merging was enabled
+     */
     std::vector<uint8_t *>& get_results();
 
-	/**
-	 * @brief Get the number of processed flows (for progress reporting)
-	 */
+    /**
+     * @brief Get the number of processed flows (for progress reporting)
+     */
     uint64_t get_processed_flows() const;
 
-	/**
-	 * @brief Get the number of processed files (for progress reporting)
-	 */
+    /**
+     * @brief Get the number of processed files (for progress reporting)
+     */
     uint64_t get_processed_files() const;
 
-	/**
-	 * @brief Get the total number of flows that will be processed (for progress reporting)
-	 */
+    /**
+     * @brief Get the total number of flows that will be processed (for progress reporting)
+     */
     uint64_t get_total_flows() const;
 
-	/**
-	 * @brief Get the total number of files that will be processed (for progress reporting)
-	 */
+    /**
+     * @brief Get the total number of files that will be processed (for progress reporting)
+     */
     uint64_t get_total_files() const;
 
-	/**
-	 * @brief Get the current aggregator state
-	 */
+    /**
+     * @brief Get the current aggregator state
+     */
     AggregatorState get_aggregator_state() const;
 
-	/**
-	 * @brief Get the per-thread tables if merging was disabled
-	 */
+    /**
+     * @brief Get the per-thread tables if merging was disabled
+     */
     std::vector<HashTable *> get_tables();
 
-	/**
-	 * @brief Cancel the aggregation
-	 */
+    /**
+     * @brief Cancel the aggregation
+     */
     void cancel();
 
-	/**
-	 * @brief Block until the execution of all aggregator threads finishes
-	 */
+    /**
+     * @brief Block until the execution of all aggregator threads finishes
+     */
     void join();
 
 private:
@@ -144,7 +144,7 @@ class ThreadedAggregator {
     std::string m_order_by;
     bool m_biflow_autoignore;
     bool m_merge_results;
-	unsigned int m_merge_topk;
+    unsigned int m_merge_topk;
 
     std::thread m_main_thread;
 
@@ -167,15 +167,15 @@ class ThreadedAggregator {
 
     std::exception_ptr m_exception = nullptr;
 
-	std::unique_ptr<ThresholdAlgorithm> m_threshold_algorithm;
+    std::unique_ptr<ThresholdAlgorithm> m_threshold_algorithm;
 
     void thread_worker(unsigned int thread_id);
 
     void run();
 
     void perform_all_merge();
 
-	void perform_topk_merge();
+    void perform_topk_merge();
 };
 
 } // aggregator