queues.rst

Queues

Python :doc:`Java <queues-java>`

Challenge

Allow efficient operations on a shared queue by multiple clients acting concurrently.

Explanation

We can model a queue by assigning increasing integers that encode the order of items. To minimize conflicts for concurrent operations, we combine the integers in a tuple with a random element to make the final key unique.

Ordering

The ordering of keys preserves the FIFO order of items and therefore lets us identify the next item to be dequeued without maintaining a pointer to it.

Pattern

We store each item in the queue within a subspace, which takes care of packing our integer indexes into byte strings.

queue = fdb.Subspace(('Q',))

As a first cut, we could store each item with a single key-value pair using increasing integer indexes for subsequent items:

tr[queue[index]] = value

However, this would leave concurrent enqueue operations vulnerable to conflicts. To minimize these conflicts, we can add a random integer to the key.

tr[queue[index][random_int]] = value

With this data model, items enqueued concurrently may be assigned the same index, but the keys as a whole will still be ordered (in this case, randomly). By using a :ref:`snapshot read <snapshot isolation>`, we guarantee that enqueuing will be conflict-free.

To implement this model, we need an efficient way of finding the first and last index presently in use. FoundationDB's range reads have limit and reverse options that let us accomplish this. Given the range of the subspace:

r = queue.range()

we can find the first and last key-value pairs in the range with:

tr.get_range(r.start, r.stop, limit=1) # first
tr.get_range(r.start, r.stop, limit=1, reverse=True) # last

Extensions

High-Contention Dequeue Operations

To minimize conflicts during dequeue operations, we can use a staging technique to service the requests. If a dequeue operation doesn't initially succeed, it registers a dequeue request in a semi-ordered set of such requests. It then enters a retry loop in which it attempts to fulfill outstanding requests.

Code

The following is a simple implementation of the basic pattern:

import os

queue = fdb.Subspace(('Q',))

@fdb.transactional
def dequeue(tr):
    item = first_item(tr)
    if item is None: return None
    del tr[item.key]
    return item.value

@fdb.transactional
def enqueue(tr, value):
    tr[queue[last_index(tr) + 1][os.urandom(20)]] = value

@fdb.transactional
def last_index(tr):
    r = queue.range()
    for key, _ in tr.snapshot.get_range(r.start, r.stop, limit=1, reverse=True):
        return queue.unpack(key)[0]
    return 0

@fdb.transactional
def first_item(tr):
    r = queue.range()
    for kv in tr.get_range(r.start, r.stop, limit=1):
        return kv