Simple mechanism for collecting statistics during execution

This adds a simple class that can be used to (optionally) collect statistics
about a program execution. This includes information bout the program (e.g.
number of reactors, reactions, posts, etc) and about the execution (e.g. number
of processed events and reactions as well as the number of actions triggered).

CMakeLists.txt

@@ -40,6 +40,7 @@ if(NOT CMAKE_BUILD_TYPE AND NOT CMAKE_CONFIGURATION_TYPES)
   set_property(CACHE CMAKE_BUILD_TYPE PROPERTY STRINGS "Debug" "Release" "MinSizeRel" "RelWithDebInfo")
 endif()
 
+option(REACTOR_CPP_PRINT_STATISTICS "Print statistics after execution" OFF)
 option(REACTOR_CPP_TRACE "Enable tracing" OFF)
 option(REACTOR_CPP_VALIDATE "Enable runtime validation" ON)
 if (NOT DEFINED REACTOR_CPP_LOG_LEVEL)

include/reactor-cpp/config.hh.in

@@ -1,6 +1,8 @@
 #ifndef REACTOR_CPP_CONFIG_HH
 #define REACTOR_CPP_CONFIG_HH
 
+// NOLINTNEXTLINE
+#cmakedefine REACTOR_CPP_PRINT_STATISTICS
 // NOLINTNEXTLINE
 #cmakedefine REACTOR_CPP_TRACE
 // NOLINTNEXTLINE

include/reactor-cpp/logging.hh

@@ -9,7 +9,7 @@
 #ifndef REACTOR_CPP_LOGGING_HH
 #define REACTOR_CPP_LOGGING_HH
 
-#include "reactor-cpp/config.hh" //NOLINT
+#include "reactor-cpp/config.hh"
 #include "reactor-cpp/time.hh"
 #include <chrono>
 #include <iostream>

include/reactor-cpp/scheduler.hh

@@ -165,6 +165,8 @@ private:
   void terminate_all_workers();
   void set_port_helper(BasePort* port);
 
+  void advance_logical_time_to(const Tag& tag);
+
 public:
   explicit Scheduler(Environment* env);
   ~Scheduler();

include/reactor-cpp/statistics.hh

@@ -0,0 +1,89 @@
+/*
+ * Copyright (C) 2023 TU Dresden
+ * All rights reserved.
+ *
+ * Authors:
+ *   Christian Menard
+ */
+
+#ifndef REACTOR_CPP_STATISTICS_HH
+#define REACTOR_CPP_STATISTICS_HH
+
+#include <atomic>
+
+#include "reactor-cpp/config.hh"
+#include "reactor-cpp/logging.hh"
+
+namespace reactor {
+
+class Statistics {
+private:
+#ifdef REACTOR_CPP_PRINT_STATISTICS
+  constexpr static bool enabled_{true};
+#else
+  constexpr static bool enabled_{false};
+#endif
+
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+  inline static std::atomic_size_t reactor_instances_{0};
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+  inline static std::atomic_size_t connections_{0};
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+  inline static std::atomic_size_t reactions_{0};
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+  inline static std::atomic_size_t actions_{0};
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+  inline static std::atomic_size_t ports_{0};
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+  inline static std::atomic_size_t processed_events_{0};
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+  inline static std::atomic_size_t processed_reactions_{0};
+  // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+  inline static std::atomic_size_t triggered_actions_{0};
+
+  inline static void increment(std::atomic_size_t& counter) {
+    if constexpr (enabled_) {
+      counter.fetch_add(1, std::memory_order_release);
+    }
+  }
+
+public:
+  inline static void increment_reactor_instances() { increment(reactor_instances_); }
+  inline static void increment_connections() { increment(connections_); }
+  inline static void increment_reactions() { increment(reactions_); }
+  inline static void increment_actions() { increment(actions_); }
+  inline static void increment_ports() { increment(ports_); }
+  inline static void increment_processed_events() { increment(processed_events_); }
+  inline static void increment_processed_reactions() { increment(processed_reactions_); }
+  inline static void increment_triggered_actions() { increment(triggered_actions_); }
+
+  inline static auto reactor_instances() { return reactor_instances_.load(std::memory_order_acquire); }
+  inline static auto connections() { return connections_.load(std::memory_order_acquire); }
+  inline static auto reactions() { return reactions_.load(std::memory_order_acquire); }
+  inline static auto actions() { return actions_.load(std::memory_order_acquire); }
+  inline static auto ports() { return ports_.load(std::memory_order_acquire); }
+  inline static auto processed_events() { return processed_events_.load(std::memory_order_acquire); }
+  inline static auto processed_reactions() { return processed_reactions_.load(std::memory_order_acquire); }
+  inline static auto triggered_actions() { return triggered_actions_.load(std::memory_order_acquire); }
+
+  inline static void print() {
+    if constexpr (enabled_) {
+      reactor::log::Info() << "-----------------------------------------------------------";
+      reactor::log::Info() << "Program statistics:";
+      reactor::log::Info() << "  - number of reactors:    " << reactor_instances();
+      reactor::log::Info() << "  - number of connections: " << connections();
+      reactor::log::Info() << "  - number of reactions    " << reactions();
+      reactor::log::Info() << "  - number of actions:     " << actions();
+      reactor::log::Info() << "  - number of ports:       " << ports();
+      reactor::log::Info() << "Execution statistics:";
+      reactor::log::Info() << "  - processed events:      " << processed_events();
+      reactor::log::Info() << "  - triggered actions:     " << triggered_actions();
+      reactor::log::Info() << "  - processed reactions:   " << processed_reactions();
+      reactor::log::Info() << "-----------------------------------------------------------";
+    }
+  }
+};
+
+} // namespace reactor
+
+#endif // REACTOR_CPP_STATISTICS_HH

lib/environment.cc

@@ -19,6 +19,7 @@
 #include "reactor-cpp/logging.hh"
 #include "reactor-cpp/port.hh"
 #include "reactor-cpp/reaction.hh"
+#include "reactor-cpp/statistics.hh"
 #include "reactor-cpp/time.hh"
 
 namespace reactor {
@@ -285,6 +286,11 @@ auto Environment::startup(const TimePoint& start_time) -> std::thread {
     for (auto& thread : threads) {
       thread.join();
     }
+
+    // If this is the top level environment, then print some execution statistics
+    if (this->containing_environment_ == nullptr) {
+      Statistics::print();
+    }
   });
 }
 

lib/port.cc

@@ -11,6 +11,7 @@
 #include "reactor-cpp/assert.hh"
 #include "reactor-cpp/environment.hh"
 #include "reactor-cpp/reaction.hh"
+#include "reactor-cpp/statistics.hh"
 
 namespace reactor {
 
@@ -42,6 +43,8 @@ void BasePort::base_bind_to(BasePort* port) {
 
   port->inward_binding_ = this;
   reactor_assert(this->outward_bindings_.insert(port).second);
+
+  Statistics::increment_connections();
 }
 
 void BasePort::register_dependency(Reaction* reaction, bool is_trigger) noexcept {

lib/reactor.cc

@@ -14,6 +14,7 @@
 #include "reactor-cpp/logging.hh"
 #include "reactor-cpp/port.hh"
 #include "reactor-cpp/reaction.hh"
+#include "reactor-cpp/statistics.hh"
 
 namespace reactor {
 
@@ -66,6 +67,17 @@ ReactorElement::ReactorElement(const std::string& name, ReactorElement::Type typ
   validate(type == Type::Reactor || type == Type::Action, "Only reactors and actions can be owned by the environment!");
   validate(this->environment_->phase() == Environment::Phase::Construction,
            "Reactor elements can only be created during construction phase!");
+
+  switch (type) {
+  case Type::Action:
+    Statistics::increment_actions();
+    break;
+  case Type::Reactor:
+    Statistics::increment_reactor_instances();
+    break;
+  default:
+    break;
+  }
 }
 
 Reactor::Reactor(const std::string& name, Reactor* container)
@@ -81,20 +93,23 @@ void Reactor::register_action([[maybe_unused]] BaseAction* action) {
                         this->environment()->phase() == Environment::Phase::Assembly,
                     "Actions can only be registered during construction phase!");
   reactor_assert(actions_.insert(action).second);
+  Statistics::increment_actions();
 }
 
 void Reactor::register_input(BasePort* port) {
   reactor_assert(port != nullptr);
   reactor::validate(this->environment()->phase() == Environment::Phase::Construction,
                     "Ports can only be registered during construction phase!");
   reactor_assert(inputs_.insert(port).second);
+  Statistics::increment_ports();
 }
 
 void Reactor::register_output(BasePort* port) {
   reactor_assert(port != nullptr);
   reactor::validate(this->environment()->phase() == Environment::Phase::Construction,
                     "Ports can only be registered during construction phase!");
   reactor_assert(outputs_.insert(port).second);
+  Statistics::increment_ports();
 }
 
 void Reactor::register_reaction([[maybe_unused]] Reaction* reaction) {
@@ -103,13 +118,15 @@ void Reactor::register_reaction([[maybe_unused]] Reaction* reaction) {
   validate(this->environment()->phase() == Environment::Phase::Construction,
            "Reactions can only be registered during construction phase!");
   reactor_assert(reactions_.insert(reaction).second);
+  Statistics::increment_reactions();
 }
 
 void Reactor::register_reactor([[maybe_unused]] Reactor* reactor) {
   reactor_assert(reactor != nullptr);
   validate(this->environment()->phase() == Environment::Phase::Construction,
            "Reactions can only be registered during construction phase!");
   reactor_assert(reactors_.insert(reactor).second);
+  Statistics::increment_reactor_instances();
 }
 
 void Reactor::startup() {

lib/scheduler.cc

@@ -19,6 +19,7 @@
 #include "reactor-cpp/logging.hh"
 #include "reactor-cpp/port.hh"
 #include "reactor-cpp/reaction.hh"
+#include "reactor-cpp/statistics.hh"
 #include "reactor-cpp/time_barrier.hh"
 #include "reactor-cpp/trace.hh"
 
@@ -82,6 +83,8 @@ void Worker::execute_reaction(Reaction* reaction) const {
   tracepoint(reactor_cpp, reaction_execution_starts, identity_, reaction->fqn(), scheduler_.logical_time());
   reaction->trigger();
   tracepoint(reactor_cpp, reaction_execution_finishes, identity_, reaction->fqn(), scheduler_.logical_time());
+
+  Statistics::increment_processed_reactions();
 }
 
 void Scheduler::schedule() noexcept {
@@ -297,6 +300,12 @@ void Scheduler::start() {
   }
 }
 
+void Scheduler::advance_logical_time_to(const Tag& tag) {
+  log_.debug() << "advance logical time to tag " << tag;
+  logical_time_.advance_to(tag);
+  Statistics::increment_processed_events();
+}
+
 void Scheduler::next() { // NOLINT
   // Notify other environments and let them know that we finished processing the
   // current tag
@@ -348,8 +357,7 @@ void Scheduler::next() { // NOLINT
           log_.debug() << "Schedule the last round of reactions including all "
                           "termination reactions";
           triggered_actions_ = event_queue_.extract_next_event();
-          log_.debug() << "advance logical time to tag " << t_next;
-          logical_time_.advance_to(t_next);
+          advance_logical_time_to(t_next);
         } else {
           return;
         }
@@ -390,9 +398,7 @@ void Scheduler::next() { // NOLINT
         // queue.
         triggered_actions_ = event_queue_.extract_next_event();
 
-        // advance logical time
-        log_.debug() << "advance logical time to tag " << t_next;
-        logical_time_.advance_to(t_next);
+        advance_logical_time_to(t_next);
 
         // If there are no triggered actions at the event, then release the
         // current tag and go back to the start of the loop
@@ -407,6 +413,7 @@ void Scheduler::next() { // NOLINT
   log_.debug() << "events: " << triggered_actions_->size();
   for (auto* action : *triggered_actions_) {
     log_.debug() << "Action " << action->fqn();
+    Statistics::increment_triggered_actions();
     action->setup();
     for (auto* reaction : action->triggers()) {
       // There is no need to acquire the mutex. At this point the scheduler