[master 1/1] Use dictionaries is thread-safe manner.

dashea installerbot-noreply at redhat.com
Wed Jul 22 16:04:27 UTC 2015


From: David Shea <dshea at redhat.com>

One of the changes in Python 3 was in the way that dictionaries are
accessed: the dict.iter{keys,items,values} methods were removed, and the
return type of dict.{keys,items,values} was changed from returning a
copy of the dictionary data as a list to returning an iterable view of
the dictionary data that is modified along with the dictionary. one of
the implications of this is that it is now impossible to create an
iterable representation of dictionary data that is safe from
modifications from other threads, because dictionaries (and now the
views into the dictionaries) cannot be modified during iteration and
list(x) works using iteration. Way to go, Python.

This is most visible in the threads module: sometimes wait_all() will
crash with "RuntimeError: dictionary changed size during iteration."
Fixed that and some other potential multi-threaded dictionary access in
threads, dnfpayload and xkl_wrapper through the use of more locks, and
added a bunch of comments where I needed to convince myself that the
dictionary access is thread safe.
---
 pyanaconda/anaconda.py                       |  4 ++
 pyanaconda/bootloader.py                     |  3 ++
 pyanaconda/kickstart.py                      |  6 +++
 pyanaconda/localization.py                   |  6 +--
 pyanaconda/packaging/dnfpayload.py           | 70 +++++++++++++++++-----------
 pyanaconda/storage_utils.py                  |  2 +
 pyanaconda/threads.py                        | 26 +++++++----
 pyanaconda/ui/gui/spokes/advstorage/iscsi.py |  2 +
 pyanaconda/ui/gui/xkl_wrapper.py             | 19 +++++---
 9 files changed, 93 insertions(+), 45 deletions(-)

diff --git a/pyanaconda/anaconda.py b/pyanaconda/anaconda.py
index 1235646..59399cd 100644
--- a/pyanaconda/anaconda.py
+++ b/pyanaconda/anaconda.py
@@ -197,6 +197,10 @@ def dumpState(self):
 
         # gather up info on the running threads
         threads = "\nThreads\n-------\n"
+
+        # Every call to sys._current_frames() returns a new dict, so it is not
+        # modified when threads are created or destroyed. Iterating over it is
+        # thread safe.
         for thread_id, frame in sys._current_frames().items():
             threads += "\nThread %s\n" % (thread_id,)
             threads += "".join(format_stack(frame))
diff --git a/pyanaconda/bootloader.py b/pyanaconda/bootloader.py
index 7537626..0f62d73 100644
--- a/pyanaconda/bootloader.py
+++ b/pyanaconda/bootloader.py
@@ -798,6 +798,9 @@ def set_boot_args(self, *args, **kwargs):
         rootdev = storage.rootDevice
         if any(rootdev.dependsOn(netdev) for netdev in netdevs):
             dracut_devices = set(dracut_devices)
+            # By this time this thread should be the only one running, and also
+            # mountpoints is a property function that returns a new dict every
+            # time, so iterating over the values is safe.
             for dev in storage.mountpoints.values():
                 if any(dev.dependsOn(netdev) for netdev in netdevs):
                     dracut_devices.add(dev)
diff --git a/pyanaconda/kickstart.py b/pyanaconda/kickstart.py
index 13a729a..983425e 100644
--- a/pyanaconda/kickstart.py
+++ b/pyanaconda/kickstart.py
@@ -248,6 +248,8 @@ def getAvailableDiskSpace(storage):
     """
 
     free_space = storage.freeSpaceSnapshot
+    # blivet creates a new free space dict to instead of modifying the old one,
+    # so there is no worry about the dictionary changing during iteration.
     return sum(disk_free for disk_free, fs_free in free_space.values())
 
 def refreshAutoSwapSize(storage):
@@ -1116,6 +1118,8 @@ def execute(self, storage, ksdata, instClass):
         storage.doAutoPart = False
 
         if self.onbiosdisk != "":
+            # eddDict is only modified during storage.reset(), so don't do that
+            # while executing storage.
             for (disk, biosdisk) in storage.eddDict.items():
                 if "%x" % biosdisk == self.onbiosdisk:
                     self.disk = disk
@@ -1884,6 +1888,8 @@ def __init__(self):
     def __str__(self):
         """Return the %anaconda section"""
         retval = ""
+        # This dictionary should only be modified during __init__, so if it
+        # changes during iteration something has gone horribly wrong.
         lst = sorted(self._writeOrder.keys())
         for prio in lst:
             for obj in self._writeOrder[prio]:
diff --git a/pyanaconda/localization.py b/pyanaconda/localization.py
index 480293f..f6f05bf 100644
--- a/pyanaconda/localization.py
+++ b/pyanaconda/localization.py
@@ -140,7 +140,7 @@ def find_best_locale_match(locale, langcodes):
 
     """
 
-    score_map = {"language" : 1000,
+    SCORE_MAP = {"language" : 1000,
                  "territory":  100,
                  "script"   :   10,
                  "encoding" :    1}
@@ -153,7 +153,7 @@ def get_match_score(locale, langcode):
         if not locale_parts or not langcode_parts:
             return score
 
-        for part, part_score in score_map.items():
+        for part, part_score in SCORE_MAP.items():
             if locale_parts[part] and langcode_parts[part]:
                 if locale_parts[part] == langcode_parts[part]:
                     # match
@@ -177,7 +177,7 @@ def get_match_score(locale, langcode):
     sorted_langcodes = sorted(scores, key=lambda item_score: item_score[1], reverse=True)
 
     # matches matching only script or encoding or both are not useful
-    if sorted_langcodes and sorted_langcodes[0][1] > score_map["territory"]:
+    if sorted_langcodes and sorted_langcodes[0][1] > SCORE_MAP["territory"]:
         return sorted_langcodes[0][0]
     else:
         return None
diff --git a/pyanaconda/packaging/dnfpayload.py b/pyanaconda/packaging/dnfpayload.py
index 19dc4ad..3a5ee87 100644
--- a/pyanaconda/packaging/dnfpayload.py
+++ b/pyanaconda/packaging/dnfpayload.py
@@ -43,6 +43,7 @@
 import shutil
 import sys
 import time
+import threading
 from pyanaconda.iutil import ProxyString, ProxyStringError
 from pyanaconda.iutil import open   # pylint: disable=redefined-builtin
 
@@ -194,6 +195,12 @@ def __init__(self, data):
         self._download_location = None
         self._configure()
 
+        # Protect access to _base.repos to ensure that the dictionary is not
+        # modified while another thread is attempting to iterate over it. The
+        # lock only needs to be held during operations that change the number
+        # of repos or that iterate over the repos.
+        self._repos_lock = threading.RLock()
+
     def unsetup(self):
         super(DNFPayload, self).unsetup()
         self._base = None
@@ -259,12 +266,14 @@ def _add_repo(self, ksrepo):
             if not url and not mirrorlist:
                 self._base.repos[repo.id].enable()
             else:
-                self._base.repos.pop(repo.id)
-                self._base.repos.add(repo)
+                with self._repos_lock:
+                    self._base.repos.pop(repo.id)
+                    self._base.repos.add(repo)
                 repo.enable()
         # If the repo's not already known, we've got to add it.
         else:
-            self._base.repos.add(repo)
+            with self._repos_lock:
+                self._base.repos.add(repo)
             repo.enable()
 
         # Load the metadata to verify that the repo is valid
@@ -438,8 +447,9 @@ def _pick_download_location(self):
             raise packaging.PayloadError(msg)
 
         pkgdir = '%s/%s' % (mpoint, DNF_PACKAGE_CACHE_DIR_SUFFIX)
-        for repo in self._base.repos.iter_enabled():
-            repo.pkgdir = pkgdir
+        with self._repos_lock:
+            for repo in self._base.repos.iter_enabled():
+                repo.pkgdir = pkgdir
 
         return pkgdir
 
@@ -492,9 +502,10 @@ def _sync_metadata(self, dnf_repo):
     def baseRepo(self):
         # is any locking needed here?
         repo_names = [constants.BASE_REPO_NAME] + self.DEFAULT_REPOS
-        for repo in self._base.repos.iter_enabled():
-            if repo.id in repo_names:
-                return repo.id
+        with self._repos_lock:
+            for repo in self._base.repos.iter_enabled():
+                if repo.id in repo_names:
+                    return repo.id
         return None
 
     @property
@@ -513,7 +524,8 @@ def mirrorEnabled(self):
     @property
     def repos(self):
         # known repo ids
-        return [r.id for r in self._base.repos.values()]
+        with self._repos_lock:
+            return [r.id for r in self._base.repos.values()]
 
     @property
     def spaceRequired(self):
@@ -611,8 +623,9 @@ def groupDescription(self, grpid):
         return (grp.ui_name, grp.ui_description)
 
     def gatherRepoMetadata(self):
-        for repo in self._base.repos.iter_enabled():
-            self._sync_metadata(repo)
+        with self._repos_lock:
+            for repo in self._base.repos.iter_enabled():
+                self._sync_metadata(repo)
         self._base.fill_sack(load_system_repo=False)
         self._base.read_comps()
         self._refreshEnvironmentAddons()
@@ -745,9 +758,10 @@ def updateBaseRepo(self, fallback=True, checkmount=True):
         self._base.read_all_repos()
 
         enabled = []
-        for repo in self._base.repos.iter_enabled():
-            enabled.append(repo.id)
-            repo.disable()
+        with self._repos_lock:
+            for repo in self._base.repos.iter_enabled():
+                enabled.append(repo.id)
+                repo.disable()
 
         # If askmethod was specified on the command-line, leave all the repos
         # disabled and return
@@ -771,10 +785,12 @@ def updateBaseRepo(self, fallback=True, checkmount=True):
             except (packaging.MetadataError, packaging.PayloadError) as e:
                 log.error("base repo (%s/%s) not valid -- removing it",
                           method.method, url)
-                self._base.repos.pop(constants.BASE_REPO_NAME, None)
+                with self._repos_lock:
+                    self._base.repos.pop(constants.BASE_REPO_NAME, None)
                 if not fallback:
-                    for repo in self._base.repos.iter_enabled():
-                        self.disableRepo(repo.id)
+                    with self._repos_lock:
+                        for repo in self._base.repos.iter_enabled():
+                            self.disableRepo(repo.id)
                     return
 
                 # this preserves the method details while disabling it
@@ -788,9 +804,10 @@ def updateBaseRepo(self, fallback=True, checkmount=True):
                 return
 
             # Otherwise, fall back to the default repos that we disabled above
-            for (id_, repo) in self._base.repos.items():
-                if id_ in enabled:
-                    repo.enable()
+            with self._repos_lock:
+                for (id_, repo) in self._base.repos.items():
+                    if id_ in enabled:
+                        repo.enable()
 
         for ksrepo in self.data.repo.dataList():
             log.debug("repo %s: mirrorlist %s, baseurl %s",
@@ -804,12 +821,13 @@ def updateBaseRepo(self, fallback=True, checkmount=True):
 
         ksnames = [r.name for r in self.data.repo.dataList()]
         ksnames.append(constants.BASE_REPO_NAME)
-        for repo in self._base.repos.iter_enabled():
-            id_ = repo.id
-            if 'source' in id_ or 'debuginfo' in id_:
-                self.disableRepo(id_)
-            elif constants.isFinal and 'rawhide' in id_:
-                self.disableRepo(id_)
+        with self._repos_lock:
+            for repo in self._base.repos.iter_enabled():
+                id_ = repo.id
+                if 'source' in id_ or 'debuginfo' in id_:
+                    self.disableRepo(id_)
+                elif constants.isFinal and 'rawhide' in id_:
+                    self.disableRepo(id_)
 
     def _writeDNFRepo(self, repo, repo_path):
         """ Write a repo object to a DNF repo.conf file
diff --git a/pyanaconda/storage_utils.py b/pyanaconda/storage_utils.py
index d047986..342dfb5 100644
--- a/pyanaconda/storage_utils.py
+++ b/pyanaconda/storage_utils.py
@@ -216,6 +216,8 @@ def sanity_check(storage, min_ram=isys.MIN_RAM):
                             % {'mount': mount, 'size': size,
                                'productName': productName}))
 
+    # storage.mountpoints is a property that returns a new dict each time, so
+    # iterating over it is thread-safe.
     for (mount, device) in filesystems.items():
         problem = filesystems[mount].checkSize()
         if problem < 0:
diff --git a/pyanaconda/threads.py b/pyanaconda/threads.py
index 3f98082..cec2620 100644
--- a/pyanaconda/threads.py
+++ b/pyanaconda/threads.py
@@ -42,6 +42,7 @@ def __init__(self):
         self._objs = {}
         self._objs_lock = threading.RLock()
         self._errors = {}
+        self._errors_lock = threading.RLock()
         self._main_thread = threading.current_thread()
 
     def __call__(self):
@@ -121,7 +122,7 @@ def wait_all(self):
         """Wait for all threads to exit and if there was an error re-raise it.
         """
         with self._objs_lock:
-            names = self._objs.keys()
+            names = list(self._objs.keys())
 
         for name in names:
             if self.get(name) == threading.current_thread():
@@ -130,8 +131,9 @@ def wait_all(self):
             self.wait(name)
 
         if self.any_errors:
-            thread_names = ", ".join(thread_name for thread_name in self._errors.keys()
-                                     if self._errors[thread_name])
+            with self._errors_lock:
+                thread_names = ", ".join(thread_name for thread_name in self._errors.keys()
+                                         if self._errors[thread_name])
             msg = "Unhandled errors from the following threads detected: %s" % thread_names
             raise RuntimeError(msg)
 
@@ -140,7 +142,8 @@ def set_error(self, name, *exc_info):
 
            The exception data is expected to be the tuple from sys.exc_info()
         """
-        self._errors[name] = exc_info
+        with self._errors_lock:
+            self._errors[name] = exc_info
 
     def get_error(self, name):
         """Get the error data for a thread using its name
@@ -151,7 +154,8 @@ def get_error(self, name):
     def any_errors(self):
         """Return True of there have been any errors in any threads
         """
-        return any(self._errors.values())
+        with self._errors_lock:
+            return any(self._errors.values())
 
     def raise_if_error(self, name):
         """If a thread has failed due to an exception, raise it into the main
@@ -161,7 +165,8 @@ def raise_if_error(self, name):
             # no errors found for the thread
             return
 
-        exc_info = self._errors.pop(name)
+        with self._errors_lock:
+            exc_info = self._errors.pop(name)
         if exc_info:
             raise exc_info[0](exc_info[1]).with_traceback(exc_info[2])
 
@@ -189,7 +194,7 @@ def names(self):
             :rtype:   list of strings
         """
         with self._objs_lock:
-            return self._objs.keys()
+            return list(self._objs.keys())
 
     def wait_for_error_threads(self):
         """
@@ -198,9 +203,10 @@ def wait_for_error_threads(self):
 
         """
 
-        for thread_name in self._errors.keys():
-            thread = self._objs[thread_name]
-            thread.join()
+        with self._errors_lock:
+            for thread_name in self._errors.keys():
+                thread = self._objs[thread_name]
+                thread.join()
 
 class AnacondaThread(threading.Thread):
     """A threading.Thread subclass that exists only for a couple purposes:
diff --git a/pyanaconda/ui/gui/spokes/advstorage/iscsi.py b/pyanaconda/ui/gui/spokes/advstorage/iscsi.py
index ef15ff6..e3819f4 100644
--- a/pyanaconda/ui/gui/spokes/advstorage/iscsi.py
+++ b/pyanaconda/ui/gui/spokes/advstorage/iscsi.py
@@ -203,6 +203,8 @@ def _discover(self, credentials, bind):
         elif (self.iscsi.mode == "bind"
               or self.iscsi.mode == "none" and bind):
             activated = set(nm.nm_activated_devices())
+            # The only place iscsi.ifaces is modified is create_interfaces(),
+            # right below, so iteration is safe.
             created = set(self.iscsi.ifaces.values())
             self.iscsi.create_interfaces(activated - created)
 
diff --git a/pyanaconda/ui/gui/xkl_wrapper.py b/pyanaconda/ui/gui/xkl_wrapper.py
index 6ee3199..b559eae 100644
--- a/pyanaconda/ui/gui/xkl_wrapper.py
+++ b/pyanaconda/ui/gui/xkl_wrapper.py
@@ -117,7 +117,9 @@ def __init__(self):
         self.configreg.load(False)
 
         self._layout_infos = dict()
+        self._layout_infos_lock = threading.RLock()
         self._switch_opt_infos = dict()
+        self._switch_opt_infos_lock = threading.RLock()
 
         #this might take quite a long time
         self.configreg.foreach_language(self._get_language_variants, None)
@@ -137,7 +139,8 @@ def _get_lang_variant(self, c_reg, item, subitem, lang):
         #if this layout has already been added for some other language,
         #do not add it again (would result in duplicates in our lists)
         if name not in self._layout_infos:
-            self._layout_infos[name] = LayoutInfo(lang, description)
+            with self._layout_infos_lock:
+                self._layout_infos[name] = LayoutInfo(lang, description)
 
     def _get_country_variant(self, c_reg, item, subitem, country):
         if subitem:
@@ -149,7 +152,8 @@ def _get_country_variant(self, c_reg, item, subitem, country):
 
         # if the layout was not added with any language, add it with a country
         if name not in self._layout_infos:
-            self._layout_infos[name] = LayoutInfo(country, description)
+            with self._layout_infos_lock:
+                self._layout_infos[name] = LayoutInfo(country, description)
 
     def _get_language_variants(self, c_reg, item, user_data=None):
         lang_name, lang_desc = item.get_name(), item.get_description()
@@ -167,7 +171,8 @@ def _get_switch_option(self, c_reg, item, user_data=None):
         desc = item.get_description()
         name = item.get_name()
 
-        self._switch_opt_infos[name] = desc
+        with self._switch_opt_infos_lock:
+            self._switch_opt_infos[name] = desc
 
     def get_current_layout(self):
         """
@@ -200,14 +205,16 @@ def get_current_layout(self):
         return join_layout_variant(layout, variant)
 
     def get_available_layouts(self):
-        """A generator yielding layouts (no need to store them as a bunch)"""
+        """A list of layouts"""
 
-        return self._layout_infos.keys()
+        with self._layout_infos_lock:
+            return list(self._layout_infos.keys())
 
     def get_switching_options(self):
         """Method returning list of available layout switching options"""
 
-        return self._switch_opt_infos.keys()
+        with self._switch_opt_infos_lock:
+            return list(self._switch_opt_infos.keys())
 
     def get_layout_variant_description(self, layout_variant, with_lang=True, xlated=True):
         """


-- 
To view this commit on github, visit https://github.com/rhinstaller/anaconda/commit/79b79378cd483a3527c9c720ef429a186222eee7


More information about the anaconda-patches mailing list