File make-reactor-engine-less-blocking-the-eventpublisher.patch of Package salt

From d4fdca81f8e011f75ded1081bcb6e33ee044569d Mon Sep 17 00:00:00 2001
From: Victor Zhestkov <vzhestkov@suse.com>
Date: Wed, 15 May 2024 09:44:21 +0200
Subject: [PATCH] Make reactor engine less blocking the EventPublisher

---
 salt/utils/reactor.py | 44 ++++++++++++++++++++++++++-----------------
 1 file changed, 27 insertions(+), 17 deletions(-)

diff --git a/salt/utils/reactor.py b/salt/utils/reactor.py
index 5b9138636b..afee77c2fd 100644
--- a/salt/utils/reactor.py
+++ b/salt/utils/reactor.py
@@ -6,6 +6,7 @@ import fnmatch
 import glob
 import logging
 import os
+from threading import Lock
 
 import salt.client
 import salt.defaults.exitcodes
@@ -204,13 +205,6 @@ class Reactor(salt.utils.process.SignalHandlingProcess, salt.state.Compiler):
         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
@@ -228,7 +222,7 @@ class Reactor(salt.utils.process.SignalHandlingProcess, salt.state.Compiler):
         ) 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
@@ -278,15 +272,9 @@ class Reactor(salt.utils.process.SignalHandlingProcess, salt.state.Compiler):
                     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:
@@ -307,6 +295,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"]
@@ -492,3 +481,24 @@ class ReactWrap:
         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)
+        )
-- 
2.47.0