queue documentation

Directory based queue.

A port of Perl module Directory::Queue http://search.cpan.org/dist/Directory-Queue/ The documentation from Directory::Queue module was adapted for Python.

The goal of this module is to offer a simple queue system using the underlying filesystem for storage, security and to prevent race conditions via atomic operations. It focuses on simplicity, robustness and scalability.

This module allows multiple concurrent readers and writers to interact with the same queue.

Provides

Classes:

Documentation

Queue class

dirq.queue.Queue - directory based queue.

Usage:

from dirq.queue import Queue

# simple schema:
#  - there must be a "body" which is a string
#  - there can be a "header" which is a table/dictionary

schema = {"body": "string", "header": "table?"}
queuedir = "/tmp/test"

# sample producer

dirq = Queue(queuedir, schema=schema)
import os
for count in range(1,101):
    name = dirq.add({"body"  : "element %i"%count,
                     "header": dict(os.environ)})
    print("# added element %i as %s" % (count, name))

# sample consumer

dirq = Queue(queuedir, schema=schema)
name = dirq.first()
while name:
    if not dirq.lock(name):
        name = dirq.next()
        continue
    print("# reading element %s" % name)
    data = dirq.get(name)
    # one can use data['body'] and data['header'] here...
    # one could use dirq.unlock(name) to only browse the queue...
    dirq.remove(name)
    name = dirq.next()

Terminology

An element is something that contains one or more pieces of data. A simple string may be an element but more complex schemas can also be used, see the Schema section for more information.

A queue is a “best effort FIFO” collection of elements.

It is very hard to guarantee pure FIFO behavior with multiple writers using the same queue. Consider for instance:

. Writer1: calls the add() method . Writer2: calls the add() method . Writer2: the add() method returns . Writer1: the add() method returns

Who should be first in the queue, Writer1 or Writer2?

For simplicity, this implementation provides only “best effort FIFO”, i.e. there is a very high probability that elements are processed in FIFO order but this is not guaranteed. This is achieved by using a high-resolution time function and having elements sorted by the time the element’s final directory gets created.

Locking

Adding an element is not a problem because the add() method is atomic.

In order to support multiple processes interacting with the same queue, advisory locking is used. Processes should first lock an element before working with it. In fact, the get() and remove() methods raise an exception if they are called on unlocked elements.

If the process that created the lock dies without unlocking the ele- ment, we end up with a staled lock. The purge() method can be used to remove these staled locks.

An element can basically be in only one of two states: locked or unlocked.

A newly created element is unlocked as a writer usually does not need to do anything more with the element once dropped in the queue.

Iterators return all the elements, regardless of their states.

There is no method to get an element state as this information is usu- ally useless since it may change at any time. Instead, programs should directly try to lock elements to make sure they are indeed locked.

Constructor

For the signature of the Queue constructor see documentation to the respective __init__() method.

Schema

The schema defines how user supplied data is stored in the queue. It is only required by the add() and get() methods.

The schema must be a dictionary containing key/value pairs.

The key must contain only alphanumerical characters. It identifies the piece of data and will be used as file name when storing the data inside the element directory.

The value represents the type of the given piece of data. It can be:

binary
the data is a sequence of binary bytes, it will be stored directly in a plain file with no further encoding
string
the data is a text string (i.e. a sequence of characters), it will be UTF-8 encoded
table
the data is a reference to a hash of text strings, it will be seri- alized and UTF-8 encoded before being stored in a file

By default, all pieces of data are mandatory. If you append a question mark to the type, this piece of data will be marked as optional. See the comments in the Usage section for more information.

To comply with Directory::Queue implementation it is allowed to append ‘*’ (asterisk) to data type specification, which in Directory::Queue means switching to working with element references in add() and get() operations. This is irrelevant for the Python implementation.

Directory Structure

All the directories holding the elements and all the files holding the data pieces are located under the queue toplevel directory. This direc- tory can contain:

temporary
the directory holding temporary elements, i.e. the elements being added
obsolete
the directory holding obsolete elements, i.e. the elements being removed
NNNNNNNN
an intermediate directory holding elements; NNNNNNNN is an 8-digits long hexadecimal number

In any of the above directories, an element is stored as a single directory with a 14-digits long hexadecimal name SSSSSSSSMMMMMR where:

SSSSSSSS
represents the number of seconds since the Epoch
MMMMM
represents the microsecond part of the time since the Epoch

R is a random digit used to reduce name collisions

Finally, inside an element directory, the different pieces of data are stored into different files, named according to the schema. A locked element contains in addition a directory named “locked”.

Security

There are no specific security mechanisms in this module.

The elements are stored as plain files and directories. The filesystem security features (owner, group, permissions, ACLs...) should be used to adequately protect the data.

By default, the process’ umask is respected. See the class constructor documentation if you want an other behavior.

If multiple readers and writers with different uids are expected, the easiest solution is to have all the files and directories inside the toplevel directory world-writable (i.e. umask=0). Then, the permissions of the toplevel directory itself (e.g. group-writable) are enough to control who can access the queue.

QueueSet class

dirq.queue.QueueSet - interface to a set of Queue objects

Usage:

from dirq.queue import Queue, QueueSet

dq1 = Queue("/tmp/q1")
dq2 = Queue("/tmp/q2")
dqset = QueueSet(dq1, dq2)
# dqs = [dq1, dq2]
# dqset = QueueSet(dqs)

(dq, elt) = dqset.first()
while dq:
    # you can now process the element elt of queue dq...
    (dq, elt) = dqset.next()

Description

This class can be used to put different queues into a set and browse them as one queue. The elements from all queues are merged together and sorted independently from the queue they belong to.

Constructor

For the signature of the QueueSet constructor see documentation to the respective dirq.queue.QueueSet.__init__() method.

Author

Konstantin Skaburskas <konstantin.skaburskas@gmail.com>