I needed to impose on a Python method the following locking semantics: the method can only be run by one thread at a time. If thread B tries to run the method while it is already being run by thread A, then thread B immediately receives a return value of None
.
I wrote the following decorator to apply these semantics:
from threading import Lock
def non_blocking_lock(fn):
fn.lock = Lock()
@wraps(fn)
def locker(*args, **kwargs):
if fn.lock.acquire(False):
try:
return fn(*args, **kwargs)
finally:
fn.lock.release()
return locker
This works in my testing so far. Does anyone notice any gotchas or have any suggestions for improvements?
Revised version
After the suggestions by @RemcoGerlich, I have added a docstring and kept the lock local to the decorator:
from threading import Lock
def non_blocking_lock(fn):
"""Decorator. Prevents the function from being called multiple times simultaneously.
If thread A is executing the function and thread B attempts to call the
function, thread B will immediately receive a return value of None instead.
"""
lock = Lock()
@wraps(fn)
def locker(*args, **kwargs):
if lock.acquire(False):
try:
return fn(*args, **kwargs)
finally:
lock.release()
return locker