OddThinking

You can use your thread per work item in a slightly different way.

Create a thread that exits when you’re not processing something and the queue is empty. Then start it when you get a new item to log only if it’s not running already.

As long as you’re processing anything, you should have a thread running which will prevent your actual processing thread from being killed. When you aren’t processing, there won’t be such a thread and the program will be able to exit.

Pseudocode:

ConsumerThread:
. . while (true)
. . . . if (queue.empty)
. . . . . . ProtectorThread.Release
. . . . queue.GetOne
. . . . ProtectorThread.Protect
. . . . process item

ProtectorThread:
. . Protect:
. . . . clear semaphore
. . . . t = new Thread
. . Thread function:
. . . . Wait for a semaphore to be set.
. . Release:
. . . . set a semaphore to end the event


@configurator,

I did actually think about (but didn’t document) an approach where I’d start a consumer thread only if a consumer thread wasn’t already started, but I could see a fatal flaw in my design. I think your design is much more clever that what I came up with – it avoids both a plethora of threads and the client having to wait for them to start – but it might suffer the same flaw:

A application that is dying, and wants to log the cause, will push an item into the queue, and terminate. There is then a race condition – will the consumer thread get a time-slice, pull the record, call the protector thread and initiate the new Thread before the interpreter realises there are only daemon threads remaining, and terminates them all?

A way to fix this is to move the call to protect into the proxy, where it would be run from the client’s thread. Then, once again, the client needs to wait for the thread to be started before it can return. I still consider starting a thread to be expensive, but I haven’t actually profiled it for years.

Starting a thread is pretty quick AFAIK. At least in .Net it is, where threads are fake 🙂

The advantage is that the slowdown won’t be repeated with every logging request, as long as the queue is full. You could put a small delay when creating a thread in my example, to make sure that repeated logging is all done in one batch and a thread isn’t created and destroted for each call. Or perhaps a delay before releasing the thread as in:
if (queue.empty && queue.GetOne(timeout=100)) ProtectorThread.Release.
That way as long as the thread exists the delay on the logging caller will be minimal.

Configurator,

I believed you, but I didn’t really believe you, until I did my own (moderately dodgy) profiling test.

Python 2.6.4 on Vista (my dev machine, typical load), creating a thread took an average of 354 microseconds.

Python 2.6.4 on Ubuntu (target machine, minimum spec virtual private server, typical load), creating a thread took an average of 78 microseconds.

I’ve got to get over this out-dated aversion of waiting for a thread to start.

Thanks for the ideas. I think I came up with a solution.

When creating the consumer, pass it a handle to the parent thread:

c = Consumer(parent=threading.current_thread())


Then, in the consumer’s run loop you can check if the parent is still alive. If the parent is not alive, and the queue is empty, terminate:


class Consumer(threading.Thread):
. def __init__(self, parent):
. . threading.Thread.__init__(self)
. . . self.parent = parent
. . . self.q = Queue()
. def run(self):
. . while True:
. . . if not self.parent.isAlive() and self.q.empty():
. . . . break
. . else:
. . . pass #do stuff


If you have multiple producers, you could have a list of parents, and require them to all be dead. The consumer could then be a singleton, which adds a parent to the list each time its constructor is called.

Thanks for considering this, Smaddox.

If the “do stuff” section of your code is blocking, waiting for input to the queue, then the consumer thread will block forever when the parent dies, waiting for input.

If the “do stuff” section of your code is non-blocking, and the loop will occur ever when there is no input, then this becomes the same as the Polling section listed in the article above, with the same performance trade-off.

In another post, I pointed out the configurator’s idea, which I eventually implemented, may lead to a thread explosion.

In the comments, configurator expressed surprise:

I thought my suggestion would only create a thread while there isn’t one already alive? How would you have an exploding number of threads, then? You’d just have the one thread, busy in a pop-from-queue-and-send-email loop. […] Looking at the source code I see there is only one consumer thread that will ever get started. Maybe I missed something, but I fail to see the thread explosion.

Let me start with the easy part.

In the source code available here, there is a _delegate() method that is called each time a log message is handled. Its last line is this:

Thread(name="NonblockingLogHandler.protector",
    target=NonblockingLogHandler._protect,
    args=(self.message_queue,)).start()

So, each call to handle a log message results in a new Thread. If it gets in a tight loop, that is lots of new threads.

So the potential explosion is there and waiting.

Now, the hard part…

The hard part is deciding whether what I implemented should actually be attributed to configurator! In my mind, I implemented a small variation of configurator’s idea. Perhaps it was a bigger change than I realised, and configurator shouldn’t be blamed!

You can see in the comments above, I was clearly influenced by configurator’s suggestion that threads were cheap.

I also went for a solution that, inspired by configurator, involves a protector thread that does nothing but stall the runtime from shutting down the program before the messages were logged.

However, unlike configurator’s solution, which has a race condition, I have a protector thread per message, which survives until the message is emitted.

I cannot recall why I made such a significant change to multiple protector threads, rather than a single Protector thread, except that I do remember scribbling on pieces of paper trying to make sure I can covered all the potential race conditions that were sprinkled around. I convinced myself that the solution I adopted wasn’t going to have any such race conditions.

So, in conclusion:

Credit to configurator for the idea of fast threads and protector threads.

Credit/blame to me for the idea of thread-per-log-message.

And I am still open to better solutions, but be aware that this area is trickier than it looks.

Ah, I see, it’s the protector threads!
Could you point out if there’s a race condition I’m missing in this pseudo-code example:


ConsumerThread:
. . while (true)
. . . . if (queue.empty)
. . . . . . lock (locker)
. . . . . . . . if (queue.empty)
. . . . . . . . . . ProtectorThread.Release
. . . . queue.GetOne
. . . . process item

LogFunction:
. . lock (locker)
. . . . ProtectorThread.Protect
. . . . queue.Put(item)


Assume the lock function works like C#’s lock. The LogFunction will be quick except in the case of multiple run-away thread emitting log messages in an infinite loop. This is similar to the problem in your current design, but won’t gobble up system resources – it would just slow down the logging function (because the locker is saturated)

Also, doesn’t python have a Tasks or ThreadPool library?

I appreciate your continued interest. It is a tougher nut to crack than it should be.

Your pseudo-code is very similar in many ways to the actual implementation I have used (although I have used some language features, such as Queue.queue.join() that produces the same effect as your semaphore, without explicitly requiring one.

I believe my implementation and this one suffer the same downside: When you get multiple calls to the log function, it makes multiple calls (sequential, due to your lock statement) to the Protect method, which creates a new thread for each.

Now, the protect threads are consuming minimal system resources each, and, as proved before, can be created quickly, but enough of them being created quickly could theoretically cause problems. I’m sorry but I don’t see how this improves on the actual implementation.

Does Python have Tasks or ThreadPool libraries?

So many different ones, it makes your eyes water.

The basic building blocks are threading.Thread and queue.Queue, which together normally give you most of what you need.

What seems to be special with this logging application, is that the client never wants to see the result of the processing, and never closes down the pool cleanly, making the timing of the pool shutdown more ambiguous than it normally would be.

Multiprocessing is another built-in library, offering pools, but it is about processes, not threads, and therefore has larger overhead.

Many third-party libraries go the next step to offer process pools across multiple machines, using your favourite messaging stack.

Some smaller third-party threadpool libraries are available, such as this one that I believe abandons work when the client shuts-down, or this one which requires the client to explicitly close up the threadpool resources.

I haven’t found anything that has low-overhead and can handle shut-down of the client without notice, and then close itself down when it has finished the processing.

Sorry, I was thinking one thing and typing another. My example doesn’t improve anything, in fact. Let’s make a small change to the protect method.


ConsumerThread:
. . while (true)
. . . . if (queue.empty)
. . . . . . lock (locker)
. . . . . . . . if (queue.empty)
. . . . . . . . . . ProtectorThread.Release
. . . . queue.GetOne
. . . . process item

LogFunction:
. . lock (locker)
. . . . ProtectorThread.Protect
. . . . queue.Put(item)


Now, only one thread will ever exist. This was the entire point of the semaphore (and why you can’t use queue.join()) here – I just forgot to write down that line.

Note that since protect-and-insert-to-queue is inside a lock, and check-queue-is-empty-and-release is in another, we should be race-condition free, I think.

Nice. I stared at it for a while, and it looks good to me.

I had a play today with this suggestion. I wanted to use this technique, but I wasn’t entirely happy with the encapsulation – detailed knowledge about the solution has been spread over three different parts of the code.

I finally had a breakthrough.

I now have a subclass of queue.Queue, called autoclosingqueue.Queue. It encapsulates the semaphore, the lock and the protector thread away from both the producer and the consumer.

There is a requirement that the consumer call task_done(), which is an existing method on the queue.Queue object. Normally, it is optional to call this, depending on your design. With the new class, it is mandatory, so that the Protector thread can be notified when it is time to shut-down.

It is a reusable class for anyone who has this same, rather unusual, problem with consumer threads.

I will issue a new version of the NonblockingHandler containing this class, in due course.

Excellent. Let me know how it goes when you’ve got the new version ready.