Persistent Queue implementations?

[Karl A. Krueger]

Kevin Altis <altis at semi-retired.com> wrote:
> I think you should be able to do what you want with a subclass of the Queue
> class using the pickle module. For example, here's a test in the shell just
> using pickle with get() and put() from the Queue instance.

This example seems to be pickling objects as they come off the queue.
What I'm after is, rather, a queue that persists over program restarts,
so I can write a multithreaded daemon program that persistently keeps a
queue of things it needs to do.

Consider a mail server's mail queues, for one example -- except mail
queues are kept -entirely- on disk, with only perhaps an index in memory.


> So, in the __init__ method of your Queue subclass you'll need to have a
> try/except block that opens the filename you pass into the init and in the
> except just initialize a new Queue to deal with when the file doesn't exist.
> You'll need save and load methods for your subclass which basically just do
> what I've shown above.

Using pickle in the most obvious way (pickling the whole queue) would
have to be done in every put() and get() call, since the copy on disk
needs to reflect the copy in memory.  And pickling the whole queue would
be -slow-, since it would have to write the whole queue each time --
keeping it locked for read, and thus blocking all the threads that want
to write to it.

I'm looking a little closer at the shelve module.  I've worked with Perl
"tied hashes" before, and shelves seem to be similar.  The problem then
is wrapping a semaphore around the shelf so it's thread safe, and
putting a Queue-like interface around it.  Something like:

import threading, shelve
class QueueShelf:
	def __init__(self, file):
		# Open shelf and create a queue list on it.
		self.shelf = shelve.open(file)
		if not self.shelf.has_key('queue'):
			self.shelf['queue'] = []
		# Make us a semaphore.
		self.sem = threading.Semaphore()
	def __del__(self):
		# If we're lucky enough to get finalized, close the shelf.
		self.shelf.close()
	def put(self, obj, block=0):
		# Works like Queue.put ... can be block or non-block.
		self.sem.acquire(block)
		self.shelf['queue'] = self.shelf['queue'] + [obj]
		self.sem.release()
	def get(self, block=0):
		# Works like Queue.get ... can be block or non-block.
		self.sem.acquire(block)
		obj = self.shelf['queue'][-1]
		self.shelf['queue'] = self.shelf['queue'][:-1]
		self.sem.release()
		return obj

This, too, involves making copies of the queue ... and is messy in a few
other places, too.

-- 
Karl A. Krueger <kkrueger at example.edu>
Woods Hole Oceanographic Institution
Email address is spamtrapped.  s/example/whoi/
"Outlook not so good." -- Magic 8-Ball Software Reviews