Skip to content

anonymous /dogpiletest.py

Embed URL

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Script I've been using to test background_runner with dogpile.cache
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.