Advanced loop management, low-level scheduling, and platform-aware asyncio usage.
Work with the running loop and low-level scheduling APIs.
import asyncio
def on_timer():
print("timer fired")
async def main():
loop = asyncio.get_running_loop()
loop.call_later(0.2, on_timer)
await asyncio.sleep(0.3)
asyncio.run(main())For plain callbacks rather than coroutines, use loop scheduling methods.
Use `call_soon_threadsafe()` when another thread must notify the loop.
import asyncio
import threading
async def main():
loop = asyncio.get_running_loop()
event = asyncio.Event()
def notify():
loop.call_soon_threadsafe(event.set)
threading.Thread(target=notify).start()
await event.wait()
print("notified")
asyncio.run(main())This is the thread-safe callback bridge into the event loop.
Use `run_coroutine_threadsafe()` with a loop reference.
import asyncio
import threading
async def echo():
return "ok"
async def main():
loop = asyncio.get_running_loop()
result_holder = {}
def worker():
future = asyncio.run_coroutine_threadsafe(echo(), loop)
result_holder["value"] = future.result()
t = threading.Thread(target=worker)
t.start()
await asyncio.sleep(0.1)
t.join()
print(result_holder["value"])
asyncio.run(main())Useful when integrating threaded code with an existing event loop.
Modern patterns and common platform limitations.
Modern replacement for older loop access patterns.
import asyncio
async def main():
loop = asyncio.get_running_loop()
print(loop)
asyncio.run(main())In modern async code, prefer `get_running_loop()` over relying on policy-driven implicit loop creation.
Document the limitation around file descriptor watching for ordinary file I/O.
# asyncio can monitor sockets and many OS handles, but regular file I/O
# is not made non-blocking just by using asyncio. Use threads or subprocesses
# for blocking file operations when needed.Asyncio is primarily designed for network I/O and subprocess orchestration; ordinary file reads and writes can still block.