I am learning how to test threads in Python as I have never done it before. I have put a lock in but it says that I have an assertion error which is fine. I am wondering if the following code is correct
import threading
i = 0
def test():
global i
for _ in range(100000):
with threading.Lock():
i += 1
threads = [threading.Thread(target=test) for t in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
assert i == 1000000, i
Your problem is that you create a new Lock on every iteration, which is always unlocked.
This way it'll work, because your Threads will try to aquire the same lock.
import threading
i = 0
lock = threading.Lock()
def test():
global i
for _ in range(100000):
with lock:
i += 1
threads = [threading.Thread(target=test) for t in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
print(i)
assert at the end of the code.
i, it just locks access to the lock itself, until it's released. So the 1,000,000 locks your code creates don't do anything useful, they just waste time.