Return to Revisions

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

1. 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)

1. 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):

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

2. 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.)

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

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).

1. 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.

1. 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.