Simple test for dogpile.cache concurrent refresh.

test.py

import time
import threading

import dogpile.cache

flag = 0
results = [0, 0]
lock = threading.RLock()

region = dogpile.cache.CacheRegion().configure("dogpile.cache.memory")


def main():
    @region.cache_on_arguments(expiration_time=60)
    def create():
        global flag
        global lock
        global results
        with lock:
            orig_flag = flag
            if flag == 0:
                flag += 1

        if orig_flag == 0:
            time.sleep(1)

        results[orig_flag] = time.time()
        print orig_flag, results[orig_flag]
        return results[orig_flag]

    create.invalidate()

    threads = [threading.Thread(target=create.refresh) for _ in range(2)]
    for t in threads:
        t.start()
        time.sleep(0.1)

    for t in threads:
        t.join()

    assert(create() == results[1])


if __name__ == "__main__":
    main()

test_multi.py

import time
import threading

import dogpile.cache

flag = 0
results = [0, 0]
lock = threading.RLock()

region = dogpile.cache.CacheRegion().configure("dogpile.cache.memory")


def main():
    @region.cache_multi_on_arguments(expiration_time=60)
    def create(*args):
        global flag
        global lock
        global results
        with lock:
            orig_flag = flag
            if flag == 0:
                flag += 1

        if orig_flag == 0:
            time.sleep(1)

        results[orig_flag] = time.time()
        print orig_flag, results[orig_flag]
        return [(arg, results[orig_flag]) for arg in args]

    create.invalidate(1, 2)

    threads = [threading.Thread(target=create.refresh, args=(1, 2))
               for _ in range(2)]
    for t in threads:
        t.start()
        time.sleep(0.1)

    for t in threads:
        t.join()

    assert(create(1)[0][1] == results[1])
    assert(create(2)[0][1] == results[1])


if __name__ == "__main__":
    main()