Recipe9.3.Using a Queue.Queue as a Priority Queue

Recipe 9.3. Using a Queue.Queue as a Priority Queue

Credit: Simo Salminen, Lee Harr, Mark Moraes, Chris Perkins, Greg Klanderman

Problem

You want to use a Queue.Queue instance, since it is the best way to communicate among threads. However, you need the additional functionality of being able to specify a priority value associated with each item on the queue, so that items with a lower (more urgent) priority value are fetched before others with a higher (less urgent) priority value.

Solution

Among its many advantages, Queue.Queue offers an elegant architecture that eases subclassing for purposes of specializing queueing behavior. Specifically, Queue.Queue exposes several methods specifically designed to be overridden in a subclass, to get specialized queueing behavior without worrying about synchronization issues.

We can exploit this elegant architecture and module heapq from the Python Standard Library to build the needed priority-queue functionality pretty easily. However, we also need to shadow and wrap Queue.Queue's put and get methods, to decorate each item with its priority and posting time upon put, and strip off these decorations upon get:

import Queue, heapq, time class PriorityQueue(Queue.Queue):    # Initialize the queue     def _init(self, maxsize):         self.maxsize = maxsize         self.queue = [  ]     # Return the number of items that are currently enqueued     def _qsize(self):         return len(self.queue)     # Check whether the queue is empty     def _empty(self):         return not self.queue     # Check whether the queue is full     def _full(self):         return self.maxsize > 0 and len(self.queue) >= self.maxsize     # Put a new item in the queue     def _put(self, item):         heapq.heappush(self.queue, item)     # Get an item from the queue     def _get(self):         return heapq.heappop(self.queue)     # shadow and wrap Queue.Queue's own `put' to allow a 'priority' argument     def put(self, item, priority=0, block=True, timeout=None):         decorated_item = priority, time.time( ), item         Queue.Queue.put(self, decorated_item, block, timeout)     # shadow and wrap Queue.Queue's own `get' to strip auxiliary aspects     def get(self, block=True, timeout=None):         priority, time_posted, item = Queue.Queue.get(self, block, timeout)         return item

Discussion

Given an instance q of this recipe's PriorityQueue class, you can call q.put(anitem) to enqueue an item with "normal" priority (here defined as 0), or q.put(anitem, prio) to enqueue an item with a specific priority prio. At the time q.get() gets called (presumably in another thread), items with the lowest priority will be returned first, bypassing items with higher priority. Negative priorities are lower than "normal", thus suitable for "urgent" items; positive priorities, higher than "normal", indicate items that may wait longer, since other items with "normal" priority will get fetched before them. Of course, if you're not comfortable with this conception of priorities, nothing stops you from altering this recipe's code accordingly: for example, by changing sign to the priority value when you build the decorated_item at the start of method put. If you do so, items posted with positive priority will become the urgent ones and items posted with negative priority will become the can-wait-longer ones.

Queue.Queue's architecture deserves study, admiration, and imitation. Not only is Queue.Queue, all on its own, the best way to architect communication among threads, but this same class is also designed to make it easy for you to subclass and specialize it with queueing disciplines different from its default FIFO (first-in, first-out), such as the priority-based queueing discipline implemented in this recipe. Specifically, Queue.Queue uses the wonderful Template Method Design Pattern (http://www.aleax.it/Python/os03_template_dp.pdf ). This DP enables Queue.Queue itself to take care of the delicate problems connected with locking, while delegating the queueing discipline to specific methods _put, _get, and so on, which may be overridden by subclasses; such hook methods then get called in a context where synchronization issues are not a concern.

In this recipe, we also need to override Queue.Queue's put and get methods, because we need to add a priority optional argument to put's signature, decorate the item before we put it on the queue (so that the heapq module's mechanisms will produce the order we wantlowest priority first, and, among items posted with equal priority, FIFO ordering), and undecorate each decorated item that we get back from the queue to return the naked item. All of these auxiliary tweaks use nothing but local variables, however, so they introduce no synchronization worries whatsoever. Each thread gets its own stack; therefore, any code that uses nothing but local variables (and thus cannot possibly alter any state accessible from other threads, or access any state that other threads might alter) is inherently thread-safe.

Recipe9.3.Using a Queue.Queue as a Priority Queue