public
anonymous / dogpiletest.py
Created

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

  • Download Gist
dogpiletest.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63
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)

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.