asyncio sleep cooperative scheduling demo in Python
This demo shows how asyncio.sleep yields control between concurrent tasks, letting multiple workers interleave their ticks.
Python code
20 linesimport asyncio
async def worker(name, delay):
for i in range(3):
print(f"{name}: tick {i}")
await asyncio.sleep(delay)
return f"{name} done"
async def main():
tasks = [
asyncio.create_task(worker("A", 0.1)),
asyncio.create_task(worker("B", 0.2)),
asyncio.create_task(worker("C", 0.3)),
]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
if __name__ == "__main__":
asyncio.run(main())
Output
A: tick 0
B: tick 0
C: tick 0
A: tick 1
B: tick 1
C: tick 1
A: tick 2
B: tick 2
C: tick 2
A done
B done
C done
How it works
asyncio.create_task schedules each worker coroutine to run concurrently on the event loop. Inside worker, await asyncio.sleep(delay) suspends the current task and lets the event loop switch to other ready tasks, so the delays interleave the output. asyncio.gather awaits all tasks and collects their return values in order. The event loop resumes each task after its sleep completes, demonstrating cooperative multitasking. Using asyncio.run starts the loop and cleans up resources automatically.
Common mistakes
- Calling `asyncio.sleep` without `await`, which just creates a coroutine and never suspends.
- Running blocking `time.sleep` inside `async def`, which blocks the whole event loop and defeats concurrency.
- Forgetting `await asyncio.gather(*tasks)`, leaving tasks to run without their results being collected.
Variations
- Use `asyncio.as_completed` to process results as each task finishes instead of waiting for all.
- Replace fixed delays with `await asyncio.sleep(random.uniform(...))` for more realistic jittered workloads.
Real-world use cases
- Making concurrent HTTP requests from an async web client and interleaving rate-limited calls.
- Batching database writes across multiple tables without blocking the event loop on I/O.
- Running background health checks for several services at staggered intervals in one process.
Sponsored
More from Concurrency & performance
- Benchmark list.append vs deque.append in Python medium
- Build a Python Performance Profiler That Generates Readable Reports medium
- Graceful Shutdown Executor Context Manager in Python medium
- How to Build a Producer-Consumer Pattern with asyncio.Queue in Python medium
- How to Cancel an asyncio Task with Graceful Cleanup in Python medium
- How to Convert Data in Parallel with ThreadPoolExecutor in Python easy
Keep learning
Related tutorials and quizzes for this topic.