We can't run this code. Where are the import statements for Queue and Empty?
As ChrisWue points out, there's no such thing as "reasonably safe against deadlocks". Either you're safe or you're not.
There's no documentation for the ThreadsnQueues and ThreadScore classes. What do these classes do and how am I supposed to use them?
What does the name ThreadsnQueues even mean?
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.
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
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)
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):
In ThreadsnQueues.__init__ you initialize a member self.args, but this is never used.
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.)
Your use of StandardError limits your code to Python 2.
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).
- 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.
- 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.)
