Examples¶
Interprocess locks¶
Note
Launch multiple of these at the same time to see the lock(s) in action.
Warning
There are no guarantees regarding usage by multiple threads in a single process with these locks (you will have to ensure single process safety yourself using traditional thread based locks). In other words this lock works only between processes.
import time
import fasteners
@fasteners.interprocess_locked('/tmp/tmp_lock_file')
def test():
for i in range(10):
print('I have the lock')
time.sleep(1)
print('Waiting for the lock')
test()
import time
import fasteners
def test():
for i in range(10):
with fasteners.InterProcessLock('/tmp/tmp_lock_file'):
print('I have the lock')
time.sleep(1)
test()
import time
import fasteners
def test():
a_lock = fasteners.InterProcessLock('/tmp/tmp_lock_file')
for i in range(10):
gotten = a_lock.acquire(blocking=False)
try:
if gotten:
print('I have the lock')
time.sleep(0.2)
else:
print('I do not have the lock')
time.sleep(0.1)
finally:
if gotten:
a_lock.release()
test()
Lock decorator¶
import threading
import fasteners
class NotThreadSafeThing(object):
def __init__(self):
self._lock = threading.Lock()
@fasteners.locked
def do_something(self):
print("Doing something in a thread safe manner")
o = NotThreadSafeThing()
o.do_something()
Multi-lock decorator¶
import threading
import fasteners
class NotThreadSafeThing(object):
def __init__(self):
self._locks = [threading.Lock(), threading.Lock()]
@fasteners.locked(lock='_locks')
def do_something(self):
print("Doing something in a thread safe manner")
o = NotThreadSafeThing()
o.do_something()
Try lock¶
import threading
import fasteners
t = threading.Lock()
with fasteners.try_lock(t) as gotten:
if gotten:
print("I got the lock")
else:
print("I did not get the lock")