Fix publisher leak

salt/config/__init__.py

@@ -41,6 +41,13 @@
 import salt.utils.sdb
 from salt.utils.locales import sdecode
 
+try:
+    import psutil
+    HAS_PSUTIL = True
+except ImportError:
+    HAS_PSUTIL = False
+    import salt.grains.core
+
 log = logging.getLogger(__name__)
 
 _DFLT_LOG_DATEFMT = '%H:%M:%S'
@@ -60,6 +67,30 @@
     _DFLT_IPC_MODE = 'ipc'
     _MASTER_TRIES = 1
 
+def _gather_buffer_space():
+    '''
+    Gather some system data and then calculate
+    buffer space.
+
+    Result is in bytes.
+    '''
+    if HAS_PSUTIL:
+        # Oh good, we have psutil. This will be quick.
+        total_mem = psutil.virtual_memory().total
+    else:
+        # We need to load up some grains. This will be slow.
+        os_data = salt.grains.core.os_data()
+        grains = salt.grains.core._memdata(os_data)
+        total_mem = grains['mem_total']
+    # Return the higher number between 5% of the system memory and 100MB
+    return min([total_mem * 0.05, 10 << 20])
+
+# For the time being this will be a fixed calculation
+# TODO: Allow user configuration
+_DFLT_IPC_WBUFFER = _gather_buffer_space() * .5
+# TODO: Reserved for future use
+_DFLT_IPC_RBUFFER = _gather_buffer_space() * .5
+
 FLO_DIR = os.path.join(
         os.path.dirname(os.path.dirname(__file__)),
         'daemons', 'flo')
@@ -444,6 +475,10 @@
     # ZMQ HWM for EventPublisher pub socket
     'event_publisher_pub_hwm': int,
 
+    # IPC buffer size
+    # Refs https://github.com/saltstack/salt/issues/34215
+    'ipc_write_buffer': int,
+
     # The number of MWorker processes for a master to startup. This number needs to scale up as
     # the number of connected minions increases.
     'worker_threads': int,
@@ -1176,6 +1211,7 @@
     'minion_data_cache': True,
     'enforce_mine_cache': False,
     'ipc_mode': _DFLT_IPC_MODE,
+    'ipc_write_buffer': _DFLT_IPC_WBUFFER,
     'ipv6': False,
     'tcp_master_pub_port': 4512,
     'tcp_master_pull_port': 4513,
@@ -1541,6 +1577,7 @@ def _read_conf_file(path):
         return conf_opts
 
 
+
 def _absolute_path(path, relative_to=None):
     '''
     Return an absolute path. In case ``relative_to`` is passed and ``path`` is

salt/transport/ipc.py

@@ -431,9 +431,10 @@ class IPCMessagePublisher(object):
     A Tornado IPC Publisher similar to Tornado's TCPServer class
     but using either UNIX domain sockets or TCP sockets
     '''
-    def __init__(self, socket_path, io_loop=None):
+    def __init__(self, opts, socket_path, io_loop=None):
         '''
         Create a new Tornado IPC server
+        :param dict opts: Salt options
         :param str/int socket_path: Path on the filesystem for the
                                     socket to bind to. This socket does
                                     not need to exist prior to calling
@@ -444,6 +445,7 @@ def __init__(self, socket_path, io_loop=None):
                                     for a tcp localhost connection.
         :param IOLoop io_loop: A Tornado ioloop to handle scheduling
         '''
+        self.opts = opts
         self.socket_path = socket_path
         self._started = False
 
@@ -505,10 +507,12 @@ def publish(self, msg):
 
     def handle_connection(self, connection, address):
         log.trace('IPCServer: Handling connection to address: {0}'.format(address))
+        log.trace('IPCPublisher write buffer set to: {0}'.format(self.opts['ipc_write_buffer']))
         try:
             stream = IOStream(
                 connection,
                 io_loop=self.io_loop,
+                max_write_buffer_size = self.opts['ipc_write_buffer']
             )
             self.streams.add(stream)
         except Exception as exc:

salt/utils/event.py

@@ -865,6 +865,7 @@ def __init__(self, opts, publish_handler, io_loop=None):
                         raise
 
         self.publisher = salt.transport.ipc.IPCMessagePublisher(
+            self.opts,
             epub_uri,
             io_loop=self.io_loop
         )
@@ -953,6 +954,7 @@ def run(self):
                 )
 
             self.publisher = salt.transport.ipc.IPCMessagePublisher(
+                self.opts,
                 epub_uri,
                 io_loop=self.io_loop
             )