Feature/counter (#732)

* add counter no tests

* some refactoring

* add tests + fix docu for counter

* counter tests

* rename test function

* fix format

* change aggregation default for counter

* fix doxygen

Author: mschimek

include/kamping/measurements/aggregated_tree_node.hpp

@@ -19,19 +19,18 @@
 #include "kamping/measurements/measurement_aggregation_definitions.hpp"
 
 namespace kamping::measurements {
-/// @brief Class representing a node in the timer tree. Each node represents a time measurement (or multiple with
-/// the
-///  same key). A node can have multiple children which represent nested time measurements. The measurements
-///  associated with a node's children are executed while the node's measurement is still active.
+
+/// @brief Class representing a node in an (globally) aggregated tree, i.e., a node of a timer (or counter) tree
+/// where the global aggregation operations has been performed and which can be printed.
 ///
-/// @tparam Duration  Type of a duration.
-template <typename Duration>
-class AggregatedTreeNode : public internal::TreeNode<AggregatedTreeNode<Duration>> {
+/// @tparam DataType  Underlying data type.
+template <typename DataType>
+class AggregatedTreeNode : public internal::TreeNode<AggregatedTreeNode<DataType>> {
 public:
-    using internal::TreeNode<AggregatedTreeNode<Duration>>::TreeNode;
+    using internal::TreeNode<AggregatedTreeNode<DataType>>::TreeNode;
 
     ///@brief Type into which the aggregated data is stored together with the applied aggregation operation.
-    using StorageType = std::unordered_map<GlobalAggregationMode, std::vector<ScalarOrContainer<Duration>>>;
+    using StorageType = std::unordered_map<GlobalAggregationMode, std::vector<ScalarOrContainer<DataType>>>;
 
     /// @brief Access to stored aggregated data.
     /// @return Reference to aggregated data.
@@ -41,9 +40,9 @@ class AggregatedTreeNode : public internal::TreeNode<AggregatedTreeNode<Duration
 
     /// @brief Add scalar of type T to aggregated data storage together with the name of the  applied aggregation
     /// operation.
-    /// @param aggregation_mode Aggregation mode that has been applied to the duration data.
+    /// @param aggregation_mode Aggregation mode that has been applied to the data.
     /// @param data Scalar resulted from applying the given aggregation operation.
-    void add(GlobalAggregationMode aggregation_mode, std::optional<Duration> data) {
+    void add(GlobalAggregationMode aggregation_mode, std::optional<DataType> data) {
         if (data) {
             _aggregated_data[aggregation_mode].emplace_back(data.value());
         }
@@ -53,11 +52,128 @@ class AggregatedTreeNode : public internal::TreeNode<AggregatedTreeNode<Duration
     /// operation.
     /// @param aggregation_mode Aggregation mode that has been applied to the duration data.
     /// @param data Vector of Scalars resulted from applying the given aggregation operation.
-    void add(GlobalAggregationMode aggregation_mode, std::vector<Duration> const& data) {
+    void add(GlobalAggregationMode aggregation_mode, std::vector<DataType> const& data) {
         _aggregated_data[aggregation_mode].emplace_back(data);
     }
 
 public:
     StorageType _aggregated_data; ///< Storage of the aggregated data.
 };
+
+/// @brief Class representing an aggregated measurement tree, i.e., a measurement tree for which the global aggregation
+/// has been performed.
+///
+/// @tparam DataType Type of interanlly stored data.
+template <typename DataType>
+class AggregatedTree {
+public:
+    /// @brief Globally aggregates the measurement tree provided with \param measurement_root_node across all ranks in
+    /// \param comm .
+    ///
+    /// @tparam MeasurementNode Type of the measurement tree to aggregate.
+    /// @tparam Communicator Communicator defining the scope for the global aggregation.
+    template <typename MeasurementNode, typename Communicator>
+    AggregatedTree(MeasurementNode const& measurement_root_node, Communicator const& comm) : _root{"root"} {
+        aggregate(_root, measurement_root_node, comm);
+    }
+
+    /// @brief Access to the root of the aggregated tree.
+    /// @return Reference to root node of aggregated tree.
+    auto& root() {
+        return _root;
+    }
+
+    /// @brief Access to the root of the aggregated tree.
+    /// @return Reference to root node of aggregated tree.
+    auto const& root() const {
+        return _root;
+    }
+
+private:
+    AggregatedTreeNode<DataType> _root; ///< Root node of aggregated tree.
+    /// @brief Traverses and evaluates the given (Measurement)TreeNode and stores the result in the corresponding
+    /// AggregatedTreeNode
+    ///
+    /// param aggregation_tree_node Node where the aggregated data points are stored.
+    /// param measurement_tree_node Node where the raw (not aggregated) data points are stored.
+    template <typename MeasurementNode, typename Communciator>
+    void aggregate(
+        AggregatedTreeNode<DataType>& aggregation_tree_node,
+        MeasurementNode&              measurement_tree_node,
+        Communciator const&           comm
+    ) {
+        KASSERT(
+            internal::is_string_same_on_all_ranks(measurement_tree_node.name(), comm),
+            "Currently processed MeasurementTreeNode has not the same name on all ranks -> measurement trees have "
+            "diverged",
+            assert::heavy_communication
+        );
+        KASSERT(
+            comm.is_same_on_all_ranks(measurement_tree_node.measurements().size()),
+            "Currently processed MeasurementTreeNode has not the same number of measurements on all ranks -> "
+            "measurement trees have "
+            "diverged",
+            assert::light_communication
+        );
+
+        // gather all durations at once as gathering all durations individually may deteriorate
+        // the performance of the evaluation operation significantly.
+        auto       recv_buf      = comm.gatherv(send_buf(measurement_tree_node.measurements()));
+        auto const num_durations = measurement_tree_node.measurements().size();
+        for (size_t duration_idx = 0; duration_idx < num_durations; ++duration_idx) {
+            if (!comm.is_root()) {
+                continue;
+            }
+            std::vector<DataType> cur_durations;
+            cur_durations.reserve(comm.size());
+            // gather the durations belonging to the same measurement
+            for (size_t rank = 0; rank < comm.size(); ++rank) {
+                cur_durations.push_back(recv_buf[duration_idx + rank * num_durations]);
+            }
+
+            for (auto const& aggregation_mode: measurement_tree_node.measurements_aggregation_operations()) {
+                aggregate_measurements_globally(aggregation_mode, cur_durations, aggregation_tree_node);
+            }
+        }
+        for (auto& measurement_tree_child: measurement_tree_node.children()) {
+            auto& aggregation_tree_child = aggregation_tree_node.find_or_insert(measurement_tree_child->name());
+            aggregate(aggregation_tree_child, *measurement_tree_child.get(), comm);
+        }
+    }
+
+    /// @brief Computes the specified aggregation operation on an already gathered range of values.
+    ///
+    /// @param mode Aggregation operation to perform.
+    /// @param gathered_data Durations gathered from all participating ranks.
+    /// @param evaluation_node Object where the aggregated and evaluated measurements are stored.
+    void aggregate_measurements_globally(
+        GlobalAggregationMode                                mode,
+        std::vector<DataType> const&                         gathered_data,
+        kamping::measurements::AggregatedTreeNode<DataType>& evaluation_node
+    ) {
+        switch (mode) {
+            case GlobalAggregationMode::max: {
+                using Operation = internal::Max;
+                evaluation_node.add(mode, Operation::compute(gathered_data));
+                break;
+            }
+            case GlobalAggregationMode::min: {
+                using Operation = internal::Min;
+                evaluation_node.add(mode, Operation::compute(gathered_data));
+                break;
+            }
+            case GlobalAggregationMode::sum: {
+                using Operation = internal::Sum;
+                evaluation_node.add(mode, Operation::compute(gathered_data));
+                break;
+            }
+            case GlobalAggregationMode::gather: {
+                using Operation = internal::Gather;
+                evaluation_node.add(mode, Operation::compute(gathered_data));
+                break;
+            }
+        }
+    }
+};
+
 } // namespace kamping::measurements

include/kamping/measurements/counter.hpp

@@ -0,0 +1,147 @@
+// This file is part of KaMPIng.
+//
+// Copyright 2023 The KaMPIng Authors
+//
+// KaMPIng is free software : you can redistribute it and/or modify it under the
+// terms of the GNU Lesser General Public License as published by the Free
+// Software Foundation, either version 3 of the License, or (at your option) any
+// later version. KaMPIng is distributed in the hope that it will be useful, but
+// WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+// FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License
+// for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with KaMPIng.  If not, see <https://www.gnu.org/licenses/>.
+
+/// @file
+/// This file contains a (distributed) counter class.
+
+#pragma once
+
+#include <kassert/kassert.hpp>
+#include <mpi.h>
+
+#include "kamping/collectives/barrier.hpp"
+#include "kamping/collectives/gather.hpp"
+#include "kamping/communicator.hpp"
+#include "kamping/environment.hpp"
+#include "kamping/measurements/aggregated_tree_node.hpp"
+#include "kamping/measurements/internal/measurement_utils.hpp"
+
+namespace kamping::measurements {
+
+/// @brief Distributed counter object.
+/// @tparam CommunicatorType Communicator in which the measurements are
+/// executed.
+template <typename CommunicatorType = Communicator<>>
+class Counter {
+public:
+    using DataType = std::int64_t; ///< Data type of the stored measurements.
+    /// @brief Constructs a timer using the \c MPI_COMM_WORLD communicator.
+    Counter() : _tree{}, _comm{comm_world()} {}
+
+    /// @brief Constructs a timer using a given communicator.
+    ///
+    /// @param comm Communicator in which the time measurements are executed.
+    Counter(CommunicatorType const& comm) : _tree{}, _comm{comm} {}
+
+    /// @brief Creates a measurement entry with name \param name and stores \param data therein. If such an entry
+    /// already exists with associated data entry `data_prev`, \c data will be added to it, i.e. `data_prev +
+    /// data`.
+    /// @param global_aggregation_modi Specify how the measurement entry is aggregated over all participationg PEs when
+    /// Counter::aggregate() is called.
+    void
+    add(std::string const&                        name,
+        DataType const&                           data,
+        std::vector<GlobalAggregationMode> const& global_aggregation_modi = std::vector<GlobalAggregationMode>{}) {
+        add_measurement(name, data, LocalAggregationMode::accumulate, global_aggregation_modi);
+    }
+
+    /// @brief Looks for a measurement entry with name \param name and appends \param data to the list of previously
+    /// stored data. If no such entry exists, a new measurement entry with \c data as first entry will be created. entry
+    /// `data_prev`, \c data will be added to it, i.e. `data_prev + data`.
+    /// @param global_aggregation_modi Specify how the measurement entry is aggregated over all participationg PEs when
+    /// Counter::aggregate() is called.
+    void append(
+        std::string const&                        name,
+        DataType const&                           data,
+        std::vector<GlobalAggregationMode> const& global_aggregation_modi = std::vector<GlobalAggregationMode>{}
+    ) {
+        add_measurement(name, data, LocalAggregationMode::append, global_aggregation_modi);
+    }
+
+    /// @brief Aggregate the measurement entries globally.
+    /// @return AggregatedTree object which encapsulates the aggregated data in a tree structure representing the
+    /// measurements.
+    auto aggregate() {
+        AggregatedTree<DataType> aggregated_tree(_tree.root, _comm);
+        return aggregated_tree;
+    }
+
+    /// @brief Clears all stored measurements.
+    void clear() {
+        _tree.reset();
+    }
+
+    /// @brief Aggregates and outputs the the executed measurements. The output is
+    /// done via the print() method of a given Printer object.
+    ///
+    /// The print() method must accept an object of type AggregatedTreeNode and
+    /// receives the root of the evaluated timer tree as parameter. The print()
+    /// method is only called on the root rank of the communicator. See
+    /// EvaluationTreeNode for the accessible data. The
+    /// EvaluationTreeNode::children() member function can be used to navigate the
+    /// nested measurement structure.
+    ///
+    /// @tparam Printer Type of printer which is used to output the aggregated
+    /// timing data. Printer must possess a member print() which accepts a
+    /// EvaluationTreeNode as parameter.
+    /// @param printer Printer object used to output the aggregated timing data.
+    template <typename Printer>
+    void aggregate_and_print(Printer&& printer) {
+        auto const aggregated_tree = aggregate();
+        if (_comm.is_root()) {
+            printer.print(aggregated_tree.root());
+        }
+    }
+
+private:
+    internal::Tree<internal::CounterTreeNode<DataType>>
+        _tree; ///< Tree structure in which the counted values are stored. Note that unlike for Timer, the tree is
+               ///< always a star as there is currently no functionality to allow for "nested" counting, e.g. by
+               ///< defining different phase within your algorithm.
+    CommunicatorType const& _comm; ///< Communicator in which the time measurements take place.
+
+    /// @brief Adds a new measurement to the tree
+    /// @param local_aggregation_mode Specifies how the measurement duration is
+    /// locally aggregated when there are multiple measurements at the same level
+    /// with identical key.
+    /// @param global_aggregation_modi Specifies how the measurement data is
+    /// aggregated over all participationg ranks when Timer::aggregate() is
+    /// called.
+    void add_measurement(
+        std::string const&                        name,
+        DataType const&                           data,
+        LocalAggregationMode                      local_aggregation_mode,
+        std::vector<GlobalAggregationMode> const& global_aggreation_modi
+    ) {
+        auto& child = _tree.current_node->find_or_insert(name);
+        child.aggregate_measurements_locally(data, local_aggregation_mode);
+        if (!global_aggreation_modi.empty()) {
+            child.measurements_aggregation_operations() = global_aggreation_modi;
+        }
+    }
+};
+
+/// @brief A basic Counter that uses kamping::Communicator<> as underlying
+/// communicator type.
+using BasicCounter = Counter<Communicator<>>;
+
+/// @brief Gets a reference to a kamping::measurements::BasicTimer.
+///
+/// @return A reference to a kamping::measurements::BasicCounter.
+inline Counter<Communicator<>>& counter() {
+    static Counter<Communicator<>> counter;
+    return counter;
+}
+} // namespace kamping::measurements