Persistent Queue implementations?

Karl A. Krueger <kkrueger at example.edu> wrote:
> I'm working on a cPickle-based queue now ... will post it when it's done> (as well as a corrected version of the shelf-based queue I posted, with> more of the Queue protocol working).

I shan't bother with the shelf, since Kevin Altis's advice was the right
way to do it.  Here's my cPickle-based queue, which appears to be both
the right thing and fast as heck.  I don't diddle around with making my
own semaphores -- inheriting from Queue.Queue does the job:


# PickleQueue -- persistent queue using cPickle

import Queue, cPickle

class PickleQueue(Queue.Queue):
    """A multi-producer, multi-consumer, persistent queue."""

    def __init__(self, filename, maxsize=0):
        """Initialize a persistent queue with a filename and maximum size.

        The filename is used as a persistent data store for the queue.
        If maxsize <= 0, the queue size is infinite.
        """
        self.filename = filename
        Queue.Queue.__init__(self, maxsize)
        if self.queue:
            self.esema.release()
        if self._full():
            self.fsema.acquire()

    def _init(self, maxsize):
        # Implements Queue protocol _init for persistent queue.
        # Sets up the pickle files.
        self.maxsize = maxsize
        try:
            self.readfile = file(self.filename, 'r')
            self.queue = cPickle.load(self.readfile)
            self.readfile.close()
        except IOError, err:
            if err.errno == 2:
                # File doesn't exist, continue ...
                self.queue = []
            else:
                # Some other I/O problem, reraise error
                raise err
        except EOFError:
            # File was null?  Continue ...
            self.queue = []

        # Rewrite file, so it's created if it doesn't exist,
        # and raises an exception now if we aren't allowed
        self.writefile = file(self.filename, 'w')
        cPickle.dump(self.queue, self.writefile, 1)

    def __sync(self):
        # Writes the queue to the pickle file.
        self.writefile.seek(0)
        cPickle.dump(self.queue, self.writefile, 1)
        self.writefile.flush()

    def _put(self, item):
        # Implements Queue protocol _put for persistent queue.
        self.queue.append(item)
        self.__sync()

    def _get(self):
        # Implements Queue protocol _get for persistent queue.
        item = self.queue[0]
        del self.queue[0]
        self.__sync()
        return item


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