Scheduler Optimizations

.github/workflows/cpp-tests.yml

@@ -29,6 +29,11 @@ jobs:
           repository: lf-lang/lingua-franca
           submodules: true
           ref: ${{ inputs.compiler-ref }}
+      - name: Install boost on Ubuntu
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y boost boost-dev
+        if: matrix.platform == 'ubuntu-latest'
       - name: Set cpp runtime version
         run: |
             echo ${{ inputs.runtime-ref }} > org.lflang/src/org/lflang/generator/cpp/cpp-runtime-version.txt

.github/workflows/main.yaml

@@ -14,7 +14,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-latest, windows-latest, macos-latest]
+        os: [ubuntu-latest]
 
     steps:
     - uses: actions/checkout@v1
@@ -23,6 +23,11 @@ jobs:
         sudo apt-get update
         sudo apt-get install -y clang-tidy
       if: matrix.os == 'ubuntu-latest'
+    - name: Install boost on Ubuntu
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y libboost-all-dev
+      if: matrix.os == 'ubuntu-latest'
     - name: configure
       run: |
         mkdir build
@@ -35,7 +40,7 @@ jobs:
         cmake --build build --target examples
 
   lf-tests-pull-request:
-    uses: lf-lang/lingua-franca/.github/workflows/cpp-tests.yml@master
+    uses: lf-lang/lingua-franca/.github/workflows/cpp-tests.yml@cpp-boost
     with:
       runtime-ref: ${{github.ref}}
       compiler-ref: master

CMakeLists.txt

@@ -26,6 +26,7 @@ if (NOT DEFINED LF_REACTOR_CPP_SUFFIX)
 endif()
 
 find_package (Threads)
+find_package (Boost 1.71 COMPONENTS graph REQUIRED)
 
 set(DEFAULT_BUILD_TYPE "Release")
 if(NOT CMAKE_BUILD_TYPE AND NOT CMAKE_CONFIGURATION_TYPES)

include/reactor-cpp/action.hh

@@ -44,7 +44,7 @@ public:
   [[nodiscard]] auto inline min_delay() const noexcept -> Duration { return min_delay_; }
 
   friend class Reaction;
-  friend class Scheduler;
+  template<class SchedulingPolicy> friend class Scheduler;
 };
 
 template <class T> class Action : public BaseAction {
@@ -154,6 +154,6 @@ public:
 
 } // namespace reactor
 
-#include "impl/action_impl.hh"
+#include "reactor-cpp/impl/action_impl.hh"
 
 #endif // REACTOR_CPP_ACTION_HH

include/reactor-cpp/assert.hh

@@ -24,6 +24,7 @@ constexpr bool runtime_assertion = true;
 #include "environment.hh"
 
 #include <cassert>
+#include <map>
 #include <sstream>
 #include <stdexcept>
 #include <string>

include/reactor-cpp/base_scheduler.hh

@@ -0,0 +1,60 @@
+/*
+ * Copyright (C) 2022 TU Dresden
+ * All rights reserved.
+ *
+ * Authors:
+ *   Christian Menard
+ */
+
+#ifndef REACTOR_CPP_BASE_SCHEDULER_HH
+#define REACTOR_CPP_BASE_SCHEDULER_HH
+
+#include <atomic>
+#include <condition_variable>
+#include <functional>
+#include <map>
+#include <mutex>
+
+#include "reactor-cpp/environment.hh"
+#include "reactor-cpp/fwd.hh"
+#include "reactor-cpp/logical_time.hh"
+
+namespace reactor {
+
+using EventMap = std::map<BaseAction*, std::function<void(void)>>;
+
+class BaseScheduler {
+protected:
+  Environment* environment_; // NOLINT
+
+  LogicalTime logical_time_{}; // NOLINT
+  const bool using_workers_;   // NOLINT
+
+  std::mutex scheduling_mutex_;                                                      // NOLINT
+  std::unique_lock<std::mutex> scheduling_lock_{scheduling_mutex_, std::defer_lock}; // NOLINT
+  std::condition_variable cv_schedule_;                                              // NOLINT
+
+  std::mutex lock_event_queue_;         // NOLINT
+  std::map<Tag, EventMap> event_queue_; // NOLINT
+
+public:
+  BaseScheduler(Environment* env);
+  virtual ~BaseScheduler() = default;
+  BaseScheduler(const BaseScheduler&) = delete;
+  BaseScheduler(BaseScheduler&&) = delete;
+  auto operator=(const BaseScheduler&) -> BaseScheduler& = delete;
+  auto operator=(BaseScheduler&&) -> BaseScheduler& = delete;
+
+  [[nodiscard]] inline auto logical_time() const noexcept -> const auto& { return logical_time_; }
+  void schedule_sync(const Tag& tag, BaseAction* action, std::function<void(void)> pre_handler);
+  void schedule_async(const Tag& tag, BaseAction* action, std::function<void(void)> pre_handler);
+
+  virtual void set_port(BasePort* port) = 0;
+
+  void inline lock() noexcept { scheduling_lock_.lock(); }
+  void inline unlock() noexcept { scheduling_lock_.unlock(); }
+};
+
+} // namespace reactor
+
+#endif // REACTOR_CPP_BASE_SCHEDULER_HH

include/reactor-cpp/default_scheduling_policy.hh

@@ -0,0 +1,86 @@
+/*
+ * Copyright (C) 2022 TU Dresden
+ * All rights reserved.
+ *
+ * Authors:
+ *   Christian Menard
+ */
+
+#ifndef REACTOR_CPP_DEFAULT_SCHEDULING_POLICY_HH
+#define REACTOR_CPP_DEFAULT_SCHEDULING_POLICY_HH
+
+#include <cstddef>
+#include <vector>
+
+#include "reactor-cpp/fwd.hh"
+#include "reactor-cpp/reaction.hh"
+#include "reactor-cpp/semaphore.hh"
+
+namespace reactor {
+
+class DefaultSchedulingPolicy {
+  Scheduler<DefaultSchedulingPolicy>& scheduler_;
+  Environment& environment_;
+  std::size_t identity_counter{0};
+
+  class ReadyQueue {
+  private:
+    std::vector<Reaction*> queue_{};
+    std::atomic<std::ptrdiff_t> size_{0};
+    Semaphore sem_{0};
+    std::ptrdiff_t waiting_workers_{0};
+    const unsigned int num_workers_;
+
+  public:
+    explicit ReadyQueue(unsigned num_workers)
+        : num_workers_(num_workers) {}
+
+    /**
+     * Retrieve a ready reaction from the queue.
+     *
+     * This method may be called concurrently. In case the queue is empty, the
+     * method blocks and waits until a ready reaction becomes available.
+     */
+    auto pop() -> Reaction*;
+
+    /**
+     * Fill the queue up with ready reactions.
+     *
+     * This method assumes that the internal queue is empty. It moves all
+     * reactions from the provided `ready_reactions` vector to the internal
+     * queue, leaving `ready_reactions` empty.
+     *
+     * Note that this method is not thread-safe. The caller needs to ensure that
+     * no other thread will try to read from the queue during this operation.
+     */
+    void fill_up(std::vector<Reaction*>& ready_reactions);
+  };
+
+  ReadyQueue ready_queue_;
+
+  std::vector<std::vector<Reaction*>> reaction_queue_;
+  unsigned int reaction_queue_pos_{std::numeric_limits<unsigned>::max()};
+
+  std::atomic<std::ptrdiff_t> reactions_to_process_{0};
+  std::vector<std::vector<Reaction*>> triggered_reactions_;
+
+  bool continue_execution_{true};
+
+  void schedule() noexcept;
+  void terminate_all_workers();
+  auto schedule_ready_reactions() -> bool;
+
+public:
+  DefaultSchedulingPolicy(Scheduler<DefaultSchedulingPolicy>& scheduler, Environment& env);
+
+  void init();
+  auto create_worker() -> Worker<DefaultSchedulingPolicy>;
+  void worker_function(const Worker<DefaultSchedulingPolicy>& worker);
+
+  void trigger_reaction_from_next(Reaction* reaction);
+  void trigger_reaction_from_set_port(Reaction* reaction);
+};
+
+} // namespace reactor
+
+#endif // REACTOR_CPP_DEFAULT_SCHEDULING_POLICY_HH

include/reactor-cpp/dependency_graph.hh

@@ -0,0 +1,139 @@
+#ifndef REACTOR_CPP_DEPENDENCY_GRAPH_HH
+#define REACTOR_CPP_DEPENDENCY_GRAPH_HH
+
+#include "reactor-cpp/reaction.hh"
+#include "reactor-cpp/reactor.hh"
+
+#include <boost/graph/breadth_first_search.hpp>
+#include <boost/graph/directed_graph.hpp>
+#include <boost/graph/graph_selectors.hpp>
+#include <boost/graph/properties.hpp>
+#include <boost/pending/property.hpp>
+#include <vector>
+
+namespace reactor {
+
+class GroupedDependencyGraph;
+
+enum class DependencyType { Undefined, Priority, Trigger, Effect };
+
+class ReactionDependencyGraph {
+private:
+  struct reaction_info_t {
+    using kind = boost::vertex_property_tag;
+  };
+  struct dependency_info_t {
+    using kind = boost::edge_property_tag;
+  };
+  using ReactionProperty = boost::property<reaction_info_t, Reaction*>;
+  using DependencyProperty = boost::property<dependency_info_t, DependencyType>;
+  using ReactionGraph = boost::directed_graph<ReactionProperty, DependencyProperty>;
+  using ReactionToVertexMap = std::map<Reaction*, ReactionGraph::vertex_descriptor>;
+  using ReactionPropertyMap = boost::property_map<ReactionGraph, reaction_info_t>::type;
+  using DependencyPropertyMap = boost::property_map<ReactionGraph, dependency_info_t>::type;
+
+  ReactionGraph graph{};
+  ReactionToVertexMap vertex_map{};
+
+  // helper functions
+  void populate_graph_with_reactions(const Reactor* reactor);
+  void populate_graph_with_priority_edges(const Reactor* reactor);
+  void populate_graph_with_dependency_edges(const Reactor* reactor);
+
+  [[nodiscard]] auto get_reaction_property_map() -> ReactionPropertyMap { return boost::get(reaction_info_t{}, graph); }
+  [[nodiscard]] auto get_dependency_property_map() -> DependencyPropertyMap {
+    return boost::get(dependency_info_t{}, graph);
+  }
+
+  ReactionDependencyGraph() = default;
+
+public:
+  ReactionDependencyGraph(const std::set<Reactor*>& top_level_reactors);
+
+  void export_graphviz(const std::string& file_name);
+
+  // TODO: This should be const, but I don't know how to get immutable access to the reaction graph properties...
+  [[nodiscard]] auto transitive_reduction() -> ReactionDependencyGraph;
+
+  friend GroupedDependencyGraph;
+};
+
+class GroupedDependencyGraph {
+public:
+  struct group_info_t {
+    using kind = boost::vertex_property_tag;
+  };
+  using Group = std::vector<Reaction*>;
+  using GroupProperty = boost::property<group_info_t, Group>;
+  using GroupGraph = boost::directed_graph<GroupProperty>;
+  using ReactionToVertexMap = std::map<Reaction*, GroupGraph::vertex_descriptor>;
+  using GroupPropertyMap = boost::property_map<GroupGraph, group_info_t>::type;
+
+private:
+  GroupGraph graph{};
+  ReactionToVertexMap vertex_map{};
+
+  struct ReachabilityVisitor : public boost::default_bfs_visitor {
+  private:
+    GroupGraph::vertex_descriptor to;
+
+  public:
+    ReachabilityVisitor(GroupGraph::vertex_descriptor to)
+        : to{to} {}
+
+    // this exception is used to stop the BFS early if a matching vertex is found
+    struct PathExists : public std::exception {};
+
+    // this throws PathExists on success
+    void discover_vertex(GroupGraph::vertex_descriptor u, [[maybe_unused]] const GroupGraph& g) const {
+      if (u == to) {
+        throw PathExists();
+      }
+    }
+  };
+
+  class NonEmptyGoupFilter {
+  private:
+    GroupPropertyMap property_map;
+
+  public:
+    NonEmptyGoupFilter() = default;
+    NonEmptyGoupFilter(const GroupPropertyMap& map)
+        : property_map{map} {}
+
+    auto operator()(GroupGraph::vertex_descriptor vertex) const -> bool {
+      return !boost::get(property_map, vertex).empty();
+    }
+  };
+
+  void group_reactions_by_container_helper(const Reactor* reactor);
+
+  void clear_all_empty_vertices();
+
+public:
+  // TODO: This should be a const reference, but I don't know how to get immutable access to the reaction graph
+  // properties...
+  GroupedDependencyGraph(ReactionDependencyGraph& reactionGraph);
+  GroupedDependencyGraph() = default;
+
+  void export_graphviz(const std::string& file_name);
+
+  auto has_path(GroupGraph::vertex_descriptor va, GroupGraph::vertex_descriptor vb) const -> bool;
+
+  void try_contract_edge(GroupGraph::vertex_descriptor va, GroupGraph::vertex_descriptor vb);
+
+  void group_reactions_by_container(const std::set<Reactor*>& top_level_reactors);
+
+  void group_chains();
+
+  // TODO: This should be const, but I don't know how to get immutable access to the reaction graph properties...
+  [[nodiscard]] auto transitive_reduction() -> GroupedDependencyGraph;
+
+  [[nodiscard]] auto get_graph() const -> const GroupGraph& { return graph; }
+
+  [[nodiscard]] auto get_group_property_map() -> GroupPropertyMap { return boost::get(group_info_t{}, graph); }
+};
+
+} // namespace reactor
+
+#endif