Is this code thread-safe? 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() &gt; 0:
        l, r = out_queue.get() #unpack tuples of Stock, int
        tmp[l] = r #tmp[&lt;Stock&gt;] = 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
"reasonably safe against deadlocks" is a strange expression :)
In ThreadsnQueues the if func is not None: can be turned into an else.
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.
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.
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?
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?