Skip to content

Instantly share code, notes, and snippets.

@zztalker
Created March 24, 2021 05:15
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save zztalker/8599689fc37b04e4f21e7518b57402f1 to your computer and use it in GitHub Desktop.
Save zztalker/8599689fc37b04e4f21e7518b57402f1 to your computer and use it in GitHub Desktop.
asyncio locks with timeout
""" $ python3 multi-lock.py
I'm 1 and want to lock, wait no more than 1 sec
I'm 2 and want to lock, wait no more than 1 sec
I'm 3 and want to lock, wait no more than 1 sec
I'm 4 and want to lock, wait no more than 1 sec
I'm 5 and want to lock, wait no more than 1 sec
I'm 6 and want to lock, wait no more than 1 sec
I'm 7 and want to lock, wait no more than 1 sec
I'm 8 and want to lock, wait no more than 1 sec
I'm 9 and want to lock, wait no more than 1 sec
I'm 1 and I'm locked!
I'm 2 and I'm timeout to lock!
I'm 3 and I'm timeout to lock!
I'm 4 and I'm timeout to lock!
I'm 5 and I'm timeout to lock!
I'm 6 and I'm timeout to lock!
I'm 7 and I'm timeout to lock!
I'm 8 and I'm timeout to lock!
I'm 9 and I'm timeout to lock!
I'm 1 and I'm sleep well for 2 sec.
I'm 1 and I'm release lock!
I'm 2 and I'm sleep well for 2 sec.
I'm 2 and I'm wasn't locked!
I'm 3 and I'm sleep well for 2 sec.
I'm 3 and I'm wasn't locked!
I'm 4 and I'm sleep well for 2 sec.
I'm 4 and I'm wasn't locked!
I'm 5 and I'm sleep well for 2 sec.
I'm 5 and I'm wasn't locked!
I'm 6 and I'm sleep well for 2 sec.
I'm 6 and I'm wasn't locked!
I'm 7 and I'm sleep well for 2 sec.
I'm 7 and I'm wasn't locked!
I'm 8 and I'm sleep well for 2 sec.
I'm 8 and I'm wasn't locked!
I'm 9 and I'm sleep well for 2 sec.
I'm 9 and I'm wasn't locked!
"""
import asyncio
lock = asyncio.Lock()
async def i_want_to_lock(timeout, my_number):
print(f"I'm {my_number} and want to lock, wait no more than {timeout} sec")
try:
await asyncio.wait_for(lock.acquire(), timeout)
except asyncio.TimeoutError:
print(f"I'm {my_number} and I'm timeout to lock!")
else:
print(f"I'm {my_number} and I'm locked!")
await asyncio.sleep(timeout + 1)
print(f"I'm {my_number} and I'm sleep well for {timeout + 1} sec.")
if lock.locked():
lock.release()
print(f"I'm {my_number} and I'm release lock!")
else:
print(f"I'm {my_number} and I'm wasn't locked!")
tasks = []
for i in range(1, 10):
tasks.append(i_want_to_lock(1, i))
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(*tasks))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment