Script I've been using to test background_runner with dogpile.cache

dogpiletest.py

from __future__ import print_function

from dogpile.cache import make_region
import time
import random
import threading

log_lock = threading.Lock()


def log(*args):
    """ This exists just to make print statements readable. """
    with log_lock:
        identity = threading.current_thread().ident
        print(identity, *args)


def background_runner(mutex, creator):
    log(repr(creator))
    def f():
        try:
            creator()
        finally:
            mutex.release()

    t = threading.Thread(target=f)
    t.start()


region = make_region(
    background_runner=background_runner,
).configure(
    'dogpile.cache.memcached',
    expiration_time=5,
    arguments={
        'url': '127.0.0.1:11211',
        'distributed_lock': True,
    }
)


@region.cache_on_arguments()
def my_func():
    log("Really calling func.  sleeping.")
    time.sleep(5)
    value = random.randint(0, 100)
    log("Waking.  Cache gets new val", value)
    return value


if __name__ == '__main__':
    for i in range(3):
        log("Calling.  This is try", i)
        value = my_func()
        log("Returned with", value)

    log("Waiting the expiration time")
    time.sleep(5)

    for i in range(3, 6):
        log("Calling.  This is try", i)
        value = my_func()
        log("Returned with", value)