Fix rules engine unsubscribe race condition

production/rules/event_manager/inc/event_manager.hpp

@@ -96,7 +96,7 @@ class event_manager_t
     std::recursive_mutex m_init_lock;
 
     // Hash table of all rules registered with the system.
-    // The key is the rulset_name::rule_name.
+    // The key is the ruleset_name::rule_name.
     std::unordered_map<std::string, std::unique_ptr<_rule_binding_t>> m_rules;
 
     // List of rules that are invoked when an event is logged.
@@ -133,6 +133,10 @@ class event_manager_t
     // members, so the thread pool must be initialized last and destroyed first.
     std::unique_ptr<rule_thread_pool_t> m_invocations;
 
+    // Commit trigger function pointer that the database calls
+    // whenever a transaction is committed.
+    gaia::db::triggers::commit_trigger_fn m_trigger_fn;
+
 private:
     // Only internal static creation is allowed.
     event_manager_t() = default;

production/rules/event_manager/inc/rule_thread_pool.hpp

@@ -62,6 +62,12 @@ class rule_thread_pool_t
      */
     rule_thread_pool_t(size_t num_threads, uint32_t max_rule_retries, rule_stats_manager_t& stats_manager);
 
+    /**
+     * Wait for the current rules "graph" to execute. Wait for all rules to finish
+     * executing as well as any rules that these rules enqueue.
+     */
+    void wait_for_rules_to_finish();
+
     /**
      * Notify and wait for all workers in the thread pool
      * to finish executing their last work item before destroying
@@ -93,9 +99,9 @@ class rule_thread_pool_t
 
 private:
     void rule_worker(int32_t& count_busy_workers);
-
     void invoke_rule(invocation_t& invocation);
     void process_pending_invocations(bool should_schedule);
+    void wait_for_rules_to_finish(std::unique_lock<std::mutex>& lock);
 
     // Each thread has a copy of these two variables to determine
     // whether pending rule invocations can be scheduled or they

production/rules/event_manager/src/event_manager.cpp

@@ -93,10 +93,10 @@ void event_manager_t::init(const event_manager_settings_t& settings)
         m_rule_checker = make_unique<rule_checker_t>();
     }
 
-    auto fn = [](const trigger_event_list_t& event_list) {
+    m_trigger_fn = [](const trigger_event_list_t& event_list) {
         event_manager_t::get().commit_trigger(event_list);
     };
-    set_commit_trigger(fn);
+    set_commit_trigger(m_trigger_fn);
 
     m_is_initialized = true;
 }
@@ -381,8 +381,29 @@ bool event_manager_t::unsubscribe_rule(
 void event_manager_t::unsubscribe_rules()
 {
     gaia_log::rules().debug("Unsubscribing all rules.");
+
+    // Because a rule may cause further invocations on commit which cause access of the
+    // rule subscriptions, we need to wait for the current rules graph to finish executing.
+    if (m_invocations)
+    {
+        m_invocations->wait_for_rules_to_finish();
+
+        // Detach the commit trigger so that any new events that come in do not try
+        // to look for rule subscriptions while we are removing them.
+        gaia::db::set_commit_trigger(nullptr);
+
+        // Since an invocation might have snuck in between the time we finished executing
+        // and the time we detached the commit trigger, wait for these to finish.
+        m_invocations->wait_for_rules_to_finish();
+    }
+
+    // Now it is safe to clear out the subscriptions and rule bindings.
     m_subscriptions.clear();
     m_rules.clear();
+
+    // Reset the commit trigger so we are ready to go on any new
+    // rule subscriptions.
+    gaia::db::set_commit_trigger(m_trigger_fn);
 }
 
 void event_manager_t::list_subscribed_rules(

production/rules/event_manager/src/rule_stats.cpp

@@ -23,7 +23,7 @@ rule_stats_t::rule_stats_t(const char* a_rule_id)
     if (a_rule_id)
     {
         rule_id = a_rule_id;
-        // For pretting formatting, truncate long rule ids.
+        // Truncate long rule ids for pretty formatting.
         if (rule_id.length() > c_max_rule_id_len)
         {
             truncated_rule_id = rule_id.substr(0, c_max_rule_id_len);

production/rules/event_manager/src/rule_thread_pool.cpp

@@ -44,21 +44,19 @@ rule_thread_pool_t::~rule_thread_pool_t()
     shutdown();
 }
 
-void rule_thread_pool_t::shutdown()
+void rule_thread_pool_t::wait_for_rules_to_finish()
 {
-    if (m_threads.size() == 0)
-    {
-        return;
-    }
-
-    // Wait for any scheduled rules to finish executing. Once the rule
-    // has finished, its worker thread will go into a wait state (not busy).
-    // Once all workers are not busy, then we can exit if there are no pending
-    // rule invocations.  If there are invocations left, then wait for them to
-    // be executed and check again.
-    auto start_shutdown_time = gaia::common::timer_t::get_time_point();
-
     unique_lock lock(m_lock, defer_lock);
+    wait_for_rules_to_finish(lock);
+    lock.unlock();
+}
+
+// NOTE:  Callers must unlock the passed in mutex.
+//
+// TODO[GAIAPLAT-1020]: Add a configuration setting to limit the time
+// we will wait for all rules to execute.
+void rule_thread_pool_t::wait_for_rules_to_finish(std::unique_lock<std::mutex>& lock)
+{
     while (true)
     {
         lock.lock();
@@ -70,6 +68,23 @@ void rule_thread_pool_t::shutdown()
         lock.unlock();
         std::this_thread::yield();
     }
+}
+
+void rule_thread_pool_t::shutdown()
+{
+    if (m_threads.size() == 0)
+    {
+        return;
+    }
+
+    auto start_shutdown_time = gaia::common::timer_t::get_time_point();
+
+    unique_lock lock(m_lock, defer_lock);
+
+    // Wait for the currently executing rules "graph" to finish executing.  This will
+    // drain the work queues of any rules executing AND any rules that these rules
+    // may trigger.
+    wait_for_rules_to_finish(lock);
 
     m_exit = true;
     lock.unlock();