plot_early_recompute.py

"""Early Recomputation Measurements
"""

import functools as ft
import multiprocessing.pool
import shutil
import threading
import time

import diskcache as dc


def make_timer(times):
    """Make a decorator which accumulates (start, end) in `times` for function
    calls.

    """
    lock = threading.Lock()

    def timer(func):
        @ft.wraps(func)
        def wrapper(*args, **kwargs):
            start = time.time()
            func(*args, **kwargs)
            pair = start, time.time()
            with lock:
                times.append(pair)

        return wrapper

    return timer


def make_worker(times, delay=0.2):
    """Make a worker which accumulates (start, end) in `times` and sleeps for
    `delay` seconds.

    """

    @make_timer(times)
    def worker():
        time.sleep(delay)

    return worker


def make_repeater(func, total=10, delay=0.01):
    """Make a repeater which calls `func` and sleeps for `delay` seconds
    repeatedly until `total` seconds have elapsed.

    """

    def repeat(num):
        start = time.time()
        while time.time() - start < total:
            func()
            time.sleep(delay)

    return repeat


def frange(start, stop, step=1e-3):
    """Generator for floating point values from `start` to `stop` by `step`."""
    while start < stop:
        yield start
        start += step


def plot(option, filename, cache_times, worker_times):
    """Plot concurrent workers and latency."""
    import matplotlib.pyplot as plt

    fig, (workers, latency) = plt.subplots(2, sharex=True)

    fig.suptitle(option)

    changes = [(start, 1) for start, _ in worker_times]
    changes.extend((stop, -1) for _, stop in worker_times)
    changes.sort()
    start = (changes[0][0] - 1e-6, 0)
    counts = [start]

    for mark, diff in changes:
        # Re-sample between previous and current data point for a nicer-looking
        # line plot.

        for step in frange(counts[-1][0], mark):
            pair = (step, counts[-1][1])
            counts.append(pair)

        pair = (mark, counts[-1][1] + diff)
        counts.append(pair)

    min_x = min(start for start, _ in cache_times)
    max_x = max(start for start, _ in cache_times)
    for step in frange(counts[-1][0], max_x):
        pair = (step, counts[-1][1])
        counts.append(pair)

    x_counts = [x - min_x for x, y in counts]
    y_counts = [y for x, y in counts]

    workers.set_title('Concurrency')
    workers.set_ylabel('Workers')
    workers.set_ylim(0, 11)
    workers.plot(x_counts, y_counts)

    latency.set_title('Latency')
    latency.set_ylabel('Seconds')
    latency.set_ylim(0, 0.5)
    latency.set_xlabel('Time')
    x_latency = [start - min_x for start, _ in cache_times]
    y_latency = [stop - start for start, stop in cache_times]
    latency.scatter(x_latency, y_latency)

    plt.savefig(filename)


def main():
    shutil.rmtree('/tmp/cache')
    cache = dc.Cache('/tmp/cache')

    count = 10

    cache_times = []
    timer = make_timer(cache_times)

    options = {
        ('No Caching', 'no-caching.png'): [
            timer,
        ],
        ('Traditional Caching', 'traditional-caching.png'): [
            timer,
            cache.memoize(expire=1),
        ],
        ('Synchronized Locking', 'synchronized-locking.png'): [
            timer,
            cache.memoize(expire=0),
            dc.barrier(cache, dc.Lock),
            cache.memoize(expire=1),
        ],
        ('Early Recomputation', 'early-recomputation.png'): [
            timer,
            dc.memoize_stampede(cache, expire=1),
        ],
        ('Early Recomputation (beta=0.5)', 'early-recomputation-05.png'): [
            timer,
            dc.memoize_stampede(cache, expire=1, beta=0.5),
        ],
        ('Early Recomputation (beta=0.3)', 'early-recomputation-03.png'): [
            timer,
            dc.memoize_stampede(cache, expire=1, beta=0.3),
        ],
    }

    for (option, filename), decorators in options.items():
        print('Simulating:', option)
        worker_times = []
        worker = make_worker(worker_times)
        for decorator in reversed(decorators):
            worker = decorator(worker)

        worker()
        repeater = make_repeater(worker)

        with multiprocessing.pool.ThreadPool(count) as pool:
            pool.map(repeater, [worker] * count)

        plot(option, filename, cache_times, worker_times)

        cache.clear()
        cache_times.clear()


if __name__ == '__main__':
    main()