[PATCH] Make thread manager operations atomic (#1029898)

Vratislav Podzimek vpodzime at redhat.com
Fri Nov 15 19:06:48 UTC 2013


On Fri, 2013-11-15 at 18:06 +0100, Martin Kolman wrote:
> The thread manipulation in thread manager needs to
> be atomic to get rid of nasty race conditions.
> For example, without the lock, it was possible
> to get a thread object that was not yet started
> (as threads are first added to the tracking dict
> and then started) and try to join it,
> resulting in an exception.
> And it was also possible that wait would fail,
> as exists & get were called separately,
> introducing yet other race condition.
> Also for clarity, rename raise_error -> raise_if_error.
> 
> Signed-off-by: Martin Kolman <mkolman at redhat.com>
> ---
>  pyanaconda/threads.py | 56 ++++++++++++++++++++++++++++++++++++---------------
>  1 file changed, 40 insertions(+), 16 deletions(-)
> 
> diff --git a/pyanaconda/threads.py b/pyanaconda/threads.py
> index 77cc306..29290df 100644
> --- a/pyanaconda/threads.py
> +++ b/pyanaconda/threads.py
> @@ -40,6 +40,7 @@ class ThreadManager(object):
>      """
>      def __init__(self):
>          self._objs = {}
> +        self._objs_lock = threading.RLock()
>          self._errors = {}
>          self._main_thread = threading.current_thread()
>  
> @@ -50,12 +51,16 @@ class ThreadManager(object):
>          """Given a Thread or Process object, add it to the list of known objects
>             and start it.  It is assumed that obj.name is unique and descriptive.
>          """
> -        if obj.name in self._objs:
> -            raise KeyError("Cannot add thread '%s', a thread with the same name already running" % obj.name)
>  
> -        self._objs[obj.name] = obj
> -        self._errors[obj.name] = None
> -        obj.start()
> +        # we need to lock the thread dictionary when adding a new thread,
> +        # so that callers can't get & join threads that are not yet started
> +        with self._objs_lock:
> +            if obj.name in self._objs:
> +                raise KeyError("Cannot add thread '%s', a thread with the same name already running" % obj.name)
> +
> +            self._objs[obj.name] = obj
> +            self._errors[obj.name] = None
> +            obj.start()
>  
>          return obj.name
>  
> @@ -64,31 +69,48 @@ class ThreadManager(object):
>             be called when a thread exits, or there will be no way to get a
>             handle on it.
>          """
> -        self._objs.pop(name)
> +        with self._objs_lock:
> +            self._objs.pop(name)
>  
>      def exists(self, name):
>          """Determine if a thread or process exists with the given name."""
> -        return name in self._objs
> +
> +        # thread in the ThreadManager only officially exists once started
> +        with self._objs_lock:
> +            return name in self._objs
>  
>      def get(self, name):
>          """Given an object name, see if it exists and return the object.
>             Return None if no such object exists.  Additionally, this method
>             will re-raise any uncaught exception in the thread.
>          """
> -        obj = self._objs.get(name)
> -        if obj:
> -            self.raise_error(name)
>  
> -        return obj
> +        # without the lock it would be possible to get & join
> +        # a thread that was not yet started
> +        with self._objs_lock:
> +            obj = self._objs.get(name)
> +            if obj:
> +                self.raise_if_error(name)
> +
> +            return obj
>  
>      def wait(self, name):
>          """Wait for the thread to exit and if the thread exited with an error
>             re-raise it here.
>          """
> -        if self.exists(name):
> +        # we don't need a lock here,
> +        # because get() acquires it itself
> +        try:
>              self.get(name).join()
> +        except AttributeError:
> +            pass
> +        # - if there is a thread object for the given name,
> +        #   we join it
> +        # - if there is not a thread object for the given name,
> +        #   we get None, try to join it, suppress the AttributeError
> +        #   and return immediately
>  
> -        self.raise_error(name)
> +        self.raise_if_error(name)
>  
>      def wait_all(self):
>          """Wait for all threads to exit and if there was an error re-raise it.
> @@ -116,7 +138,7 @@ class ThreadManager(object):
>          """
>          return any(self._errors.values())
>  
> -    def raise_error(self, name):
> +    def raise_if_error(self, name):
>          """If a thread has failed due to an exception, raise it into the main
>             thread.
>          """
> @@ -136,7 +158,8 @@ class ThreadManager(object):
>              :returns: number of running threads
>              :rtype:   int
>          """
> -        return len(self._objs)
> +        with self._objs_lock:
> +            return len(self._objs)
>  
>      @property
>      def names(self):
> @@ -145,7 +168,8 @@ class ThreadManager(object):
>              :returns: list of thread names
>              :rtype:   list of strings
>          """
> -        return self._objs.keys()
> +        with self._objs_lock:
> +            return self._objs.keys()
ACK.

-- 
Vratislav Podzimek

Anaconda Rider | Red Hat, Inc. | Brno - Czech Republic



More information about the anaconda-patches mailing list