Is this code thread-safe?

Question

EDIT: The full source code in question can be found here.

Is this code reasonably safe against dead/livelocks, and if not, how can I fix it?

• I know processes are usually recommended over threads for operations like this, but processing's version of queue doesn't do what I need.
• The choose function will be called on 100 objects, but in sequence. I think each one should be finished before the next, but I'm not certain.

• The appraise function that choose calls only gets values from dicts, and does math on them.

  ---thread_queue.py---

  class ThreadsnQueues(threading.Thread):
      def __init__(self, queue, out_queue=None, func=None, args=None, semaphore=None):
          super(ThreadsnQueues, self).__init__()
          self.queue = queue
          self.out_queue = out_queue
          self.semaphore = semaphore
          if func is None:
              try:
                  func = self.queue.get(block=False)
              except Empty:
                  self.func = None
                  return
          if func is not None:
              if type(func) is list:
                  self.func = func[0]
                  self.args = func[1:]
              else:
                  self.func = func
          if args != [] and args != None: self.args = args
  
  class ThreadScore(ThreadsnQueues):
      def __init__(self, queue, out_queue, func, semaphore):
          super(ThreadScore, self).__init__(queue, out_queue, func, None, semaphore)
      def run(self):
          try:
              s = self.queue.get() #should be a stock object
          except Empty:
              print "Empty queue?!"
              return
          except:
              return
          while True:
              try:
                  self.semaphore.acquire()
                  ret = self.func(s) #should return a score (int)
                  self.out_queue.put((s, ret)) #tuple with Stock, int
                  self.queue.task_done()
                  self.semaphore.release()
                  return
              except KeyboardInterrupt:
                  raise
              except ArithmeticError:
                  a, b, c = exc_info()
                  print a, b
                  pdb.post_mortem(c)
              except StandardError:
                  pass
  

  ---genome.py---

  def choose(self, stocks):
      """
      Scores the whole list of stocks, and puts the top twelve on a list to be
      purchased.
      """
      tmp = {}
      queue = Queue.Queue()
      out_queue = Queue.Queue()
      procs = []
      thread_limit = 10
      self.prog = ProgressBar(widgets=[Percentage(), ' ', Bar(marker=RotatingMarker()), ' ', ETA()],
          maxval=len(stocks)).start()
      sym = threading.BoundedSemaphore(thread_limit)
      for stock in stocks:
          queue.put(stock)
      while queue.empty() != True:
          try:
              for j in range(thread_limit):
                  t = ThreadScore(queue=queue, out_queue=out_queue, func=self.appraise, semaphore=sym)
                  t.daemon = True
                  t.start()
                  procs.append(t)
          except ArithmeticError:
              raise
          except threrror:
              print "Encounted a thread error, blocking until threads complete..."
              [ p.join(5.0) for p in procs ]
              procs = []
          except KeyboardInterrupt:
              raise
          except StandardError as e:
              print e
              [ p.join(5.0) for p in procs ]
              procs = []
      [ p.join(timeout=5.0) for p in procs ]
      print "Finished scoring stocks, sorting..."
      self.prog.finish()
      while out_queue.qsize() > 0:
          l, r = out_queue.get() #unpack tuples of Stock, int
          tmp[l] = r #tmp[<Stock>] = score
      pmt = sorted(tmp, key=lambda x: tmp.get(x), reverse=True)
      self.prefers = { key: value for (key, value) in zip(pmt, sorted(tmp.values(), reverse=True))}
      self.scores = self.prefers.copy()
      return self.prefers
  

Answer 1

1. Comments on your code

1. We can't run this code. Where are the import statements for Queue and Empty?

2. As ChrisWue points out, there's no such thing as "reasonably safe against deadlocks". Either you're safe or you're not.

3. There's no documentation for the ThreadsnQueues and ThreadScore classes. What do these classes do and how am I supposed to use them?

4. What does the name ThreadsnQueues even mean?

5. I can't see what use there can possibly be for the ThreadsnQueues class. You don't override the run method, so this class is useless by itself.

6. What kind of objects are supposed to be in the queue? In ThreadsnQueues.__init__ you have:

   func = self.queue.get(block=False)
   

   which suggests that the objects in the queue are functions, but in ThreadScore.run you have:

   s = self.queue.get() #should be a stock object
   

7. ThreadsnQueues.__init__ calls its superclass method with no arguments:

   super(ThreadsnQueues, self).__init__()
   

   but in fact the threading.Thread class takes several keyword arguments (group, target, name, args, kwargs, daemon). By calling the superclass method with no arguments you make it impossible to use any of these features. This means that later on in the code you have to resort to writing:

   t.daemon = True
   

   because your interface doesn't allow you to pass the keyword argument daemon=True to your constructor.

   The proper way to handle this is for the subclass method to take arbitrary keyword arguments and pass them to the superclass method. Like this:

   def __init__(self, queue, out_queue=None, func=None, args=None, semaphore=None, **kwargs):
       super(ThreadsnQueues, self).__init__(**kwargs)
   

8. It's rarely correct in Python to insist on types matching exactly, like you do here:

   if type(func) is list:
   

   What you care about here is that func supports the sequence interface, and the way to test for that is:

   if isinstance(func, collections.abc.Sequence):
   

9. In ThreadsnQueues.__init__ you initialize a member self.args, but this is never used.

10. In ThreadsnQueues.__init__, semaphore defaults to None, but in ThreadScore.run you just call

    self.semaphore.acquire()
    

    which will raise AttributeError if self.semaphore is None. If a semaphore is required, you should detect its absence in the constructor and raise an exception. (Or make the semaphore argument a required positional argument instead of a keyword argument.)

11. Your use of StandardError limits your code to Python 2.

12. Ignoring generic classes of exceptions like this:

    except StandardError:
        pass
    

    is almost always a bad idea. When you get an unexpected exception, you need to be informed about it so that you can fix the problem that caused it. If you have to suppress a particular exception, do so locally around the code that might generate it (and explain why you are doing so).

13. This all seems way too complex to me. It looks as though you have a collection of stock objects, and you want to call self.appraise on each stock object (using a fixed-size pool of worker threads), wait for them all to complete, and then sort the results in reverse order by score.

    In Python 3 you'd accomplish this using the built-in concurrent.futures.ThreadPoolExecutor, like this:

    sorted(ThreadPoolExecutor(max_workers=thread_limit)
           .map(lambda s:(self.appraise(s), s), stocks, timeout=5.0),
           reverse=True)
    

    In Python 2 you can use the (poorly documented) multiprocessing.pool.ThreadPool.

    Why doesn't this work for you? You write, "I know processes are usually recommended over threads for operations like this, but processing's version of queue doesn't do what I need." But what exactly do you need? Perhaps if you explained the problem then we could see if your reasoning makes any sense.

14. Do you actually get any benefit from using threads here? If your appraise method spends most of its time running Python code, then it seems unlikely that you will get any benefit, because all the worker threads will queue up waiting for the global interpreter lock.

    (On the other hand, if appraise spends most of its time waiting for file or network I/O, then your approach makes sense.)

2. Responses to comments

ThreadsnQueues is an ABC and parent to both ThreadScore, and another class called ThreadTrue

It's no good explaining this to me now! Your code needs to make this clear to its readers. Think of the future when someone may need to maintain this code, and you might not be around to answer questions.

The name was just meant to say it worked with queues.

It did not convey that information to me. (Also, "work with queues" how?)

Isn't it sufficient to just keep testing my code, (e.g., by adding code to call sys.exc_info() and using that to run pdb's post mortem analysis, or simply capturing StandardException as e and printing it) and ignore exceptions for actual releases? It seems like if I run into an exception I've never seen before, it couldn't be handled by some premeditated heuristic.

My point is, what are you going to do if you get an unexpected exception in an actual release? At the moment you just suppress the exception and keep running. But that could be disastrous: maybe the exception is telling you something important that needs to be fixed, such as the computer running out of disk space, or some Python library getting corrupted, or who knows? By suppressing exceptions you prevent anyone being informed of problems in a timely manner.

The GIL probably is getting in the way. I began using threads to handle network and file I/O. This is likely out of their domain.

I had a quick look at the source for the appraise method and it executes database queries over a network connection, so it will spend some (maybe most) of its time waiting for the responses from the database, and so the GIL might not be a bottleneck.

P.S. Although it's out of scope for this code review, I couldn't help noticing this line:

vals = {key: conn.execute(select([a[1] for a in s.this.c.items() if a[0] != 'key']).where(s.this.c.key == key)).fetchall() for key in self.keys }

This approach is a bad idea: it executes a database query for every key. It would be much more efficient to execute one query that fetches the results for all the keys at once.

Answer 2

1. "reasonably safe against deadlocks" is a strange expression :)
2. In ThreadsnQueues the if func is not None: can be turned into an else.
3. Your code in ThreadScore indicates that some exceptions might be expected because you handle them and continue. The problem is if that happens while you have the semaphore acquired then it will not release it. You should move the release into finally block.
4. In choose you join on all threads in multiple places and this code [ p.join(5.0) for p in procs ] gets repeated several times. Should be refactored into a method.

5. I'm not sure what the point of the semaphore is. The way I read you code is:

   • You initialize the semaphore with N
   • Then you spawn N threads each calling acquire/release in it's run method.

   This means that all threads can simultaneously enter the block protected by the semaphore. So what's the point of having it?

6. In ThreadScore.run() the main part is the while True loop. However it never seems to take anything of the queue (it only seems to do it once at the beginning of the run method) so how does it work?