[master] Make reactor engine less blocking the EventPublisher

changelog/66158.changed.md

@@ -0,0 +1 @@
+Make reactor engine less blocking the EventPublisher

salt/utils/reactor.py

@@ -6,6 +6,7 @@
 import glob
 import logging
 import os
+from threading import Lock
 
 import salt.client
 import salt.defaults.exitcodes
@@ -195,13 +196,6 @@ def reactions(self, tag, data, reactors):
         self.resolve_aliases(chunks)
         return chunks
 
-    def call_reactions(self, chunks):
-        """
-        Execute the reaction state
-        """
-        for chunk in chunks:
-            self.wrap.run(chunk)
-
     def run(self):
         """
         Enter into the server loop
@@ -219,7 +213,7 @@ def run(self):
         ) as event:
             self.wrap = ReactWrap(self.opts)
 
-            for data in event.iter_events(full=True):
+            for data in event.iter_events(full=True, auto_reconnect=True):
                 # skip all events fired by ourselves
                 if data["data"].get("user") == self.wrap.event_user:
                     continue
@@ -269,15 +263,9 @@ def run(self):
                     if not self.is_leader:
                         continue
                     else:
-                        reactors = self.list_reactors(data["tag"])
-                        if not reactors:
-                            continue
-                        chunks = self.reactions(data["tag"], data["data"], reactors)
-                        if chunks:
-                            try:
-                                self.call_reactions(chunks)
-                            except SystemExit:
-                                log.warning("Exit ignored by reactor")
+                        self.wrap.call_reactions(
+                            data, self.list_reactors, self.reactions
+                        )
 
 
 class ReactWrap:
@@ -298,6 +286,7 @@ class ReactWrap:
 
     def __init__(self, opts):
         self.opts = opts
+        self._run_lock = Lock()
         if ReactWrap.client_cache is None:
             ReactWrap.client_cache = salt.utils.cache.CacheDict(
                 opts["reactor_refresh_interval"]
@@ -483,3 +472,24 @@ def caller(self, fun, **kwargs):
         Wrap LocalCaller to execute remote exec functions locally on the Minion
         """
         self.client_cache["caller"].cmd(fun, *kwargs["arg"], **kwargs["kwarg"])
+
+    def _call_reactions(self, data, list_reactors, get_reactions):
+        reactors = list_reactors(data["tag"])
+        if not reactors:
+            return
+        chunks = get_reactions(data["tag"], data["data"], reactors)
+        if not chunks:
+            return
+        with self._run_lock:
+            try:
+                for chunk in chunks:
+                    self.run(chunk)
+            except Exception as exc:  # pylint: disable=broad-except
+                log.error(
+                    "Exception while calling the reactions: %s", exc, exc_info=True
+                )
+
+    def call_reactions(self, data, list_reactors, get_reactions):
+        return self.pool.fire_async(
+            self._call_reactions, args=(data, list_reactors, get_reactions)
+        )

tests/pytests/unit/utils/test_reactor.py

@@ -1,6 +1,9 @@
+import threading
+
 import pytest
 
 import salt.utils.data
+import salt.utils.event
 import salt.utils.reactor as reactor
 from tests.support.mock import MagicMock, call, patch
 
@@ -55,3 +58,37 @@ def test_run_master_reactor(master_opts, master_reactor):
                 master_reactor.run()
                 calls = [call(9)]
                 os_nice_mock.assert_has_calls(calls)
+
+
+@pytest.mark.skip_on_windows(reason="Reactors unavailable on Windows")
+def test_reactors_list_in_separate_thread(master_opts, master_reactor):
+    """
+    Test that some of the tasks for reactors are performed in the separate thread
+    """
+
+    list_reactors_thread_id = None
+    get_reactions_thread_id = None
+
+    def list_reactors(*args, **kwargs):
+        nonlocal list_reactors_thread_id
+        list_reactors_thread_id = threading.get_ident()
+        return ["test_reactor"]
+
+    def get_reactions(*args, **kwargs):
+        nonlocal get_reactions_thread_id
+        get_reactions_thread_id = threading.get_ident()
+        return []
+
+    with patch.object(
+        salt.utils.event.SaltEvent,
+        "iter_events",
+        return_value=iter([{"tag": "test", "data": {}}]),
+    ), patch.object(master_reactor, "list_reactors", list_reactors), patch.object(
+        master_reactor, "reactions", get_reactions
+    ):
+        main_thread_id = threading.get_ident()
+        master_reactor.run()
+        assert main_thread_id != list_reactors_thread_id
+        assert main_thread_id != get_reactions_thread_id
+        assert list_reactors_thread_id == get_reactions_thread_id
+        assert None not in (list_reactors_thread_id, get_reactions_thread_id)