Fix race condition in Salt loader.

There was a race condition in the salt loader when injecting global
values (e.g. "__pillar__" or "__salt__") into modules. One effect of
this race condition was that in a setup with multiple threads, some
threads may see pillar data intended for other threads or the pillar
data seen by a thread might even change spuriously.

There have been earlier attempts to fix this problem (#27937, #29397).
These patches tried to fix the problem by storing the dictionary that
keeps the relevant data in a thread-local variable and referencing this
thread-local variable from the variables that are injected into the
modules.

These patches did not fix the problem completely because they only
work when a module is loaded through a single loader instance only.
When there is more than one loader, there is more than one
thread-local variable and the variable injected into a module is
changed to point to another thread-local variable when the module is
loaded again. Thus, the problem resurfaced while working on #39670.

This patch attempts to solve the problem from a slightly different
angle, complementing the earlier patches: The value injected into the
modules now is a proxy that internally uses a thread-local variable to
decide to which object it points. This means that when loading a module
again through a different loader (possibly passing different pillar
data), the data is actually only changed in the thread in which the
loader is used. Other threads are not affected by such a change.

This means that it will work correctly in the current situation where
loaders are possibly created by many different modules and these
modules do not necessary know in which context they are executed. Thus
it is much more flexible and reliable than the more explicit approach
used by the two earlier patches.

Unfortunately, the stand JSON and Msgpack serialization code cannot
handle proxied objects, so they have to be unwrapped before passing them
to that code.

The salt.utils.json module has been modified to  takes care of
unwrapping objects that are proxied using the ThreadLocalProxy.

The salt.utils.msgpack module has been added and basically provides the
same functions as the salt.utils.json module, but for msgpack. Like the
json module, it takes care of unwrapping proxies.

salt/cloud/clouds/ec2.py

@@ -82,7 +82,6 @@
 import binascii
 import datetime
 import base64
-import msgpack
 import re
 import decimal
 
@@ -91,6 +90,7 @@
 import salt.utils.files
 import salt.utils.hashutils
 import salt.utils.json
+import salt.utils.msgpack
 import salt.utils.stringutils
 import salt.utils.yaml
 from salt._compat import ElementTree as ET
@@ -4840,7 +4840,7 @@ def _parse_pricing(url, name):
         __opts__['cachedir'], 'ec2-pricing-{0}.p'.format(name)
     )
     with salt.utils.files.fopen(outfile, 'w') as fho:
-        msgpack.dump(regions, fho)
+        salt.utils.msgpack.dump(regions, fho)
 
     return True
 
@@ -4908,7 +4908,8 @@ def show_pricing(kwargs=None, call=None):
         update_pricing({'type': name}, 'function')
 
     with salt.utils.files.fopen(pricefile, 'r') as fhi:
-        ec2_price = salt.utils.stringutils.to_unicode(msgpack.load(fhi))
+        ec2_price = salt.utils.stringutils.to_unicode(
+            salt.utils.msgpack.load(fhi))
 
     region = get_location(profile)
     size = profile.get('size', None)

salt/cloud/clouds/gce.py

@@ -53,7 +53,6 @@
 import re
 import pprint
 import logging
-import msgpack
 from ast import literal_eval
 from salt.utils.versions import LooseVersion as _LooseVersion
 
@@ -91,6 +90,7 @@
 import salt.utils.cloud
 import salt.utils.files
 import salt.utils.http
+import salt.utils.msgpack
 import salt.config as config
 from salt.cloud.libcloudfuncs import *  # pylint: disable=redefined-builtin,wildcard-import,unused-wildcard-import
 from salt.exceptions import (
@@ -2619,7 +2619,7 @@ def update_pricing(kwargs=None, call=None):
         __opts__['cachedir'], 'gce-pricing.p'
     )
     with salt.utils.files.fopen(outfile, 'w') as fho:
-        msgpack.dump(price_json['dict'], fho)
+        salt.utils.msgpack.dump(price_json['dict'], fho)
 
     return True
 
@@ -2658,7 +2658,7 @@ def show_pricing(kwargs=None, call=None):
         update_pricing()
 
     with salt.utils.files.fopen(pricefile, 'r') as fho:
-        sizes = msgpack.load(fho)
+        sizes = salt.utils.msgpack.load(fho)
 
     per_hour = float(sizes['gcp_price_list'][size][region])
 

salt/engines/stalekey.py

@@ -28,11 +28,11 @@
 import salt.key
 import salt.utils.files
 import salt.utils.minions
+import salt.utils.msgpack
 import salt.wheel
 
 # Import 3rd-party libs
 from salt.ext import six
-import msgpack
 
 log = logging.getLogger(__name__)
 
@@ -60,7 +60,7 @@ def start(interval=3600, expire=604800):
         if os.path.exists(presence_file):
             try:
                 with salt.utils.files.fopen(presence_file, 'r') as f:
-                    minions = msgpack.load(f)
+                    minions = salt.utils.msgpack.load(f)
             except IOError as e:
                 log.error('Could not open presence file %s: %s', presence_file, e)
                 time.sleep(interval)
@@ -95,7 +95,7 @@ def start(interval=3600, expire=604800):
 
         try:
             with salt.utils.files.fopen(presence_file, 'w') as f:
-                msgpack.dump(minions, f)
+                salt.utils.msgpack.dump(minions, f)
         except IOError as e:
             log.error('Could not write to presence file %s: %s', presence_file, e)
         time.sleep(interval)

salt/key.py

@@ -38,9 +38,10 @@
 from salt.ext.six.moves import input
 # pylint: enable=import-error,no-name-in-module,redefined-builtin
 
-# Import third party libs
+# We do not always need msgpack, so we do not want to fail here if msgpack is
+# not available.
 try:
-    import msgpack
+    import salt.utils.msgpack
 except ImportError:
     pass
 
@@ -1035,7 +1036,7 @@ def check_minion_cache(self, preserve_minions=False):
                     if ext == '.json':
                         data = salt.utils.json.load(fp_)
                     elif ext == '.msgpack':
-                        data = msgpack.load(fp_)
+                        data = salt.utils.msgpack.load(fp_)
                     role = salt.utils.stringutils.to_unicode(data['role'])
                     if role not in minions:
                         os.remove(path)

salt/loader.py

@@ -13,6 +13,7 @@
 import logging
 import inspect
 import tempfile
+import threading
 import functools
 import types
 from collections import MutableMapping
@@ -28,6 +29,7 @@
 import salt.utils.lazy
 import salt.utils.odict
 import salt.utils.platform
+import salt.utils.thread_local_proxy
 import salt.utils.versions
 from salt.exceptions import LoaderError
 from salt.template import check_render_pipe_str
@@ -1003,6 +1005,76 @@ def _mod_type(module_path):
     return 'ext'
 
 
+def _inject_into_mod(mod, name, value, force_lock=False):
+    '''
+    Inject a variable into a module. This is used to inject "globals" like
+    ``__salt__``, ``__pillar``, or ``grains``.
+
+    Instead of injecting the value directly, a ``ThreadLocalProxy`` is created.
+    If such a proxy is already present under the specified name, it is updated
+    with the new value. This update only affects the current thread, so that
+    the same name can refer to different values depending on the thread of
+    execution.
+
+    This is important for data that is not truly global. For example, pillar
+    data might be dynamically overriden through function parameters and thus
+    the actual values available in pillar might depend on the thread that is
+    calling a module.
+
+    mod:
+        module object into which the value is going to be injected.
+
+    name:
+        name of the variable that is injected into the module.
+
+    value:
+        value that is injected into the variable. The value is not injected
+        directly, but instead set as the new reference of the proxy that has
+        been created for the variable.
+
+    force_lock:
+        whether the lock should be acquired before checking whether a proxy
+        object for the specified name has already been injected into the
+        module. If ``False`` (the default), this function checks for the
+        module's variable without acquiring the lock and only acquires the lock
+        if a new proxy has to be created and injected.
+    '''
+    from salt.utils.thread_local_proxy import ThreadLocalProxy
+    old_value = getattr(mod, name, None)
+    # We use a double-checked locking scheme in order to avoid taking the lock
+    # when a proxy object has already been injected.
+    # In most programming languages, double-checked locking is considered
+    # unsafe when used without explicit memory barriers because one might read
+    # an uninitialized value. In CPython it is safe due to the global
+    # interpreter lock (GIL). In Python implementations that do not have the
+    # GIL, it could be unsafe, but at least Jython also guarantees that (for
+    # Python objects) memory is not corrupted when writing and reading without
+    # explicit synchronization
+    # (http://www.jython.org/jythonbook/en/1.0/Concurrency.html).
+    # Please note that in order to make this code safe in a runtime environment
+    # that does not make this guarantees, it is not sufficient. The
+    # ThreadLocalProxy must also be created with fallback_to_shared set to
+    # False or a lock must be added to the ThreadLocalProxy.
+    if force_lock:
+        with _inject_into_mod.lock:
+            if isinstance(old_value, ThreadLocalProxy):
+                ThreadLocalProxy.set_reference(old_value, value)
+            else:
+                setattr(mod, name, ThreadLocalProxy(value, True))
+    else:
+        if isinstance(old_value, ThreadLocalProxy):
+            ThreadLocalProxy.set_reference(old_value, value)
+        else:
+            _inject_into_mod(mod, name, value, True)
+
+
+# Lock used when injecting globals. This is needed to avoid a race condition
+# when two threads try to load the same module concurrently. This must be
+# outside the loader because there might be more than one loader for the same
+# namespace.
+_inject_into_mod.lock = threading.RLock()
+
+
 # TODO: move somewhere else?
 class FilterDictWrapper(MutableMapping):
     '''
@@ -1493,7 +1565,7 @@ def _load_module(self, name):
 
         # pack whatever other globals we were asked to
         for p_name, p_value in six.iteritems(self.pack):
-            setattr(mod, p_name, p_value)
+            _inject_into_mod(mod, p_name, p_value)
 
         module_name = mod.__name__.rsplit('.', 1)[-1]
 

salt/log/handlers/fluent_mod.py

@@ -97,14 +97,17 @@
 try:
     # Attempt to import msgpack
     import msgpack
+    import salt.utils.msgpack
     # There is a serialization issue on ARM and potentially other platforms
     # for some msgpack bindings, check for it
     if msgpack.loads(msgpack.dumps([1, 2, 3]), use_list=True) is None:
         raise ImportError
+    import salt.utils.msgpack
 except ImportError:
     # Fall back to msgpack_pure
     try:
         import msgpack_pure as msgpack
+        import salt.utils.msgpack
     except ImportError:
         # TODO: Come up with a sane way to get a configured logfile
         #       and write to the logfile when this error is hit also
@@ -456,7 +459,7 @@ def _make_packet(self, label, timestamp, data):
         packet = (tag, timestamp, data)
         if self.verbose:
             print(packet)
-        return msgpack.packb(packet)
+        return salt.utils.msgpack.packb(packet, _msgpack_module=msgpack)
 
     def _send(self, bytes_):
         self.lock.acquire()

salt/modules/saltcheck.py

@@ -51,7 +51,7 @@
 import logging
 import os
 import time
-from json import loads, dumps
+from salt.utils.json import loads, dumps
 try:
     import salt.utils.files
     import salt.utils.path

salt/modules/state.py

@@ -34,6 +34,7 @@
 import salt.utils.hashutils
 import salt.utils.jid
 import salt.utils.json
+import salt.utils.msgpack
 import salt.utils.platform
 import salt.utils.state
 import salt.utils.stringutils
@@ -44,7 +45,6 @@
 
 # Import 3rd-party libs
 from salt.ext import six
-import msgpack
 
 __proxyenabled__ = ['*']
 
@@ -184,7 +184,7 @@ def _get_pause(jid, state_id=None):
             data[state_id] = {}
     if os.path.exists(pause_path):
         with salt.utils.files.fopen(pause_path, 'rb') as fp_:
-            data = msgpack.loads(fp_.read())
+            data = salt.utils.msgpack.loads(fp_.read())
     return data, pause_path
 
 
@@ -255,7 +255,7 @@ def soft_kill(jid, state_id=None):
     data, pause_path = _get_pause(jid, state_id)
     data[state_id]['kill'] = True
     with salt.utils.files.fopen(pause_path, 'wb') as fp_:
-        fp_.write(msgpack.dumps(data))
+        fp_.write(salt.utils.msgpack.dumps(data))
 
 
 def pause(jid, state_id=None, duration=None):
@@ -290,7 +290,7 @@ def pause(jid, state_id=None, duration=None):
     if duration:
         data[state_id]['duration'] = int(duration)
     with salt.utils.files.fopen(pause_path, 'wb') as fp_:
-        fp_.write(msgpack.dumps(data))
+        fp_.write(salt.utils.msgpack.dumps(data))
 
 
 def resume(jid, state_id=None):
@@ -324,7 +324,7 @@ def resume(jid, state_id=None):
     if state_id == '__all__':
         data = {}
     with salt.utils.files.fopen(pause_path, 'wb') as fp_:
-        fp_.write(msgpack.dumps(data))
+        fp_.write(salt.utils.msgpack.dumps(data))
 
 
 def orchestrate(mods,

salt/modules/win_repo.py

@@ -31,11 +31,8 @@
     PER_REMOTE_ONLY
 )
 from salt.ext import six
-try:
-    import msgpack
-except ImportError:
-    import msgpack_pure as msgpack  # pylint: disable=import-error
 import salt.utils.gitfs
+import salt.utils.msgpack
 # pylint: enable=unused-import
 
 log = logging.getLogger(__name__)