test_deque_pipeline.py

from collections import deque
import time

from statsd import defaults
from statsd.client import StatsClient


class DequePipeline(StatsClient):
    def __init__(self, client):
        self._client = client
        self._prefix = client._prefix
        self._maxudpsize = client._maxudpsize
        self._stats = deque()

    def _after(self, data):
        if data is not None:
            self._stats.append(data)

    def __enter__(self):
        return self

    def __exit__(self, typ, value, tb):
        self.send()

    def send(self):
        # Use popleft to preserve the order of the stats.
        if not self._stats:
            return
        data = self._stats.popleft()
        while self._stats:
            stat = self._stats.popleft()
            if len(stat) + len(data) + 1 >= self._maxudpsize:
                self._client._after(data)
                data = stat
            else:
                data += '\n' + stat
        self._client._after(data)


class CustomStatsClient(StatsClient):

    def pipeline(self):
        return DequePipeline(self)

stock_client = StatsClient('127.0.0.1', defaults.PORT, defaults.PREFIX,
                           defaults.MAXUDPSIZE)
def test_stock():
    # 16 patterns
    with stock_client.pipeline() as pipe:
        begin = time.time()
        for i in range(15):
            pipe.incr('foo.bar')
        dt = (time.time() - begin) * 1000
        pipe.timing('foo.baz', dt)


deque_client = CustomStatsClient('127.0.0.1', defaults.PORT,
                                 defaults.PREFIX, defaults.MAXUDPSIZE)
def test_deque():
    # 16 patterns
    with deque_client.pipeline() as pipe:
        begin = time.time()
        for i in range(15):
            pipe.incr('foo.bar')
        dt = (time.time() - begin) * 1000
        pipe.timing('foo.baz', dt)

if __name__ == '__main__':
    import timeit
    print('Stock Pipeline (1000000 iterations):')
    print(timeit.timeit("test_stock()",
                        setup="from __main__ import test_stock"))
    print('Deque Pipeline (1000000 iterations):')
    print(timeit.timeit("test_deque()",
                        setup="from __main__ import test_deque"))