python – efficient circular buffer?

python – efficient circular buffer?

I would use collections.deque with a maxlen arg

>>> import collections
>>> d = collections.deque(maxlen=10)
>>> d
deque([], maxlen=10)
>>> for i in xrange(20):
...     d.append(i)
>>> d
deque([10, 11, 12, 13, 14, 15, 16, 17, 18, 19], maxlen=10)

There is a recipe in the docs for deque that is similar to what you want. My assertion that its the most efficient rests entirely on the fact that its implemented in C by an incredibly skilled crew that is in the habit of cranking out top notch code.

Although there are already a great number of great answers here, I could not find any direct comparison of timings for the options mentioned. Therefore, please find my humble attempt at a comparison below.

For testing purposes only, the class can switch between a list-based buffer, a collections.deque-based buffer, and a Numpy.roll-based buffer.

Note that the update method adds only one value at a time, to keep it simple.

import numpy
import timeit
import collections

class CircularBuffer(object):
    buffer_methods = (list, deque, roll)

    def __init__(self, buffer_size, buffer_method):
        self.content = None
        self.size = buffer_size
        self.method = buffer_method

    def update(self, scalar):
        if self.method == self.buffer_methods[0]:
            # Use list
            except AttributeError:
                self.content = [0.] * self.size
        elif self.method == self.buffer_methods[1]:
            # Use collections.deque
            except AttributeError:
                self.content = collections.deque([0.] * self.size,
        elif self.method == self.buffer_methods[2]:
            # Use Numpy.roll
                self.content = numpy.roll(self.content, -1)
                self.content[-1] = scalar
            except IndexError:
                self.content = numpy.zeros(self.size, dtype=float)

# Testing and Timing
circular_buffer_size = 100
circular_buffers = [CircularBuffer(buffer_size=circular_buffer_size,
                    for method in CircularBuffer.buffer_methods]
timeit_iterations = 1e4
timeit_setup = from __main__ import circular_buffers
timeit_results = []
for i, cb in enumerate(circular_buffers):
    # We add a convenient number of convenient values (see equality test below)
    code = [circular_buffers[{}].update(float(j)) for j in range({})].format(
        i, circular_buffer_size)
    # Testing
    buffer_content = [item for item in cb.content]
    assert buffer_content == range(circular_buffer_size)
    # Timing
        timeit.timeit(code, setup=timeit_setup, number=int(timeit_iterations)))
    print {}: total {:.2f}s ({:.2f}ms per iteration).format(
        cb.method, timeit_results[-1],
        timeit_results[-1] / timeit_iterations * 1e3)

On my system this yields:

list:  total 1.06s (0.11ms per iteration)
deque: total 0.87s (0.09ms per iteration)
roll:  total 6.27s (0.63ms per iteration)

python – efficient circular buffer?

popping from the head of a list causes the whole list to be copied, so is inefficient

You should instead use a list/array of fixed size and an index which moves through the buffer as you add/remove items

Leave a Reply

Your email address will not be published. Required fields are marked *